diff --git a/.env.example b/.env.example index b08cd14..3673d64 100644 --- a/.env.example +++ b/.env.example @@ -60,6 +60,9 @@ RAKUTEN_PROXY_SERVER= RAKUTEN_PROXY_USERNAME= # 代理密码（如代理需要认证则填写） RAKUTEN_PROXY_PASSWORD= +# 不经代理的主机名（逗号分隔，支持 *.internal 这类通配符）。默认覆盖本机与 Docker +# 服务间通信；内部网关或抓取服务使用其它域名时，将其显式加入此列表。 +RAKUTEN_PROXY_BYPASS=localhost,127.0.0.1,::1,rakuten-api,rakuten-trading,rakuten-gateway # ---- OpenTelemetry traces（可选；默认关闭）---- # 启用后把抓取-解析链路以 span 导出到 OTLP/HTTP endpoint， diff --git a/README.md b/README.md index 57c4ec2..b5e5eaa 100644 --- a/README.md +++ b/README.md @@ -644,6 +644,8 @@ trading 加购时的字段选择策略：多规格挑第一个非售罄的 varia - 鉴权：`RAKUTEN_BEARER_TOKEN`（三个服务共用） - 抓取：`RAKUTEN_MAX_SITE_CONCURRENCY`（默认 `8`，两站各自独立计数）、`RAKUTEN_HTTP_MAX_ATTEMPTS`（默认 `3`）、`RAKUTEN_SESSION_TTL_SECONDS`（默认 `1800`，仅乐天） - 浏览器兜底（仅乐天）：`RAKUTEN_BROWSER_FALLBACK_ENABLED`、`RAKUTEN_BROWSER_HEADLESS`、`RAKUTEN_BROWSER_CHANNEL` -- 代理（需日本 IP 时）：`RAKUTEN_PROXY_SERVER`、`RAKUTEN_PROXY_USERNAME`、`RAKUTEN_PROXY_PASSWORD` +- 代理（需日本 IP 时）：`RAKUTEN_PROXY_SERVER`、`RAKUTEN_PROXY_USERNAME`、`RAKUTEN_PROXY_PASSWORD`、`RAKUTEN_PROXY_BYPASS` -> 从中国大陆直连实测可用、无需代理；`RAKUTEN_PROXY_SERVER` 留空即可。 +> 从中国大陆直连实测可用、无需代理；`RAKUTEN_PROXY_SERVER` 留空即可。设置后，所有面向 +> 外部站点的 HTTPX 与 Playwright 流量都会经代理；本机和 Docker 服务间地址由 +> `RAKUTEN_PROXY_BYPASS` 直连。该设置不影响 OTel 上报。 diff --git a/app/scraping/services/browser_fallback.py b/app/scraping/services/browser_fallback.py index 90c52a7..bcdf801 100644 --- a/app/scraping/services/browser_fallback.py +++ b/app/scraping/services/browser_fallback.py @@ -16,6 +16,7 @@ from dataclasses import dataclass, field from typing import Any from app.shared.config import Settings +from app.shared.proxy import playwright_launch_proxy from app.scraping.core import site logger = logging.getLogger(__name__) @@ -100,7 +101,7 @@ class BrowserFallback: self._browser = await self._playwright.chromium.launch( headless=self._settings.browser_headless_effective, channel=self._settings.browser_channel or None, - proxy=self._settings.playwright_proxy, + proxy=playwright_launch_proxy(self._settings), timeout=self._settings.browser_launch_timeout_seconds * 1000, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) diff --git a/app/scraping/services/rakuma_session.py b/app/scraping/services/rakuma_session.py index 9619bcc..b78f658 100644 --- a/app/scraping/services/rakuma_session.py +++ b/app/scraping/services/rakuma_session.py @@ -22,6 +22,7 @@ from opentelemetry import trace from app.scraping.core import rakuma_site as site from app.shared.config import Settings +from app.shared.proxy import httpx_client_options from app.shared.errors import ( ItemNotFoundError, ResourceBusyError, @@ -51,7 +52,7 @@ class RakumaSession: headers=site.default_headers(), timeout=self._settings.request_timeout_seconds, follow_redirects=True, - proxy=self._settings.httpx_proxy, + **httpx_client_options(self._settings), http2=True, ) logger.info( diff --git a/app/scraping/services/site_session.py b/app/scraping/services/site_session.py index 09f8959..30ed7de 100644 --- a/app/scraping/services/site_session.py +++ b/app/scraping/services/site_session.py @@ -23,6 +23,7 @@ from opentelemetry import trace from app.scraping.core import site from app.shared.config import Settings +from app.shared.proxy import httpx_client_options from app.shared.errors import ( ItemNotFoundError, ResourceBusyError, @@ -91,7 +92,7 @@ class SiteSession: headers=site.default_headers(mobile=mobile), timeout=self._settings.request_timeout_seconds, follow_redirects=True, - proxy=self._settings.httpx_proxy, + **httpx_client_options(self._settings), http2=True, ), ) diff --git a/app/shared/config.py b/app/shared/config.py index 8b7e015..bb10064 100644 --- a/app/shared/config.py +++ b/app/shared/config.py @@ -8,9 +8,11 @@ 「通用 / 仅抓取 / 仅交易 / 仅网关 / 仅交易 worker」分区标注，各进程只读自己那部分。 """ import socket +from fnmatch import fnmatchcase from functools import lru_cache from pathlib import Path from typing import Literal +from urllib.parse import quote, urlsplit from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict @@ -88,6 +90,8 @@ class Settings(BaseSettings): proxy_server: str | None = None proxy_username: str | None = None proxy_password: str | None = None + # 逗号分隔的直连主机名或通配符。服务间请求不应绕到公网代理。 + proxy_bypass: str = "localhost,127.0.0.1,::1,rakuten-api,rakuten-trading,rakuten-gateway" # ---- OpenTelemetry traces（通用，可选；默认关闭）---- # 启用后把抓取-解析链路以 span 导出到 OTLP/HTTP endpoint，用于排查"抓到 @@ -192,6 +196,8 @@ class Settings(BaseSettings): proxy["username"] = self.proxy_username if self.proxy_password: proxy["password"] = self.proxy_password + if self.proxy_bypass.strip(): + proxy["bypass"] = self.proxy_bypass return proxy @property @@ -205,9 +211,29 @@ class Settings(BaseSettings): scheme, _, rest = self.proxy_server.partition("://") if not rest: return self.proxy_server - credentials = f"{self.proxy_username}:{self.proxy_password or ''}" + credentials = f"{quote(self.proxy_username, safe='')}:{quote(self.proxy_password or '', safe='')}" return f"{scheme}://{credentials}@{rest}" + def proxy_bypasses(self, url: str) -> bool: + """目标 URL 的主机是否应绕过外部代理。""" + host = urlsplit(url).hostname + if not host: + return False + + normalized_host = host.rstrip(".").lower() + for raw_pattern in self.proxy_bypass.split(","): + pattern = raw_pattern.strip().rstrip(".").lower() + if not pattern: + continue + if pattern.startswith("."): + suffix = pattern[1:] + if normalized_host == suffix or normalized_host.endswith(f".{suffix}"): + return True + continue + if fnmatchcase(normalized_host, pattern): + return True + return False + @property def auth_state_path(self) -> Path: """登录态目录的绝对路径，不存在时创建""" diff --git a/app/shared/proxy.py b/app/shared/proxy.py new file mode 100644 index 0000000..5e00023 --- /dev/null +++ b/app/shared/proxy.py @@ -0,0 +1,23 @@ +"""项目出站代理策略；OTel exporter 不使用这里的配置。""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from app.shared.config import Settings + + +def httpx_client_options( + settings: "Settings", *, target_url: str | None = None +) -> dict[str, object]: + """返回 HTTPX 客户端统一的代理与环境变量策略。""" + proxy = None if target_url and settings.proxy_bypasses(target_url) else settings.httpx_proxy + options: dict[str, object] = {"trust_env": False} + if proxy is not None: + options["proxy"] = proxy + return options + + +def playwright_launch_proxy(settings: "Settings") -> dict[str, str] | None: + """返回 Playwright 启动参数使用的统一代理设置。""" + return settings.playwright_proxy diff --git a/app/trading/main.py b/app/trading/main.py index 481d38f..b9fd480 100644 --- a/app/trading/main.py +++ b/app/trading/main.py @@ -61,6 +61,7 @@ def build_container() -> TradingContainer: client = GatewayClient( settings.order_gateway_url, settings.bearer_token, + settings=settings, timeout=max(60.0, settings.lease_max_wait_seconds + 10), ) local_db = LocalDB(settings.trading_db_path_resolved) diff --git a/app/trading/services/auth_session.py b/app/trading/services/auth_session.py index cd55dd6..6c5639a 100644 --- a/app/trading/services/auth_session.py +++ b/app/trading/services/auth_session.py @@ -37,6 +37,7 @@ from typing import Any import httpx from app.shared.config import Settings +from app.shared.proxy import httpx_client_options from app.shared.errors import NotLoggedInError, UpstreamRequestError from app.trading.core import auth_site @@ -110,7 +111,7 @@ class AuthSession: headers=profile.headers(), timeout=self._settings.request_timeout_seconds, follow_redirects=True, - proxy=self._settings.httpx_proxy, + **httpx_client_options(self._settings), http2=True, ) self._sites[name] = _SiteAuth(name=name, client=client) diff --git a/app/trading/services/login_runner.py b/app/trading/services/login_runner.py index 6aa88f9..40e9cac 100644 --- a/app/trading/services/login_runner.py +++ b/app/trading/services/login_runner.py @@ -25,6 +25,7 @@ from typing import Any import yaml from app.shared.config import BASE_DIR, Settings +from app.shared.proxy import playwright_launch_proxy from app.trading.core import auth_site logger = logging.getLogger(__name__) @@ -355,7 +356,7 @@ async def login_one( user_data_dir=str(account.user_data_dir), headless=False, # 自动登录仍需有头：撞验证码要人工接管 channel=settings.browser_channel or None, - proxy=settings.playwright_proxy, + proxy=playwright_launch_proxy(settings), user_agent=profile.user_agent, locale="ja-JP", timezone_id="Asia/Tokyo", diff --git a/app/trading/worker/client.py b/app/trading/worker/client.py index b085e58..1c6ade9 100644 --- a/app/trading/worker/client.py +++ b/app/trading/worker/client.py @@ -16,7 +16,9 @@ from typing import Any import httpx +from app.shared.config import Settings from app.shared.errors import AppError +from app.shared.proxy import httpx_client_options from app.shared.task_state import OrderState, TaskStatus from app.trading.worker.models import LeaseTask @@ -29,13 +31,16 @@ class GatewayClient: 一份 AsyncClient 实例贯穿 worker 整个生命周期，连接池由 httpx 管理。 """ - def __init__(self, base_url: str, bearer_token: str, *, timeout: float = 60.0): + def __init__( + self, base_url: str, bearer_token: str, *, settings: Settings, timeout: float = 60.0 + ): # 末尾去斜杠，避免 base + "/api/..." 拼出双斜杠 self._base_url = base_url.rstrip("/") self._client = httpx.AsyncClient( base_url=self._base_url, headers={"Authorization": f"Bearer {bearer_token}"}, timeout=timeout, + **httpx_client_options(settings, target_url=self._base_url), ) async def aclose(self) -> None: diff --git a/app/trading/worker/site_interact.py b/app/trading/worker/site_interact.py index 699937c..9f85b6d 100644 --- a/app/trading/worker/site_interact.py +++ b/app/trading/worker/site_interact.py @@ -97,6 +97,7 @@ from app.shared.purchase_contract import ( basket_domain_of, inventory_flag_for, ) +from app.shared.proxy import playwright_launch_proxy from app.shared.task_state import OrderState from app.trading.core import auth_site from app.trading.worker.models import LeaseTask @@ -556,6 +557,7 @@ class SiteInteractor: # 状态的操作（加购/结算/支付）都必须走非无头浏览器 headless=False, channel=self._settings.browser_channel or None, + proxy=playwright_launch_proxy(self._settings), args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) self._context = await self._browser.new_context( diff --git a/scripts/probe_address_confirm.py b/scripts/probe_address_confirm.py index ea5290c..01ae1f8 100644 --- a/scripts/probe_address_confirm.py +++ b/scripts/probe_address_confirm.py @@ -48,6 +48,7 @@ async def _start_visible(site: SiteInteractor, settings) -> None: site._browser = await site._playwright.chromium.launch( headless=False, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) site._context = await site._browser.new_context( diff --git a/scripts/probe_checkout.py b/scripts/probe_checkout.py index 4ecb1d3..a6d9267 100644 --- a/scripts/probe_checkout.py +++ b/scripts/probe_checkout.py @@ -24,7 +24,8 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import httpx -from app.shared.config import get_settings +from app.shared.config import Settings, get_settings +from app.shared.proxy import httpx_client_options from app.trading.core import auth_site PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "checkout" @@ -42,12 +43,13 @@ def load_cookies(state_path: Path) -> list[dict]: return state.get("cookies", []) -def build_client(cookies: list[dict]) -> httpx.AsyncClient: +def build_client(cookies: list[dict], settings: Settings) -> httpx.AsyncClient: client = httpx.AsyncClient( headers=auth_site.PROFILES["rakuten"].headers(), timeout=30.0, follow_redirects=True, http2=True, + **httpx_client_options(settings), ) for cookie in cookies: name = cookie.get("name") @@ -210,7 +212,7 @@ async def main() -> int: cookies = load_cookies(state_path) print(f"加载 {len(cookies)} 条 cookie 自 {state_path}") - async with build_client(cookies) as client: + async with build_client(cookies, settings) as client: await probe_readonly(client) if args.mutate: if not args.item_url: diff --git a/scripts/probe_checkout_playwright.py b/scripts/probe_checkout_playwright.py index 5abd3ad..22db7df 100644 --- a/scripts/probe_checkout_playwright.py +++ b/scripts/probe_checkout_playwright.py @@ -56,6 +56,7 @@ async def run(item_url: str, *, headful: bool) -> int: browser = await pw.chromium.launch( headless=not headful, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) context = await browser.new_context( diff --git a/scripts/probe_new_card.py b/scripts/probe_new_card.py index 005e921..7c08b45 100644 --- a/scripts/probe_new_card.py +++ b/scripts/probe_new_card.py @@ -48,6 +48,7 @@ async def _start_visible(site: SiteInteractor, settings) -> None: site._browser = await site._playwright.chromium.launch( headless=False, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) site._context = await site._browser.new_context( diff --git a/scripts/probe_order_list.py b/scripts/probe_order_list.py index 380364f..be5f9ae 100644 --- a/scripts/probe_order_list.py +++ b/scripts/probe_order_list.py @@ -48,6 +48,7 @@ async def run(*, headful: bool) -> int: browser = await pw.chromium.launch( headless=not headful, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) context = await browser.new_context( diff --git a/scripts/probe_payment_method.py b/scripts/probe_payment_method.py index 414e772..e2a577a 100644 --- a/scripts/probe_payment_method.py +++ b/scripts/probe_payment_method.py @@ -53,6 +53,7 @@ async def _start_visible(site: SiteInteractor, settings) -> None: site._browser = await site._playwright.chromium.launch( headless=False, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) site._context = await site._browser.new_context( diff --git a/scripts/probe_purchase_block.py b/scripts/probe_purchase_block.py index 5f7772c..2c0f339 100644 --- a/scripts/probe_purchase_block.py +++ b/scripts/probe_purchase_block.py @@ -164,7 +164,11 @@ async def main() -> int: log: list[str] = [] async with async_playwright() as pw: - browser = await pw.chromium.launch(headless=True, args=["--no-first-run"]) + browser = await pw.chromium.launch( + headless=True, + proxy=settings.playwright_proxy, + args=["--no-first-run"], + ) context = await browser.new_context( storage_state=state_path, user_agent=auth_site.RAKUTEN_USER_AGENT, diff --git a/scripts/probe_purchase_block_v2.py b/scripts/probe_purchase_block_v2.py index 5c8fdb4..06881c3 100644 --- a/scripts/probe_purchase_block_v2.py +++ b/scripts/probe_purchase_block_v2.py @@ -179,7 +179,11 @@ async def main() -> int: log: list[str] = [] async with async_playwright() as pw: - browser = await pw.chromium.launch(headless=True, args=["--no-first-run"]) + browser = await pw.chromium.launch( + headless=True, + proxy=settings.playwright_proxy, + args=["--no-first-run"], + ) context = await browser.new_context( storage_state=state_path, user_agent=auth_site.RAKUTEN_USER_AGENT, diff --git a/scripts/probe_search_add.py b/scripts/probe_search_add.py index 8785515..8624e82 100644 --- a/scripts/probe_search_add.py +++ b/scripts/probe_search_add.py @@ -121,6 +121,7 @@ async def main() -> int: browser = await pw.chromium.launch( headless=True, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) context = await browser.new_context( diff --git a/scripts/probe_subsite_checkout.py b/scripts/probe_subsite_checkout.py index b1b1301..23de701 100644 --- a/scripts/probe_subsite_checkout.py +++ b/scripts/probe_subsite_checkout.py @@ -723,6 +723,7 @@ async def _start_visible(site: SiteInteractor, settings) -> None: site._browser = await site._playwright.chromium.launch( # noqa: SLF001 headless=False, channel=settings.browser_channel or None, + proxy=settings.playwright_proxy, args=["--no-first-run", "--disable-blink-features=AutomationControlled"], ) site._context = await site._browser.new_context( # noqa: SLF001 diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 0000000..a0dc634 --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,115 @@ +"""统一出站代理策略的单元测试。""" +from __future__ import annotations + +import asyncio +import ast +from pathlib import Path + +from app.shared.config import Settings +from app.shared.proxy import httpx_client_options, playwright_launch_proxy +from app.trading.worker import client as worker_client +from app.trading.worker.client import GatewayClient + + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + + +def _settings(**overrides) -> Settings: + return Settings(_env_file=None, **overrides) + + +def test_disabled_proxy_does_not_inherit_host_environment() -> None: + settings = _settings() + + assert httpx_client_options(settings) == {"trust_env": False} + assert playwright_launch_proxy(settings) is None + + +def test_authenticated_proxy_encodes_httpx_credentials() -> None: + settings = _settings( + proxy_server="http://proxy.example:8080", + proxy_username="user@example.com", + proxy_password="pa:ss /#", + proxy_bypass="localhost,.internal.example", + ) + + assert httpx_client_options(settings) == { + "trust_env": False, + "proxy": "http://user%40example.com:pa%3Ass%20%2F%23@proxy.example:8080", + } + assert playwright_launch_proxy(settings) == { + "server": "http://proxy.example:8080", + "username": "user@example.com", + "password": "pa:ss /#", + "bypass": "localhost,.internal.example", + } + + +def test_internal_hosts_bypass_httpx_proxy() -> None: + settings = _settings( + proxy_server="http://proxy.example:8080", + proxy_bypass="localhost,.internal.example,*.svc", + ) + + assert settings.proxy_bypasses("http://localhost:31109") + assert settings.proxy_bypasses("https://api.internal.example/path") + assert settings.proxy_bypasses("https://orders.svc") + assert not settings.proxy_bypasses("https://www.rakuten.co.jp/") + assert httpx_client_options(settings, target_url="http://localhost:31109") == { + "trust_env": False + } + assert httpx_client_options(settings, target_url="https://www.rakuten.co.jp/") == { + "trust_env": False, + "proxy": "http://proxy.example:8080", + } + + +def test_gateway_client_uses_its_base_url_for_proxy_policy(monkeypatch) -> None: + settings = _settings(proxy_server="http://proxy.example:8080") + captured: dict[str, object] = {} + + def fake_httpx_options(actual_settings: Settings, *, target_url: str | None = None) -> dict[str, object]: + captured["settings"] = actual_settings + captured["target_url"] = target_url + return {"trust_env": False} + + monkeypatch.setattr(worker_client, "httpx_client_options", fake_httpx_options) + client = GatewayClient("https://gateway.example", "token", settings=settings) + try: + assert captured == {"settings": settings, "target_url": "https://gateway.example"} + finally: + asyncio.run(client.aclose()) + + +def test_every_project_http_client_and_browser_launch_has_proxy_policy() -> None: + """新增出站入口时，强制它显式接入统一代理策略。""" + missing_httpx_policy: list[Path] = [] + missing_browser_proxy: list[Path] = [] + + for root_name in ("app", "scripts"): + for path in (PROJECT_ROOT / root_name).rglob("*.py"): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): + continue + if ( + node.func.attr == "AsyncClient" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "httpx" + ): + uses_policy = any( + keyword.arg is None + and isinstance(keyword.value, ast.Call) + and isinstance(keyword.value.func, ast.Name) + and keyword.value.func.id == "httpx_client_options" + for keyword in node.keywords + ) + if not uses_policy: + missing_httpx_policy.append(path) + if node.func.attr in {"launch", "launch_persistent_context"} and not any( + keyword.arg == "proxy" for keyword in node.keywords + ): + missing_browser_proxy.append(path) + + assert not missing_httpx_policy + assert not missing_browser_proxy