Spaces:
Paused
Paused
File size: 3,501 Bytes
1ae289a 7044ec8 78fa9aa 1ae289a 7044ec8 1ae289a 7044ec8 78fa9aa 7044ec8 1ae289a 7044ec8 1ae289a 2d5ff33 1ae289a 2d5ff33 1ae289a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | import os
from urllib.parse import urlparse, urlsplit, urlunsplit
POS_ORIGIN = "https://pos.hengdianfilm.com"
POS_PROXY_URL_ENV = "POS_CASHIER_PROXY_URL"
POS_PROXY_TOKEN_ENV = "POS_CASHIER_PROXY_TOKEN"
POS_PROXY_AUTH_QUERY_ENV = "POS_CASHIER_PROXY_AUTH_QUERY"
POS_PROXY_TOKEN_HEADER = "X-POS-Proxy-Token"
TMS_PROXY_URL_ENV = "TMS_CF_WORKER_URL"
TMS_PROXY_TOKEN_ENV = "TMS_CF_WORKER_TOKEN"
POS_PROXY_USE_TMS_ENV = "POS_CASHIER_USE_TMS_PROXY"
def _env_flag(name, default=False):
raw = str(os.getenv(name, "")).strip().lower()
if not raw:
return default
return raw not in {"0", "false", "no", "off"}
def _normalize_url(raw):
raw = str(raw or "").strip().strip('"').strip("'")
if not raw:
return ""
if "://" not in raw:
raw = f"https://{raw}"
return raw.rstrip("/")
def _derive_pos_proxy_from_tms_proxy():
tms_proxy_base_url = _normalize_url(os.getenv(TMS_PROXY_URL_ENV, ""))
if not tms_proxy_base_url:
return ""
parts = urlsplit(tms_proxy_base_url)
base_path = parts.path.rstrip("/")
pos_path = f"{base_path}/pos-api" if base_path and base_path != "/" else "/pos-api"
return urlunsplit((parts.scheme, parts.netloc, pos_path, parts.query, parts.fragment)).rstrip("/")
def get_pos_proxy_base_url(proxy_url=None):
if proxy_url is not None:
return _normalize_url(proxy_url)
explicit_proxy_url = _normalize_url(os.getenv(POS_PROXY_URL_ENV, ""))
if explicit_proxy_url:
return explicit_proxy_url
if _env_flag(POS_PROXY_USE_TMS_ENV, False):
return _derive_pos_proxy_from_tms_proxy()
return ""
def get_pos_origin(origin_url=None):
return _normalize_url(origin_url if origin_url is not None else os.getenv("POS_CASHIER_BASE_URL", POS_ORIGIN)) or POS_ORIGIN
def build_pos_url(origin_url, proxy_url=None, origin_base_url=None):
proxy_base_url = get_pos_proxy_base_url(proxy_url)
if not proxy_base_url:
return origin_url
parsed = urlparse(origin_url)
origin = f"{parsed.scheme}://{parsed.netloc}"
if origin != get_pos_origin(origin_base_url):
return origin_url
base_parts = urlsplit(proxy_base_url)
base_path = base_parts.path.rstrip("/")
target_path = parsed.path
final_path = f"{base_path}{target_path}" if base_path and base_path != "/" else target_path
extra_query = os.getenv(POS_PROXY_AUTH_QUERY_ENV, "").strip().lstrip("?")
query_parts = [q for q in (base_parts.query, extra_query, parsed.query) if q]
return urlunsplit(
(
base_parts.scheme,
base_parts.netloc,
final_path,
"&".join(query_parts),
base_parts.fragment,
)
)
def with_pos_proxy_headers(headers=None, proxy_url=None):
headers = dict(headers or {})
if get_pos_proxy_base_url(proxy_url):
pos_proxy_token = os.getenv(POS_PROXY_TOKEN_ENV, "").strip()
tms_proxy_token = os.getenv(TMS_PROXY_TOKEN_ENV, "").strip()
proxy_token = pos_proxy_token or tms_proxy_token or os.getenv("NEW_API_PROXY_TOKEN", "").strip()
if proxy_token:
headers[POS_PROXY_TOKEN_HEADER] = proxy_token
if tms_proxy_token:
headers["X-TMS-Proxy-Token"] = tms_proxy_token
return headers
def mask_pos_proxy_url(proxy_url=None):
proxy_base_url = get_pos_proxy_base_url(proxy_url)
if not proxy_base_url:
return ""
return proxy_base_url.split("?", 1)[0] + ("?..." if "?" in proxy_base_url else "")
|