zkxw/backend/app/crud/qc_sms_notification.py

"""样本质检短信通知 CRUD"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_, case, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizQcSmsNotification,
    BizQcTask,
    BizQcWorkOrder,
    SysUser,
)

# biz_qc_work_order / biz_qc_task 为 general_ci；与 unicode_ci 表 JOIN 时需 collate 对齐
_UNICODE_CI = "utf8mb4_unicode_ci"


def _unicode_ci(col):
    return col.collate(_UNICODE_CI)


def _eq_ci(left, right):
    """列与列比较时两侧 collate；与字面量比较时仅对列侧 collate。"""
    if isinstance(left, str):
        return left == (_unicode_ci(right) if not isinstance(right, str) else right)
    if isinstance(right, str):
        return _unicode_ci(left) == right
    return _unicode_ci(left) == _unicode_ci(right)


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
    val = (value or "").strip()
    if len(val) == 10:
        return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
    return val


def _derive_qc_conclusion(tasks: List[BizQcTask]) -> str:
    if not tasks:
        return ""
    if any((t.conclusion or "").strip() == "btg" for t in tasks):
        return "btg"
    if all((t.conclusion or "").strip() == "tg" for t in tasks):
        return "tg"
    return ""


def _qc_status_label(
    wo: Optional[BizQcWorkOrder],
    *,
    qc_conclusion: str = "",
) -> Tuple[str, str]:
    if not wo:
        return "dzj", "待质检"
    st = (wo.work_order_status or "").strip()
    if st == "yzf":
        return "yzf", "已作废"
    if st == "ypz":
        return "zjz", "质检中"
    if st == "ywc":
        if qc_conclusion == "btg":
            return "btg", "不通过"
        if qc_conclusion == "tg":
            return "tg", "通过"
        return "zjz", "质检中"
    return "dzj", "待质检"


def _fetch_completed_tasks(db: Session, work_order_id: str, tenant_no: str) -> List[BizQcTask]:
    if not work_order_id:
        return []
    return (
        db.query(BizQcTask)
        .filter(
            BizQcTask.work_order_id == work_order_id,
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcTask.task_status == "ywc",
        )
        .all()
    )


def _qc_complete_time(tasks: List[BizQcTask], wo: Optional[BizQcWorkOrder]) -> Optional[datetime]:
    times = [t.completed_time for t in tasks if t.completed_time]
    if times:
        return max(times)
    return wo.update_time if wo else None


def _row_to_dict(
    *,
    notification_id: str = "",
    qc_work_order_id: str = "",
    protocol_sample_id: str = "",
    protocol_id: str = "",
    sms_notified: int = 0,
    sms_content: str = "",
    protocol_no: str = "",
    mother_name: str = "",
    newborn_name: str = "",
    sample_code: str = "",
    lab_storage_code: str = "",
    sample_type_name: str = "",
    qc_complete_time: str = "",
    qc_status: str = "",
    qc_status_label: str = "",
    qc_conclusion: str = "",
    collection_protocol_id: str = "",
    protocol_status: str = "",
) -> Dict[str, Any]:
    return {
        "id": notification_id,
        "qcWorkOrderId": qc_work_order_id,
        "protocolSampleId": protocol_sample_id,
        "protocolId": protocol_id,
        "protocolNo": protocol_no,
        "protocolStatus": protocol_status,
        "motherName": mother_name,
        "newbornName": newborn_name,
        "sampleCode": sample_code,
        "labStorageCode": lab_storage_code,
        "sampleTypeName": sample_type_name,
        "qcCompleteTime": qc_complete_time,
        "qcStatus": qc_status,
        "qcStatusLabel": qc_status_label,
        "qcConclusion": qc_conclusion,
        "qcConclusionLabel": qc_status_label if qc_status in ("tg", "btg") else "",
        "smsNotified": sms_notified,
        "smsContent": sms_content,
        "collectionProtocolId": collection_protocol_id,
    }


def _base_query(db: Session, tenant_no: str):
    return (
        db.query(
            BizProtocolSample,
            BizQcWorkOrder,
            BizQcSmsNotification,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
        )
        .outerjoin(
            BizQcWorkOrder,
            and_(
                _eq_ci(BizQcWorkOrder.sample_code, BizProtocolSample.sample_code),
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.reception_source == "cell_storage",
                BizQcWorkOrder.qc_biz_type == "inbound",
                BizQcWorkOrder.trigger_source == "reception",
            ),
        )
        .outerjoin(
            BizQcSmsNotification,
            and_(
                _eq_ci(BizQcSmsNotification.protocol_sample_id, BizProtocolSample.id),
                BizQcSmsNotification.tenant_no == tenant_no,
                BizQcSmsNotification.del_flag == "0",
            ),
        )
        .join(
            BizCollectionProtocol,
            and_(
                BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .outerjoin(
            BizProtocolPartyCustomer,
            and_(
                BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id,
                BizProtocolPartyCustomer.del_flag == 0,
            ),
        )
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
        )
    )


def _ensure_notification(
    db: Session,
    psample: BizProtocolSample,
    tenant_no: str,
    user_id: Optional[str] = None,
    *,
    wo: Optional[BizQcWorkOrder] = None,
) -> BizQcSmsNotification:
    existing = (
        db.query(BizQcSmsNotification)
        .filter(
            BizQcSmsNotification.protocol_sample_id == psample.id,
            BizQcSmsNotification.tenant_no == tenant_no,
            BizQcSmsNotification.del_flag == "0",
        )
        .first()
    )
    if existing:
        if wo and not existing.qc_work_order_id:
            existing.qc_work_order_id = wo.id
            existing.protocol_id = wo.collection_protocol_id or psample.collection_protocol_id or ""
            existing.update_time = datetime.now()
            db.commit()
            db.refresh(existing)
        return existing

    obj = BizQcSmsNotification(
        id=generate_uuid(),
        protocol_sample_id=psample.id,
        qc_work_order_id=wo.id if wo else None,
        protocol_id=(wo.collection_protocol_id if wo else psample.collection_protocol_id) or "",
        sms_notified=0,
        tenant_no=tenant_no,
        create_by=user_id,
        update_by=user_id,
    )
    db.add(obj)
    db.flush()
    db.commit()
    db.refresh(obj)
    return obj


def resolve_notification(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    qc_work_order_id: Optional[str] = None,
    protocol_sample_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[BizQcSmsNotification]:
    notif = (
        db.query(BizQcSmsNotification)
        .filter(
            BizQcSmsNotification.id == notification_id,
            BizQcSmsNotification.tenant_no == tenant_no,
            BizQcSmsNotification.del_flag == "0",
        )
        .first()
    )
    if notif:
        return notif

    psample = None
    if protocol_sample_id:
        psample = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.id == protocol_sample_id,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
            .first()
        )
    elif qc_work_order_id:
        wo = (
            db.query(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.id == qc_work_order_id,
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
            )
            .first()
        )
        if wo and wo.sample_code:
            psample = (
                db.query(BizProtocolSample)
                .filter(
                    _eq_ci(BizProtocolSample.sample_code, wo.sample_code),
                    BizProtocolSample.tenant_no == tenant_no,
                    BizProtocolSample.del_flag == 0,
                )
                .first()
            )

    if not psample:
        return None
    wo = None
    if qc_work_order_id:
        wo = (
            db.query(BizQcWorkOrder)
            .filter(
                BizQcWorkOrder.id == qc_work_order_id,
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
            )
            .first()
        )
    return _ensure_notification(db, psample, tenant_no=tenant_no, user_id=user_id, wo=wo)


def _task_stats_subquery(db: Session, tenant_no: str):
    return (
        db.query(
            BizQcTask.work_order_id.label("work_order_id"),
            func.max(case((BizQcTask.conclusion == "btg", 1), else_=0)).label("has_btg"),
            func.count(BizQcTask.id).label("task_count"),
            func.max(BizQcTask.completed_time).label("max_completed_time"),
        )
        .filter(
            BizQcTask.tenant_no == tenant_no,
            BizQcTask.del_flag == 0,
            BizQcTask.task_status == "ywc",
        )
        .group_by(BizQcTask.work_order_id)
        .subquery()
    )


def _apply_qc_status_filter(q, qc_status: str, task_stats):
    if qc_status == "dzj":
        return q.filter(
            or_(
                BizQcWorkOrder.id.is_(None),
                BizQcWorkOrder.work_order_status == "dpz",
            )
        )
    if qc_status == "zjz":
        return q.filter(BizQcWorkOrder.work_order_status == "ypz")
    if qc_status == "yzf":
        return q.filter(BizQcWorkOrder.work_order_status == "yzf")
    if qc_status == "btg":
        return q.filter(
            BizQcWorkOrder.work_order_status == "ywc",
            task_stats.c.has_btg == 1,
        )
    if qc_status == "tg":
        return q.filter(
            BizQcWorkOrder.work_order_status == "ywc",
            task_stats.c.has_btg == 0,
            task_stats.c.task_count > 0,
        )
    return q


def list_notifications(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    tenant_no: str,
    keyword: Optional[str] = None,
    qc_status: Optional[str] = None,
    qc_conclusion: Optional[str] = None,
    sms_notified: Optional[int] = None,
    qc_complete_time_start: Optional[str] = None,
    qc_complete_time_end: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0

    status_filter = (qc_status or qc_conclusion or "").strip()
    task_stats = _task_stats_subquery(db, tenant_no)
    q = _base_query(db, tenant_no).outerjoin(
        task_stats,
        _eq_ci(task_stats.c.work_order_id, BizQcWorkOrder.id),
    )

    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        kw_conds = [
            BizProtocolPartyCustomer.mother_name.like(kw),
            BizProtocolSample.sample_code.like(kw),
        ]
        from app.crud import sample_reception as reception_svc

        lab_codes = reception_svc.sample_codes_matching_lab_storage_keyword(
            db, tenant_no, keyword, reception_source="cell_storage"
        )
        if lab_codes:
            kw_conds.append(BizProtocolSample.sample_code.in_(lab_codes))
        q = q.filter(or_(*kw_conds))

    if status_filter:
        q = _apply_qc_status_filter(q, status_filter, task_stats)

    if sms_notified is not None:
        if sms_notified == 0:
            q = q.filter(
                or_(
                    BizQcSmsNotification.sms_notified == 0,
                    BizQcSmsNotification.id.is_(None),
                )
            )
        else:
            q = q.filter(BizQcSmsNotification.sms_notified == sms_notified)

    if qc_complete_time_start:
        q = q.filter(
            BizProtocolSample.create_time >= _normalize_date_bound(qc_complete_time_start)
        )
    if qc_complete_time_end:
        q = q.filter(
            BizProtocolSample.create_time
            <= _normalize_date_bound(qc_complete_time_end, end_of_day=True)
        )

    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizProtocolSample, user_info, db, tenant_no)

    total = q.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)

    rows = (
        q.order_by(
            BizProtocolSample.create_time.desc(),
            BizProtocolSample.id.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    sample_type_ids = list({
        psample.sample_type_id
        for psample, _, _, _, _ in rows
        if psample and psample.sample_type_id
    })
    type_map: Dict[str, str] = {}
    if sample_type_ids:
        for t in (
            db.query(BizCellSampleType)
            .filter(BizCellSampleType.id.in_(sample_type_ids), BizCellSampleType.del_flag == 0)
            .all()
        ):
            type_map[t.id] = t.type_name or ""

    items: List[Dict[str, Any]] = []
    for psample, wo, notif, proto, party in rows:
        if notif is None:
            notif = _ensure_notification(db, psample, tenant_no=tenant_no, wo=wo)

        tasks = _fetch_completed_tasks(db, wo.id if wo else "", tenant_no)
        conclusion = _derive_qc_conclusion(tasks) if wo else ""
        status_code, status_label = _qc_status_label(wo, qc_conclusion=conclusion)
        complete_dt = _qc_complete_time(tasks, wo) if status_code in ("tg", "btg") else None

        sample_type_name = ""
        if psample.sample_type_id:
            sample_type_name = type_map.get(psample.sample_type_id, "")

        items.append(
            _row_to_dict(
                notification_id=notif.id,
                qc_work_order_id=(wo.id if wo else ""),
                protocol_sample_id=psample.id,
                protocol_id=(notif.protocol_id or "") if notif else "",
                sms_notified=(notif.sms_notified or 0) if notif else 0,
                sms_content=(notif.sms_content or "") if notif else "",
                protocol_no=(proto.protocol_no or "") if proto else "",
                mother_name=(party.mother_name or "") if party else "",
                newborn_name=(party.newborn_name or "") if party else "",
                sample_code=psample.sample_code or (wo.sample_code if wo else "") or "",
                sample_type_name=sample_type_name,
                qc_complete_time=_fmt_dt(complete_dt) if complete_dt else "",
                qc_status=status_code,
                qc_status_label=status_label,
                qc_conclusion=conclusion,
                collection_protocol_id=(proto.id or "") if proto else "",
                protocol_status=(proto.protocol_status or "") if proto else "",
            )
        )
    if items:
        from app.crud import sample_reception as reception_svc

        codes = [i["sampleCode"] for i in items if i.get("sampleCode")]
        code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
            db, tenant_no, codes, reception_source="cell_storage"
        )
        for item in items:
            sc = (item.get("sampleCode") or "").strip()
            item["labStorageCode"] = code_map.get(sc, "") if sc else ""
    return items, int(total)


def toggle_notification_status(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    sms_notified: int,
    user_id: Optional[str] = None,
    qc_work_order_id: Optional[str] = None,
    protocol_sample_id: Optional[str] = None,
) -> Optional[BizQcSmsNotification]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        qc_work_order_id=qc_work_order_id,
        protocol_sample_id=protocol_sample_id,
        user_id=user_id,
    )
    if not notif:
        return None

    notif.sms_notified = sms_notified
    if sms_notified == 1:
        notif.notified_by = user_id
        notif.notified_time = datetime.now()
        _user = db.query(SysUser.nick_name).filter(SysUser.id == user_id).first()
        notif.notified_by_name = _user.nick_name if _user else None
    else:
        notif.notified_by = None
        notif.notified_by_name = None
        notif.notified_time = None
    notif.update_by = user_id
    notif.update_time = datetime.now()
    db.commit()
    db.refresh(notif)
    return notif


def generate_sms_content(
    db: Session,
    *,
    notification_id: str,
    tenant_no: str,
    qc_work_order_id: Optional[str] = None,
    protocol_sample_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[str]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        qc_work_order_id=qc_work_order_id,
        protocol_sample_id=protocol_sample_id,
        user_id=user_id,
    )
    if not notif:
        return None

    psample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.id == notif.protocol_sample_id,
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    wo = None
    if notif.qc_work_order_id:
        wo = (
            db.query(BizQcWorkOrder)
            .filter(
                _eq_ci(BizQcWorkOrder.id, notif.qc_work_order_id),
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
            )
            .first()
        )
    if not wo and psample and psample.sample_code:
        wo = (
            db.query(BizQcWorkOrder)
            .filter(
                _eq_ci(BizQcWorkOrder.sample_code, psample.sample_code),
                BizQcWorkOrder.tenant_no == tenant_no,
                BizQcWorkOrder.del_flag == 0,
                BizQcWorkOrder.reception_source == "cell_storage",
                BizQcWorkOrder.qc_biz_type == "inbound",
                BizQcWorkOrder.trigger_source == "reception",
            )
            .first()
        )

    tasks = _fetch_completed_tasks(db, wo.id if wo else "", tenant_no)
    conclusion = _derive_qc_conclusion(tasks)
    if not conclusion:
        status_code, status_label = _qc_status_label(wo, qc_conclusion="")
        if status_code == "dzj":
            return "暂无短信内容，样本尚未开始质检"
        if status_code == "zjz":
            return "暂无短信内容，样本质检尚未完成"
        return "暂无短信内容，样本尚未完成质检确认"

    mother_name = ""
    proto_id = (wo.collection_protocol_id if wo else psample.collection_protocol_id) if psample else ""
    if proto_id:
        party = (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id == proto_id,
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .first()
        )
        if party:
            mother_name = party.mother_name or ""

    sample_code = (psample.sample_code if psample else "") or (wo.sample_code if wo else "") or ""
    if conclusion == "tg":
        content = (
            f"尊敬的{mother_name}女士，您的样本（编码：{sample_code}）"
            f"已完成质检，结果合格。如有疑问请联系客服。"
        )
    else:
        content = (
            f"尊敬的{mother_name}女士，您的样本（编码：{sample_code}）"
            f"质检未通过。如有疑问请联系客服。"
        )

    notif.sms_content = content
    if wo and not notif.qc_work_order_id:
        notif.qc_work_order_id = wo.id
    notif.update_time = datetime.now()
    db.commit()
    db.refresh(notif)
    return content