"""采集地区及下属医院 CRUD""" from __future__ import annotations import math from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.models.models import ( BizCollectionHospital, BizCollectionRegion, BizCollectionRegionStaff, SysDept, SysUser, ) def staff_display_name(nick_name: Optional[str], user_no: Optional[str]) -> str: n = (nick_name or "").strip() if n: return n return (user_no or "").strip() def build_staff_leader_name_lists(rows: List[Dict[str, Any]]) -> Tuple[List[str], List[str]]: """rows: dicts with nickName/isLeader (already sorted leaders-first preferred).""" leaders = [staff_display_name(r.get("nickName"), r.get("userNo")) for r in rows if r.get("isLeader")] # Keep leader-first order if caller sorted; otherwise partition non_leaders = [staff_display_name(r.get("nickName"), r.get("userNo")) for r in rows if not r.get("isLeader")] staff_names = leaders + non_leaders return staff_names, leaders def _region_to_dict( r: BizCollectionRegion, hospital_names: Optional[List[str]] = None, staff_names: Optional[List[str]] = None, leader_names: Optional[List[str]] = None, ) -> Dict[str, Any]: return { "id": r.id, "regionName": r.region_name, "hospitalNames": hospital_names or [], "staffNames": staff_names or [], "leaderNames": leader_names or [], "tenantNo": r.tenant_no, "orderNum": r.order_num if r.order_num is not None else 0, "remark": r.remark or "", "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "", "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "", } def _hospital_names_by_region_ids(db: Session, region_ids: List[str]) -> Dict[str, List[str]]: """批量查询地区下属医院名称，按 order_num、create_time 排序。""" hospital_map: Dict[str, List[str]] = {} if not region_ids: return hospital_map hosp_rows = ( db.query(BizCollectionHospital.region_id, BizCollectionHospital.hospital_name) .filter( and_( BizCollectionHospital.region_id.in_(region_ids), BizCollectionHospital.del_flag == 0, ) ) .order_by(BizCollectionHospital.order_num.asc(), BizCollectionHospital.create_time.desc()) .all() ) for h in hosp_rows: hospital_map.setdefault(h.region_id, []).append(h.hospital_name) return hospital_map def _staff_names_by_region_ids( db: Session, region_ids: List[str] ) -> Dict[str, Tuple[List[str], List[str]]]: """Return region_id -> (staffNames, leaderNames). Leaders first in staffNames.""" out: Dict[str, Tuple[List[str], List[str]]] = {rid: ([], []) for rid in region_ids} if not region_ids: return out rows = ( db.query(BizCollectionRegionStaff, SysUser) .outerjoin(SysUser, SysUser.id == BizCollectionRegionStaff.user_id) .filter( and_( BizCollectionRegionStaff.region_id.in_(region_ids), BizCollectionRegionStaff.del_flag == 0, ) ) .order_by( BizCollectionRegionStaff.is_leader.desc(), BizCollectionRegionStaff.order_num.asc(), BizCollectionRegionStaff.create_time.desc(), ) .all() ) buckets: Dict[str, List[Dict[str, Any]]] = {rid: [] for rid in region_ids} for staff, user in rows: buckets.setdefault(staff.region_id, []).append( { "nickName": (user.nick_name if user else "") or "", "userNo": (user.user_no if user else "") or "", "isLeader": bool(staff.is_leader), } ) for rid, items in buckets.items(): out[rid] = build_staff_leader_name_lists(items) return out def _hospital_to_dict(h: BizCollectionHospital) -> Dict[str, Any]: return { "id": h.id, "regionId": h.region_id, "hospitalName": h.hospital_name, "tenantNo": h.tenant_no, "orderNum": h.order_num if h.order_num is not None else 0, "remark": h.remark or "", "createTime": h.create_time.strftime("%Y-%m-%d %H:%M:%S") if h.create_time else "", "updateTime": h.update_time.strftime("%Y-%m-%d %H:%M:%S") if h.update_time else "", } def _staff_to_dict( s: BizCollectionRegionStaff, *, nick_name: str = "", user_no: str = "", dept_name: str = "", ) -> Dict[str, Any]: return { "id": s.id, "regionId": s.region_id, "userId": s.user_id, "userName": staff_display_name(nick_name, user_no), "nickName": nick_name or "", "userNo": user_no or "", "deptName": dept_name or "", "isLeader": bool(s.is_leader), "tenantNo": s.tenant_no, "orderNum": s.order_num if s.order_num is not None else 0, "remark": s.remark or "", "createTime": s.create_time.strftime("%Y-%m-%d %H:%M:%S") if s.create_time else "", "updateTime": s.update_time.strftime("%Y-%m-%d %H:%M:%S") if s.update_time else "", } def list_collection_regions( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizCollectionRegion).filter(BizCollectionRegion.del_flag == 0) if tenant_no: q0 = q0.filter(BizCollectionRegion.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter(or_(BizCollectionRegion.region_name.like(like), BizCollectionRegion.remark.like(like))) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCollectionRegion.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCollectionRegion.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) region_ids = [r.id for r in rows] hospital_map = _hospital_names_by_region_ids(db, region_ids) staff_map = _staff_names_by_region_ids(db, region_ids) return [ _region_to_dict( x, hospital_names=hospital_map.get(x.id, []), staff_names=staff_map.get(x.id, ([], []))[0], leader_names=staff_map.get(x.id, ([], []))[1], ) for x in rows ], int(total) def get_collection_region(db: Session, region_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionRegion]: q = db.query(BizCollectionRegion).filter(and_(BizCollectionRegion.id == region_id, BizCollectionRegion.del_flag == 0)) if tenant_no: q = q.filter(BizCollectionRegion.tenant_no == tenant_no) return q.first() def create_collection_region( db: Session, *, region_name: str, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, ) -> BizCollectionRegion: name = (region_name or "").strip() if not name: raise ValueError("地区名称不能为空") obj = BizCollectionRegion( id=generate_uuid(), region_name=name, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_collection_region( db: Session, region_id: str, *, region_name: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizCollectionRegion: obj = get_collection_region(db, region_id, tenant_no=tenant_no) if not obj: raise ValueError("采集地区不存在") if region_name is not None: n = region_name.strip() if not n: raise ValueError("地区名称不能为空") obj.region_name = n if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark.strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_collection_region(db: Session, region_id: str, tenant_no: Optional[str] = None) -> None: obj = get_collection_region(db, region_id, tenant_no=tenant_no) if not obj: raise ValueError("采集地区不存在") now = datetime.now() obj.del_flag = 1 obj.update_time = now db.add(obj) subs = ( db.query(BizCollectionHospital) .filter( and_( BizCollectionHospital.region_id == region_id, BizCollectionHospital.del_flag == 0, ) ) .all() ) for h in subs: h.del_flag = 1 h.update_time = now db.add(h) staff_subs = ( db.query(BizCollectionRegionStaff) .filter( and_( BizCollectionRegionStaff.region_id == region_id, BizCollectionRegionStaff.del_flag == 0, ) ) .all() ) for s in staff_subs: s.del_flag = 1 s.update_time = now db.add(s) db.commit() def list_hospitals( db: Session, *, region_id: str, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: region = get_collection_region(db, region_id, tenant_no=tenant_no) if not region: raise ValueError("采集地区不存在") def _filtered(): q0 = db.query(BizCollectionHospital).filter( and_(BizCollectionHospital.region_id == region_id, BizCollectionHospital.del_flag == 0) ) if tenant_no: q0 = q0.filter(BizCollectionHospital.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter(BizCollectionHospital.hospital_name.like(like)) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCollectionHospital.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCollectionHospital.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return [_hospital_to_dict(x) for x in rows], int(total) def search_hospitals_by_keyword( db: Session, *, keyword: str, tenant_no: Optional[str] = None, limit: int = 20, ) -> List[Dict[str, Any]]: """按医院名称模糊检索，返回医院及所属采集地区信息。""" kw = (keyword or "").strip() if not kw: return [] like = f"%{kw}%" q = ( db.query(BizCollectionHospital, BizCollectionRegion) .join(BizCollectionRegion, BizCollectionHospital.region_id == BizCollectionRegion.id) .filter( BizCollectionHospital.del_flag == 0, BizCollectionRegion.del_flag == 0, BizCollectionHospital.hospital_name.like(like), ) ) if tenant_no: q = q.filter(BizCollectionHospital.tenant_no == tenant_no) rows = ( q.order_by(BizCollectionHospital.order_num.asc(), BizCollectionHospital.create_time.desc()) .limit(min(max(1, limit), 50)) .all() ) items: List[Dict[str, Any]] = [] for hospital, region in rows: item = _hospital_to_dict(hospital) item["regionName"] = region.region_name or "" items.append(item) return items def get_hospital(db: Session, hospital_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionHospital]: q = db.query(BizCollectionHospital).filter(and_(BizCollectionHospital.id == hospital_id, BizCollectionHospital.del_flag == 0)) if tenant_no: q = q.filter(BizCollectionHospital.tenant_no == tenant_no) return q.first() def create_hospital( db: Session, *, region_id: str, hospital_name: str, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, ) -> BizCollectionHospital: region = get_collection_region(db, region_id, tenant_no=tenant_no) if not region: raise ValueError("采集地区不存在") hn = (hospital_name or "").strip() if not hn: raise ValueError("医院名称不能为空") obj = BizCollectionHospital( id=generate_uuid(), region_id=region_id, hospital_name=hn, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_hospital( db: Session, hospital_id: str, *, hospital_name: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizCollectionHospital: obj = get_hospital(db, hospital_id, tenant_no=tenant_no) if not obj: raise ValueError("医院不存在") if hospital_name is not None: hn = hospital_name.strip() if not hn: raise ValueError("医院名称不能为空") obj.hospital_name = hn if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark.strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_hospital(db: Session, hospital_id: str, tenant_no: Optional[str] = None) -> None: obj = get_hospital(db, hospital_id, tenant_no=tenant_no) if not obj: raise ValueError("医院不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit() def map_staff_region_rows_to_list(rows: List[Tuple[str, str]]) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] for rid, name in rows: rid = (rid or "").strip() if not rid: continue out.append({"id": rid, "regionName": (name or "").strip()}) return out def list_regions_by_staff_user( db: Session, user_id: str, tenant_no: Optional[str] = None ) -> List[Dict[str, Any]]: uid = (user_id or "").strip() if not uid: return [] q = ( db.query(BizCollectionRegionStaff.region_id, BizCollectionRegion.region_name) .join( BizCollectionRegion, BizCollectionRegion.id == BizCollectionRegionStaff.region_id, ) .filter( BizCollectionRegionStaff.user_id == uid, BizCollectionRegionStaff.del_flag == 0, BizCollectionRegion.del_flag == 0, ) ) if tenant_no: q = q.filter( BizCollectionRegionStaff.tenant_no == tenant_no, BizCollectionRegion.tenant_no == tenant_no, ) q = q.order_by( BizCollectionRegionStaff.order_num.asc(), BizCollectionRegionStaff.create_time.desc(), ) return map_staff_region_rows_to_list([(r[0], r[1]) for r in q.all()]) def list_active_region_leaders(db: Session, *, tenant_no: str) -> List[Dict[str, Any]]: """协议表单候选：租户内有效地区负责人，按用户去重并保留负责地区。""" if not tenant_no: raise ValueError("租户信息缺失") rows = ( db.query(SysUser.id, SysUser.nick_name, BizCollectionRegionStaff.region_id) .join(BizCollectionRegionStaff, BizCollectionRegionStaff.user_id == SysUser.id) .join(BizCollectionRegion, BizCollectionRegion.id == BizCollectionRegionStaff.region_id) .filter( BizCollectionRegion.tenant_no == tenant_no, BizCollectionRegion.del_flag == 0, BizCollectionRegionStaff.tenant_no == tenant_no, BizCollectionRegionStaff.del_flag == 0, BizCollectionRegionStaff.is_leader == 1, SysUser.tenant_no == tenant_no, SysUser.del_flag == 0, SysUser.status == "1", ) .distinct() .order_by(SysUser.nick_name.asc(), SysUser.id.asc(), BizCollectionRegionStaff.region_id.asc()) .all() ) leaders: Dict[str, Dict[str, Any]] = {} for user_id, nick_name, region_id in rows: item = leaders.setdefault(user_id, {"id": user_id, "nickName": nick_name or "", "regionIds": []}) item["regionIds"].append(region_id) return list(leaders.values()) def list_staff( db: Session, *, region_id: str, page: int = 1, page_size: int = 10, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: region = get_collection_region(db, region_id, tenant_no=tenant_no) if not region: raise ValueError("采集地区不存在") def _filtered(): q0 = db.query(BizCollectionRegionStaff).filter( and_(BizCollectionRegionStaff.region_id == region_id, BizCollectionRegionStaff.del_flag == 0) ) if tenant_no: q0 = q0.filter(BizCollectionRegionStaff.tenant_no == tenant_no) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCollectionRegionStaff.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by( BizCollectionRegionStaff.is_leader.desc(), BizCollectionRegionStaff.order_num.asc(), BizCollectionRegionStaff.create_time.desc(), ) .offset((page - 1) * page_size) .limit(page_size) .all() ) user_ids = [r.user_id for r in rows] user_map: Dict[str, SysUser] = {} dept_map: Dict[str, str] = {} if user_ids: for u in db.query(SysUser).filter(SysUser.id.in_(user_ids)).all(): user_map[u.id] = u dept_ids = [u.dept_id for u in user_map.values() if u.dept_id] if dept_ids: for d in db.query(SysDept).filter(SysDept.id.in_(dept_ids), SysDept.del_flag == 0).all(): dept_map[d.id] = d.dept_name or "" items: List[Dict[str, Any]] = [] for s in rows: u = user_map.get(s.user_id) items.append( _staff_to_dict( s, nick_name=(u.nick_name if u else "") or "", user_no=(u.user_no if u else "") or "", dept_name=dept_map.get((u.dept_id if u else "") or "", ""), ) ) return items, int(total) def get_staff(db: Session, staff_id: str, tenant_no: Optional[str] = None) -> Optional[BizCollectionRegionStaff]: q = db.query(BizCollectionRegionStaff).filter( and_(BizCollectionRegionStaff.id == staff_id, BizCollectionRegionStaff.del_flag == 0) ) if tenant_no: q = q.filter(BizCollectionRegionStaff.tenant_no == tenant_no) return q.first() def create_staff( db: Session, *, region_id: str, user_id: str, is_leader: bool = False, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id_actor: Optional[str] = None, ) -> BizCollectionRegionStaff: region = get_collection_region(db, region_id, tenant_no=tenant_no) if not region: raise ValueError("采集地区不存在") uid = (user_id or "").strip() if not uid: raise ValueError("请选择系统人员") user = ( db.query(SysUser) .filter( and_( SysUser.id == uid, SysUser.del_flag == 0, SysUser.tenant_no == tenant_no, ) ) .first() ) if not user: raise ValueError("用户不存在") if (user.status or "") != "1": raise ValueError("用户已停用，无法绑定") exists = ( db.query(BizCollectionRegionStaff) .filter( and_( BizCollectionRegionStaff.region_id == region_id, BizCollectionRegionStaff.user_id == uid, BizCollectionRegionStaff.del_flag == 0, BizCollectionRegionStaff.tenant_no == tenant_no, ) ) .first() ) if exists: raise ValueError("该人员已绑定到此地区") obj = BizCollectionRegionStaff( id=generate_uuid(), region_id=region_id, user_id=uid, is_leader=1 if is_leader else 0, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id_actor, create_dept_id="", create_time=datetime.now(), update_by=user_id_actor, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_staff( db: Session, staff_id: str, *, is_leader: Optional[bool] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizCollectionRegionStaff: obj = get_staff(db, staff_id, tenant_no=tenant_no) if not obj: raise ValueError("绑定记录不存在") if is_leader is not None: obj.is_leader = 1 if is_leader else 0 if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = remark.strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_staff(db: Session, staff_id: str, tenant_no: Optional[str] = None) -> None: obj = get_staff(db, staff_id, tenant_no=tenant_no) if not obj: raise ValueError("绑定记录不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit() def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size)))