zkxw/backend/app/core/protocol_status_display.py

"""协议状态展示（含作废类型后缀）。"""
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