AutoRpa/docs/superpowers/plans/2026-10-05-contract-job-interface4-envelope.md

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