# 合同班合单写表 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` 改动。