From e8ef1a77fcd218b2f39e2da596c08d27e11b460f Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Mon, 5 Oct 2026 22:55:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=90=88=E5=90=8C=E7=8F=AD=E6=8C=89?= =?UTF-8?q?=E4=BE=9B=E5=BA=94=E5=95=86=E5=90=88=E5=8D=95=E5=B9=B6=E6=98=A0?= =?UTF-8?q?=E5=B0=84=E8=8D=89=E7=A8=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rpa_contract/group.py | 89 ++++++++++++++++++++++++++++++ rpa_contract/repo.py | 36 ++++++++++++ rpa_contract/tests/test_group.py | 95 ++++++++++++++++++++++++++++++++ 3 files changed, 220 insertions(+) create mode 100644 rpa_contract/group.py create mode 100644 rpa_contract/repo.py create mode 100644 rpa_contract/tests/test_group.py diff --git a/rpa_contract/group.py b/rpa_contract/group.py new file mode 100644 index 0000000..6f311ad --- /dev/null +++ b/rpa_contract/group.py @@ -0,0 +1,89 @@ +"""合同班合单：可录判定、按供应商分组、草稿映射。""" +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, + ) diff --git a/rpa_contract/repo.py b/rpa_contract/repo.py new file mode 100644 index 0000000..c6c7531 --- /dev/null +++ b/rpa_contract/repo.py @@ -0,0 +1,36 @@ +"""表 3 合同行与仓库。Memory / Oracle 在后续任务补全。""" +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date +from decimal import Decimal + + +@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 = "" diff --git a/rpa_contract/tests/test_group.py b/rpa_contract/tests/test_group.py new file mode 100644 index 0000000..da802b9 --- /dev/null +++ b/rpa_contract/tests/test_group.py @@ -0,0 +1,95 @@ +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"