"""样本质检短信通知 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