zlwuandCursor
db31990ebe
汇总近期业务改动：催缴菜单与后端、样本信息快捷登记与进度、运费关联、客户档案扩展、协议优惠/期次与押金划转，并同步 SQL 迁移与管理端页面收敛。 Co-authored-by: Cursor <cursoragent@cursor.com>
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
"""样本信息管理：进度展示码、证/保险已办理判定、办理字段同步。"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
|
|
def collection_progress(sample_status: str) -> Tuple[str, str]:
|
|
st = (sample_status or "").strip()
|
|
if not st:
|
|
return "uncollected", "未采集"
|
|
if st == "yzf":
|
|
return "voided", "已作废"
|
|
if st in ("ycj", "jcz", "yrk"):
|
|
return "collected", "已采集"
|
|
return "uncollected", "未采集"
|
|
|
|
|
|
def reception_progress(reception: Optional[Dict[str, Any]]) -> Tuple[str, str]:
|
|
if not reception:
|
|
return "none", "未接收"
|
|
c = (reception.get("acceptConclusion") or "").strip()
|
|
if c == "js":
|
|
return "js", "已接收"
|
|
if c == "bj":
|
|
return "bj", "拒收"
|
|
return "none", "未接收"
|
|
|
|
|
|
def qc_progress(qc: Optional[Dict[str, Any]]) -> Tuple[str, str]:
|
|
if not qc:
|
|
return "incomplete", "未完成"
|
|
if (qc.get("qcWorkOrderStatus") or "").strip() == "yzf":
|
|
return "voided", "已作废"
|
|
conc = (qc.get("qcConclusion") or "").strip()
|
|
if conc in ("tg", "btg"):
|
|
return "complete", "已完成"
|
|
return "incomplete", "未完成"
|
|
|
|
|
|
def storage_progress(review: Optional[Dict[str, Any]]) -> Tuple[str, str]:
|
|
if not review:
|
|
return "none", "未入库"
|
|
if (review.get("storageStatus") or "").strip() == "yrk":
|
|
return "yrk", "已入库"
|
|
if (review.get("reviewConclusion") or "").strip() == "btg":
|
|
return "rejected", "拒绝入库"
|
|
return "none", "未入库"
|
|
|
|
|
|
def storage_cert_handled(row: Any) -> bool:
|
|
if row is None:
|
|
return False
|
|
if getattr(row, "handle_time", None):
|
|
return True
|
|
if (getattr(row, "make_status", None) or "").strip() == "1":
|
|
return True
|
|
return bool((getattr(row, "make_date", None) or "").strip())
|
|
|
|
|
|
def insurance_handled(row: Any) -> bool:
|
|
if row is None:
|
|
return False
|
|
if getattr(row, "handle_time", None):
|
|
return True
|
|
if (getattr(row, "todo_status", None) or "").strip() == "1":
|
|
return True
|
|
return bool((getattr(row, "handle_date", None) or "").strip())
|
|
|
|
|
|
def sync_handle_status_fields(handle_time: Optional[datetime]) -> Tuple[str, str]:
|
|
if handle_time is None:
|
|
return "0", ""
|
|
return "1", handle_time.strftime("%Y%m%d")
|
|
|
|
|
|
def apply_handle_payload(
|
|
*,
|
|
handle_time: Optional[datetime],
|
|
handler_user_id: Optional[str],
|
|
attachments: str = "",
|
|
remark: str = "",
|
|
) -> Dict[str, Any]:
|
|
"""规范化办理写入字段。有办理时间则办理人员必填。"""
|
|
uid = (handler_user_id or "").strip() or None
|
|
if handle_time is not None and not uid:
|
|
raise ValueError("填写办理时间时必须指定办理人员")
|
|
status, legacy_date = sync_handle_status_fields(handle_time)
|
|
if handle_time is None:
|
|
uid = None
|
|
return {
|
|
"handle_time": handle_time,
|
|
"handler_user_id": uid,
|
|
"attachments": (attachments or "").strip(),
|
|
"remark": (remark or "").strip(),
|
|
"make_or_todo_status": status,
|
|
"legacy_date": legacy_date,
|
|
}
|