130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
"""协议状态展示（含作废类型后缀）。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.models import SysDictData
|
|
|
|
VOID_TYPE_DICT = "xyzflx"
|
|
|
|
# 协议/捐赠单无效态（与样本接收、生产工单列表过滤口径一致）
|
|
INACTIVE_PROTOCOL_STATUSES = ("yzf", "sdfq")
|
|
|
|
_PROTOCOL_STATUS_LABELS = {
|
|
"dsx": "待生效",
|
|
"sxz": "生效中",
|
|
"ygq": "已过期",
|
|
"yzf": "已作废",
|
|
"sdfq": "属地废弃",
|
|
"zky": "转科研",
|
|
}
|
|
_DONATION_STATUS_LABELS = {
|
|
"dsx": "待生效",
|
|
"sxz": "生效中",
|
|
"ygq": "已过期",
|
|
"yzf": "已作废",
|
|
"sdfq": "属地废弃",
|
|
"zckc": "转细胞存储",
|
|
}
|
|
|
|
_VOID_TYPE_FALLBACK_LABELS = {
|
|
"khhc": "客户毁存",
|
|
"jcbhg": "检测不合格",
|
|
}
|
|
|
|
# 非法来源状态视为终态，不允许再转出
|
|
_PROTOCOL_STATUS_TRANSITIONS = {
|
|
"dsx": ("sxz", "yzf", "sdfq"),
|
|
"sxz": ("ygq", "yzf", "zky", "sdfq"),
|
|
}
|
|
_DONATION_STATUS_TRANSITIONS = {
|
|
"dsx": ("sxz", "yzf", "sdfq"),
|
|
"sxz": ("ygq", "yzf", "zckc", "sdfq"),
|
|
}
|
|
|
|
|
|
def resolve_void_type_label(
|
|
db: Optional[Session],
|
|
tenant_no: Optional[str],
|
|
void_type: Optional[str],
|
|
) -> str:
|
|
code = (void_type or "").strip()
|
|
if not code:
|
|
return ""
|
|
if db is not None and tenant_no:
|
|
row = (
|
|
db.query(SysDictData.dict_label)
|
|
.filter(
|
|
SysDictData.dict_type == VOID_TYPE_DICT,
|
|
SysDictData.dict_value == code,
|
|
SysDictData.tenant_no == tenant_no,
|
|
SysDictData.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if row and (row.dict_label or "").strip():
|
|
return (row.dict_label or "").strip()
|
|
return _VOID_TYPE_FALLBACK_LABELS.get(code, code)
|
|
|
|
|
|
def status_transition_denied_message(current: str, target: str, *, kind: str = "protocol") -> str:
|
|
"""状态流转被拒绝时的提示，状态码展示为中文名称。"""
|
|
current_label = format_protocol_status_label(current, kind=kind)
|
|
target_label = format_protocol_status_label(target, kind=kind)
|
|
return f"状态{current_label}不可转为{target_label}"
|
|
|
|
|
|
def assert_protocol_status_transition(current: str, target: str, *, kind: str = "protocol") -> None:
|
|
cur = (current or "").strip()
|
|
nxt = (target or "").strip()
|
|
if cur == nxt:
|
|
return
|
|
table = _DONATION_STATUS_TRANSITIONS if kind == "donation" else _PROTOCOL_STATUS_TRANSITIONS
|
|
if nxt not in table.get(cur, ()):
|
|
raise ValueError(status_transition_denied_message(cur, nxt, kind=kind))
|
|
|
|
|
|
def format_protocol_status_label(
|
|
status: str,
|
|
void_type: Optional[str] = None,
|
|
*,
|
|
kind: str = "protocol",
|
|
db: Optional[Session] = None,
|
|
tenant_no: Optional[str] = None,
|
|
) -> str:
|
|
st = (status or "").strip()
|
|
labels = _DONATION_STATUS_LABELS if kind == "donation" else _PROTOCOL_STATUS_LABELS
|
|
base = labels.get(st, st)
|
|
if st != "yzf":
|
|
return base
|
|
vt_label = resolve_void_type_label(db, tenant_no, void_type)
|
|
if vt_label:
|
|
return f"{base}({vt_label})"
|
|
return base
|
|
|
|
|
|
def build_void_type_label_map(
|
|
db: Session,
|
|
tenant_no: str,
|
|
codes: list[str],
|
|
) -> Dict[str, str]:
|
|
normalized = list({(c or "").strip() for c in codes if (c or "").strip()})
|
|
if not normalized:
|
|
return {}
|
|
rows = (
|
|
db.query(SysDictData.dict_value, SysDictData.dict_label)
|
|
.filter(
|
|
SysDictData.dict_type == VOID_TYPE_DICT,
|
|
SysDictData.dict_value.in_(normalized),
|
|
SysDictData.tenant_no == tenant_no,
|
|
SysDictData.del_flag == 0,
|
|
)
|
|
.all()
|
|
)
|
|
out = {r.dict_value: (r.dict_label or "").strip() for r in rows if (r.dict_value or "").strip()}
|
|
for code in normalized:
|
|
out.setdefault(code, _VOID_TYPE_FALLBACK_LABELS.get(code, code))
|
|
return out
|