From f58c7a73ab63bf01eab15d4f5a5870590ccd2b3e Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Sat, 3 Oct 2026 19:51:13 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E9=80=89=E5=93=81=E5=A4=8D=E7=94=A8?= =?UTF-8?q?=E4=BB=93=E5=BA=93=E5=A4=96=20Chrome=20=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=EF=BC=8C=E5=87=8F=E5=B0=91=E6=9E=81=E9=AA=8C?= =?UTF-8?q?=E6=BB=91=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 登录改为 persistent context，默认目录在仓库外 rpa-chrome-profile/selection。 --- rpa_selection/auth_login.py | 87 ++++++++++++--- rpa_selection/tests/test_chrome_profile.py | 122 +++++++++++++++++++++ 2 files changed, 192 insertions(+), 17 deletions(-) create mode 100644 rpa_selection/tests/test_chrome_profile.py diff --git a/rpa_selection/auth_login.py b/rpa_selection/auth_login.py index b3eea10..4020894 100644 --- a/rpa_selection/auth_login.py +++ b/rpa_selection/auth_login.py @@ -17,6 +17,7 @@ from search_yaoex import YaoexAuth from util import normalize_text PROJECT_ROOT = Path(__file__).resolve().parent +REPO_ROOT = PROJECT_ROOT.parent AUTH_DIR = PROJECT_ROOT / "data" / ".auth" DEFAULT_YAOEX_AUTH_PATH = AUTH_DIR / "yaoex.json" DEFAULT_YBM_COOKIE_PATH = AUTH_DIR / "ybm.json" @@ -99,37 +100,88 @@ def _require_account(mall_label: str, username: str, password: str) -> None: ) +def _is_inside_repo(path: Path, repo_root: Path) -> bool: + try: + path.resolve().relative_to(repo_root.resolve()) + except ValueError: + return False + return True + + +def resolve_selection_chrome_profile( + requested: str | None = None, + *, + repo_root: Path | None = None, +) -> Path: + """选品 Chrome 用户目录固定在仓库外，避免每次新开空浏览器触发极验滑块。""" + root = (repo_root or REPO_ROOT).resolve() + outside = (root.parent / "rpa-chrome-profile" / "selection").resolve() + raw = os.getenv("RPA_SELECTION_CHROME_PROFILE") if requested is None else requested + if not raw: + base = os.getenv("RPA_CHROME_PROFILE") + raw = str(Path(base).expanduser() / "selection") if base else "" + if not raw: + return outside + candidate = Path(raw).expanduser().resolve() + if _is_inside_repo(candidate, root): + print(f"选品 Chrome 用户目录指向仓库内 {candidate}，已改用 {outside}") + return outside + return candidate + + class BrowserRuntime: - """同一轮脚本共用一套 Playwright。keep=True 时搜完仍留着窗口。""" + """同一轮脚本共用一套已登录的 Chrome 用户目录。keep=True 时搜完仍留着窗口。""" def __init__(self, keep: bool = False, hold_seconds: int = 180) -> None: self.keep = keep self.hold_seconds = hold_seconds self._playwright: Any = None - self._browsers: list[Any] = [] + self._context: Any = None + self._virgin_page: Any = None - def start(self) -> None: + def start(self, *, viewport: dict[str, int] | None = None) -> None: if self._playwright is not None: return from playwright.sync_api import sync_playwright self._playwright = sync_playwright().start() + profile = resolve_selection_chrome_profile() + profile.mkdir(parents=True, exist_ok=True) + print(f"选品 Chrome 用户目录：{profile}") + try: + self._context = self._playwright.chromium.launch_persistent_context( + str(profile), + headless=False, + channel="chrome", + viewport=viewport or {"width": 1440, "height": 900}, + args=[ + "--start-maximized", + "--disable-blink-features=AutomationControlled", + "--no-first-run", + "--no-default-browser-check", + ], + ignore_default_args=["--enable-automation"], + ) + except Exception as exc: + raise RuntimeError( + f"无法打开选品 Chrome 用户目录 {profile}（是否已有实例占用）：{exc}" + ) from exc + pages = list(self._context.pages) + self._virgin_page = pages[0] if pages else None def new_page(self, *, viewport: dict[str, int] | None = None) -> tuple[Any, Any]: - self.start() - browser = self._playwright.chromium.launch( - headless=False, - channel="chrome", - args=["--new-window", "--start-maximized"], - ) - self._browsers.append(browser) - context = browser.new_context(viewport=viewport or {"width": 1440, "height": 900}) - page = context.new_page() + self.start(viewport=viewport) + assert self._context is not None + if self._virgin_page is not None: + page = self._virgin_page + self._virgin_page = None + else: + page = self._context.new_page() page.bring_to_front() - return context, page + return self._context, page def close(self) -> None: - if self.keep and self._browsers: + if self.keep and self._context is not None: if self.hold_seconds <= 0: print("浏览器保持打开，不会自动关。要关时在终端停掉进程。") while True: @@ -137,12 +189,13 @@ class BrowserRuntime: else: print(f"浏览器保持打开 {self.hold_seconds} 秒，请看屏幕上的 Chrome。") time.sleep(self.hold_seconds) - for browser in self._browsers: + if self._context is not None: try: - browser.close() + self._context.close() except Exception: pass - self._browsers.clear() + self._context = None + self._virgin_page = None if self._playwright is not None: try: self._playwright.stop() diff --git a/rpa_selection/tests/test_chrome_profile.py b/rpa_selection/tests/test_chrome_profile.py new file mode 100644 index 0000000..049a1ce --- /dev/null +++ b/rpa_selection/tests/test_chrome_profile.py @@ -0,0 +1,122 @@ +from pathlib import Path + +import playwright.sync_api as playwright_sync + +import auth_login +from auth_login import BrowserRuntime, resolve_selection_chrome_profile + +REPO_ROOT = Path(__file__).resolve().parents[2] + + +def _is_relative_to(path: Path, root: Path) -> bool: + try: + path.resolve().relative_to(root.resolve()) + except ValueError: + return False + return True + + +def test_default_chrome_profile_is_outside_repo_selection_subdir(monkeypatch): + monkeypatch.delenv("RPA_SELECTION_CHROME_PROFILE", raising=False) + monkeypatch.delenv("RPA_CHROME_PROFILE", raising=False) + profile = resolve_selection_chrome_profile(repo_root=REPO_ROOT) + assert profile == (REPO_ROOT.parent / "rpa-chrome-profile" / "selection").resolve() + assert not _is_relative_to(profile, REPO_ROOT) + + +def test_env_path_inside_repo_is_rejected(monkeypatch): + inside = REPO_ROOT / "rpa_selection" / "data" / "chrome-profile" + monkeypatch.delenv("RPA_CHROME_PROFILE", raising=False) + profile = resolve_selection_chrome_profile(requested=str(inside), repo_root=REPO_ROOT) + assert profile == (REPO_ROOT.parent / "rpa-chrome-profile" / "selection").resolve() + assert not _is_relative_to(profile, REPO_ROOT) + + +def test_env_path_outside_repo_is_kept(tmp_path): + outside = tmp_path / "custom-chrome" + outside.mkdir() + profile = resolve_selection_chrome_profile(requested=str(outside), repo_root=REPO_ROOT) + assert profile == outside.resolve() + + +def test_rpa_chrome_profile_uses_selection_subdir(monkeypatch, tmp_path): + base = tmp_path / "work-profile" + monkeypatch.delenv("RPA_SELECTION_CHROME_PROFILE", raising=False) + monkeypatch.setenv("RPA_CHROME_PROFILE", str(base)) + profile = resolve_selection_chrome_profile(repo_root=REPO_ROOT) + assert profile == (base / "selection").resolve() + + +class _FakePage: + url = "about:blank" + + def bring_to_front(self) -> None: + return None + + +class _FakeContext: + def __init__(self) -> None: + self.pages = [_FakePage()] + self.closed = False + self.new_page_calls = 0 + + def new_page(self) -> _FakePage: + self.new_page_calls += 1 + page = _FakePage() + self.pages.append(page) + return page + + def close(self) -> None: + self.closed = True + + +class _FakeChromium: + def __init__(self) -> None: + self.persistent_calls: list[tuple[str, dict]] = [] + self.launch_calls: list[dict] = [] + self.context = _FakeContext() + + def launch_persistent_context(self, user_data_dir: str, **kwargs): + self.persistent_calls.append((user_data_dir, kwargs)) + return self.context + + def launch(self, **kwargs): + self.launch_calls.append(kwargs) + raise AssertionError("选品登录必须复用用户目录，禁止 launch 空 Chrome") + + +class _FakePlaywright: + def __init__(self, chromium: _FakeChromium) -> None: + self.chromium = chromium + + def start(self) -> "_FakePlaywright": + return self + + def stop(self) -> None: + return None + + +def test_browser_runtime_reuses_persistent_profile(monkeypatch, tmp_path): + profile = tmp_path / "selection-profile" + monkeypatch.setenv("RPA_SELECTION_CHROME_PROFILE", str(profile)) + chromium = _FakeChromium() + fake = _FakePlaywright(chromium) + monkeypatch.setattr(playwright_sync, "sync_playwright", lambda: fake) + + runtime = BrowserRuntime(keep=False) + ctx1, page1 = runtime.new_page() + ctx2, page2 = runtime.new_page() + + assert chromium.launch_calls == [] + assert len(chromium.persistent_calls) == 1 + user_data_dir, kwargs = chromium.persistent_calls[0] + assert user_data_dir == str(profile.resolve()) + assert kwargs.get("channel") == "chrome" + assert "--disable-blink-features=AutomationControlled" in kwargs.get("args", []) + assert ctx1 is ctx2 is chromium.context + assert page1 is not page2 + assert chromium.context.new_page_calls == 1 + + runtime.close() + assert chromium.context.closed + assert auth_login.resolve_selection_chrome_profile() == profile.resolve()