70 lines
1.9 KiB
Python
70 lines
1.9 KiB
Python
"""表 3 领取、表 4 写入。Oracle 仓库后续任务补全。"""
|
|
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 = ""
|
|
|
|
|
|
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: Any) -> 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
|
|
|