From 2e93b2ceac853b4edad1e663b40c0a512c2d4109 Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Mon, 5 Oct 2026 21:58:14 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E5=90=88=E5=90=8C=E7=8F=AD=E5=90=88?= =?UTF-8?q?=E5=8D=95=E5=86=99=E8=A1=A84=E4=B8=8E=E6=8E=A5=E5=8F=A34?= =?UTF-8?q?=E7=BB=84=E5=8C=85=E5=AE=9E=E6=96=BD=E8=AE=A1=E5=88=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按已锁定 spec 拆成接口4组包、尽调/日期、合单、落表、班次入口八个 TDD 任务。 --- ...-10-05-contract-job-interface4-envelope.md | 1402 +++++++++++++++++ 1 file changed, 1402 insertions(+) create mode 100644 docs/superpowers/plans/2026-10-05-contract-job-interface4-envelope.md diff --git a/docs/superpowers/plans/2026-10-05-contract-job-interface4-envelope.md b/docs/superpowers/plans/2026-10-05-contract-job-interface4-envelope.md new file mode 100644 index 0000000..1935489 --- /dev/null +++ b/docs/superpowers/plans/2026-10-05-contract-job-interface4-envelope.md @@ -0,0 +1,1402 @@ +# 合同班合单写表 4 + 接口 4 组包 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. Use @superpowers:test-driven-development for every code task. + +**Goal:** 合同班从表 3 冻结可录行、按 `supplierId` 合单、注入 runner 录合同并写表 4；`rpa_oms` 从表 4 组接口 4 报文（不发 HTTP）。 + +**Architecture:** 作业扁平放在 `rpa_contract/`（`group.py` / `job.py` / `repo.py` / `persist.py`），单测走内存仓库、注入 runner，不打开浏览器、不连库。接口 4 组包对标 `app/outbound/cart_settlement.py`，合同班不 import `rpa_oms`。 + +**Tech Stack:** Python 3.12、pytest、oracledb（仅 Oracle 仓库）、现有 Selenium 页面流（本轮只加尽调扫描和 S08 日期回退）。 + +**Spec:** `docs/superpowers/specs/2026-10-05-contract-job-interface4-envelope-design.md` + +--- + +## File structure + +**Create** + +| 文件 | 职责 | +| --- | --- | +| `rpa_oms/app/outbound/contract_agreement.py` | `load_` / `build_contract_agreement_envelope` | +| `rpa_oms/tests/test_contract_agreement_outbound.py` | 接口 4 组包点名用例 | +| `rpa_oms/examples/contract_agreement_batch.json` | 报文示例（`omsId` 整数） | +| `rpa_contract/group.py` | 可录判定、按供应商分组、草稿映射 | +| `rpa_contract/persist.py` | `ContractOutcome` → 表 4 dict + 表 3 状态 | +| `rpa_contract/job.py` | `claim` → 分组 → runner → `save_group` | +| `rpa_contract/repo.py` | `PurchaseContractRow`、`MemoryContractRepo`、`OracleContractRepo` | +| `rpa_contract/db.py` | 拷 `rpa_selection/db.py` 连接；戳在 repo 里写「合同班」 | +| `rpa_contract/run_batch.py` | 班次入口 | +| `rpa_contract/tests/test_group.py` | 合单 / 可录 / 草稿 | +| `rpa_contract/tests/test_job.py` | 点名合单写表 4、未尽调 | +| `rpa_contract/tests/test_flow_gates.py` | 尽调扫描、S08 日期回退（假 browser，不开页） | + +**Modify** + +| 文件 | 改什么 | +| --- | --- | +| `rpa_oms/app/outbound/memory_repo.py` | 加 `MemoryContractAgreementRepo` | +| `rpa_oms/app/outbound/oracle_repo.py` | 加 `OracleContractAgreementRepo` | +| `rpa_oms/app/outbound/__init__.py` | 注释改为接口 2、4 都只组包 | +| `rpa_oms/README.md` | 接口 4 组包已做，path 空 | +| `rpa_contract/errors.py` | `DiligenceError` | +| `rpa_contract/contract_flow_steps.py` | S06 后尽调；S08 日期回退；`ContractFlowContext` 两个可选日期 | +| `rpa_contract/contract_draft.py` | `contract_sign_date` / `contract_start_date` | +| `rpa_contract/contract_create_node.py` | `build_flow_context` 传入两个日期 | +| `rpa_contract/pyproject.toml` | `oracledb>=2.4.1` | +| `rpa_contract/README.md` | 合单/尽调/写表 4 已接 | +| `README.md` | 合同班不再写「尚未接」 | + +不要改：`rpa_selection/`、`rpa_cart/`、`rpa_rebate/`、接口 1/2/3、表 DDL、`main.py --input-json`。 + +--- + +### Task 1: 接口 4 组包（点名信封） + +**Files:** +- Create: `rpa_oms/app/outbound/contract_agreement.py` +- Modify: `rpa_oms/app/outbound/memory_repo.py` +- Test: `rpa_oms/tests/test_contract_agreement_outbound.py` + +@superpowers:test-driven-development + +- [ ] **Step 1: Write the failing tests** + +在 `memory_repo.py` 末尾先加空类会让 import 过；测试文件先写，类和函数尚不存在。 + +`rpa_oms/tests/test_contract_agreement_outbound.py`： + +```python +"""接口 4：从表 4 取 push_status=0，组装文档报文。不连 Oracle，不发 HTTP。""" +from __future__ import annotations + +from app.outbound.contract_agreement import ( + MAX_BATCH, + build_contract_agreement_envelope, + load_contract_agreement_envelope, +) +from app.outbound.memory_repo import MemoryContractAgreementRepo + + +def _row(**overrides): + row = { + "fid": 300001, + "oms_id": 20260928000001, + "action_code": 1, + "contract_status": 1, + "contract_fail_reason": None, + "contract_no": "HT1", + "agreement_status": None, + "agreement_fail_reason": None, + "agreement_id": None, + "agreement_dtl_id": None, + "org_code": "901", + "push_status": 0, + } + row.update(overrides) + return row + + +def test_assembles_unpushed_rows_as_document_envelope(): + repo = MemoryContractAgreementRepo( + [ + _row(), + _row( + fid=300002, + action_code=2, + contract_status=None, + contract_no="HT1", + agreement_status=1, + agreement_id="AGR1", + agreement_dtl_id="AGRD1", + ), + _row(fid=300003, push_status=1, contract_no="HT-SKIP"), + ] + ) + envelope = load_contract_agreement_envelope(repo) + assert envelope == { + "data": [ + { + "fid": 300001, + "omsId": 20260928000001, + "actionCode": 1, + "contractStatus": 1, + "contractFailReason": None, + "contractNo": "HT1", + "agreementStatus": None, + "agreementFailReason": None, + "agreementId": None, + "agreementDtlId": None, + "orgCode": "901", + }, + { + "fid": 300002, + "omsId": 20260928000001, + "actionCode": 2, + "contractStatus": None, + "contractFailReason": None, + "contractNo": "HT1", + "agreementStatus": 1, + "agreementFailReason": None, + "agreementId": "AGR1", + "agreementDtlId": "AGRD1", + "orgCode": "901", + }, + ] + } + assert isinstance(envelope["data"][0]["omsId"], int) + + +def test_oversized_contract_no_group_enters_when_batch_empty(): + rows = [ + _row(fid=1, oms_id=1, contract_no="HT-BIG"), + _row(fid=2, oms_id=2, contract_no="HT-BIG"), + _row(fid=3, oms_id=3, contract_no="HT-BIG"), + ] + envelope = build_contract_agreement_envelope(rows, limit=2) + assert [item["fid"] for item in envelope["data"]] == [1, 2, 3] + + +def test_does_not_split_contract_no_when_batch_already_has_rows(): + rows = [ + _row(fid=1, oms_id=1, contract_no="HT-A"), + _row(fid=2, oms_id=2, contract_no="HT-B"), + _row(fid=3, oms_id=3, contract_no="HT-B"), + ] + envelope = build_contract_agreement_envelope(rows, limit=2) + assert [item["fid"] for item in envelope["data"]] == [1] + + +def test_empty_contract_no_rows_are_independent(): + rows = [ + _row(fid=1, oms_id=1, contract_no=None, contract_status=3), + _row(fid=2, oms_id=2, contract_no="", contract_status=3), + _row(fid=3, oms_id=3, contract_no="HT-C"), + ] + envelope = build_contract_agreement_envelope(rows, limit=1) + assert [item["fid"] for item in envelope["data"]] == [1] + + +def test_batch_cap_is_500(): + assert MAX_BATCH == 500 + + +def test_payload_omits_push_status(): + item = build_contract_agreement_envelope([_row()])["data"][0] + assert "pushStatus" not in item + assert "push_status" not in item +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py::test_assembles_unpushed_rows_as_document_envelope -q` + +Expected: FAIL import error（`contract_agreement` 或 `MemoryContractAgreementRepo` 不存在） + +- [ ] **Step 3: Minimal implementation** + +`rpa_oms/app/outbound/memory_repo.py` 追加： + +```python +class MemoryContractAgreementRepo: + def __init__(self, rows: list[dict] | None = None): + self._rows = [dict(row) for row in (rows or [])] + + def list_unpushed(self) -> list[dict]: + return [dict(row) for row in self._rows if row.get("push_status", 0) == 0] +``` + +`rpa_oms/app/outbound/contract_agreement.py`（对标 `cart_settlement.py` 的 `_take_complete_groups`，分组键改为非空 `contract_no`；空号用 `("fid", fid)`）： + +```python +"""接口 4：从表 4 取未推行，组装文档报文。 + +同一非空 contractNo 必须在同一批 data 里。一批最多 500 条。 +空 contractNo（失败行）按行独立。不改 push_status，不发 HTTP。 +""" +from __future__ import annotations + +from collections import OrderedDict + +MAX_BATCH = 500 + + +def load_contract_agreement_envelope(repo, limit: int = MAX_BATCH) -> dict: + return build_contract_agreement_envelope(repo.list_unpushed(), limit=limit) + + +def build_contract_agreement_envelope(rows: list[dict], limit: int = MAX_BATCH) -> dict: + chosen = _take_complete_groups(rows, limit) + return {"data": [_to_item(row) for row in chosen]} + + +def _group_key(row: dict): + number = row.get("contract_no") + if number is None or str(number).strip() == "": + return ("fid", row["fid"]) + return ("no", str(number).strip()) + + +def _take_complete_groups(rows: list[dict], limit: int) -> list[dict]: + groups: OrderedDict = OrderedDict() + for row in rows: + groups.setdefault(_group_key(row), []).append(row) + batch: list[dict] = [] + for group in groups.values(): + if len(batch) + len(group) > limit and batch: + break + batch.extend(group) + return batch + + +def _to_item(row: dict) -> dict: + return { + "fid": _as_int(row["fid"]), + "omsId": _as_int(row["oms_id"]), + "actionCode": _as_int(row["action_code"]), + "contractStatus": _as_int(row.get("contract_status")), + "contractFailReason": _as_str(row.get("contract_fail_reason")), + "contractNo": _as_str(row.get("contract_no")), + "agreementStatus": _as_int(row.get("agreement_status")), + "agreementFailReason": _as_str(row.get("agreement_fail_reason")), + "agreementId": _as_str(row.get("agreement_id")), + "agreementDtlId": _as_str(row.get("agreement_dtl_id")), + "orgCode": _as_str(row.get("org_code")), + } + + +def _as_int(value): + if value is None: + return None + return int(value) + + +def _as_str(value): + if value is None: + return None + text = str(value) + if text.strip() == "": + return None + return text +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py -q` + +Expected: PASS（6 passed） + +- [ ] **Step 5: Commit** + +```bash +git add rpa_oms/app/outbound/contract_agreement.py rpa_oms/app/outbound/memory_repo.py rpa_oms/tests/test_contract_agreement_outbound.py +git commit -m "feat: 接口4从表4组装合同协议报文" +``` + +--- + +### Task 2: 接口 4 Oracle 只读 + 示例 + README + +**Files:** +- Modify: `rpa_oms/app/outbound/oracle_repo.py` +- Modify: `rpa_oms/app/outbound/__init__.py` +- Create: `rpa_oms/examples/contract_agreement_batch.json` +- Modify: `rpa_oms/README.md` + +本任务无真库单测。SQL 与 `OracleCartSettlementRepo` 同构，只换表和列。 + +- [ ] **Step 1: Add `OracleContractAgreementRepo`** + +在 `oracle_repo.py` 追加（文件头注释改为接口 2/4）： + +```python +_LIST_AGREEMENT_SQL = """ +SELECT + fid, oms_id, action_code, contract_status, contract_fail_reason, + contract_no, agreement_status, agreement_fail_reason, + agreement_id, agreement_dtl_id, org_code, push_status +FROM rpa_contract_agreement_result +WHERE push_status = 0 +ORDER BY fid +""" + + +class OracleContractAgreementRepo: + def __init__(self, connect: Callable): + self._connect = connect + + @classmethod + def from_env(cls) -> "OracleContractAgreementRepo": + user = os.getenv("RPA_ORACLE_USER", "") + password = os.getenv("RPA_ORACLE_PASSWORD", "") + host = os.getenv("RPA_ORACLE_HOST", "") + if not user or not password or not host: + raise RepoUnavailable("未配置 RPA_ORACLE_HOST / RPA_ORACLE_USER / RPA_ORACLE_PASSWORD") + port = os.getenv("RPA_ORACLE_PORT", "1521") + service = os.getenv("RPA_ORACLE_SERVICE", "rpa") + dsn = f"{host}:{port}/{service}" + + def connect(): + _ensure_client() + import oracledb + + return oracledb.connect(user=user, password=password, dsn=dsn) + + return cls(connect=connect) + + def list_unpushed(self) -> list[dict]: + conn = self._connect() + try: + with conn.cursor() as cur: + cur.execute(_LIST_AGREEMENT_SQL) + cols = [item[0].lower() for item in cur.description] + return [dict(zip(cols, row)) for row in cur.fetchall()] + except Exception as exc: + raise RepoUnavailable(str(exc)) from exc + finally: + conn.close() +``` + +`__init__.py` 改成：「本期接口 2、4 只取数组装，不发 HTTP。」 + +`examples/contract_agreement_batch.json`（Excel 请求示例，`omsId` 一律整数）： + +```json +{ + "data": [ + { + "fid": 300001, + "omsId": 20260928000001, + "actionCode": 1, + "contractStatus": 2, + "contractFailReason": null, + "contractNo": "HT20260928001", + "agreementStatus": null, + "agreementFailReason": null, + "agreementId": null, + "agreementDtlId": null, + "orgCode": "901" + }, + { + "fid": 300002, + "omsId": 20260928000001, + "actionCode": 2, + "contractStatus": null, + "contractFailReason": null, + "contractNo": "HT20260928001", + "agreementStatus": 1, + "agreementFailReason": null, + "agreementId": "AGR20260928001", + "agreementDtlId": "AGRD20260928001", + "orgCode": "901" + } + ] +} +``` + +README 接口 4 那一行改为：`app/outbound/contract_agreement.py` 取 `push_status=0` 组装报文。路径仍空，不发 HTTP。在接口 2 示例代码后补接口 4 调用示例： + +```python +from app.outbound.contract_agreement import load_contract_agreement_envelope +from app.outbound.oracle_repo import OracleContractAgreementRepo + +envelope = load_contract_agreement_envelope(OracleContractAgreementRepo.from_env()) +``` + +- [ ] **Step 2: Re-run envelope tests** + +Run: `cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py -q` + +Expected: PASS + +- [ ] **Step 3: Commit** + +```bash +git add rpa_oms/app/outbound/oracle_repo.py rpa_oms/app/outbound/__init__.py rpa_oms/examples/contract_agreement_batch.json rpa_oms/README.md +git commit -m "feat: 接口4只读表4未推行" +``` + +--- + +### Task 3: 尽调闸 + S08 日期回退 + +**Files:** +- Modify: `rpa_contract/errors.py` +- Modify: `rpa_contract/contract_flow_steps.py`（`ContractFlowContext` 约 71 行后；`run_contract_create_flow` 106–108 行之间；`fill_contract_amount_and_dates` 534–557 行） +- Modify: `rpa_contract/contract_draft.py`（`ContractDraft` 加两个可选日期字段；`contract_draft_from_dict` 读入） +- Modify: `rpa_contract/contract_create_node.py` `build_flow_context`（约 143 行，传入两个日期） +- Test: `rpa_contract/tests/test_flow_gates.py` + +页面函数仍要 Selenium；日期和尽调抽纯函数 / 假 browser，单测不开页。 + +- [ ] **Step 1: Write the failing tests** + +`rpa_contract/tests/test_flow_gates.py`： + +```python +from contract_flow_steps import ( + ContractFlowContext, + check_customer_diligence, + sign_date_for, + begin_date_for, +) +from errors import DiligenceError +import pytest + + +def _ctx(**overrides) -> ContractFlowContext: + data = dict( + contract_url="http://example", + username="u", + password="p", + supplier_name="示例", + order_date="2026-09-28", + source_type="", + ) + data.update(overrides) + return ContractFlowContext(**data) + + +class _FakeBrowser: + def __init__(self, text: str): + self._text = text + + def read_body_text(self) -> str: + return self._text + + +def test_sign_and_begin_fall_back_to_order_date(): + ctx = _ctx() + assert sign_date_for(ctx) == "2026-09-28" + assert begin_date_for(ctx) == "2026-09-28" + + +def test_sign_and_begin_use_contract_dates_when_set(): + ctx = _ctx(contract_sign_date="2026-10-01", contract_start_date="2026-10-02") + assert sign_date_for(ctx) == "2026-10-01" + assert begin_date_for(ctx) == "2026-10-02" + + +def test_diligence_raises_when_page_says_not_done(): + with pytest.raises(DiligenceError, match="客户未尽调"): + check_customer_diligence(_FakeBrowser("客户未尽调，不能起草"), _ctx()) + + +def test_diligence_passes_when_text_absent(): + check_customer_diligence(_FakeBrowser("客户已尽调"), _ctx()) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_contract && uv run pytest tests/test_flow_gates.py -q` + +Expected: FAIL（无 `DiligenceError` / `sign_date_for` / `check_customer_diligence`，或 `ContractFlowContext` 不收 `contract_sign_date`） + +- [ ] **Step 3: Minimal implementation** + +`errors.py` 追加： + +```python +class DiligenceError(NodeError): + error_code = "BIZ_CUSTOMER_NOT_DILIGENCE" + retryable = False + route_role = "PURCHASER" +``` + +`ContractDraft` 在 `contract_end_date` 旁增加 `contract_sign_date: str = ""`、`contract_start_date: str = ""`；`contract_draft_from_dict` 从 `contractSignDate` / `contractStartDate` 填入。 + +`ContractFlowContext` 同样在 `contract_end_date` 旁加这两个字段，默认 `""`。 + +`build_flow_context` 传入： + +```python + contract_end_date=draft.contract_end_date, + contract_sign_date=draft.contract_sign_date, + contract_start_date=draft.contract_start_date, +``` + +`contract_flow_steps.py`： + +```python +def sign_date_for(ctx: ContractFlowContext) -> str: + return (ctx.contract_sign_date or ctx.order_date or "").strip() + + +def begin_date_for(ctx: ContractFlowContext) -> str: + return (ctx.contract_start_date or ctx.order_date or "").strip() + + +def check_customer_diligence(browser: BrowserAdapter, ctx: ContractFlowContext) -> None: + text = browser.read_body_text() or "" + if "未尽调" in text: + raise DiligenceError("客户未尽调") +``` + +`run_contract_create_flow`：`fill_contract_signatory_info` 之后、`fill_contract_basic_info` 之前调用 `check_customer_diligence(browser, ctx)`。 + +`fill_contract_amount_and_dates`：用 `sign_date_for(ctx)` / `begin_date_for(ctx)` 填签订日、开始日，ACCEPT 期望值同步改成这两个函数的返回值。 + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_contract && uv run pytest tests/test_flow_gates.py tests/test_accept.py tests/test_errors.py tests/test_apa_browser_adapter.py -q` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add rpa_contract/errors.py rpa_contract/contract_flow_steps.py rpa_contract/contract_draft.py rpa_contract/contract_create_node.py rpa_contract/tests/test_flow_gates.py +git commit -m "feat: 合同页尽调闸和签订日开始日回退" +``` + +--- + +### Task 4: 可录判定、合单、草稿映射 + +**Files:** +- Create: `rpa_contract/group.py` +- Create: `rpa_contract/repo.py`（本任务只放 `PurchaseContractRow` dataclass；Memory/Oracle 下一任务） +- Test: `rpa_contract/tests/test_group.py` + +- [ ] **Step 1: Write the failing tests** + +`repo.py` 先只定义行模型（测试 import 它）。若希望测试先红，可把 dataclass 放进 Step 3 一起写。 + +`tests/test_group.py`： + +```python +from datetime import date +from decimal import Decimal + +from group import ( + draft_from_group, + group_by_supplier, + is_claimable, + missing_mdm_reason, + money4, +) +from repo import PurchaseContractRow + + +def _row(**overrides) -> PurchaseContractRow: + data = dict( + fid=1, + oms_id=2026100501, + supplier_id="SUP001", + supplier_name="示例供应商", + contract_sign_date=date(2026, 10, 1), + contract_start_date=date(2026, 10, 2), + contract_end_date=date(2027, 9, 30), + payment_term=90, + goods_id="G0001", + goods_name="阿莫西林胶囊", + reg_no="国药准字H20201234", + spec="0.25g*24粒", + unit="盒", + qty=Decimal("10.0000"), + unit_price=Decimal("100.0000"), + purchaser_id="P001", + purchaser_name="张三", + planner_id="PL001", + planner_name="李四", + contract_entry_status=None, + contract_no=None, + process_status=0, + org_code="901", + goods_mdm_code="QX0100167583", + supplier_mdm_code="DW0100046361", + ) + data.update(overrides) + return PurchaseContractRow(**data) + + +def test_protocol_rerun_is_not_claimable(): + row = _row(contract_entry_status=0, contract_no="HT-OLD") + assert is_claimable(row) is False + + +def test_blank_contract_no_is_still_claimable(): + assert is_claimable(_row(contract_entry_status=0, contract_no=" ")) is True + assert is_claimable(_row(contract_entry_status=0, contract_no="")) is True + assert is_claimable(_row(contract_entry_status=0, contract_no=None)) is True + + +def test_processing_or_done_is_not_claimable(): + assert is_claimable(_row(process_status=3)) is False + assert is_claimable(_row(process_status=1)) is False + + +def test_group_by_supplier_keeps_fid_order_and_first_header(): + rows = [ + _row(fid=2, supplier_id="B", payment_term=30, supplier_mdm_code="MDM-B"), + _row(fid=1, supplier_id="A", payment_term=90, supplier_mdm_code="MDM-A"), + _row(fid=3, supplier_id="A", payment_term=60, oms_id=2026100502, goods_id="G0002"), + ] + groups = group_by_supplier(rows) + assert [g[0].supplier_id for g in groups] == ["A", "B"] + assert [r.fid for r in groups[0]] == [1, 3] + draft = draft_from_group(groups[0]) + assert draft.customer_code == "MDM-A" + assert draft.payment_days == "90" + assert draft.order_date == "2026-10-01" + assert draft.contract_sign_date == "2026-10-01" + assert draft.contract_start_date == "2026-10-02" + assert draft.contract_end_date == "2027-09-30" + assert draft.amount == "2000.0000" + assert draft.source_type == "" + assert draft.is_prepayment == "否" + assert len(draft.goods_lines) == 2 + assert draft.goods_lines[0].goods_code == "G0001" + assert draft.goods_lines[0].operation_code == "G0001" + assert draft.goods_lines[0].unit_price == "100.0000" + assert draft.belong_year == "2026" + + +def test_missing_mdm_reason(): + rows = [_row(supplier_mdm_code="")] + assert missing_mdm_reason(rows) == "缺少供应商MDM码" + assert missing_mdm_reason([_row()]) is None + + +def test_money4(): + assert money4(Decimal("10.5")) == "10.5000" +``` + +金额：两行各 `10 * 100` → `2000.0000`。 + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_contract && uv run pytest tests/test_group.py -q` + +Expected: FAIL import + +- [ ] **Step 3: Minimal implementation** + +`repo.py` 行模型（可变，领取时改 `process_status`）： + +```python +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from typing import Any + + +@dataclass +class PurchaseContractRow: + fid: int + oms_id: int + supplier_id: str + supplier_name: str + contract_sign_date: date | None + contract_start_date: date | None + contract_end_date: date | None + payment_term: int + goods_id: str + goods_name: str + reg_no: str + spec: str + unit: str + qty: Decimal | float | int + unit_price: Decimal | float | int + purchaser_id: str + purchaser_name: str + planner_id: str + planner_name: str + contract_entry_status: int | None + contract_no: str | None + process_status: int + org_code: str | None + goods_mdm_code: str | None + supplier_mdm_code: str | None + process_result: str = "" +``` + +`group.py`： + +```python +from __future__ import annotations + +from collections import OrderedDict +from datetime import date, datetime +from decimal import Decimal, ROUND_HALF_UP + +from contract_draft import ContractDraft +from contract_flow_steps import ContractGoodsLine +from repo import PurchaseContractRow + + +def is_claimable(row: PurchaseContractRow) -> bool: + if int(row.process_status or 0) != 0: + return False + if row.contract_entry_status is not None and int(row.contract_entry_status) == 0: + if (row.contract_no or "").strip(): + return False + return True + + +def group_by_supplier(rows: list[PurchaseContractRow]) -> list[list[PurchaseContractRow]]: + groups: OrderedDict[str, list[PurchaseContractRow]] = OrderedDict() + for row in sorted(rows, key=lambda item: item.fid): + groups.setdefault(row.supplier_id, []).append(row) + return list(groups.values()) + + +def missing_mdm_reason(rows: list[PurchaseContractRow]) -> str | None: + if not rows: + return "缺少供应商MDM码" + if not (rows[0].supplier_mdm_code or "").strip(): + return "缺少供应商MDM码" + return None + + +def money4(value) -> str: + amount = Decimal(str(value)).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP) + return f"{amount:.4f}" + + +def iso_date(value) -> str: + if value is None: + return "" + if isinstance(value, datetime): + return value.date().isoformat() + if isinstance(value, date): + return value.isoformat() + return str(value)[:10] + + +def draft_from_group(rows: list[PurchaseContractRow]) -> ContractDraft: + first = rows[0] + sign = iso_date(first.contract_sign_date) + total = sum((Decimal(str(row.qty)) * Decimal(str(row.unit_price)) for row in rows), Decimal("0")) + goods = [ + ContractGoodsLine( + operation_code=row.goods_id, + goods_code=row.goods_id, + goods_name=row.goods_name, + approval_no=row.reg_no, + spec=row.spec, + unit=row.unit, + quantity=float(row.qty), + mdm_code=row.goods_mdm_code or "", + unit_price=money4(row.unit_price), + ) + for row in rows + ] + return ContractDraft( + supplier_name=first.supplier_name, + order_date=sign, + source_type="", + customer_code=(first.supplier_mdm_code or "").strip(), + customer_name=first.supplier_name, + contract_category="采购协议", + amount=money4(total), + contract_sign_date=sign, + contract_start_date=iso_date(first.contract_start_date), + contract_end_date=iso_date(first.contract_end_date), + is_prepayment="否", + belong_year=sign[:4], + payment_method="账期支付", + payment_days=str(first.payment_term), + goods_lines=goods, + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_contract && uv run pytest tests/test_group.py -q` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add rpa_contract/group.py rpa_contract/repo.py rpa_contract/tests/test_group.py rpa_contract/contract_draft.py +git commit -m "feat: 合同班按供应商合单并映射草稿" +``` + +（若 Task 3 已提交 `contract_draft.py` 的新字段，本 commit 不必再加该文件。） + +--- + +### Task 5: 内存仓库 + persist + +**Files:** +- Modify: `rpa_contract/repo.py`（`MemoryContractRepo`） +- Create: `rpa_contract/persist.py` +- Test: `rpa_contract/tests/test_job.py` 先写领取 / persist 小用例；点名合单在 Task 6 + +- [ ] **Step 1: Write the failing tests** + +在 `tests/test_job.py`： + +```python +from datetime import date +from decimal import Decimal + +from persist import KIND_DILIGENCE, KIND_FAIL, KIND_OK, ContractOutcome, build_save +from repo import MemoryContractRepo, PurchaseContractRow + + +def _row(**overrides) -> PurchaseContractRow: + data = dict( + fid=1, + oms_id=2026100501, + supplier_id="SUP001", + supplier_name="示例供应商", + contract_sign_date=date(2026, 10, 1), + contract_start_date=date(2026, 10, 2), + contract_end_date=date(2027, 9, 30), + payment_term=90, + goods_id="G0001", + goods_name="阿莫西林胶囊", + reg_no="国药准字H20201234", + spec="0.25g*24粒", + unit="盒", + qty=Decimal("10.0000"), + unit_price=Decimal("100.0000"), + purchaser_id="P001", + purchaser_name="张三", + planner_id="PL001", + planner_name="李四", + contract_entry_status=None, + contract_no=None, + process_status=0, + org_code="901", + goods_mdm_code="QX0100167583", + supplier_mdm_code="DW0100046361", + ) + data.update(overrides) + return PurchaseContractRow(**data) + + +def test_claim_pending_sets_processing_and_skips_protocol_rerun(): + repo = MemoryContractRepo( + [ + _row(), + _row(fid=2, oms_id=2026100502, process_status=1), + _row(fid=3, oms_id=2026100503, contract_entry_status=0, contract_no="HT-OLD"), + ] + ) + claimed = repo.claim_pending() + assert [row.fid for row in claimed] == [1] + assert repo.table3[1].process_status == 3 + assert repo.table3[3].process_status == 0 + + +def test_save_group_inserts_one_table4_row_per_oms_id(): + repo = MemoryContractRepo( + [ + _row(), + _row(fid=2, oms_id=2026100501, goods_id="G0002"), + _row(fid=3, oms_id=2026100502, goods_id="G0003"), + ] + ) + repo.claim_pending() + outcome = ContractOutcome(KIND_OK, "HT20261005001", None) + payload = build_save(list(repo.table3.values())[:3], outcome) + repo.save_group(payload) + assert [row["oms_id"] for row in repo.table4] == [2026100501, 2026100502] + assert {row["contract_no"] for row in repo.table4} == {"HT20261005001"} + assert {row["action_code"] for row in repo.table4} == {1} + assert {row["contract_status"] for row in repo.table4} == {1} + assert {row["push_status"] for row in repo.table4} == {0} + assert repo.table3[1].process_status == 1 + assert repo.table3[3].process_status == 1 +``` + +`build_save` 返回一个小对象或 dict，含 `fids`、`process_status`、`process_result`、`table4_rows`。同一 `oms_id` 多货品只留一行，顺序为首次出现。 + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py::test_claim_pending_sets_processing_and_skips_protocol_rerun tests/test_job.py::test_save_group_inserts_one_table4_row_per_oms_id -q` + +Expected: FAIL + +- [ ] **Step 3: Minimal implementation** + +`persist.py`： + +```python +from __future__ import annotations + +from dataclasses import dataclass + +from repo import PurchaseContractRow + +KIND_OK = "ok" +KIND_DILIGENCE = "diligence" +KIND_FAIL = "fail" +STAMP = "合同班" +PROCESS_RESULT_MAX = 1000 +FAIL_REASON_MAX = 500 + + +@dataclass(frozen=True) +class ContractOutcome: + kind: str + contract_no: str | None + reason: str | None + + +@dataclass(frozen=True) +class GroupSave: + fids: list[int] + process_status: int + process_result: str + table4_rows: list[dict] + + +def clip(text: str, max_len: int) -> str: + text = (text or "").strip() + if len(text) <= max_len: + return text + return text[: max_len - 1] + "…" + + +def build_save(rows: list[PurchaseContractRow], outcome: ContractOutcome) -> GroupSave: + kind = outcome.kind + number = (outcome.contract_no or "").strip() or None + if kind == KIND_OK and number: + process_status = 1 + process_result = "合同已提交" + contract_status = 1 + fail_reason = None + contract_no = number + elif kind == KIND_DILIGENCE: + process_status = 2 + process_result = clip("客户未尽调", PROCESS_RESULT_MAX) + contract_status = 3 + fail_reason = "客户未尽调" + contract_no = None + else: + reason = outcome.reason or "合同录入失败" + if kind == KIND_OK and not number: + reason = "submit 模式未读取到真实合同编号" + process_status = 2 + process_result = clip(reason, PROCESS_RESULT_MAX) + contract_status = 3 + fail_reason = clip(reason, FAIL_REASON_MAX) + contract_no = None + + seen: dict[int, PurchaseContractRow] = {} + for row in rows: + seen.setdefault(row.oms_id, row) + table4 = [ + { + "oms_id": row.oms_id, + "action_code": 1, + "contract_status": contract_status, + "contract_fail_reason": fail_reason, + "contract_no": contract_no, + "agreement_status": None, + "agreement_fail_reason": None, + "agreement_id": None, + "agreement_dtl_id": None, + "push_status": 0, + "org_code": row.org_code, + "create_user_id": 0, + "create_user_name": STAMP, + "update_user_id": 0, + "update_user_name": STAMP, + } + for row in seen.values() + ] + return GroupSave( + fids=[row.fid for row in rows], + process_status=process_status, + process_result=process_result, + table4_rows=table4, + ) +``` + +`MemoryContractRepo`： + +```python +class MemoryContractRepo: + def __init__(self, rows: list[PurchaseContractRow]): + self.table3 = {row.fid: row for row in rows} + self.table4: list[dict] = [] + self._next_fid = 1 + + def claim_pending(self, limit: int | None = None) -> list[PurchaseContractRow]: + from group import is_claimable + + pending = sorted( + (row for row in self.table3.values() if is_claimable(row)), + key=lambda row: row.fid, + ) + if limit is not None: + pending = pending[:limit] + for row in pending: + row.process_status = 3 + return pending + + def save_group(self, payload: GroupSave) -> None: + for row in payload.table4_rows: + item = dict(row) + item["fid"] = self._next_fid + self._next_fid += 1 + self.table4.append(item) + for fid in payload.fids: + target = self.table3[fid] + target.process_status = payload.process_status + target.process_result = payload.process_result +``` + +把 `GroupSave` 的 import 放进 `repo.py` 会循环。`save_group` 的类型用字符串注解，`from persist import GroupSave` 放函数内，或让 `persist` 不 import Memory。`repo.py` 顶层不要 import `persist`；`save_group(self, payload)` 不注类型即可。 + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py::test_claim_pending_sets_processing_and_skips_protocol_rerun tests/test_job.py::test_save_group_inserts_one_table4_row_per_oms_id tests/test_group.py -q` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add rpa_contract/persist.py rpa_contract/repo.py rpa_contract/tests/test_job.py +git commit -m "feat: 合同班内存领取并写表4" +``` + +--- + +### Task 6: 班次 job 点名用例 + +**Files:** +- Create: `rpa_contract/job.py` +- Modify: `rpa_contract/tests/test_job.py` + +- [ ] **Step 1: Write the failing tests**（追加到 `test_job.py`） + +```python +from job import run_batch +from persist import KIND_DILIGENCE, KIND_OK, ContractOutcome + + +def test_job_merges_same_supplier_and_writes_table4(): + repo = MemoryContractRepo( + [ + _row(), + _row(fid=2, oms_id=2026100502, goods_id="G0002"), + _row(fid=3, oms_id=2026100503, contract_entry_status=0, contract_no="HT-OLD"), + ] + ) + calls: list = [] + + def runner(draft): + calls.append(draft) + return ContractOutcome(KIND_OK, "HT20261005001", None) + + run_batch(repo, runner=runner) + assert len(calls) == 1 + assert [line.goods_code for line in calls[0].goods_lines] == ["G0001", "G0002"] + assert calls[0].customer_code == "DW0100046361" + assert [row["oms_id"] for row in repo.table4] == [2026100501, 2026100502] + assert {row["contract_no"] for row in repo.table4} == {"HT20261005001"} + assert {row["action_code"] for row in repo.table4} == {1} + assert {row["contract_status"] for row in repo.table4} == {1} + assert {row["push_status"] for row in repo.table4} == {0} + assert repo.table3[1].process_status == 1 + assert repo.table3[2].process_status == 1 + assert repo.table3[3].process_status == 0 + + +def test_job_marks_diligence_fail(): + repo = MemoryContractRepo([_row()]) + + def runner(draft): + return ContractOutcome(KIND_DILIGENCE, None, "客户未尽调") + + run_batch(repo, runner=runner) + assert repo.table3[1].process_status == 2 + assert "客户未尽调" in repo.table3[1].process_result + assert repo.table4[0]["contract_status"] == 3 + assert repo.table4[0]["contract_no"] is None + assert repo.table4[0]["contract_fail_reason"] == "客户未尽调" + + +def test_job_skips_runner_when_mdm_missing(): + repo = MemoryContractRepo([_row(supplier_mdm_code="")]) + called = {"n": 0} + + def runner(draft): + called["n"] += 1 + return ContractOutcome(KIND_OK, "HT-X", None) + + run_batch(repo, runner=runner) + assert called["n"] == 0 + assert repo.table3[1].process_status == 2 + assert "缺少供应商MDM码" in repo.table3[1].process_result +``` + +注意：`_row()` 默认 `supplier_mdm_code="DW0100046361"`，与 spec 9.1「客商编码为第一行 MDM」一致。 + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py::test_job_merges_same_supplier_and_writes_table4 -q` + +Expected: FAIL（`job` / `run_batch` 不存在） + +- [ ] **Step 3: Minimal implementation** + +`job.py`： + +```python +"""合同班次：领取表 3 → 按供应商合单 → runner 录合同 → 写表 4。""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable + +from contract_draft import ContractDraft +from group import draft_from_group, group_by_supplier, missing_mdm_reason +from persist import KIND_FAIL, KIND_OK, ContractOutcome, build_save + +Runner = Callable[[ContractDraft], ContractOutcome] + + +@dataclass +class JobSummary: + claimed: int = 0 + groups: int = 0 + ok: int = 0 + failed: int = 0 + + +def run_batch(repo, *, runner: Runner, limit: int | None = None) -> JobSummary: + claimed = repo.claim_pending(limit=limit) + summary = JobSummary(claimed=len(claimed), groups=0) + for rows in group_by_supplier(claimed): + summary.groups += 1 + outcome = _run_group(rows, runner) + repo.save_group(build_save(rows, outcome)) + if outcome.kind == KIND_OK and (outcome.contract_no or "").strip(): + summary.ok += 1 + else: + summary.failed += 1 + return summary + + +def _run_group(rows, runner: Runner) -> ContractOutcome: + reason = missing_mdm_reason(rows) + if reason: + return ContractOutcome(KIND_FAIL, None, reason) + try: + outcome = runner(draft_from_group(rows)) + except Exception as exc: + return ContractOutcome(KIND_FAIL, None, str(exc)) + if outcome.kind == KIND_OK and not (outcome.contract_no or "").strip(): + return ContractOutcome(KIND_FAIL, None, "submit 模式未读取到真实合同编号") + return outcome +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py tests/test_group.py tests/test_flow_gates.py -q` + +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add rpa_contract/job.py rpa_contract/tests/test_job.py +git commit -m "feat: 合同班合单录入写表4" +``` + +--- + +### Task 7: Oracle 仓库 + 生产 runner + 班次入口 + +**Files:** +- Create: `rpa_contract/db.py`（拷 `rpa_selection/db.py`，删掉 `DEAL_USER_NAME`，保留 `load_runtime_env` / `connect_from_env` / `RepoUnavailable`） +- Modify: `rpa_contract/repo.py`（`OracleContractRepo`） +- Create: `rpa_contract/run_batch.py` +- Modify: `rpa_contract/pyproject.toml`（`oracledb>=2.4.1`） +- Test: `rpa_contract/tests/test_job.py` 追加 `test_outcome_from_run_*`（不连库、不开页） + +生产 runner 捕获 `DiligenceError` → `KIND_DILIGENCE`，其它异常 → `KIND_FAIL`。抽 `outcome_from_run` 方便单测。 + +- [ ] **Step 1: Write the failing tests** + +```python +from errors import DiligenceError, NodeError +from job import outcome_from_run +from persist import KIND_DILIGENCE, KIND_FAIL, KIND_OK, ContractOutcome + + +def test_outcome_from_run_maps_diligence(): + def run(): + raise DiligenceError("客户未尽调") + + out = outcome_from_run(run) + assert out.kind == KIND_DILIGENCE + assert out.reason == "客户未尽调" + assert out.contract_no is None + + +def test_outcome_from_run_maps_empty_number_to_fail(): + out = outcome_from_run(lambda: "") + assert out.kind == KIND_FAIL + assert "未读取到真实合同编号" in (out.reason or "") + + +def test_outcome_from_run_ok(): + out = outcome_from_run(lambda: "HT9") + assert out == ContractOutcome(KIND_OK, "HT9", None) + + +def test_outcome_from_run_maps_node_error(): + def run(): + raise NodeError("客商加载失败") + + out = outcome_from_run(run) + assert out.kind == KIND_FAIL + assert "客商加载失败" in (out.reason or "") +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py::test_outcome_from_run_ok -q` + +Expected: FAIL（无 `outcome_from_run`） + +- [ ] **Step 3: Implement runner helper, Oracle repo, CLI** + +`job.py` 追加： + +```python +from errors import DiligenceError, NodeError + + +def outcome_from_run(run: Callable[[], str]) -> ContractOutcome: + try: + number = (run() or "").strip() + except DiligenceError: + return ContractOutcome(KIND_DILIGENCE, None, "客户未尽调") + except NodeError as exc: + return ContractOutcome(KIND_FAIL, None, str(exc)) + except Exception as exc: + return ContractOutcome(KIND_FAIL, None, str(exc)) + if not number: + return ContractOutcome(KIND_FAIL, None, "submit 模式未读取到真实合同编号") + return ContractOutcome(KIND_OK, number, None) +``` + +`production_runner` 放 `run_batch.py`，避免 `job.py` 在单测里拉起 Selenium： + +```python +from __future__ import annotations + +from apa_browser_adapter import ApaBrowserAdapter +from contract_create_node import CONTRACT_URL, build_flow_context +from contract_create_node import ContractCreateInput +from contract_draft import ContractDraft +from contract_flow_steps import run_contract_create_flow +from db import load_runtime_env +from job import outcome_from_run, run_batch +from persist import ContractOutcome +from repo import OracleContractRepo + + +def production_runner(draft: ContractDraft) -> ContractOutcome: + spec = ContractCreateInput( + job_id="contract-batch", + supplier_name=draft.supplier_name, + order_date=draft.order_date, + source_type=draft.source_type or "", + submit_mode="submit", + ) + ctx = build_flow_context(spec, draft) + browser = ApaBrowserAdapter(url=ctx.contract_url, flow="CONTRACT_CREATE") + + def run() -> str: + return run_contract_create_flow(browser, ctx) + + return outcome_from_run(run) + + +def main() -> int: + load_runtime_env() + repo = OracleContractRepo.from_env() + summary = run_batch(repo, runner=production_runner) + print(f"claimed={summary.claimed} groups={summary.groups} ok={summary.ok} failed={summary.failed}") + return 0 if summary.failed == 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) +``` + +`OracleContractRepo`：`from_env` 用 `db.connect_from_env`。领取 SQL 按 spec 4.2；`rowcount==1` 才进 claimed。`save_group` 同一连接先 INSERT 表 4（不绑 `fid`），再 UPDATE 表 3 `process_status=3` 的那些 `fid`。审计字段 `0` / `合同班`。`create_time`/`update_time`/`process_time` 用 `SYSTIMESTAMP`。 + +SELECT 列与 `PurchaseContractRow` 字段一一对应。日期列用 cursor 原样（`date`/`datetime`）。 + +`pyproject.toml` dependencies 加 `"oracledb>=2.4.1"`，然后 `cd rpa_contract && uv lock`。 + +`db.py`：整文件拷 `rpa_selection/db.py`，删除 `DEAL_USER_NAME`。 + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rpa_contract && uv run pytest tests/test_job.py tests/test_group.py tests/test_flow_gates.py -q` + +Expected: PASS + +不要跑带页面等待的套件当回归。不要连真库。 + +- [ ] **Step 5: Commit** + +```bash +git add rpa_contract/db.py rpa_contract/repo.py rpa_contract/job.py rpa_contract/run_batch.py rpa_contract/pyproject.toml rpa_contract/uv.lock rpa_contract/tests/test_job.py +git commit -m "feat: 合同班 Oracle 领取与班次入口" +``` + +--- + +### Task 8: 文档收口 + +**Files:** +- Modify: `rpa_contract/README.md` +- Modify: `README.md`（合同班那一行） +- Modify: `rpa_oms/README.md`（若 Task 2 已写接口 4，本任务只补合同班怎么用组包） +- Modify: `Agents.md` 合同班落点表：合单/写表 4 已接；接口 4 组包在 oms，path 空。不要改选品/加购节。 + +- [ ] **Step 1: Update READMEs** + +`rpa_contract/README.md`： + +- 读表 3 `process_status=0`（跳过协议重录行）；写表 4 `actionCode=1` +- 合单 / 尽调 / 写表 4 本轮已接 +- 接口 4 组包在 `rpa_oms`，本班不推、不改 `push_status` +- 班次：`uv run python run_batch.py`；JSON 干跑入口仍是 `uv run python main.py --input-json ...` +- 验收：`uv run pytest tests/test_job.py::test_job_merges_same_supplier_and_writes_table4 -q` + +根 `README.md`：`rpa_contract/` 那一行改为「合同班。表 3 合单录入 → 表 4；接口 4 组包在 oms，path 空」。`rpa_oms/` 那一行补「接口 2、4 组装报文」。 + +不要改 `rpa_rebate/` 文案。不要把密码写进 README。 + +- [ ] **Step 2: Re-run named accept tests** + +```bash +cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py::test_assembles_unpushed_rows_as_document_envelope -q +cd rpa_contract && uv run pytest tests/test_job.py::test_job_merges_same_supplier_and_writes_table4 tests/test_job.py::test_job_marks_diligence_fail -q +``` + +Expected: PASS + +- [ ] **Step 3: Commit** + +```bash +git add rpa_contract/README.md README.md rpa_oms/README.md +git commit -m "docs: 合同班合单与接口4组包已接" +``` + +--- + +## 完成标准 + +1. `cd rpa_contract && uv run pytest tests/test_job.py::test_job_merges_same_supplier_and_writes_table4 -q` 绿。 +2. `cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py::test_assembles_unpushed_rows_as_document_envelope -q` 绿。 +3. JSON 干跑 `main.py --input-json` 仍在。 +4. 无 HTTP 客户端，无 `push_status` 更新，无返利代码，无 `rpa_selection` / `rpa_cart` 改动。