Spaces:
Paused
Paused
| 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 "") | |