45 KiB

Raw Permalink Blame History

合同班合单写表 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：

"""接口 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 追加：

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)）：

"""接口 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

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）：

_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 一律整数）：

{
  "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 调用示例：

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

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：

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 追加：

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 传入：

        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：

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

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：

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）：

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：

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

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：

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：

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：

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

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）

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：

"""合同班次：领取表 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

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

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 追加：

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：

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

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

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

git add rpa_contract/README.md README.md rpa_oms/README.md
git commit -m "docs: 合同班合单与接口4组包已接"

完成标准

cd rpa_contract && uv run pytest tests/test_job.py::test_job_merges_same_supplier_and_writes_table4 -q 绿。
cd rpa_oms && uv run pytest tests/test_contract_agreement_outbound.py::test_assembles_unpushed_rows_as_document_envelope -q 绿。
JSON 干跑 main.py --input-json 仍在。
无 HTTP 客户端，无 push_status 更新，无返利代码，无 rpa_selection / rpa_cart 改动。

45 KiB Raw Permalink Blame History Unescape Escape

合同班合单写表 4 + 接口 4 组包 Implementation Plan

File structure

Task 1: 接口 4 组包（点名信封）

Task 2: 接口 4 Oracle 只读 + 示例 + README

Task 3: 尽调闸 + S08 日期回退

Task 4: 可录判定、合单、草稿映射

Task 5: 内存仓库 + persist

Task 6: 班次 job 点名用例

Task 7: Oracle 仓库 + 生产 runner + 班次入口

Task 8: 文档收口

完成标准

45 KiB

Raw Permalink Blame History