用于价格监控的移动代理

export PROXY_URL="http://USERNAME-mbl-de-rot-auto10:[email protected]:7000"

# 替换国家参数（de -> fr、es、gb ……）即可切换市场。
# 代码的其他部分无需任何改动。

import os, time, random
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

PROXY = os.environ["PROXY_URL"]

session = requests.Session()
session.proxies = {"http": PROXY, "https": PROXY}
session.headers["User-Agent"] = (
    "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/125.0 Mobile Safari/537.36"
)

# 遇到 429/5xx 时退避并遵守 Retry-After，不要反复猛冲。
retry = Retry(
    total=4,
    backoff_factor=1.5,
    status_forcelist=(429, 500, 502, 503, 504),
    respect_retry_after_header=True,
    allowed_methods=frozenset(["GET"]),
)
session.mount("https://", HTTPAdapter(max_retries=retry, pool_maxsize=20))

def fetch_price(url: str) -> str:
    r = session.get(url, timeout=30)
    r.raise_for_status()
    # 防范静默过期问题：如果页面返回的
    # 不是预期币种，就视为未命中，而不是一个价格。
    if "€" not in r.text:
        raise ValueError(f"unexpected currency for DE exit: {url}")
    return r.text

for url in ["https://shop.example/p/1", "https://shop.example/p/2"]:
    html = fetch_price(url)
    print(len(html), url)
    time.sleep(random.uniform(1.5, 4.0))   # 控制请求节奏

币种断言是最关键的一行。它把代价高昂的静默失败变成了可见的失败。

# 运费和税费通常要在商品加入购物车后才会显示，
# 因此整个流程必须在同一个 IP 上完成。请使用粘性会话：
#   ...-mbl-de-sid-basket01-rot-sticky:PAK_KEY@...
import os, requests

s = requests.Session()
s.proxies = {"http": os.environ["PROXY_URL"], "https": os.environ["PROXY_URL"]}

s.post("https://shop.example/cart/add", data={"sku": "ABC", "qty": 1}, timeout=30)
quote = s.get("https://shop.example/cart/shipping", timeout=30).json()
print(quote["currency"], quote["total"], quote["eta_days"])

每个并发购物车对应一个 -sid-。两个 worker 复用同一个 sid 会让它们共用同一个购物车。