hotel-manage-stack/rates
1
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

Add IPRoyal proxy support alongside DataImpulse

Browse source 
Auto-detect provider from hostname. DataImpulse appends session/country to
the username (LOGIN__cr.gb;sessid.ID); IPRoyal appends them to the password
(PASS_country-gb_session-ID_lifetime-30m). Both providers use the same
host/port/username/password/country config fields in Settings.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
jtricerolph 2026-07-10 02:37:53 +00:00
parent b44e688b3e
commit 2b55b0d6a5
1 changed files with 28 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

34
backend/services/proxy.py
View file

@ -69,12 +69,30 @@ def is_enabled(cfg: dict) -> bool:
return bool(cfg.get('host') and cfg.get('username')) return bool(cfg.get('host') and cfg.get('username'))
def username(cfg: dict, session_id: Optional[str] = None) -> str: def _provider(cfg: dict) -> str:
"""DataImpulse username: LOGIN__cr.<country>[;sessid.<id>].""" """Detect proxy provider from hostname."""
user = f"{cfg['username']}__cr.{cfg.get('country', 'gb')}" return 'iproyal' if 'iproyal' in cfg.get('host', '').lower() else 'dataimpulse'
def _build_auth(cfg: dict, session_id: Optional[str] = None):
"""Return (proxy_user, proxy_password) with session/country options applied.
DataImpulse: options appended to username → LOGIN__cr.gb;sessid.ID : PASS
IPRoyal: options appended to password → LOGIN : PASS_country-gb_session-ID_lifetime-30m
"""
cc = cfg.get('country', 'gb')
base_user = cfg.get('username', '')
base_pass = cfg.get('password', '')
if _provider(cfg) == 'iproyal':
pwd = f"{base_pass}_country-{cc}"
if session_id:
pwd += f"_session-{session_id}_lifetime-30m"
return base_user, pwd
else:
user = f"{base_user}__cr.{cc}"
if session_id: if session_id:
user += f";sessid.{session_id}" user += f";sessid.{session_id}"
return user return user, base_pass
def playwright_proxy(cfg: dict, session_id: Optional[str] = None) -> Optional[dict]: def playwright_proxy(cfg: dict, session_id: Optional[str] = None) -> Optional[dict]:
@ -89,10 +107,9 @@ def playwright_proxy(cfg: dict, session_id: Optional[str] = None) -> Optional[di
if not is_enabled(cfg): if not is_enabled(cfg):
return None return None
from urllib.parse import quote from urllib.parse import quote
user = quote(username(cfg, session_id), safe='') user, pwd = _build_auth(cfg, session_id)
pwd = quote(cfg.get('password', ''), safe='')
return { return {
'server': f"http://{user}:{pwd}@{cfg['host']}:{cfg['port']}", 'server': f"http://{quote(user, safe='')}:{quote(pwd, safe='')}@{cfg['host']}:{cfg['port']}",
} }
@ -108,7 +125,8 @@ def httpx_proxy_url(cfg: dict, session_id: Optional[str] = None) -> Optional[str
"""Proxy URL for httpx.AsyncClient(proxy=...). None when disabled.""" """Proxy URL for httpx.AsyncClient(proxy=...). None when disabled."""
if not is_enabled(cfg): if not is_enabled(cfg):
return None return None
return f"http://{username(cfg, session_id)}:{cfg.get('password', '')}@{cfg['host']}:{cfg['port']}" user, pwd = _build_auth(cfg, session_id)
return f"http://{user}:{pwd}@{cfg['host']}:{cfg['port']}"
def direct_httpx_proxy(db): def direct_httpx_proxy(db):