zkxw/backend/app/crud/sample_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_, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizCellSampleType,
    BizCollectionProtocol,
    BizProtocolPartyCustomer,
    BizProtocolSample,
    BizSampleReception,
    BizSampleSmsNotification,
)

# biz_sample_sms_notification may use general_ci; align with unicode_ci tables on JOIN
_UNICODE_CI = "utf8mb4_unicode_ci"


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


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 _accept_conclusion_label(v: str) -> str:
    return {"js": "接收", "bj": "拒收"}.get(v, "待接收确认")


def _row_to_dict(
    *,
    notification_id: str = "",
    sample_reception_id: str = "",
    protocol_id: str = "",
    sms_notified: int = 0,
    sms_content: str = "",
    protocol_no: str = "",
    mother_name: str = "",
    newborn_name: str = "",
    sample_code: str = "",
    sample_type_name: str = "",
    reception_time: str = "",
    accept_conclusion: str = "",
    lab_storage_code: str = "",
    collection_protocol_id: str = "",
    protocol_status: str = "",
) -> Dict[str, Any]:
    return {
        "id": notification_id,
        "sampleReceptionId": sample_reception_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,
        "receptionTime": reception_time,
        "acceptConclusion": accept_conclusion,
        "acceptConclusionLabel": _accept_conclusion_label(accept_conclusion),
        "smsNotified": sms_notified,
        "smsContent": sms_content,
        "collectionProtocolId": collection_protocol_id,
    }


