From bcf5e03d078f93988e489ef971abbbcacea90c9d Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Sat, 24 Oct 2026 11:37:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=80=89=E5=93=81=E5=87=91=E9=BD=90?= =?UTF-8?q?=E5=90=8E=E6=8C=89=E6=9C=AC=E6=AC=A1=E6=8B=86=E5=AE=9A=E7=9A=84?= =?UTF-8?q?=20sku=20=E5=8A=A0=E8=BD=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 凑齐的新行先标成加购中并取回 fid，同一班加车。表 2 已有该 omsId 时不加第二份。加购失败不回滚表 1。补偿班只补仍是系统管理员或过期加购中的行。 --- AGENTS.md | 16 +-- rpa_cart/README.md | 2 +- rpa_cart/cart_job.py | 147 +++++++++++++++++++++--- rpa_cart/tests/test_cart_job.py | 100 +++++++++++++++- rpa_selection/README.md | 3 +- rpa_selection/job.py | 125 +++++++++++++++++++- rpa_selection/main.py | 133 ++++++++++++++++++++- rpa_selection/repo.py | 71 +++++++++--- rpa_selection/search_yaoex.py | 5 +- rpa_selection/search_ybm.py | 8 +- rpa_selection/tests/test_cli.py | 31 ++++- rpa_selection/tests/test_job.py | 115 +++++++++++++++++- rpa_selection/tests/test_mapping.py | 3 + rpa_selection/tests/test_oracle_repo.py | 83 +++++++++---- 14 files changed, 763 insertions(+), 79 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6a43ae5..8696a02 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,6 +1,6 @@ # AutoRpa -洁达 RPA。当前主线是**订单中心入站**（`rpa_oms`）、**选品班次**（`rpa_selection/`）、**加购班次**（`rpa_cart/`）、**合同班次**（`rpa_contract/`）和**返利班次**（`rpa_rebate/`）：收件落 Oracle，请求里不跑选品、不录合同；选品另起进程捞 `deal_flag=0`；加购另起进程领表 2 成功行；合同另起进程领表 3；返利另起进程领表 4 起草行 / 表 3 只录协议行。根上不再有 `选品流程/`、`state_center/`、`rpa_yisaiqi/`。 +洁达 RPA。当前主线是**订单中心入站**（`rpa_oms`）、**选品班次**（`rpa_selection/`）、**加购班次**（`rpa_cart/`）、**合同班次**（`rpa_contract/`）和**返利班次**（`rpa_rebate/`）：收件落 Oracle，请求里不跑选品、不录合同；选品另起进程捞 `deal_flag=0`，凑齐后同一进程按本次拆定的 sku 加车；加购进程只补表 2 里仍是「系统管理员」或过期「加购中」的行；合同另起进程领表 3；返利另起进程领表 4 起草行 / 表 3 只录协议行。根上不再有 `选品流程/`、`state_center/`、`rpa_yisaiqi/`。 智力用在**定落点、想清楚对错**；速度和 token 靠下面这张地图定点打开，禁止用扫仓代替思考。改对应了，反工和消耗都会少。 @@ -28,7 +28,7 @@ | 接口3 收件契约 | `rpa_oms/app/inbound/purchase_contract.py` | `rpa_oms/tests/test_purchase_contract_receive.py::点名` | | 接口3 Oracle 落表 | `rpa_oms/app/inbound/oracle_contract_repo.py` | `rpa_oms/tests/test_oracle_contract_repo.py::点名` | | 选品规则 / 拆量 | `rpa_selection/rules.py`、`pipeline.py` | `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` | -| 选品班次落表 | `rpa_selection/job.py`、`persist.py`、`repo.py` | `rpa_selection/tests/test_job.py::test_job_searches_barcode_not_approval_no` | +| 选品班次落表 / 凑齐后加车 | `rpa_selection/job.py`、`persist.py`、`repo.py`；加车 `rpa_cart/cart_job.py` `add_known_lines` | `rpa_selection/tests/test_job.py::test_success_adds_each_split_sku_without_rereading_table2` | | 合同班合单落表 | `rpa_contract/job.py`、`group.py`、`persist.py`、`repo.py` | `rpa_contract/tests/test_job.py::test_job_merges_same_supplier_and_writes_table4` | | 合同页面步骤 | `rpa_contract/contract_login.py`（S02–S05）/ `contract_form.py`（S06–S08、S10）/ `contract_goods.py`（S09）/ `contract_upload.py`（S11）/ `contract_submit.py`（S12）/ `contract_number.py`（S13）；共用 `contract_common.py`；入口 `contract_flow_steps.py` | `ACCEPT SUMMARY` 或 `rpa_contract/screenshots/` 下 `ACCEPT.txt` | | 合同入参 / 客商草稿 | `rpa_contract/contract_draft.py` | `rpa_contract` 点名用例 | @@ -81,7 +81,7 @@ ## 选品落点 -班次作业在 `rpa_selection/`，不在接口 1 请求里跑。从表 1 领 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，两平台合并后按四级仓序贪心拆量，写入表 2。凑齐才落成功行；凑不齐不落部分成功。 +班次作业在 `rpa_selection/`，不在接口 1 请求里跑。从表 1 领 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，两平台合并后按四级仓序贪心拆量，写入表 2。凑齐在内存里定，才落成功行；凑不齐不落部分成功。凑齐的新行同一班按这次拆定的 sku 加车，不再去表 2 重捞来判断齐没齐。表 2 已有该 `omsId` 时不加第二份。 口径：`docs/superpowers/specs/2026-10-02-selection-split-allocation-design.md`。 @@ -89,17 +89,17 @@ | --- | --- | --- | | 领取 | `claim_pending` | `deal_flag` 0→3；`FOR UPDATE SKIP LOCKED`；中途崩溃停在 3，超过租约（`deal_time` 30 分钟）可再领 | | 规则 | `allocate` / `run_selection` | 自营仓全称匹配、毛利严格大于 4%、效期 ≥ 12 月；同仓多个 sku 都参与拆量，低价先吃；按仓序拆量 | -| 落表 | `build_result` + `save_result` | 成功每条拆量一行 `status=1`，合计=计划量；`push_status=0` | +| 落表 | `build_result` + `save_result` | 成功每条拆量一行 `status=1`，合计=计划量；`push_status=0`。要马上加车的新行更新人写成「加购中」，并 RETURNING 回 fid | | 库存不足 | `KIND_BUSINESS_FAIL` | 两边都搜到但合计不够：`deal_flag=2`，该 `omsId` 不重跑 | | 对端故障 | `KIND_SYSTEM_FAIL` | 一边故障且剩下的凑不齐：可把 `deal_flag` 改回 0；禁止写成库存不足 | -湖南壹号、甘肃壹号云不在自营集合。不加购，不推接口 2。 +湖南壹号、甘肃壹号云不在自营集合。凑齐后按本次 sku 加车，不推接口 2。加购失败不回滚表 1 的选品成功。登录或查询失败把表 2 戳放回「系统管理员」。同单有成功又有失败时，失败原因带「未加齐」。 验收：`cd rpa_selection` → `uv run pytest tests/test_pipeline.py::test_split_across_malls_is_success -q`（杭州 3 + 广东 7 = 计划 10）。 ## 加购落点 -班次作业在 `rpa_cart/`，不在选品请求里跑。领表 2 `operate_type=1 AND status=1 AND push_status=0`，戳为 `系统管理员` 或过期的 `加购中`。登录 1 号药城 / 药帮忙，按条形码再搜，对上 `supplier_name`，按行加车。 +补偿班在 `rpa_cart/`。正常凑齐的加车发生在选品班里。本班领表 2 `operate_type=1 AND status=1 AND push_status=0`，戳为 `系统管理员` 或过期的 `加购中`。这些行没有 sku，登录 1 号药城 / 药帮忙后按条形码再搜，对上 `supplier_name`，按行加车。 | 步 | 函数 | 要点 | | --- | --- | --- | @@ -171,8 +171,8 @@ - 接口1：点名 `rpa_oms/tests/test_sales_goods_receive.py` 或 `test_oracle_repo.py`；禁止为「显得全面」跑整包再去连真库。 - 接口2：点名 `rpa_oms/tests/test_cart_settlement_outbound.py::test_assembles_unpushed_rows_as_document_envelope`；SQL 截组 `tests/test_oracle_cart_repo.py::test_list_unpushed_keeps_whole_oms_group_when_limit_is_3`。 - 接口3：点名 `rpa_oms/tests/test_purchase_contract_receive.py` 或 `test_oracle_contract_repo.py`。 -- 选品：点名 `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` 或 `test_rules.py::test_margin_exactly_4_percent_passes`。 -- 加购：点名 `rpa_cart/tests/test_cart_job.py::test_job_adds_split_rows_without_insert` 或 `test_job_marks_out_of_stock`。 +- 选品：点名 `rpa_selection/tests/test_pipeline.py::test_split_across_malls_is_success` 或 `test_rules.py::test_margin_exactly_4_percent_passes`。凑齐后加车点名 `rpa_selection/tests/test_job.py::test_success_adds_each_split_sku_without_rereading_table2`。 +- 加购：点名 `rpa_cart/tests/test_cart_job.py::test_job_adds_split_rows_without_insert` 或 `test_job_marks_out_of_stock`。选品交来的 sku 点名 `test_known_lines_keep_each_sku_and_flag_incomplete`。 - 合同班：点名 `rpa_contract/tests/test_job.py::test_job_merges_same_supplier_and_writes_table4` 或 `test_job_marks_diligence_fail`。 - 返利班：点名 `rpa_rebate/tests/test_job.py::test_job_finds_contract_and_writes_agreement` 或 `test_job_leaves_not_found_for_next_shift`。 - 返利查号页：只读控制台 `ACCEPT SUMMARY REBATE_FIND` 或 `rpa_rebate/screenshots/REBATE_FIND/ACCEPT.txt`。 diff --git a/rpa_cart/README.md b/rpa_cart/README.md index 6382ac6..4782572 100644 --- a/rpa_cart/README.md +++ b/rpa_cart/README.md @@ -9,7 +9,7 @@ ## 加购 -选品已经把「哪个仓、加几件」写成表 2：`operate_type=1`、`status=1`、`push_status=0`、`update_user_name=系统管理员`。本班读这些行，登录 1 号药城 / 药帮忙，按条形码再搜一次，对上 `supplier_name`，改数量，加入购物车。同一 `omsId` 拆两仓、两平台要加齐。 +选品凑齐的行在选品班里按本次拆定的 sku 直接加车，不经过这里。本命令只补还停在 `update_user_name=系统管理员`、或租约过期的「加购中」的行。那些行表 2 里没有 sku，仍按条形码再搜，对上 `supplier_name` 再加。同一 `omsId` 拆两仓、两平台要加齐。 成功不插新行，只把 `update_user_name` 改成 `加购班`。加不进去（登录挂、库存没了、中包装加不进、对不上仓）改该行 `status=2`，写失败原因。领取时先盖 `加购中`；中途崩溃停在这一戳。`update_time` 超过 30 分钟的「加购中」下一班会再领；未过期的需改回 `系统管理员` 再跑。 diff --git a/rpa_cart/cart_job.py b/rpa_cart/cart_job.py index 8e00e41..8c78743 100644 --- a/rpa_cart/cart_job.py +++ b/rpa_cart/cart_job.py @@ -37,6 +37,23 @@ class CartSummary: failed: int = 0 +# 选品已经拆定的一行。加这行时不再搜、不再按商家名重对。 +@dataclass +class KnownLine: + fid: int + oms_id: int + platform_type: int + supplier_name: str + bar_code: str + qty: int + sku_id: str + vendor_raw: str + vendor_norm: str + stock: float | None + price: float | None + raw: dict + + def _session(repo): """有 session 就借一条，没有就空上下文。商城调用必须在这个块外面。""" session = getattr(repo, "session", None) @@ -71,28 +88,122 @@ def run_cart_batch( (row, *_add_one(row, searchers, adders, login_errors, readers)) for row in group ] - incomplete = _is_incomplete(outcomes) - if incomplete: - logger.warning("加购未加齐 oms_id=%s", oms_id) - with _session(repo): - for row, reason, code, text in outcomes: - if reason is None: - repo.mark_success(row.fid) - summary.added += 1 - continue - if incomplete: - reason = _incomplete_reason(reason) - if _retryable(code, reason): - repo.mark_release(row.fid, reason) - else: - repo.mark_fail(row.fid, reason) - summary.failed += 1 - if code and text: - audits.append(ShiftError("加购", str(row.fid), code, text)) + _apply_outcomes(repo, oms_id, outcomes, summary, audits) repo.save_shift_errors(audits) return summary +def add_known_lines( + repo, + lines: list[KnownLine], + *, + adders: dict[int, Adder], + login_errors: dict[int, str], + cart_readers: dict[int, CartReader] | None = None, +) -> CartSummary: + """按调用方已经拆定的 sku 加车并盖章。不再搜条形码，不再按商家名重对。 + + 同一 oms_id 先全部加完再盖章。成功只改戳。业务失败改 status=2。 + 同组有成功又有失败时，失败原因带「未加齐」。登录失败把戳放回系统管理员。 + 行必须已经是「加购中」，否则盖章 SQL 不会改到它。 + """ + audits: list[ShiftError] = [] + readers = cart_readers or {} + summary = CartSummary(claimed=len(lines)) + for oms_id, group in _by_known(lines).items(): + outcomes = [ + (line, *_add_known(line, adders, login_errors, readers)) for line in group + ] + _apply_outcomes(repo, oms_id, outcomes, summary, audits) + repo.save_shift_errors(audits) + return summary + + +# 按盖章结果改表 2。加车必须发生在这个 session 外面。 +def _apply_outcomes(repo, oms_id, outcomes, summary: CartSummary, audits: list) -> None: + incomplete = _is_incomplete(outcomes) + if incomplete: + logger.warning("加购未加齐 oms_id=%s", oms_id) + with _session(repo): + for item, reason, code, text in outcomes: + if reason is None: + repo.mark_success(item.fid) + summary.added += 1 + continue + if incomplete: + reason = _incomplete_reason(reason) + if _retryable(code, reason): + repo.mark_release(item.fid, reason) + else: + repo.mark_fail(item.fid, reason) + summary.failed += 1 + if code and text: + audits.append(ShiftError("加购", str(item.fid), code, text)) + + +# 按 oms_id 保序分组。同组先加完再盖章，才能看出有没有加齐。 +def _by_known(lines: list[KnownLine]) -> OrderedDict[int, list[KnownLine]]: + groups: OrderedDict[int, list[KnownLine]] = OrderedDict() + for line in lines: + groups.setdefault(int(line.oms_id), []).append(line) + return groups + + +def _add_known( + line: KnownLine, + adders: dict[int, Adder], + login_errors: dict[int, str], + cart_readers: dict[int, CartReader], +) -> tuple[str | None, str | None, str | None]: + """对一条已经拆定的行加车。返回 (失败原因, 审计短码, 原文)。成功三项都是 None。 + + 不看选品当时的库存。库存以商城加车回包为准。购物车里已经不少于本行时直接成功。 + 没有加车函数时放回待领，让补偿班再处理。 + """ + platform = line.platform_type + if platform in login_errors: + text = (login_errors[platform] or "").strip() + logger.warning("加购登录失败 platform=%s: %s", platform, login_errors[platform]) + return _system("LOGIN_FAILED"), ("LOGIN_FAILED" if text else None), (text or None) + adder = adders.get(platform) + if adder is None: + return _system("LOGIN_FAILED"), "LOGIN_FAILED", "未配置加车" + bar_code = (line.bar_code or "").strip() + if not bar_code: + return "缺少条形码", None, None + qty = int(line.qty) + if qty <= 0: + return "采购数量无效", None, None + reader = cart_readers.get(platform) + if reader is not None: + try: + already = int(reader(bar_code, line.supplier_name) or 0) + except Exception as exc: + logger.exception("加购查购物车失败 platform=%s bar_code=%s", platform, bar_code) + return _system("QUERY_FAILED"), "QUERY_FAILED", _text(exc) + if already >= qty: + return None, None, None + offer = CartOffer( + sku_id=line.sku_id, + vendor_raw=line.vendor_raw, + vendor_norm=line.vendor_norm, + stock=line.stock, + price=line.price, + extra={"raw": line.raw} if isinstance(line.raw, dict) else {}, + ) + try: + outcome = adder(offer, qty) + except Exception as exc: + logger.exception("加购加车失败 platform=%s fid=%s", platform, line.fid) + return _system("ADD_FAILED"), "ADD_FAILED", _text(exc) + if outcome.kind == KIND_OK: + return None, None, None + reason = (outcome.reason or "加购失败").strip() + if outcome.kind == KIND_SYSTEM: + return _system(reason), None, None + return clip_reason(reason), None, None + + # 按 oms_id 保序分组。同组先加完再盖章，才能看出有没有加齐。 def _by_oms(rows: list[CartRow]) -> OrderedDict[int, list[CartRow]]: groups: OrderedDict[int, list[CartRow]] = OrderedDict() diff --git a/rpa_cart/tests/test_cart_job.py b/rpa_cart/tests/test_cart_job.py index 7231ffc..e22cecb 100644 --- a/rpa_cart/tests/test_cart_job.py +++ b/rpa_cart/tests/test_cart_job.py @@ -3,7 +3,15 @@ from contextlib import contextmanager from datetime import datetime, timedelta -from cart_job import KIND_BUSINESS, KIND_OK, KIND_SYSTEM, AddOutcome, run_cart_batch +from cart_job import ( + KIND_BUSINESS, + KIND_OK, + KIND_SYSTEM, + AddOutcome, + KnownLine, + add_known_lines, + run_cart_batch, +) from cart_match import CartOffer from cart_repo import STAMP_CLAIMING, STAMP_DONE, STAMP_SELECTION, MemoryCartRepo @@ -449,3 +457,93 @@ def test_rerun_skips_add_when_cart_already_has_the_line(): assert repo.rows[1]["status"] == 1 assert repo.rows[1]["update_user_name"] == STAMP_DONE assert repo.rows[1]["fail_reason"] is None + + +def _known(fid: int, sku: str, **overrides) -> KnownLine: + """一条已经拆定的加购行。raw 里的 skuId 必须跟 sku 一起交到加车。""" + data = dict( + fid=fid, + oms_id=2026100201, + platform_type=YBM, + supplier_name=HANGZHOU, + bar_code="06901234567890", + qty=5, + sku_id=sku, + vendor_raw=HANGZHOU, + vendor_norm=HANGZHOU, + stock=5.0, + price=90.0, + raw={"skuId": sku}, + ) + data.update(overrides) + return KnownLine(**data) + + +# 同仓两条 sku 各加各的，不按商家名并到低价那条。一条失败时只给失败行标未加齐。 +def test_known_lines_keep_each_sku_and_flag_incomplete(): + repo = MemoryCartRepo( + [ + _row( + fid=11, + platform_type=YBM, + supplier_name=HANGZHOU, + purchase_qty=6, + update_user_name=STAMP_CLAIMING, + ), + _row( + fid=12, + platform_type=YBM, + supplier_name=HANGZHOU, + purchase_qty=4, + update_user_name=STAMP_CLAIMING, + ), + ] + ) + added: list[tuple[str, int]] = [] + + # 高价那条被商城拒绝。 + def add_ybm(offer: CartOffer, qty: int): + added.append((offer.sku_id, qty)) + assert offer.extra["raw"]["skuId"] == offer.sku_id + if offer.sku_id == "DEAR": + return AddOutcome(kind=KIND_BUSINESS, reason="库存不足") + return AddOutcome(kind=KIND_OK) + + summary = add_known_lines( + repo, + [_known(11, "CHEAP", qty=6), _known(12, "DEAR", qty=4, price=92.0)], + adders={YBM: add_ybm}, + login_errors={}, + ) + assert added == [("CHEAP", 6), ("DEAR", 4)] + assert summary.added == 1 + assert summary.failed == 1 + assert repo.rows[11]["status"] == 1 + assert repo.rows[11]["update_user_name"] == STAMP_DONE + assert repo.rows[12]["status"] == 2 + assert repo.rows[12]["fail_reason"] == "未加齐：库存不足" + + +# 没配加车函数时放回系统管理员，status 仍是 1。 +def test_known_line_without_adder_releases_for_retry(): + repo = MemoryCartRepo([_row(fid=1, update_user_name=STAMP_CLAIMING)]) + summary = add_known_lines( + repo, + [ + _known( + 1, + "S1", + platform_type=YAOEX, + supplier_name=GUANGDONG, + vendor_raw=GUANGDONG, + vendor_norm=GUANGDONG, + ) + ], + adders={}, + login_errors={}, + ) + assert summary.added == 0 + assert summary.failed == 1 + assert repo.rows[1]["status"] == 1 + assert repo.rows[1]["update_user_name"] == STAMP_SELECTION + assert "LOGIN_FAILED" in (repo.rows[1]["fail_reason"] or "") diff --git a/rpa_selection/README.md b/rpa_selection/README.md index d140368..58c14b4 100644 --- a/rpa_selection/README.md +++ b/rpa_selection/README.md @@ -1,6 +1,6 @@ # 洁达商城选品 -班次作业：从表 1 捞 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，规则选出仓，结果写入表 2。不加购，不推订单中心。 +班次作业：从表 1 捞 `deal_flag=0`，按**条形码**搜 1 号药城 / 药帮忙，规则选出仓，结果写入表 2。凑齐的行在本班按这次拆定的 sku 加车。不推订单中心。 口径：`docs/RPA订单中心接口开发口径_20260930.md` 第 3 节。 @@ -13,6 +13,7 @@ rpa_sales_goods_mapping_data (deal_flag=0) → 按条形码搜索（批准文号只落表，不拿去搜） → 选品规则（两平台合并，按仓序拆量） → rpa_cart_settlement_goods_data（operate_type=1, push_status=0） + → 凑齐的新行标成「加购中」，按本次 sku 加车，成功改「加购班」 → 表 1 deal_flag=1 成功 / 2 失败 ``` diff --git a/rpa_selection/job.py b/rpa_selection/job.py index 8ac54a8..e7db9ad 100644 --- a/rpa_selection/job.py +++ b/rpa_selection/job.py @@ -1,23 +1,49 @@ # 选品班次：领取表 1 → 按条形码搜平台 → 统一拆量写表 2。 # +# 凑齐在内存里定。传入 cart_finish 时，新插入的成功行标成「加购中」， +# 落库后用这次拆定的 sku 加车，不再去表 2 重捞。 # 崩在 deal_flag=3：超过租约可再领，未过期需把 3 改回 0。 # 查询失败原因列只写 QUERY_FAILED，原文进 rpa_selection_query_audit。 from __future__ import annotations import logging -from dataclasses import dataclass -from typing import Callable +from dataclasses import dataclass, field +from typing import Any, Callable -from persist import STATUS_OK, build_result +from mall_config import MALL_PLATFORM_TYPE +from persist import DEAL_OK, STATUS_OK, build_result from pipeline import KIND_BUSINESS_FAIL, MallOutcome, SelectionResult, run_selection -from repo import FinishWrite, MappingRow, QueryAudit -from rules import Candidate, PurchaseLine, validate_purchase +from repo import STAMP_SELECTION, FinishWrite, MappingRow, QueryAudit +from rules import Allocation, Candidate, PurchaseLine, validate_purchase + +STAMP_CLAIMING = "加购中" Searcher = Callable[[str], list[Candidate]] logger = logging.getLogger("rpa_selection") +# 凑齐后交给加车的一行。sku 和 raw 来自这次搜索，不从表 2 反查。 +@dataclass +class CartLine: + fid: int + oms_id: int + platform_type: int + supplier_name: str + bar_code: str + qty: int + sku_id: str + vendor_raw: str + vendor_norm: str + stock: float | None + price: float | None + raw: dict[str, Any] = field(default_factory=dict) + + +# cart_finish(lines) -> (加购成功行数, 失败行数)。未传入则只落表。 +CartFinish = Callable[[list[CartLine]], tuple[int, int]] + + # 一个班次的领取和落表计数。 @dataclass class JobSummary: @@ -25,6 +51,8 @@ class JobSummary: added: int = 0 skipped: int = 0 failed: int = 0 + cart_added: int = 0 + cart_failed: int = 0 # 把表 1 行收成规则层的采购行。缺价格或数量保持 None。 @@ -52,11 +80,14 @@ def run_batch( searchers: dict[str, Searcher], login_errors: dict[str, str], limit: int | None = None, + cart_finish: CartFinish | None = None, ) -> JobSummary: """领取待选品行，按行搜平台、拆量，一行搜完就落这一行。 每个平台开搜前刷新 deal_time。落库前由仓库再确认仍是 deal_flag=3， 且表 2 还没有该 oms_id。limit 交给领取层。查询失败原文在该行落表后另写。 + cart_finish 只在这一行新插入了凑齐的成功行之后调用，参数是这次拆定的 sku。 + 表 2 已有该 oms_id 时不调用，避免再加一份。 """ rows = repo.claim_pending(limit=limit) summary = JobSummary(claimed=len(rows)) @@ -78,7 +109,8 @@ def run_batch( ) result = build_result(purchase, selection) _count(summary, result.cart_rows) - repo.save_results( + _mark_cart_stamp(result, cart_finish) + saved = repo.save_results( [ FinishWrite( oms_id=result.oms_id, @@ -89,11 +121,92 @@ def run_batch( ) ] ) + fids = list(saved[0]) if saved else [] + if cart_finish is not None and result.deal_flag == DEAL_OK: + _finish_cart(summary, cart_finish, purchase, selection, result.cart_rows, fids) if row_audits: repo.save_query_audits(row_audits) return summary +# 要马上加车的成功行先标成加购中，补偿班才不会在加车完成前领走。其余行保持系统管理员。 +def _mark_cart_stamp(result, cart_finish: CartFinish | None) -> None: + claim = cart_finish is not None and result.deal_flag == DEAL_OK and bool(result.cart_rows) + for row in result.cart_rows: + if claim and row.get("status") == STATUS_OK: + row["update_user_name"] = STAMP_CLAIMING + else: + row["update_user_name"] = STAMP_SELECTION + + +# 新插入的凑齐行交给加车。行数对不上就留下加购中，不拿错 sku 去加。 +def _finish_cart( + summary: JobSummary, + cart_finish: CartFinish | None, + purchase: PurchaseLine, + selection: SelectionResult, + cart_rows: list[dict], + fids: list[int], +) -> None: + if cart_finish is None or not fids: + return + if purchase.oms_id is None: + return + lines = _cart_lines(purchase, selection.allocations, cart_rows, fids) + if not lines: + logger.error( + "选品加购行数对不上 oms_id=%s allocations=%s fids=%s", + purchase.oms_id, + len(selection.allocations), + len(fids), + ) + return + try: + added, failed = cart_finish(lines) + except Exception: + logger.exception("选品后加购未盖章 oms_id=%s", purchase.oms_id) + return + summary.cart_added += int(added) + summary.cart_failed += int(failed) + + +# 按拆量顺序配上刚插入的 fid。顺序或状态不一致就返回空，调用方不加车。 +def _cart_lines( + purchase: PurchaseLine, + allocations: list[Allocation], + cart_rows: list[dict], + fids: list[int], +) -> list[CartLine]: + if len(allocations) != len(fids) or len(cart_rows) != len(fids): + return [] + if any(row.get("status") != STATUS_OK for row in cart_rows): + return [] + lines: list[CartLine] = [] + for item, fid in zip(allocations, fids): + candidate = item.candidate + platform = MALL_PLATFORM_TYPE.get(candidate.mall_name) + if platform is None: + return [] + raw = candidate.extra.get("raw") if isinstance(candidate.extra, dict) else None + lines.append( + CartLine( + fid=int(fid), + oms_id=int(purchase.oms_id or 0), + platform_type=int(platform), + supplier_name=candidate.vendor_raw, + bar_code=purchase.bar_code, + qty=int(item.qty), + sku_id=candidate.sku_id, + vendor_raw=candidate.vendor_raw, + vendor_norm=candidate.vendor_norm, + stock=candidate.stock, + price=candidate.price, + raw=raw if isinstance(raw, dict) else {}, + ) + ) + return lines + + def _select_row( purchase: PurchaseLine, platforms: list[str], diff --git a/rpa_selection/main.py b/rpa_selection/main.py index 5187473..afd0624 100644 --- a/rpa_selection/main.py +++ b/rpa_selection/main.py @@ -1,8 +1,9 @@ -# 选品班次入口：登录 1 号药城 / 药帮忙，按条形码搜，结果写入表 2。不加购，不推接口 2。 +# 选品班次入口：登录 1 号药城 / 药帮忙，按条形码搜，凑齐后按本次 sku 加车。不推接口 2。 from __future__ import annotations import argparse +import sys import time from pathlib import Path @@ -21,7 +22,12 @@ from auth_login import ( ) from rpa_oracle import RepoUnavailable, load_job_env from job import run_batch -from mall_config import YAOEX_MALL_NAME, YBM_MALL_NAME +from mall_config import ( + YAOEX_MALL_NAME, + YAOEX_PLATFORM_TYPE, + YBM_MALL_NAME, + YBM_PLATFORM_TYPE, +) from repo import OracleSelectRepo from rules import Candidate from search_yaoex import YaoexAuth @@ -29,6 +35,7 @@ from search_yaoex import search_candidates as search_yaoex_candidates from search_ybm import search_candidates as search_ybm_candidates DEFAULT_CLAIM_LIMIT = 50 +_CART_ROOT = Path(__file__).resolve().parent.parent / "rpa_cart" def parse_args(argv: list[str] | None = None) -> argparse.Namespace: @@ -178,6 +185,15 @@ def main(argv: list[str] | None = None, repo=None) -> None: searchers[YBM_MALL_NAME] = search_ybm + adders, login_by_type = _cart_adders( + platforms, + yaoex_auth=yaoex_auth, + ybm_cookies=ybm_cookies, + yaoex_session=yaoex_session, + ybm_session=ybm_session, + login_errors=login_errors, + ) + cart_finish = _cart_finish(repo, adders, login_by_type) try: summary = run_batch( repo, @@ -185,6 +201,7 @@ def main(argv: list[str] | None = None, repo=None) -> None: searchers=searchers, login_errors=login_errors, limit=args.limit, + cart_finish=cart_finish, ) finally: yaoex_session.close() @@ -194,12 +211,122 @@ def main(argv: list[str] | None = None, repo=None) -> None: print( f"领取={summary.claimed} ADDED={summary.added} " f"SKIPPED={summary.skipped} FAILED={summary.failed} " + f"加购成功={summary.cart_added} 加购失败={summary.cart_failed} " f"用时 {elapsed:.2f} 秒" ) - print("已写入 rpa_cart_settlement_goods_data，push_status=0，未推订单中心，未加购。") + print("已写入 rpa_cart_settlement_goods_data，push_status=0，未推订单中心。") finally: runtime.close() +# 把加购包放进导入路径。选品目录仍排在前面，避免拿错同名模块。 +def _ensure_cart_path() -> None: + if _CART_ROOT.is_dir() and str(_CART_ROOT) not in sys.path: + sys.path.append(str(_CART_ROOT)) + + +# 用本班已登录的会话加车。登录失败的平台不配加车函数。 +def _cart_adders( + platforms: list[str], + *, + yaoex_auth, + ybm_cookies, + yaoex_session: requests.Session, + ybm_session: requests.Session, + login_errors: dict[str, str], +) -> tuple[dict, dict[int, str]]: + _ensure_cart_path() + from cart_yaoex import add_yaoex_cart + from cart_ybm import add_ybm_cart + + adders: dict = {} + if yaoex_auth is not None and YAOEX_MALL_NAME in platforms: + + # 按选品拆定的 sku 加入 1 号药城购物车。 + def add_yaoex(offer, qty: int): + return add_yaoex_cart(yaoex_auth, offer, qty, session=yaoex_session) + + adders[YAOEX_PLATFORM_TYPE] = add_yaoex + if ybm_cookies is not None and YBM_MALL_NAME in platforms: + + # 按选品拆定的 sku 加入药帮忙购物车。 + def add_ybm(offer, qty: int): + return add_ybm_cart(ybm_cookies, offer, qty, session=ybm_session) + + adders[YBM_PLATFORM_TYPE] = add_ybm + login_by_type: dict[int, str] = {} + if YAOEX_MALL_NAME in login_errors: + login_by_type[YAOEX_PLATFORM_TYPE] = login_errors[YAOEX_MALL_NAME] + if YBM_MALL_NAME in login_errors: + login_by_type[YBM_PLATFORM_TYPE] = login_errors[YBM_MALL_NAME] + return adders, login_by_type + + +# 内存选品仓库上的盖章。字段和加购班的成功、失败、放回一致。 +class _MemoryCartStamp: + # 盖章写回选品仓库里刚插入的表 2 行。 + def __init__(self, select_repo): + self._repo = select_repo + self.shift_errors: list = [] + + # 成功只改戳，status 保持 1。 + def mark_success(self, fid: int) -> None: + row = self._row(fid) + row["update_user_name"] = "加购班" + row["fail_reason"] = None + + # 失败改 status=2，戳改成加购班。 + def mark_fail(self, fid: int, reason: str) -> None: + row = self._row(fid) + row["status"] = 2 + row["fail_reason"] = reason + row["update_user_name"] = "加购班" + + # 登录或查询失败放回待领。status 不动。 + def mark_release(self, fid: int, reason: str) -> None: + row = self._row(fid) + row["update_user_name"] = "系统管理员" + row["fail_reason"] = reason + + # 记下未捕获异常原文。不改表 2。 + def save_shift_errors(self, items) -> None: + self.shift_errors.extend(items) + + # 按刚插入的 fid 找表 2 行。找不到则 KeyError。 + def _row(self, fid: int) -> dict: + for row in self._repo.cart: + if int(row.get("fid") or 0) == int(fid): + return row + raise KeyError(fid) + + +# 真库和内存选品仓库都能盖加购章。真库复用选品这条池，避免再开一条连接池。 +def _stamp_repo(select_repo): + if isinstance(select_repo, OracleSelectRepo): + _ensure_cart_path() + from cart_repo import OracleCartRepo + + return OracleCartRepo(select_repo._pool) + return _MemoryCartStamp(select_repo) + + +# 落库之后按拆定的行加车。返回加购成功行数和失败行数。 +def _cart_finish(select_repo, adders: dict, login_by_type: dict[int, str]): + _ensure_cart_path() + from cart_job import add_known_lines + + # 同一单的拆量行一次加完。 + def finish(lines): + summary = add_known_lines( + _stamp_repo(select_repo), + lines, + adders=adders, + login_errors=login_by_type, + ) + return summary.added, summary.failed + + return finish + + if __name__ == "__main__": main() diff --git a/rpa_selection/repo.py b/rpa_selection/repo.py index 0e36765..de2c0d5 100644 --- a/rpa_selection/repo.py +++ b/rpa_selection/repo.py @@ -2,7 +2,7 @@ # # 从表 1 领取 deal_flag=0 的行改成 3，选品成功后写表 2、表 1 改成 1。 # 进程中途退出时行停在 3，超过租约（deal_time）才能再领。 -# 不处理加购，不推接口 2。 +# 表 2 的更新人由落库行带入。凑齐后的加车在班次里做，不在这个仓库里。 from __future__ import annotations @@ -51,7 +51,8 @@ SET deal_flag = 3, WHERE fid IN ({placeholders}) """ -# 选品无登录用户：创建人/更新人固定 0 / 系统管理员，时间用库时钟。 +# 选品无登录用户：创建人固定 0 / 系统管理员。更新人由调用方传入。 +# 凑齐后马上加车的行传入「加购中」，避免补偿班在加车完成前领走。fid 由触发器生成，RETURNING 取回。 _INSERT_CART_SQL = """ INSERT INTO rpa_cart_settlement_goods_data ( operate_type, oms_id, platform_type, supplier_name, bar_code, @@ -64,10 +65,13 @@ INSERT INTO rpa_cart_settlement_goods_data ( :purchase_price, :purchase_qty, :purchase_total, :status, :fail_reason, :push_status, :org_code, :multi_platform_flag, SYSDATE, 0, '系统管理员', - SYSDATE, 0, '系统管理员' + SYSDATE, 0, :update_user_name ) +RETURNING fid INTO :out_fid """ +STAMP_SELECTION = "系统管理员" + _COMPLETE_SQL = """ UPDATE rpa_sales_goods_mapping_data SET deal_flag = :deal_flag, @@ -193,6 +197,7 @@ class MemorySelectRepo: self.mapping: dict[int, MappingRow] = {row.oms_id: row for row in rows} self.cart: list[dict[str, Any]] = [] self.query_audits: list[QueryAudit] = [] + self._cart_fid = 0 # 领取待选品任务，把 deal_flag 从 0 或过期 3 改成 3。 # @@ -241,23 +246,40 @@ class MemorySelectRepo: return row.deal_time = now or datetime.now() - def save_results(self, items: list[FinishWrite]) -> None: + def save_results(self, items: list[FinishWrite]) -> list[list[int]]: """把选品结果写入内存表 2，并按 fid 改表 1。 落库前再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。 已经不是本班领的，整单跳过。已有加购行时只改表 1，不插第二套。 + 返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。 """ + inserted: list[list[int]] = [] for item in items: row = self._row_by_fid(item.fid) if row.oms_id != item.oms_id: raise KeyError(item.oms_id) if row.deal_flag != 3: + inserted.append([]) continue exists = any(int(cart.get("oms_id") or 0) == int(item.oms_id) for cart in self.cart) + fids: list[int] = [] if item.cart_rows and not exists: - self.cart.extend(dict(cart_row) for cart_row in item.cart_rows) + for cart_row in item.cart_rows: + stored = dict(cart_row) + stored["fid"] = self._alloc_cart_fid() + stored.setdefault("update_user_name", STAMP_SELECTION) + self.cart.append(stored) + fids.append(int(stored["fid"])) row.deal_flag = item.deal_flag row.deal_result = item.deal_result + inserted.append(fids) + return inserted + + # 给新插入的内存表 2 行配一个 fid。已有行的 fid 不重用。 + def _alloc_cart_fid(self) -> int: + used = [int(row["fid"]) for row in self.cart if row.get("fid") is not None] + self._cart_fid = max([self._cart_fid, *used, 0]) + 1 + return self._cart_fid # 记下查询失败原文。不改表 1 / 表 2。 def save_query_audits(self, items: list[QueryAudit]) -> None: @@ -369,29 +391,29 @@ class OracleSelectRepo: except Exception as exc: raise RepoUnavailable("数据库不可用") from exc - def save_results(self, items: list[FinishWrite]) -> None: + def save_results(self, items: list[FinishWrite]) -> list[list[int]]: """把这一调用里的表 2 行和表 1 完成更新落在同一个事务。 每一单先再读：仍是 deal_flag=3，并且表 2 还没有该 oms_id，才插入。 已有加购行时只改表 1。空列表不借连接。任一行失败整笔回滚。 库故障对外「数据库不可用」，不含 ORA 原文。 + 返回值与 items 对齐：每一单新插入的表 2 fid，顺序与 cart_rows 一致。未插入则为空。 """ if not items: - return + return [] + inserted: list[list[int]] = [[] for _ in items] try: with self._pool.acquire() as conn: try: with conn.cursor() as cur: - cart_binds: list[dict] = [] complete_binds: list[dict] = [] - for item in items: + for index, item in enumerate(items): if not _claim_still_open(cur, item.fid): continue if item.cart_rows and not _cart_already_written(cur, item.oms_id): - cart_binds.extend( - {key: cart_row.get(key) for key in _CART_BIND_KEYS} - for cart_row in item.cart_rows - ) + inserted[index] = [ + _insert_cart_row(cur, cart_row) for cart_row in item.cart_rows + ] complete_binds.append( { "deal_flag": item.deal_flag, @@ -400,11 +422,10 @@ class OracleSelectRepo: "fid": item.fid, } ) - if cart_binds: - cur.executemany(_INSERT_CART_SQL, cart_binds) if complete_binds: cur.executemany(_COMPLETE_SQL, complete_binds) conn.commit() + return inserted except Exception as exc: _rollback(conn) if isinstance(exc, RepoUnavailable): @@ -451,6 +472,26 @@ def _cart_already_written(cur, oms_id: int) -> bool: return int(rows[0][0]) > 0 +# 插入一行表 2 并取回触发器生成的 fid。缺更新人时写成系统管理员。 +def _insert_cart_row(cur, cart_row: dict[str, Any]) -> int: + params = {key: cart_row.get(key) for key in _CART_BIND_KEYS} + params["update_user_name"] = cart_row.get("update_user_name") or STAMP_SELECTION + out = cur.var(int) + params["out_fid"] = out + cur.execute(_INSERT_CART_SQL, params) + return _fid_from_var(out) + + +# RETURNING 的值可能是单元素列表，也可能是标量。空值当落库失败。 +def _fid_from_var(var) -> int: + value = var.getvalue() + if isinstance(value, list): + value = value[0] if value else None + if value is None: + raise RepoUnavailable("数据库不可用") + return int(value) + + # 返回截断原文后的审计行。 def _clipped_audit(item: QueryAudit) -> QueryAudit: return QueryAudit( diff --git a/rpa_selection/search_yaoex.py b/rpa_selection/search_yaoex.py index 82ac76a..4ad6715 100644 --- a/rpa_selection/search_yaoex.py +++ b/rpa_selection/search_yaoex.py @@ -183,7 +183,10 @@ def map_yaoex_products(products: list[dict[str, Any]], user_id: Any) -> list[Can vendor_raw=vendor_raw, vendor_norm=normalize_vendor(vendor_raw), stock=parse_float_or_none(product.get("stockCount")), - extra={"stockCountDesc": product.get("stockCountDesc", "")}, + extra={ + "stockCountDesc": product.get("stockCountDesc", ""), + "raw": product, + }, ) ) return candidates diff --git a/rpa_selection/search_ybm.py b/rpa_selection/search_ybm.py index 5452fe3..b4c893d 100644 --- a/rpa_selection/search_ybm.py +++ b/rpa_selection/search_ybm.py @@ -85,6 +85,9 @@ def map_ybm_products(rows: list[dict[str, Any]]) -> list[Candidate]: raw_price = 0 if no_stock else pick_ybm_platform_price(product) vendor_raw = normalize_text(product.get("shopName")) expiry_raw = normalize_text(product.get("nearEffect")) + raw = dict(product) if isinstance(product, dict) else {} + if not raw.get("skuId") and row.get("skuId"): + raw["skuId"] = row.get("skuId") candidates.append( Candidate( mall_name=YBM_MALL_NAME, @@ -97,7 +100,10 @@ def map_ybm_products(rows: list[dict[str, Any]]) -> list[Candidate]: vendor_raw=vendor_raw, vendor_norm=normalize_vendor(vendor_raw), stock=parse_float_or_none(available_qty), - extra={"库存说明": "无货" if no_stock else "有货"}, + extra={ + "库存说明": "无货" if no_stock else "有货", + "raw": raw, + }, ) ) return candidates diff --git a/rpa_selection/tests/test_cli.py b/rpa_selection/tests/test_cli.py index 9bbd556..b6185e2 100644 --- a/rpa_selection/tests/test_cli.py +++ b/rpa_selection/tests/test_cli.py @@ -1,10 +1,19 @@ from __future__ import annotations +import sys +from pathlib import Path from types import SimpleNamespace import pytest -import main +_CART_ROOT = Path(__file__).resolve().parents[2] / "rpa_cart" +if str(_CART_ROOT) not in sys.path: + sys.path.append(str(_CART_ROOT)) + +import cart_ybm # noqa: E402 +import cart_yaoex # noqa: E402 + +import main # noqa: E402 from job import run_batch from mall_config import YAOEX_MALL_NAME, YBM_MALL_NAME from repo import MappingRow, MemorySelectRepo @@ -80,10 +89,20 @@ def test_main_writes_cart_rows_from_mapping(monkeypatch): ) ] + # 挡住真实加车。入口应在落表后按拆定的 sku 调到这里。 + def fake_add(_auth, offer, qty, **_kwargs): + from cart_add import KIND_OK, AddOutcome + + assert offer.sku_id == "S1" + assert qty == 10 + return AddOutcome(kind=KIND_OK) + monkeypatch.setattr(main, "login_yaoex", fake_login_yaoex) monkeypatch.setattr(main, "login_ybm", fake_login_ybm) monkeypatch.setattr(main, "search_yaoex_candidates", fake_search_yaoex) monkeypatch.setattr(main, "search_ybm_candidates", fake_search_ybm) + monkeypatch.setattr(cart_yaoex, "add_yaoex_cart", fake_add) + monkeypatch.setattr(cart_ybm, "add_ybm_cart", fake_add) main.main(["--fresh-login"], repo=repo) @@ -92,6 +111,7 @@ def test_main_writes_cart_rows_from_mapping(monkeypatch): assert repo.cart[0]["platform_type"] == 66 assert repo.cart[0]["status"] == 1 assert repo.cart[0]["bar_code"] == "06901234567890" + assert repo.cart[0]["update_user_name"] == "加购班" def test_default_limit_caps_claim_at_50(monkeypatch): @@ -120,7 +140,14 @@ def test_default_limit_caps_claim_at_50(monkeypatch): seen["limit"] = kwargs.get("limit") claimed = bound_repo.claim_pending(limit=kwargs.get("limit")) seen["count"] = len(claimed) - return SimpleNamespace(claimed=len(claimed), added=0, skipped=0, failed=0) + return SimpleNamespace( + claimed=len(claimed), + added=0, + skipped=0, + failed=0, + cart_added=0, + cart_failed=0, + ) def no_browser(*_args, **_kwargs): """挡住浏览器登录。""" diff --git a/rpa_selection/tests/test_job.py b/rpa_selection/tests/test_job.py index e0d0a74..5fcac5c 100644 --- a/rpa_selection/tests/test_job.py +++ b/rpa_selection/tests/test_job.py @@ -122,7 +122,7 @@ def test_job_saves_each_claimed_row_before_the_next(): def save_results(self, items): """记下这一次落库，再交给内存仓库。""" self.batches.append(list(items)) - super().save_results(items) + return super().save_results(items) repo = _Repo( [ @@ -362,3 +362,116 @@ def test_job_query_failed_writes_short_code_not_exception(): assert audit.mall_name == "1号药城" assert audit.bar_code == "06901234567890" assert audit.error_text == "ORA-03114 secret-traceback" + + +# 凑齐后按这次拆定的 sku 加车，同仓两条都不并成低价那条。 +def test_success_adds_each_split_sku_without_rereading_table2(): + repo = MemorySelectRepo([_row()]) + seen: list[tuple[str, int, str]] = [] + + # 同仓低价 6、高价 10，计划 10，应拆成 6 + 4。 + def search_ybm(_key: str): + return [ + _ok_candidate( + "药帮忙", + vendor=HANGZHOU, + sku_id="CHEAP", + price=90.0, + stock=6.0, + extra={"raw": {"skuId": "CHEAP"}}, + ), + _ok_candidate( + "药帮忙", + vendor=HANGZHOU, + sku_id="DEAR", + price=92.0, + stock=10.0, + extra={"raw": {"skuId": "DEAR"}}, + ), + ] + + # 记下加车入参。行必须已经落成加购中，fid 来自这次插入。 + def finish(lines): + assert [row["update_user_name"] for row in repo.cart] == ["加购中", "加购中"] + for line in lines: + seen.append((line.sku_id, line.qty, line.raw.get("skuId"))) + matched = next(row for row in repo.cart if row["fid"] == line.fid) + matched["update_user_name"] = "加购班" + return len(lines), 0 + + summary = run_batch( + repo, + platforms=["药帮忙"], + searchers={"药帮忙": search_ybm}, + login_errors={}, + cart_finish=finish, + ) + assert summary.added == 2 + assert summary.cart_added == 2 + assert summary.cart_failed == 0 + assert seen == [("CHEAP", 6, "CHEAP"), ("DEAR", 4, "DEAR")] + assert [row["update_user_name"] for row in repo.cart] == ["加购班", "加购班"] + assert repo.mapping[2026100201].deal_flag == 1 + + +# 凑不齐不加车，失败行戳保持系统管理员。 +def test_business_fail_does_not_add_cart(): + repo = MemorySelectRepo([_row()]) + called = {"n": 0} + + def search_ybm(_key: str): + return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)] + + def finish(_lines): + called["n"] += 1 + return 0, 0 + + summary = run_batch( + repo, + platforms=["药帮忙"], + searchers={"药帮忙": search_ybm}, + login_errors={}, + cart_finish=finish, + ) + assert called["n"] == 0 + assert summary.cart_added == 0 + assert summary.added == 0 + assert repo.mapping[2026100201].deal_flag == 2 + assert repo.cart + assert all(row["update_user_name"] == "系统管理员" for row in repo.cart) + + +# 表 2 已有这单时不再加一份。 +def test_existing_cart_rows_are_not_added_again(): + repo = MemorySelectRepo( + [_row(deal_flag=3, deal_time=datetime.now() - timedelta(minutes=31))] + ) + repo.cart.append( + { + "oms_id": 2026100201, + "platform_type": 66, + "purchase_qty": 10, + "status": 1, + "update_user_name": "系统管理员", + } + ) + called = {"n": 0} + + def search_yaoex(_key: str): + return [_ok_candidate()] + + def finish(_lines): + called["n"] += 1 + return 1, 0 + + summary = run_batch( + repo, + platforms=["1号药城"], + searchers={"1号药城": search_yaoex}, + login_errors={}, + cart_finish=finish, + ) + assert called["n"] == 0 + assert summary.cart_added == 0 + assert len(repo.cart) == 1 + assert repo.mapping[2026100201].deal_flag == 1 diff --git a/rpa_selection/tests/test_mapping.py b/rpa_selection/tests/test_mapping.py index a190843..c45b8d6 100644 --- a/rpa_selection/tests/test_mapping.py +++ b/rpa_selection/tests/test_mapping.py @@ -27,6 +27,8 @@ def test_yaoex_field_mapping_and_price_passthrough(): assert item.vendor_norm == "广东壹号药业有限公司" assert item.stock == 40.0 assert item.mall_name == "1号药城" + assert item.extra["raw"]["spuCode"] == "SPU1" + assert item.extra["stockCountDesc"] == "" def test_ybm_field_mapping_zero_stock_price_is_zero(): @@ -54,6 +56,7 @@ def test_ybm_field_mapping_zero_stock_price_is_zero(): assert item.stock == 0.0 assert item.expiry_raw == "14个月" assert item.mall_name == "药帮忙" + assert item.extra["raw"]["skuId"] == "SKU9" def test_ybm_uses_fob_when_price_empty(): diff --git a/rpa_selection/tests/test_oracle_repo.py b/rpa_selection/tests/test_oracle_repo.py index c534cdf..1039fab 100644 --- a/rpa_selection/tests/test_oracle_repo.py +++ b/rpa_selection/tests/test_oracle_repo.py @@ -47,6 +47,15 @@ def test_from_env_without_user_raises(monkeypatch): OracleSelectRepo.from_env() +# RETURNING 输出。仓库读 getvalue()[0]。 +class _OutBox: + def __init__(self): + self.value = None + + def getvalue(self): + return [self.value] + + # 录 SQL 的假游标，fetchall 返回预设行。 class _FakeCursor: # 挂上共享的 executed 列表和预设查询结果。 @@ -66,6 +75,7 @@ class _FakeCursor: self._fail_on_many = fail_on_many self._fail_update_at = fail_update_at self._updates = 0 + self._next_fid = 1000 self.description = description self.rowcount = 1 @@ -77,16 +87,27 @@ class _FakeCursor: def __exit__(self, *args): return False + # 记下 RETURNING 的输出变量。getvalue 给成单元素列表。 + def var(self, *_args, **_kwargs): + return _OutBox() + # 记下 SQL 和 bind。一条 IN 超过 1000 个 fid 时抛错，模拟 ORA-01795。 def execute(self, sql, params=None): - params = params or {} + params = dict(params or {}) + out = params.pop("out_fid", None) _reject_oversized_in(params) - if str(sql).lstrip().lower().startswith("update"): + text = str(sql).lstrip().lower() + if self._fail_on_many and "insert into rpa_cart_settlement_goods_data" in text: + raise RuntimeError("batch failed") + if text.startswith("update"): self._updates = getattr(self, "_updates", 0) + 1 if self._fail_update_at is not None and self._updates == self._fail_update_at: raise RuntimeError("chunk failed") self._last_sql = sql self._executed.append((sql, params)) + if out is not None: + self._next_fid += 1 + out.value = self._next_fid # 记下批量 SQL。fail_on_many 时抛错，用来点名整批回滚。 def executemany(self, sql, seq): @@ -247,8 +268,9 @@ def test_save_result_inserts_cart_then_updates_mapping(): assert conn.committed is True assert conn.rolled_back is False assert pool.acquires == 1 - assert len(conn.many) == 2 - sql = " ".join(conn.many[0][0].split()) + inserts = _cart_inserts(conn) + assert len(inserts) == 1 + sql = " ".join(inserts[0][0].split()) assert "rpa_cart_settlement_goods_data" in sql.lower() assert "create_user_id" in sql assert "create_user_name" in sql @@ -256,14 +278,19 @@ def test_save_result_inserts_cart_then_updates_mapping(): assert "update_user_id" in sql assert "update_user_name" in sql assert "SYSDATE, 0, '系统管理员'" in sql + assert ":update_user_name" in sql + assert "returning fid into :out_fid" in sql.lower() assert sql.count("SYSDATE") == 2 - assert conn.many[0][1][0]["oms_id"] == 2026100201 - assert "create_user_name" not in conn.many[0][1][0] - update_sql = " ".join(conn.many[1][0].split()).lower() + assert inserts[0][1]["oms_id"] == 2026100201 + assert inserts[0][1]["update_user_name"] == "系统管理员" + assert "create_user_name" not in inserts[0][1] + assert "out_fid" not in inserts[0][1] + assert len(conn.many) == 1 + update_sql = " ".join(conn.many[0][0].split()).lower() assert "rpa_sales_goods_mapping_data" in update_sql assert "where fid = :fid" in update_sql - assert conn.many[1][1][0]["fid"] == 42 - assert conn.many[1][1][0]["deal_flag"] == 1 + assert conn.many[0][1][0]["fid"] == 42 + assert conn.many[0][1][0]["deal_flag"] == 1 # 一批两行表 2 只从池里借一次连接。 @@ -279,21 +306,21 @@ def test_save_result_acquires_once_for_two_cart_rows(): fid=42, ) assert pool.acquires == 1 - inserts = [item for item in conn.many if "rpa_cart_settlement_goods_data" in item[0].lower()] + inserts = _cart_inserts(conn) updates = [item for item in conn.many if "rpa_sales_goods_mapping_data" in item[0].lower()] - assert len(inserts) == 1 - assert len(inserts[0][1]) == 2 + assert len(inserts) == 2 + assert [item[1]["purchase_qty"] for item in inserts] == [7, 3] assert len(updates) == 1 assert len(updates[0][1]) == 1 assert conn.committed is True -# 两单的表 2 和表 1 完成更新各一次 executemany，只借一条连接。 +# 两单的表 2 按行取回 fid，表 1 完成更新一次 executemany，只借一条连接。 def test_save_results_batches_two_orders_in_one_acquire(): conn = _FakeConn() pool = _FakePool(conn) repo = OracleSelectRepo(pool) - repo.save_results( + fids = repo.save_results( [ FinishWrite( oms_id=2026100201, @@ -301,8 +328,8 @@ def test_save_results_batches_two_orders_in_one_acquire(): deal_flag=1, deal_result="选品成功：1号药城、药帮忙", cart_rows=[ - _cart_row(purchase_qty=7), - _cart_row(platform_type=48, purchase_qty=3), + _cart_row(purchase_qty=7, update_user_name="加购中"), + _cart_row(platform_type=48, purchase_qty=3, update_user_name="加购中"), ], ), FinishWrite( @@ -314,18 +341,23 @@ def test_save_results_batches_two_orders_in_one_acquire(): ), ] ) + assert fids == [[1001, 1002], [1003]] assert pool.acquires == 1 assert conn.committed is True assert conn.rolled_back is False - assert len(conn.many) == 2 - insert_sql, insert_binds = conn.many[0] - update_sql, update_binds = conn.many[1] - assert "rpa_cart_settlement_goods_data" in insert_sql.lower() - assert [row["oms_id"] for row in insert_binds] == [ + assert len(conn.many) == 1 + inserts = _cart_inserts(conn) + update_sql, update_binds = conn.many[0] + assert [row["oms_id"] for _sql, row in inserts] == [ 2026100201, 2026100201, 2026100202, ] + assert [row["update_user_name"] for _sql, row in inserts] == [ + "加购中", + "加购中", + "系统管理员", + ] assert "where fid = :fid" in " ".join(update_sql.split()).lower() assert [row["fid"] for row in update_binds] == [42, 43] assert [row["deal_flag"] for row in update_binds] == [1, 2] @@ -402,6 +434,15 @@ def test_save_query_audits_swallows_write_failure(): assert conn.committed is False +def _cart_inserts(conn) -> list: + """表 2 插入语句。RETURNING 走 execute，不进 executemany。""" + return [ + item + for item in conn.executed + if "insert into rpa_cart_settlement_goods_data" in item[0].lower() + ] + + def _fid_bind_count(params: dict) -> int: """数出本段 bind 里的 fid 个数。""" return sum(