fix: 选品复用仓库外 Chrome 用户目录，减少极验滑块

登录改为 persistent context，默认目录在仓库外 rpa-chrome-profile/selection。

AGENTS.md

@@ -28,7 +28,7 @@
 | 节点入口 | `rpa_contract/contract_create_node.py`、`main.py` | 读到正式合同号 |
 | 浏览器适配 | `rpa_contract/apa_browser_adapter.py` | `rpa_contract/tests/test_apa_browser_adapter.py::点名` |
 | 文字验收 / 截图策略 | `rpa_contract/accept.py` | 只读 `ACCEPT.txt`，禁止读 png |
-| 加购 / 支付回传 | `rpa_cart/README.md` | 占位，未实现 |
+| 加购 / 支付回传 | `rpa_cart/README.md` | 加购未做；1 号药城支付按 Excel 落表，不对加购行 |
 | 返利班 | `rpa_rebate/README.md` | 占位，未实现 |
 | 四个口地图 | `rpa_oms/README.md`、根 `README.md` | — |
 | 节点业务规则（点名一节） | `docs/RPA订单中心接口开发口径_20260930.md` | 禁止通读 `docs/` |

rpa_cart/README.md

@@ -9,7 +9,7 @@
 
 ## 1 号药城支付落表
 
-从买家中心导出已付订单，按条码+商家对加购成功行，写入表 2 `operate_type=2`。
+从买家中心导出已付订单，按导出行字段写入表 2 `operate_type=2`。不对加购行。购物车和支付单没有强关联，`oms_id` 表上 NOT NULL，先填 `0`，等人后选。
 
 ```bash
 cd rpa_cart
@@ -23,7 +23,7 @@ uv run python main.py --day 2026-09-30 --days 0
 
 过滤：只落 待发货 / 待收货 / 已完成。单价用折后单价，总价 `round(折后单价×数量, 4)`。`pay_order_no` 用商城订单号，`pay_order_dtl_id` 用详情接口 `orderDetailId`。
 
-幂等：同一 `pay_order_no + pay_order_dtl_id` 不重复插入。对不上加购行或拉不到细单号的留下，不猜 `oms_id`。
+幂等：同一 `pay_order_no + pay_order_dtl_id` 不重复插入。细单号拉不到的留下，不猜。
 
 ```bash
 uv run pytest -v
@@ -31,7 +31,7 @@ uv run pytest -v
 
 ## Windows 真库
 
-Mac 打不到 `192.168.12.244`，**插入表 2 在 Windows 上测**。导出/解析在 Mac 已经跑过，Win 上要测的是登录、详情、对加购行、落表。
+Mac 打不到 `192.168.12.244`，**插入表 2 在 Windows 上测**。导出/解析在 Mac 已经跑过，Win 上要测的是登录、详情、落表。
 
 Python 3.12 + Instant Client（位数一致）+ Chrome。密码不要写进仓库。把本包和 `rpa_selection/.env` 一起带过去。
 
