124 lines
4.3 KiB
Python
124 lines
4.3 KiB
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 persist import LOOKUP_ONLY_RESULT
|
||
from repo import PurchaseContractRow
|
||
|
||
|
||
# 判断是否只录协议行：contract_entry_status=0 且已有合同号。
|
||
#
|
||
# 合同班跳过这些行，留给返利班。
|
||
def is_protocol_rerun(row: PurchaseContractRow) -> bool:
|
||
if row.contract_entry_status is not None and int(row.contract_entry_status) == 0:
|
||
return bool((row.contract_no or "").strip())
|
||
return False
|
||
|
||
|
||
# 判断是否可领：process_status=0 且不是协议重录行。
|
||
#
|
||
# 过期的 process_status=3 由仓库租约逻辑另判，本函数只认新鲜待领。
|
||
def is_claimable(row: PurchaseContractRow) -> bool:
|
||
if int(row.process_status or 0) != 0:
|
||
return False
|
||
return not is_protocol_rerun(row)
|
||
|
||
|
||
# 合单键：供应商、货主、签订日、开始日、结束日。任一不同则各自成组。
|
||
def _group_key(row: PurchaseContractRow) -> tuple:
|
||
return (
|
||
row.supplier_id,
|
||
row.org_code,
|
||
row.contract_sign_date,
|
||
row.contract_start_date,
|
||
row.contract_end_date,
|
||
)
|
||
|
||
|
||
# 按合单键分组，组内按 fid 升序，头字段取该组第一行。
|
||
def group_by_supplier(rows: list[PurchaseContractRow]) -> list[list[PurchaseContractRow]]:
|
||
groups: OrderedDict[tuple, list[PurchaseContractRow]] = OrderedDict()
|
||
for row in sorted(rows, key=lambda item: item.fid):
|
||
groups.setdefault(_group_key(row), []).append(row)
|
||
return list(groups.values())
|
||
|
||
|
||
# 检查合单是否缺供应商或货品 MDM 码。缺则返回原因，齐则返回 None。
|
||
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码"
|
||
for row in rows:
|
||
if not (row.goods_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}"
|
||
|
||
|
||
# 把日期收成 YYYY-MM-DD。空值返回空字符串。
|
||
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:
|
||
"""把同一供应商一组表 3 行收成页面草稿。头字段取第一行。
|
||
|
||
结果里已有「只查号」时，下一班只查已有合同，不再起草。
|
||
"""
|
||
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_mdm_code or "").strip(),
|
||
goods_code=(row.goods_mdm_code or "").strip(),
|
||
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 "").strip(),
|
||
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,
|
||
lookup_only=any(LOOKUP_ONLY_RESULT in (row.process_result or "") for row in rows),
|
||
)
|