"""协议新增样本：服务号模板消息通知。""" from __future__ import annotations import logging import threading import uuid from datetime import datetime from typing import Any, Dict, Iterable, List, Optional, Set from sqlalchemy.orm import Session from app.core.config import settings from app.core.database import SessionLocal from app.core.protocol_status_display import INACTIVE_PROTOCOL_STATUSES from app.models.models import ( BizCellSampleType, BizCollectionProtocol, BizProtocolSample, BizProtocolSampleTransport, BizWxNotifyLog, SysUser, ) from app.models.research_donation import ( BizResearchDonation, BizResearchDonationSample, BizResearchDonationSampleTransport, ) from app.services.web_auth_roles import list_user_ids_with_effective_role_name logger = logging.getLogger(__name__) NOTIFY_BIZ_TYPE = "sample_collect_notify" NOTIFY_DATE_ONCE = "1970-01-01" PROTOCOL_KIND_CELL = "cell_storage" PROTOCOL_KIND_DONATION = "research_donation" TASK_TYPE_CELL = "细胞存储" TASK_TYPE_DONATION = "科研捐赠" SAMPLE_STATUS_VOID = "yzf" def sample_codes_from_payload(samples: Optional[Iterable[Any]]) -> Set[str]: """从请求体样本列表提取非空 sample_code（兼容 camelCase）。""" out: Set[str] = set() for raw in samples or []: if isinstance(raw, dict): code = (raw.get("sample_code") or raw.get("sampleCode") or "").strip() else: code = (getattr(raw, "sample_code", None) or "").strip() if code: out.add(code) return out def sample_codes_from_orm_rows(rows: Optional[Iterable[Any]]) -> Set[str]: out: Set[str] = set() for row in rows or []: code = (getattr(row, "sample_code", None) or "").strip() if code: out.add(code) return out def diff_new_sample_codes( *, before_codes: Iterable[str], submitted_codes: Iterable[str], ) -> List[str]: """提交相对保存前的新增 sample_code；稳定排序便于测试。""" before = {str(c).strip() for c in before_codes if str(c).strip()} submitted = {str(c).strip() for c in submitted_codes if str(c).strip()} return sorted(submitted - before) def format_sample_collect_title(eta_time: Optional[datetime] = None) -> str: """工单标题：有 ETA → 样本采集（预计到达 M月D日）；否则到达待定。""" if eta_time is not None: title = f"样本采集（预计到达 {eta_time.month}月{eta_time.day}日）" else: title = "样本采集（到达待定）" return title def format_protocol_create_time(dt: Optional[datetime]) -> str: if not dt: return "" return dt.strftime("%Y年%m月%d日 %H:%M") def _notify_enabled() -> bool: if not bool(getattr(settings, "WX_MP_NOTIFY_ENABLED", False)): return False tid = (getattr(settings, "WX_MP_TEMPLATE_SAMPLE_COLLECT", "") or "").strip() return bool(tid) def _role_name() -> str: return ( getattr(settings, "WX_MP_SAMPLE_COLLECT_ROLE_NAME", "") or "样本采集通知人员" ).strip() or "样本采集通知人员" def protocol_status_blocks_notify(status: Optional[str]) -> bool: """已作废、属地废弃的协议单不发样本采集通知。""" return (status or "").strip() in INACTIVE_PROTOCOL_STATUSES def sample_status_blocks_notify(status: Optional[str]) -> bool: """已作废样本不发样本采集通知。""" return (status or "").strip() == SAMPLE_STATUS_VOID def _sample_status_by_code(samples: Iterable[Any]) -> Dict[str, str]: out: Dict[str, str] = {} for sample in samples or []: code = (getattr(sample, "sample_code", None) or "").strip() if not code: continue out[code] = (getattr(sample, "sample_status", None) or "").strip() return out def _task_type_for_kind(protocol_kind: str) -> str: if protocol_kind == PROTOCOL_KIND_DONATION: return TASK_TYPE_DONATION return TASK_TYPE_CELL def _sample_type_name_by_code( db: Session, *, tenant_no: str, samples: Iterable[Any] ) -> Dict[str, str]: """按样本编码取样本类型名称。无类型或类型已删除时为空串。""" rows = list(samples or []) type_ids = { (getattr(sample, "sample_type_id", None) or "").strip() for sample in rows if (getattr(sample, "sample_type_id", None) or "").strip() } name_by_id: Dict[str, str] = {} if type_ids: type_rows = ( db.query(BizCellSampleType) .filter( BizCellSampleType.id.in_(list(type_ids)), BizCellSampleType.tenant_no == tenant_no, BizCellSampleType.del_flag == 0, ) .all() ) name_by_id = { row.id: (row.type_name or "").strip() for row in type_rows } out: Dict[str, str] = {} for sample in rows: code = (getattr(sample, "sample_code", None) or "").strip() if not code: continue type_id = (getattr(sample, "sample_type_id", None) or "").strip() out[code] = name_by_id.get(type_id, "") if type_id else "" return out def _biz_id(sample_code: str, user_id: str) -> str: return f"{(sample_code or '').strip()}:{(user_id or '').strip()}" def _already_success( db: Session, *, tenant_no: str, sample_code: str, user_id: str ) -> bool: row = ( db.query(BizWxNotifyLog) .filter( BizWxNotifyLog.tenant_no == tenant_no, BizWxNotifyLog.biz_type == NOTIFY_BIZ_TYPE, BizWxNotifyLog.biz_id == _biz_id(sample_code, user_id), BizWxNotifyLog.notify_date == NOTIFY_DATE_ONCE, BizWxNotifyLog.status == "success", BizWxNotifyLog.del_flag == 0, ) .first() ) return row is not None def _write_success_log( db: Session, *, tenant_no: str, sample_code: str, user_id: str, ) -> None: now = datetime.now() db.add( BizWxNotifyLog( id=str(uuid.uuid4()).replace("-", "").upper(), biz_type=NOTIFY_BIZ_TYPE, biz_id=_biz_id(sample_code, user_id), notify_date=NOTIFY_DATE_ONCE, user_id=user_id or "", status="success", errcode=None, errmsg=None, tenant_no=tenant_no, order_num=0, remark="", create_by=user_id or "", create_dept_id="", create_time=now, update_by=user_id or "", update_time=now, del_flag=0, ) ) def list_notify_recipient_users(db: Session, *, tenant_no: str) -> List[SysUser]: """有效拥有「样本采集通知人员」角色且已绑定服务号的用户。 有效角色 = 用户直接绑角色 ∪ 经启用职位继承（sys_user_post → sys_post → sys_post_role）， 与登录权限解析一致。 """ role_name = _role_name() user_ids = list_user_ids_with_effective_role_name( db, tenant_no=tenant_no, role_name=role_name ) if not user_ids: return [] rows = ( db.query(SysUser) .filter( SysUser.id.in_(list(user_ids)), SysUser.tenant_no == tenant_no, SysUser.del_flag == 0, SysUser.wx_mp_openid.isnot(None), SysUser.wx_mp_openid != "", ) .all() ) # 按 user_id 已去重；再按 openid 去重，避免一人多职位/多账号同微信重复推送 seen_openid: Set[str] = set() out: List[SysUser] = [] for u in rows: oid = (getattr(u, "wx_mp_openid", None) or "").strip() if not oid or oid in seen_openid: continue seen_openid.add(oid) out.append(u) return out def _load_cell_context( db: Session, *, protocol_id: str, tenant_no: str, new_codes: List[str] ) -> Optional[Dict[str, Any]]: protocol = ( db.query(BizCollectionProtocol) .filter( BizCollectionProtocol.id == protocol_id, BizCollectionProtocol.tenant_no == tenant_no, BizCollectionProtocol.del_flag == 0, ) .first() ) if not protocol: return None samples = ( db.query(BizProtocolSample) .filter( BizProtocolSample.collection_protocol_id == protocol_id, BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code.in_(new_codes), ) .all() ) code_to_id = { (s.sample_code or "").strip(): s.id for s in samples if (s.sample_code or "").strip() } sample_ids = list(code_to_id.values()) eta_by_code: Dict[str, Optional[datetime]] = {c: None for c in new_codes} if sample_ids: transports = ( db.query(BizProtocolSampleTransport) .filter( BizProtocolSampleTransport.collection_protocol_id == protocol_id, BizProtocolSampleTransport.tenant_no == tenant_no, BizProtocolSampleTransport.del_flag == 0, BizProtocolSampleTransport.sample_id.in_(sample_ids), ) .all() ) id_to_code = {sid: code for code, sid in code_to_id.items()} for t in transports: code = id_to_code.get((t.sample_id or "").strip()) or ( getattr(t, "sample_code", None) or "" ).strip() if code and code in eta_by_code and t.eta_time and eta_by_code[code] is None: eta_by_code[code] = t.eta_time creator = None if protocol.create_by: creator = ( db.query(SysUser) .filter(SysUser.id == protocol.create_by, SysUser.del_flag == 0) .first() ) return { "create_time": protocol.create_time, "applicant": (creator.nick_name if creator else "") or "", "eta_by_code": eta_by_code, "protocol_status": (protocol.protocol_status or "").strip(), "sample_status_by_code": _sample_status_by_code(samples), "sample_type_by_code": _sample_type_name_by_code( db, tenant_no=tenant_no, samples=samples ), } def _load_donation_context( db: Session, *, protocol_id: str, tenant_no: str, new_codes: List[str] ) -> Optional[Dict[str, Any]]: donation = ( db.query(BizResearchDonation) .filter( BizResearchDonation.id == protocol_id, BizResearchDonation.tenant_no == tenant_no, BizResearchDonation.del_flag == 0, ) .first() ) if not donation: return None samples = ( db.query(BizResearchDonationSample) .filter( BizResearchDonationSample.research_donation_id == protocol_id, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_code.in_(new_codes), ) .all() ) code_to_id = { (s.sample_code or "").strip(): s.id for s in samples if (s.sample_code or "").strip() } sample_ids = list(code_to_id.values()) eta_by_code: Dict[str, Optional[datetime]] = {c: None for c in new_codes} if sample_ids: transports = ( db.query(BizResearchDonationSampleTransport) .filter( BizResearchDonationSampleTransport.research_donation_id == protocol_id, BizResearchDonationSampleTransport.tenant_no == tenant_no, BizResearchDonationSampleTransport.del_flag == 0, BizResearchDonationSampleTransport.sample_id.in_(sample_ids), ) .all() ) id_to_code = {sid: code for code, sid in code_to_id.items()} for t in transports: code = id_to_code.get((t.sample_id or "").strip()) or ( getattr(t, "sample_code", None) or "" ).strip() if code and code in eta_by_code and t.eta_time and eta_by_code[code] is None: eta_by_code[code] = t.eta_time creator = None if donation.create_by: creator = ( db.query(SysUser) .filter(SysUser.id == donation.create_by, SysUser.del_flag == 0) .first() ) return { "create_time": donation.create_time, "applicant": (creator.nick_name if creator else "") or "", "eta_by_code": eta_by_code, "protocol_status": (donation.donation_status or "").strip(), "sample_status_by_code": _sample_status_by_code(samples), "sample_type_by_code": _sample_type_name_by_code( db, tenant_no=tenant_no, samples=samples ), } def run_sample_collect_notify( db: Session, *, tenant_no: str, protocol_id: str, protocol_kind: str, new_sample_codes: Iterable[str], ) -> Dict[str, int]: """对新增样本向角色用户发送通知。返回计数。""" stats = {"codes": 0, "sent": 0, "skipped": 0, "failed": 0} if not _notify_enabled(): logger.info("sample collect notify skipped: notify disabled or no template") return stats codes = [c for c in (str(x).strip() for x in new_sample_codes) if c] if not codes: return stats stats["codes"] = len(codes) tenant = (tenant_no or "").strip() or "zkxw" kind = (protocol_kind or "").strip() or PROTOCOL_KIND_CELL if kind == PROTOCOL_KIND_DONATION: ctx = _load_donation_context( db, protocol_id=protocol_id, tenant_no=tenant, new_codes=codes ) else: ctx = _load_cell_context( db, protocol_id=protocol_id, tenant_no=tenant, new_codes=codes ) if not ctx: logger.info( "sample collect notify skip: protocol missing id=%s kind=%s", protocol_id, kind, ) return stats protocol_status = (ctx.get("protocol_status") or "").strip() if protocol_status_blocks_notify(protocol_status): logger.info( "sample collect notify skip: inactive protocol id=%s kind=%s status=%s", protocol_id, kind, protocol_status, ) stats["skipped"] = len(codes) return stats recipients = list_notify_recipient_users(db, tenant_no=tenant) if not recipients: logger.info( "sample collect notify skip: no recipients tenant=%s role=%s", tenant, _role_name(), ) return stats from app.services.wechat_mp_message import send_sample_collect_notify applicant = ctx.get("applicant") or "" time_text = format_protocol_create_time(ctx.get("create_time")) task_type = _task_type_for_kind(kind) eta_by_code: Dict[str, Optional[datetime]] = ctx.get("eta_by_code") or {} sample_type_by_code: Dict[str, str] = ctx.get("sample_type_by_code") or {} sample_status_by_code: Dict[str, str] = ctx.get("sample_status_by_code") or {} for code in codes: if sample_status_blocks_notify(sample_status_by_code.get(code)): logger.info( "sample collect notify skip: voided sample code=%s protocol=%s", code, protocol_id, ) stats["skipped"] += 1 continue title = format_sample_collect_title(eta_by_code.get(code)) sample_type = (sample_type_by_code.get(code) or "").strip() for user in recipients: uid = (user.id or "").strip() mp = (getattr(user, "wx_mp_openid", None) or "").strip() if not uid or not mp: stats["skipped"] += 1 continue if _already_success(db, tenant_no=tenant, sample_code=code, user_id=uid): stats["skipped"] += 1 continue result = send_sample_collect_notify( mp_openid=mp, title=title, applicant=applicant, time_text=time_text, task_type=task_type, sample_type=sample_type, pagepath="", ) if result.get("success"): _write_success_log( db, tenant_no=tenant, sample_code=code, user_id=uid ) stats["sent"] += 1 else: logger.error( "sample collect notify failed code=%s user=%s err=%s", code, uid, result, ) stats["failed"] += 1 try: db.commit() except Exception: db.rollback() logger.exception("sample collect notify commit failed") raise return stats def notify_sample_collect_sync( *, tenant_no: str, protocol_id: str, protocol_kind: str, new_sample_codes: List[str], ) -> None: db = None try: db = SessionLocal() stats = run_sample_collect_notify( db, tenant_no=tenant_no, protocol_id=protocol_id, protocol_kind=protocol_kind, new_sample_codes=new_sample_codes, ) logger.info( "sample collect notify done protocol=%s kind=%s stats=%s", protocol_id, protocol_kind, stats, ) except Exception: logger.exception( "sample collect notify batch failed protocol=%s kind=%s", protocol_id, protocol_kind, ) finally: if db is not None: db.close() def schedule_sample_collect_notify( *, tenant_no: str, protocol_id: str, protocol_kind: str, before_codes: Iterable[str], submitted_codes: Optional[Iterable[str]], ) -> None: """保存成功后调度：submitted_codes 为 None 表示本次未改样本，不推。""" if not _notify_enabled(): logger.info("sample collect notify schedule skipped: disabled") return if submitted_codes is None: return new_codes = diff_new_sample_codes( before_codes=before_codes, submitted_codes=submitted_codes ) if not new_codes: return pid = (protocol_id or "").strip() if not pid: return threading.Thread( target=notify_sample_collect_sync, kwargs={ "tenant_no": (tenant_no or "").strip() or "zkxw", "protocol_id": pid, "protocol_kind": (protocol_kind or "").strip() or PROTOCOL_KIND_CELL, "new_sample_codes": new_codes, }, daemon=True, name="sample-collect-wx-notify", ).start()