@@ -55,7 +55,7 @@ uv run python main.py --dry-run --day 2026-09-30 --days 0
 uv run python main.py --day 2026-09-30 --days 0
 ```
 
-看控制台 `inserted / unmatched`。对不上加购行是 `NO_CART_ROW`（表 2 还没有对应 `operate_type=1, status=1, platform=66` 的行）。细单号拉不到是 `NO_DTL_ID`。
+看控制台 `inserted / unmatched`。细单号拉不到是 `NO_DTL_ID`。
 
 查库：
 

rpa_cart/job.py

@@ -29,11 +29,7 @@ def run_yaoex_pay(
     dry_run: bool = False,
 ) -> PaySummary:
     attached = attach_detail_ids(lines, details)
-    result = match_pay_lines(
+    result = match_pay_lines(attached, repo.load_existing_pay_keys())
-        attached,
-        repo.load_cart_candidates(),
-        repo.load_existing_pay_keys(),
-    )
     if result.rows and not dry_run:
         repo.save_pay_rows(result.rows)
     reasons: dict[str, int] = {}

rpa_cart/main.py

@@ -1,4 +1,4 @@
-"""1 号药城支付落表：导出当天已付订单 → 对加购行 → 写入表 2 operate_type=2。"""
+"""1 号药城支付落表：导出当天已付订单 → 写入表 2 operate_type=2。不对加购行。"""
 from __future__ import annotations
 
 import argparse
@@ -27,7 +27,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         help="导出日期跨度，0=仅当天。与 --xls 互斥时仍用于无 xls 的导出",
     )
     parser.add_argument("--day", type=str, default=None, help="结束日期 YYYY-MM-DD，默认今天")
-    parser.add_argument("--dry-run", action="store_true", help="对码但不写库")
+    parser.add_argument("--dry-run", action="store_true", help="解析但不写库")
     parser.add_argument("--fresh-login", action="store_true", help="忽略登录缓存")
     parser.add_argument("--secrets", type=Path, default=None)
     parser.add_argument("--yaoex-auth", type=Path, default=None)
@@ -91,7 +91,7 @@ def main(argv: list[str] | None = None, repo=None) -> None:
             if not args.dry_run:
                 print(f"数据库不可用：{exc}")
                 raise SystemExit(2) from exc
-            print(f"无库，dry-run 只解析不对加购行：{exc}")
+            print(f"无库，dry-run 只解析：{exc}")
             repo = EmptyPayRepo()
 
     summary = run_yaoex_pay(repo, lines, details, dry_run=args.dry_run)

rpa_cart/repo.py

@@ -1,21 +1,11 @@
-"""表 2：读加购成功行，写支付行。"""
+"""表 2：写支付行。"""
 from __future__ import annotations
 
 from collections.abc import Callable
 from typing import Any
 
 from db import CREATE_USER_NAME, RepoUnavailable, connect_from_env, oracle_dsn
-from settle import CartCandidate, YAOEX_PLATFORM_TYPE
+from settle import YAOEX_PLATFORM_TYPE
-
-_SELECT_CART = """
-SELECT fid, oms_id, platform_type, supplier_name, bar_code,
-       purchase_qty, NVL(org_code, '') AS org_code, multi_platform_flag
-FROM rpa_cart_settlement_goods_data
-WHERE operate_type = 1
-  AND status = 1
-  AND platform_type = :platform_type
-ORDER BY fid
-"""
 
 _SELECT_PAY_KEYS = """
 SELECT pay_order_no, pay_order_dtl_id
@@ -60,9 +50,6 @@ _PAY_BIND_KEYS = (
 
 
 class EmptyPayRepo:
-    def load_cart_candidates(self) -> list[CartCandidate]:
-        return []
-
     def load_existing_pay_keys(self) -> set[tuple[str, str]]:
         return set()
 
@@ -79,17 +66,6 @@ class OraclePayRepo:
         oracle_dsn()
         return cls(connect=connect_from_env)
 
-    def load_cart_candidates(self) -> list[CartCandidate]:
-        conn = self._connect()
-        try:
-            with conn.cursor() as cur:
-                cur.execute(_SELECT_CART, {"platform_type": YAOEX_PLATFORM_TYPE})
-                return [_cart_from_rec(cur, rec) for rec in cur.fetchall()]
-        except Exception as exc:
-            raise RepoUnavailable(str(exc)) from exc
-        finally:
-            conn.close()
-
     def load_existing_pay_keys(self) -> set[tuple[str, str]]:
         conn = self._connect()
         try:
@@ -124,17 +100,3 @@ class OraclePayRepo:
             raise RepoUnavailable(str(exc)) from exc
         finally:
             conn.close()
-
-
-def _cart_from_rec(cur, rec) -> CartCandidate:
-    cols = [item[0].lower() for item in cur.description]
-    data = dict(zip(cols, rec))
-    return CartCandidate(
-        fid=int(data["fid"]),
-        oms_id=int(data["oms_id"]),
-        supplier_name=str(data.get("supplier_name") or ""),
-        bar_code=str(data.get("bar_code") or ""),
-        purchase_qty=int(data.get("purchase_qty") or 0),
-        org_code=str(data.get("org_code") or ""),
-        multi_platform_flag=int(data.get("multi_platform_flag") or 0),
-    )

rpa_cart/settle.py

@@ -1,4 +1,4 @@
-"""1 号药城导出行 → 表 2 operate_type=2。"""
+"""1 号药城导出行 → 表 2 operate_type=2。不对加购行。"""
 from __future__ import annotations
 
 from dataclasses import dataclass, replace
@@ -8,6 +8,7 @@ from typing import Any
 YAOEX_PLATFORM_TYPE = 66
 OPERATE_TYPE_PAY = 2
 PUSH_NOT_SENT = 0
+UNASSIGNED_OMS_ID = 0
 PAID_STATUSES = {"待发货", "待收货", "已完成"}
 
 
@@ -32,17 +33,6 @@ class MallLine:
     pay_order_dtl_id: str = ""
 
 
-@dataclass(frozen=True)
-class CartCandidate:
-    fid: int
-    oms_id: int
-    supplier_name: str
-    bar_code: str
-    purchase_qty: int
-    org_code: str
-    multi_platform_flag: int
-
-
 @dataclass
 class MatchResult:
     rows: list[dict[str, Any]]
@@ -73,11 +63,9 @@ def attach_detail_ids(
 
 def match_pay_lines(
     lines: list[MallLine],
-    cart_rows: list[CartCandidate],
     existing_keys: set[tuple[str, str]],
 ) -> MatchResult:
     result = MatchResult(rows=[], unmatched=[], skipped_dup=[], skipped_status=[])
-    unused = list(cart_rows)
     seen = set(existing_keys)
     for line in lines:
         if line.status not in PAID_STATUSES:
@@ -90,22 +78,17 @@ def match_pay_lines(
         if key in seen:
             result.skipped_dup.append(line)
             continue
-        cart = _pick_cart(unused, line)
-        if cart is None:
-            result.unmatched.append((line, "NO_CART_ROW"))
-            continue
-        unused.remove(cart)
         seen.add(key)
-        result.rows.append(build_pay_row(line, cart))
+        result.rows.append(build_pay_row(line))
     return result
 
 
-def build_pay_row(line: MallLine, cart: CartCandidate) -> dict[str, Any]:
+def build_pay_row(line: MallLine) -> dict[str, Any]:
     price = money4(line.pay_price)
     qty = int(line.qty)
     return {
         "operate_type": OPERATE_TYPE_PAY,
-        "oms_id": cart.oms_id,
+        "oms_id": UNASSIGNED_OMS_ID,
         "platform_type": YAOEX_PLATFORM_TYPE,
         "supplier_name": line.supplier_name,
         "bar_code": line.bar_code,
@@ -117,24 +100,11 @@ def build_pay_row(line: MallLine, cart: CartCandidate) -> dict[str, Any]:
         "pay_order_no": line.order_no,
         "pay_order_dtl_id": line.pay_order_dtl_id,
         "push_status": PUSH_NOT_SENT,
-        "org_code": cart.org_code or None,
+        "org_code": None,
-        "multi_platform_flag": cart.multi_platform_flag,
+        "multi_platform_flag": 0,
     }
 
 
-def _pick_cart(unused: list[CartCandidate], line: MallLine) -> CartCandidate | None:
-    matched = [
-        row
-        for row in unused
-        if row.bar_code == line.bar_code and row.supplier_name == line.supplier_name
-    ]
-    if not matched:
-        return None
-    equal_qty = [row for row in matched if row.purchase_qty == line.qty]
-    pool = equal_qty or matched
-    return sorted(pool, key=lambda row: row.fid)[0]
-
-
 def _find_product(pool: list[dict[str, Any]], line: MallLine) -> int | None:
     for idx, product in enumerate(pool):
         qty = int(Decimal(str(product.get("quantity") or 0)))

rpa_cart/tests/test_job.py

@@ -1,18 +1,14 @@
 from decimal import Decimal
 
 from job import run_yaoex_pay
-from settle import CartCandidate, MallLine
+from settle import MallLine
 
 
 class MemoryPayRepo:
-    def __init__(self, carts: list[CartCandidate], existing=None):
+    def __init__(self, existing=None):
-        self.carts = carts
         self.existing = existing or set()
         self.saved: list[dict] = []
 
-    def load_cart_candidates(self):
-        return list(self.carts)
-
     def load_existing_pay_keys(self):
         return set(self.existing)
 
@@ -40,20 +36,8 @@ def _line(**overrides) -> MallLine:
     return MallLine(**data)
 
 
-def test_job_attaches_dtl_matches_cart_and_inserts():
+def test_job_attaches_dtl_and_inserts_without_cart():
-    repo = MemoryPayRepo(
+    repo = MemoryPayRepo()
-        [
-            CartCandidate(
-                fid=1,
-                oms_id=2026100201,
-                supplier_name="广东壹号药业有限公司",
-                bar_code="6921665004933",
-                purchase_qty=10,
-                org_code="901",
-                multi_platform_flag=0,
-            )
-        ]
-    )
     details = {
         "ZQD1": {
             "productList": [
@@ -68,25 +52,14 @@ def test_job_attaches_dtl_matches_cart_and_inserts():
     }
     summary = run_yaoex_pay(repo, [_line()], details)
     assert summary.inserted == 1
-    assert repo.saved[0]["oms_id"] == 2026100201
+    assert summary.unmatched == 0
+    assert repo.saved[0]["oms_id"] == 0
     assert repo.saved[0]["pay_order_dtl_id"] == "881"
     assert repo.saved[0]["purchase_total"] == Decimal("98.5000")
 
 
 def test_job_dry_run_does_not_insert():
-    repo = MemoryPayRepo(
+    repo = MemoryPayRepo()
-        [
-            CartCandidate(
-                fid=1,
-                oms_id=1,
-                supplier_name="广东壹号药业有限公司",
-                bar_code="6921665004933",
-                purchase_qty=10,
-                org_code="901",
-                multi_platform_flag=0,
-            )
-        ]
-    )
     details = {
         "ZQD1": {
             "productList": [

rpa_cart/tests/test_repo.py

@@ -1,53 +1,6 @@
 from repo import OraclePayRepo
 
 
-def test_load_cart_candidates_filters_yaoex_add_cart_success():
-    fetched = [
-        (1, 11, 66, "广东壹号药业有限公司", "6921", 10, "901", 0),
-    ]
-    sql_holder: dict[str, str] = {}
-
-    class FakeCursor:
-        description = [
-            ("FID",),
-            ("OMS_ID",),
-            ("PLATFORM_TYPE",),
-            ("SUPPLIER_NAME",),
-            ("BAR_CODE",),
-            ("PURCHASE_QTY",),
-            ("ORG_CODE",),
-            ("MULTI_PLATFORM_FLAG",),
-        ]
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *args):
-            return False
-
-        def execute(self, sql, params=None):
-            sql_holder["sql"] = sql
-            sql_holder["params"] = params or {}
-
-        def fetchall(self):
-            return fetched
-
-    class FakeConn:
-        def cursor(self):
-            return FakeCursor()
-
-        def close(self):
-            pass
-
-    repo = OraclePayRepo(connect=lambda: FakeConn())
-    rows = repo.load_cart_candidates()
-    assert len(rows) == 1
-    assert rows[0].oms_id == 11
-    assert rows[0].bar_code == "6921"
-    assert "operate_type = 1" in sql_holder["sql"].lower()
-    assert sql_holder["params"]["platform_type"] == 66
-
-
 def test_save_pay_rows_inserts_operate_type_2():
     executed: list[tuple[str, dict]] = []
 
@@ -86,7 +39,7 @@ def test_save_pay_rows_inserts_operate_type_2():
         [
             {
                 "operate_type": 2,
-                "oms_id": 11,
+                "oms_id": 0,
                 "platform_type": 66,
                 "supplier_name": "广东壹号药业有限公司",
                 "bar_code": "6921",
@@ -98,7 +51,7 @@ def test_save_pay_rows_inserts_operate_type_2():
                 "pay_order_no": "ZQD1",
                 "pay_order_dtl_id": "881",
                 "push_status": 0,
-                "org_code": "901",
+                "org_code": None,
                 "multi_platform_flag": 0,
             }
         ]
@@ -106,6 +59,7 @@ def test_save_pay_rows_inserts_operate_type_2():
     assert conn.committed is True
     assert "rpa_cart_settlement_goods_data" in executed[0][0].lower()
     assert executed[0][1]["operate_type"] == 2
+    assert executed[0][1]["oms_id"] == 0
     assert executed[0][1]["pay_order_no"] == "ZQD1"
     assert executed[0][1]["pay_order_dtl_id"] == "881"
 

rpa_cart/tests/test_settle.py

@@ -2,7 +2,6 @@ from decimal import Decimal
 
 from settle import (
     PAID_STATUSES,
-    CartCandidate,
     MallLine,
     attach_detail_ids,
     build_pay_row,
@@ -31,20 +30,6 @@ def _line(**overrides) -> MallLine:
     return MallLine(**data)
 
 
-def _cart(**overrides) -> CartCandidate:
-    data = dict(
-        fid=1,
-        oms_id=2026100201,
-        supplier_name="广东壹号药业有限公司",
-        bar_code="6921665004933",
-        purchase_qty=200,
-        org_code="901",
-        multi_platform_flag=0,
-    )
-    data.update(overrides)
-    return CartCandidate(**data)
-
-
 def test_money4_rounds_half_up():
     assert money4("65.49") == Decimal("65.4900")
     assert money4(Decimal("9.85") * 200) == Decimal("1970.0000")
@@ -58,12 +43,7 @@ def test_drops_cancelled_and_keeps_paid_statuses():
         _line(status="已完成", order_no="P3"),
         _line(status="退款/退换货", order_no="R1"),
     ]
-    carts = [
+    result = match_pay_lines(lines, existing_keys=set())
-        _cart(fid=1, oms_id=1),
-        _cart(fid=2, oms_id=2),
-        _cart(fid=3, oms_id=3),
-    ]
-    result = match_pay_lines(lines, carts, existing_keys=set())
     kept = {row["pay_order_no"] for row in result.rows}
     assert kept == {"P1", "P2", "P3"}
     assert {line.order_no for line in result.skipped_status} == {"C1", "R1"}
@@ -71,7 +51,7 @@ def test_drops_cancelled_and_keeps_paid_statuses():
 
 
 def test_purchase_total_uses_discounted_price_times_qty():
-    result = match_pay_lines([_line()], [_cart()], existing_keys=set())
+    result = match_pay_lines([_line()], existing_keys=set())
     assert len(result.rows) == 1
     row = result.rows[0]
     assert row["purchase_price"] == Decimal("9.8500")
@@ -82,56 +62,39 @@ def test_purchase_total_uses_discounted_price_times_qty():
     assert row["status"] is None
     assert row["pay_order_no"] == "ZQD1"
     assert row["pay_order_dtl_id"] == "881"
-    assert row["oms_id"] == 2026100201
+    assert row["oms_id"] == 0
-    assert row["org_code"] == "901"
+    assert row["org_code"] is None
+    assert row["multi_platform_flag"] == 0
 
 
-def test_match_prefers_equal_qty_then_earlier_fid():
+def test_inserts_excel_line_without_cart_match():
-    carts = [
+    result = match_pay_lines([_line(bar_code="999")], existing_keys=set())
-        _cart(fid=10, oms_id=1, purchase_qty=50),
+    assert len(result.rows) == 1
-        _cart(fid=20, oms_id=2, purchase_qty=200),
+    assert result.rows[0]["bar_code"] == "999"
-        _cart(fid=5, oms_id=3, purchase_qty=200),
+    assert result.rows[0]["oms_id"] == 0
-    ]
+    assert result.unmatched == []
-    result = match_pay_lines([_line()], carts, existing_keys=set())
-    assert result.rows[0]["oms_id"] == 3
 
 
-def test_same_barcode_two_lines_consume_distinct_cart_rows():
+def test_same_barcode_two_lines_both_insert():
-    carts = [
-        _cart(fid=1, oms_id=11, purchase_qty=10),
-        _cart(fid=2, oms_id=22, purchase_qty=50),
-    ]
     lines = [
         _line(order_no="A", qty=10, pay_price=Decimal("53.40"), pay_order_dtl_id="1"),
         _line(order_no="A", qty=50, pay_price=Decimal("55.21"), pay_order_dtl_id="2"),
     ]
-    result = match_pay_lines(lines, carts, existing_keys=set())
+    result = match_pay_lines(lines, existing_keys=set())
-    oms = sorted(row["oms_id"] for row in result.rows)
+    assert [row["pay_order_dtl_id"] for row in result.rows] == ["1", "2"]
-    assert oms == [11, 22]
+    assert all(row["oms_id"] == 0 for row in result.rows)
 
 
 def test_skips_existing_pay_order_and_dtl():
-    result = match_pay_lines(
+    result = match_pay_lines([_line()], existing_keys={("ZQD1", "881")})
-        [_line()],
-        [_cart()],
-        existing_keys={("ZQD1", "881")},
-    )
     assert result.rows == []
     assert len(result.skipped_dup) == 1
 
 
-def test_unmatched_without_cart_or_dtl_is_not_inserted():
+def test_unmatched_without_dtl_is_not_inserted():
-    no_cart = match_pay_lines([_line(bar_code="999")], [_cart()], existing_keys=set())
+    result = match_pay_lines([_line(pay_order_dtl_id="")], existing_keys=set())
-    assert no_cart.rows == []
+    assert result.rows == []
-    assert no_cart.unmatched[0][1] == "NO_CART_ROW"
+    assert result.unmatched[0][1] == "NO_DTL_ID"
-
-    no_dtl = match_pay_lines(
-        [_line(pay_order_dtl_id="")],
-        [_cart()],
-        existing_keys=set(),
-    )
-    assert no_dtl.rows == []
-    assert no_dtl.unmatched[0][1] == "NO_DTL_ID"
 
 
 def test_attach_detail_ids_matches_qty_and_pay_price():
@@ -161,9 +124,11 @@ def test_attach_detail_ids_matches_qty_and_pay_price():
     assert [line.pay_order_dtl_id for line in attached] == ["1001", "1002"]
 
 
-def test_build_pay_row_copies_multi_platform_flag():
+def test_build_pay_row_from_excel_fields():
-    row = build_pay_row(_line(), _cart(multi_platform_flag=1))
+    row = build_pay_row(_line())
-    assert row["multi_platform_flag"] == 1
+    assert row["oms_id"] == 0
+    assert row["org_code"] is None
+    assert row["multi_platform_flag"] == 0
     assert row["supplier_name"] == "广东壹号药业有限公司"
     assert row["bar_code"] == "6921665004933"
     assert row["push_status"] == 0

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()
-    def new_page(self, *, viewport: dict[str, int] | None = None) -> tuple[Any, Any]:
+        profile.mkdir(parents=True, exist_ok=True)
-        self.start()
+        print(f"选品 Chrome 用户目录：{profile}")
-        browser = self._playwright.chromium.launch(
+        try:
+            self._context = self._playwright.chromium.launch_persistent_context(
+                str(profile),
                 headless=False,
                 channel="chrome",
-            args=["--new-window", "--start-maximized"],
+                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"],
             )
-        self._browsers.append(browser)
+        except Exception as exc:
-        context = browser.new_context(viewport=viewport or {"width": 1440, "height": 900})
+            raise RuntimeError(
-        page = context.new_page()
+                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 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()

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()