def _base_query(db: Session, tenant_no: str):
    return (
        db.query(
            BizProtocolSample,
            BizSampleReception,
            BizSampleSmsNotification,
            BizCollectionProtocol,
            BizProtocolPartyCustomer,
        )
        .outerjoin(
            BizSampleReception,
            and_(
                _unicode_ci(BizSampleReception.sample_code)
                == _unicode_ci(BizProtocolSample.sample_code),
                BizSampleReception.tenant_no == tenant_no,
                BizSampleReception.reception_source == "cell_storage",
                BizSampleReception.del_flag == 0,
            ),
        )
        .outerjoin(
            BizSampleSmsNotification,
            and_(
                BizSampleSmsNotification.sample_reception_id == BizSampleReception.id,
                BizSampleSmsNotification.tenant_no == tenant_no,
                BizSampleSmsNotification.del_flag == "0",
            ),
        )
        .outerjoin(
            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,
    reception: Optional[BizSampleReception],
    tenant_no: str,
    user_id: Optional[str] = None,
    *,
    psample: Optional[BizProtocolSample] = None,
) -> BizSampleSmsNotification:
    lookup_id = reception.id if reception else (psample.id if psample else None)
    if not lookup_id:
        raise ValueError("reception or psample required")

    existing = (
        db.query(BizSampleSmsNotification)
        .filter(
            BizSampleSmsNotification.sample_reception_id == lookup_id,
            BizSampleSmsNotification.tenant_no == tenant_no,
            BizSampleSmsNotification.del_flag == "0",
        )
        .first()
    )
    if existing:
        return existing

    obj = BizSampleSmsNotification(
        id=generate_uuid(),
        sample_reception_id=lookup_id,
        protocol_id=(reception.collection_protocol_id if reception 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,
    sample_reception_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[BizSampleSmsNotification]:
    notif = (
        db.query(BizSampleSmsNotification)
        .filter(
            BizSampleSmsNotification.id == notification_id,
            BizSampleSmsNotification.tenant_no == tenant_no,
            BizSampleSmsNotification.del_flag == "0",
        )
        .first()
    )
    if notif:
        return notif

    if not sample_reception_id:
        return None

    rec = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.id == sample_reception_id,
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .first()
    )
    if rec:
        return _ensure_notification(db, rec, tenant_no=tenant_no, user_id=user_id)

    psample = (
        db.query(BizProtocolSample)
        .filter(
            BizProtocolSample.id == sample_reception_id,
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
        )
        .first()
    )
    if psample:
        return _ensure_notification(db, None, tenant_no=tenant_no, user_id=user_id, psample=psample)

    return None


def _fetch_sample_type_name(db: Session, sample_type_id: Optional[str], tenant_no: str) -> str:
    if not sample_type_id:
        return ""
    t = (
        db.query(BizCellSampleType)
        .filter(
            BizCellSampleType.id == sample_type_id,
            BizCellSampleType.del_flag == 0,
        )
        .first()
    )
    return (t.type_name or "") if t else ""


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

    q = _base_query(db, tenant_no)

    if keyword and keyword.strip():
        kw = f"%{keyword.strip()}%"
        kw_conds = [
            BizProtocolPartyCustomer.mother_name.like(kw),
            BizProtocolSample.sample_code.like(kw),
            BizSampleReception.lab_storage_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 accept_conclusion:
        if accept_conclusion == "dqr":
            q = q.filter(BizSampleReception.id.is_(None))
        else:
            q = q.filter(BizSampleReception.accept_conclusion == accept_conclusion)

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

    if reception_time_start:
        q = q.filter(
            BizSampleReception.reception_time >= _normalize_date_bound(reception_time_start)
        )
    if reception_time_end:
        q = q.filter(
            BizSampleReception.reception_time
            <= _normalize_date_bound(reception_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)

    # MySQL 5.7 不支持 NULLS LAST；用 IS NULL 升序把空接收时间排到最后
    # 末尾加 BizProtocolSample.id 作为确定性排序，避免相同时间的记录顺序不确定
    rows = (
        q.order_by(
            BizSampleReception.reception_time.is_(None),
            BizSampleReception.reception_time.desc(),
            BizProtocolSample.create_time.desc(),
            BizProtocolSample.id.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    sample_type_ids = list({
        ps.sample_type_id
        for ps, _, _, _, _ in rows
        if ps and ps.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, rec, notif, proto, party in rows:
        if notif is None:
            notif = _ensure_notification(db, rec, tenant_no=tenant_no, psample=psample)

        sample_type_name = ""
        if psample and psample.sample_type_id:
            sample_type_name = type_map.get(psample.sample_type_id, "")
        items.append(
            _row_to_dict(
                notification_id=notif.id,
                sample_reception_id=rec.id if rec else psample.id,
                protocol_id=notif.protocol_id or "",
                sms_notified=notif.sms_notified or 0,
                sms_content=notif.sms_content or "",
                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 (rec.sample_code if rec else "") or "",
                sample_type_name=sample_type_name,
                reception_time=_fmt_dt(rec.reception_time) if rec else "",
                accept_conclusion=rec.accept_conclusion if rec else "",
                lab_storage_code=(rec.lab_storage_code or "") if rec else "",
                collection_protocol_id=(proto.id or "") if proto else "",
                protocol_status=(proto.protocol_status or "") if proto 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,
    sample_reception_id: Optional[str] = None,
) -> Optional[BizSampleSmsNotification]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        sample_reception_id=sample_reception_id,
        user_id=user_id,
    )
    if not notif:
        return None

    notif.sms_notified = sms_notified
    notif.notified_by = user_id if sms_notified == 1 else None
    notif.notified_time = datetime.now() if sms_notified == 1 else 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,
    sample_reception_id: Optional[str] = None,
    user_id: Optional[str] = None,
) -> Optional[str]:
    notif = resolve_notification(
        db,
        notification_id=notification_id,
        tenant_no=tenant_no,
        sample_reception_id=sample_reception_id,
        user_id=user_id,
    )
    if not notif:
        return None

    rec = (
        db.query(BizSampleReception)
        .filter(
            BizSampleReception.id == notif.sample_reception_id,
            BizSampleReception.tenant_no == tenant_no,
            BizSampleReception.del_flag == 0,
        )
        .first()
    )
    if not rec:
        notif.sms_content = "暂无短信内容，样本尚未完成接收确认"
        notif.update_time = datetime.now()
        db.commit()
        db.refresh(notif)
        return notif.sms_content

    mother_name = ""
    proto = (
        db.query(BizCollectionProtocol)
        .filter(
            BizCollectionProtocol.id == rec.collection_protocol_id,
            BizCollectionProtocol.del_flag == 0,
        )
        .first()
    )
    if proto:
        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 = rec.sample_code or ""
    reception_date = _fmt_dt(rec.reception_time)[:10] if rec.reception_time else ""
    conclusion_label = _accept_conclusion_label(rec.accept_conclusion or "")

    content = (
        f"尊敬的{mother_name}女士，您的样本（编码：{sample_code}）"
        f"已于{reception_date}完成接收处理，结论：{conclusion_label}。"
        f"如有疑问请联系客服。"
    )

    notif.sms_content = content
    notif.update_time = datetime.now()
    db.commit()
    db.refresh(notif)
    return content