from __future__ import annotations import json import os import time from pathlib import Path from typing import Any from http_util import parse_cookie_value from mall_config import ( YAOEX_HOME_URL, YAOEX_LOGIN_URL, YBM_HOME_URL, YBM_LOGIN_URL, ) 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" def load_dotenv(path: Path) -> None: if not path.exists(): return for line in path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, _, value = stripped.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) def load_secrets(secrets_path: Path | None = None) -> dict[str, dict[str, str]]: load_dotenv(PROJECT_ROOT / ".env") result: dict[str, dict[str, str]] = {"yaoex": {}, "ybm": {}} path = secrets_path or (PROJECT_ROOT / "secrets.json") if path.exists(): data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, dict): for mall in ("yaoex", "ybm"): block = data.get(mall) or {} if isinstance(block, dict): result[mall] = { "username": normalize_text(block.get("username")), "password": normalize_text(block.get("password")), } env_map = { "yaoex": ("YAOEX_USERNAME", "YAOEX_PASSWORD"), "ybm": ("YBM_USERNAME", "YBM_PASSWORD"), } for mall, (user_key, pass_key) in env_map.items(): result[mall]["username"] = os.environ.get(user_key) or result[mall].get("username", "") result[mall]["password"] = os.environ.get(pass_key) or result[mall].get("password", "") return result def load_yaoex_auth_file(path: Path) -> YaoexAuth: data = json.loads(path.read_text(encoding="utf-8")) return YaoexAuth( token=normalize_text(data.get("token")), user_id=normalize_text(data.get("user_id")), buyer_code=normalize_text(data.get("buyer_code")), ) def load_cookie_file(path: Path) -> dict[str, str]: data = json.loads(path.read_text(encoding="utf-8")) if not isinstance(data, dict): raise ValueError(f"cookie 文件不是对象：{path}") return {str(key): parse_cookie_value(str(value)) for key, value in data.items()} def save_yaoex_auth(auth: YaoexAuth, path: Path = DEFAULT_YAOEX_AUTH_PATH) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( {"token": auth.token, "user_id": auth.user_id, "buyer_code": auth.buyer_code}, ensure_ascii=False, indent=2, ), encoding="utf-8", ) def save_cookie_file(cookies: dict[str, str], path: Path = DEFAULT_YBM_COOKIE_PATH) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(cookies, ensure_ascii=False, indent=2), encoding="utf-8") def _require_account(mall_label: str, username: str, password: str) -> None: if not username or not password: raise SystemExit( f"缺少{mall_label}账号。请在 .env / 环境变量或 secrets.json 中配置用户名和密码。" ) 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: """同一轮脚本共用一套已登录的 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._context: Any = None self._virgin_page: Any = 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(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 self._context, page def close(self) -> None: if self.keep and self._context is not None: if self.hold_seconds <= 0: print("浏览器保持打开，不会自动关。要关时在终端停掉进程。") while True: time.sleep(3600) else: print(f"浏览器保持打开 {self.hold_seconds} 秒，请看屏幕上的 Chrome。") time.sleep(self.hold_seconds) if self._context is not None: try: self._context.close() except Exception: pass self._context = None self._virgin_page = None if self._playwright is not None: try: self._playwright.stop() except Exception: pass self._playwright = None def _ensure_yaoex_logged_in(page: Any, username: str, password: str) -> None: page.goto(YAOEX_LOGIN_URL, wait_until="domcontentloaded") page.wait_for_timeout(2000) if "/login" not in page.url: return account_input = page.locator("#username").first password_input = page.locator("#password").first captcha_button = page.locator( "#captcha-box .geetest_btn_click, #captcha-box .geetest_btn, #captcha-box .geetest_radar_btn" ).first login_button = page.locator("#login-btn").first account_input.wait_for(state="visible", timeout=30000) account_input.fill(username) password_input.fill(password) print("1号药城：已填账号。请看 Chrome——若出现滑块拼图，拖过去。窗口会等 3 分钟。") if captcha_button.count() > 0: captcha_button.click(force=True) try: page.wait_for_function( "() => !window.location.href.includes('/login')", timeout=20000, ) except Exception: try: login_button.click(force=True) except Exception: pass try: page.wait_for_function( "() => !window.location.href.includes('/login')", timeout=180000, ) except Exception: pass if "/login" in page.url: login_error = "" error_locator = page.locator(".error-tips span, .layui-layer-content, .geetest_tip") if error_locator.count() > 0: login_error = normalize_text(error_locator.first.inner_text()) shot = PROJECT_ROOT / "output" / "结果" / "yaoex_login_fail.png" shot.parent.mkdir(parents=True, exist_ok=True) page.screenshot(path=str(shot), full_page=True) detail = f"，页面提示：{login_error}" if login_error else "" raise RuntimeError( f"1号药城登录后仍停留在登录页，登录可能失败{detail}。截图：{shot}" ) def _ensure_ybm_logged_in(page: Any, username: str, password: str) -> None: page.goto(YBM_LOGIN_URL, wait_until="domcontentloaded") page.wait_for_timeout(2000) if "/new/login" not in page.url: return account_input = page.locator("form.login-form input[type='text']").first password_input = page.locator("form.login-form input[type='password']").first agree_input = page.locator(".login-section-agree input[type='checkbox']").first agree_visible_box = page.locator(".login-section-agree .el-checkbox__inner").first login_button = page.locator("button.login-button").first account_input.wait_for(state="visible", timeout=30000) account_input.fill(username) password_input.fill(password) if not agree_input.is_checked(): agree_visible_box.click() login_button.click() page.wait_for_timeout(5000) if "/new/login" in page.url: login_error = "" error_locator = page.locator(".el-message__content") if error_locator.count() > 0: login_error = normalize_text(error_locator.first.inner_text()) detail = f"，页面提示：{login_error}" if login_error else "" raise RuntimeError(f"药帮忙登录后仍停留在登录页，登录可能失败{detail}") def _playwright_login_yaoex(username: str, password: str, runtime: BrowserRuntime) -> YaoexAuth: context, page = runtime.new_page() _ensure_yaoex_logged_in(page, username, password) page.goto(YAOEX_HOME_URL, wait_until="domcontentloaded") page.wait_for_timeout(3000) print("1号药城窗口应已打开首页。") cookies = context.cookies() cookie_map = {item["name"]: item["value"] for item in cookies} token = parse_cookie_value(cookie_map.get("ycgltoken")) or parse_cookie_value( cookie_map.get("yctoken") ) user_id = parse_cookie_value(cookie_map.get("ycuserId")) buyer_code = parse_cookie_value(cookie_map.get("ycenterpriseId")) if not token or not user_id or not buyer_code: raise RuntimeError("未能从1号药城 cookie 提取完整认证字段（token/user_id/buyer_code）。") return YaoexAuth(token=token, user_id=user_id, buyer_code=buyer_code) def _playwright_login_ybm(username: str, password: str, runtime: BrowserRuntime) -> dict[str, str]: context, page = runtime.new_page() _ensure_ybm_logged_in(page, username, password) page.goto(YBM_HOME_URL, wait_until="domcontentloaded") page.wait_for_timeout(3000) print("药帮忙窗口应已打开首页。") cookies = context.cookies() return {item["name"]: parse_cookie_value(item["value"]) for item in cookies} def login_yaoex( secrets_path: Path | None = None, runtime: BrowserRuntime | None = None, ) -> YaoexAuth: secrets = load_secrets(secrets_path) username = secrets["yaoex"].get("username", "") password = secrets["yaoex"].get("password", "") _require_account("1号药城", username, password) own = runtime is None session = runtime or BrowserRuntime(keep=False) try: return _playwright_login_yaoex(username, password, session) finally: if own: session.close() def login_ybm( secrets_path: Path | None = None, runtime: BrowserRuntime | None = None, ) -> dict[str, str]: secrets = load_secrets(secrets_path) username = secrets["ybm"].get("username", "") password = secrets["ybm"].get("password", "") _require_account("药帮忙", username, password) own = runtime is None session = runtime or BrowserRuntime(keep=False) try: return _playwright_login_ybm(username, password, session) finally: if own: session.close()