"""细胞产品 CRUD(biz_cell_product)""" from __future__ import annotations import math import re 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.utils.sample_special import sample_special_to_dict from app.crud import storage_occupancy as occupancy_svc from app.models.models import BizStorageSlotOccupancy from app.models.models import ( BizCellProduct, BizCellProductSpec, BizProductionProductType, BizStorageFacility, BizStorageFacilitySlot, ) from app.services.storage_facility_code import build_full_slot_code from app.crud.storage_facility import search_slot_ids_by_keyword _CELL_PRODUCT_STATUSES = frozenset( {"frozen", "in_production", "sold", "used", "voided", "thawed"} ) _CELL_PRODUCT_RELEASE_STATUSES = frozenset({"sold", "used", "voided"}) _CELL_PRODUCT_STORE_STATUSES = frozenset({"frozen", "thawed", "in_production"}) def _apply_cell_product_status_change( db: Session, obj: BizCellProduct, to_status: str, tenant_no: str, user_id: Optional[str], now: datetime, ) -> None: """管理端手动改状态:允许在合法枚举间任意变更,并同步库位占用。""" from_status = (obj.product_status or "").strip() to_status = (to_status or "").strip() if to_status == from_status: return if to_status not in _CELL_PRODUCT_STATUSES: raise ValueError( f"状态必须为 {'/'.join(sorted(_CELL_PRODUCT_STATUSES))}" ) obj.product_status = to_status slot_id = (obj.storage_slot_id or "").strip() if to_status in _CELL_PRODUCT_RELEASE_STATUSES: if slot_id: try: occupancy_svc.remove_by_item( db, "cell_product", obj.id, user_id=user_id, ) except Exception: pass return if to_status in _CELL_PRODUCT_STORE_STATUSES and slot_id: existing = db.query(BizStorageSlotOccupancy).filter( BizStorageSlotOccupancy.item_type == "cell_product", BizStorageSlotOccupancy.item_id == obj.id, BizStorageSlotOccupancy.occupancy_status == "stored", BizStorageSlotOccupancy.del_flag == 0, ).first() if existing: return occupancy_svc.assert_slot_vacant( db, slot_id, tenant_no=tenant_no, exclude_product_id=obj.id, ) occupancy_svc._add_stored_occupancy_no_commit( db, slot_id=slot_id, item_type="cell_product", item_id=obj.id, item_code=obj.product_code or "", item_name=obj.product_name or "", item_source=obj.source_type or "manual", tenant_no=tenant_no, user_id=user_id, create_dept_id=obj.create_dept_id or "", stored_at=now, ) def _row_to_dict(c: BizCellProduct) -> Dict[str, Any]: return { "id": c.id, "productCode": c.product_code, "productName": c.product_name, "productTypeId": c.product_type_id or "", "productTypeName": "", "specId": c.spec_id or "", "specText": c.spec_text or "", "specName": c.spec_text or "", "batchNo": c.batch_no or "", "storageSlotId": c.storage_slot_id or "", "storageSlotName": "", "passage": c.passage or "", "freezeDate": c.freeze_date.strftime("%Y-%m-%d %H:%M:%S") if c.freeze_date else "", "expectedCellCount": float(c.expected_cell_count) if c.expected_cell_count is not None else None, "viability": float(c.viability) if c.viability is not None else None, "productStatus": c.product_status or "", "sourceType": c.source_type or "", "productionTaskId": c.production_task_id or "", "quantity": float(c.quantity) if c.quantity is not None else None, "unit": c.unit or "", "expiryDate": c.expiry_date.strftime("%Y-%m-%d %H:%M:%S") if c.expiry_date else "", "productImages": c.product_images or "", "tenantNo": c.tenant_no, "orderNum": c.order_num if c.order_num is not None else 0, "remark": c.remark or "", "createBy": c.create_by or "", "createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "", "updateBy": c.update_by or "", "updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "", } def _build_facility_path_label(node_id: str, by_id: Dict[str, BizStorageFacility]) -> str: parts: List[str] = [] cur_id = (node_id or "").strip() visited: set[str] = set() while cur_id and cur_id != "0" and cur_id not in visited: visited.add(cur_id) node = by_id.get(cur_id) if not node: break parts.append(node.node_name or "") cur_id = (node.parent_id or "").strip() parts.reverse() return " / ".join(p for p in parts if p) def _enrich_product_dicts( db: Session, tenant_no: str, items: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: if not items: return items type_ids = {i.get("productTypeId") for i in items if i.get("productTypeId")} spec_ids = {i.get("specId") for i in items if i.get("specId")} slot_ids = {i.get("storageSlotId") for i in items if i.get("storageSlotId")} type_map: Dict[str, str] = {} if type_ids: rows = ( db.query(BizProductionProductType) .filter( BizProductionProductType.id.in_(type_ids), BizProductionProductType.del_flag == 0, ) .all() ) type_map = {r.id: r.product_type_name or "" for r in rows} spec_map: Dict[str, str] = {} if spec_ids: rows = ( db.query(BizCellProductSpec) .filter( BizCellProductSpec.id.in_(spec_ids), BizCellProductSpec.del_flag == 0, ) .all() ) spec_map = {r.id: r.spec_name or "" for r in rows} slot_map: Dict[str, str] = {} if slot_ids: slots = ( db.query(BizStorageFacilitySlot) .filter( BizStorageFacilitySlot.id.in_(slot_ids), BizStorageFacilitySlot.del_flag == 0, ) .all() ) facilities = ( db.query(BizStorageFacility) .filter( BizStorageFacility.tenant_no == tenant_no, BizStorageFacility.del_flag == 0, ) .all() ) facility_by_id = {f.id: f for f in facilities} for slot in slots: slot_map[slot.id] = build_full_slot_code( slot.facility_node_id, slot.slot_code, facility_by_id, ) # --- 来源信息富化 --- from app.models.models import BizProductionTask, BizProductionWorkOrder task_ids = {i.get("productionTaskId") for i in items if i.get("productionTaskId")} task_map: Dict[str, Any] = {} task_to_wo: Dict[str, str] = {} wo_map: Dict[str, Any] = {} if task_ids: tasks = ( db.query(BizProductionTask) .filter( BizProductionTask.id.in_(task_ids), BizProductionTask.del_flag == 0, ) .all() ) for t in tasks: task_map[t.id] = t if t.work_order_id: task_to_wo[t.id] = t.work_order_id wo_ids = set(task_to_wo.values()) if wo_ids: wos = ( db.query(BizProductionWorkOrder) .filter( BizProductionWorkOrder.id.in_(wo_ids), BizProductionWorkOrder.del_flag == 0, ) .all() ) wo_map = {w.id: w for w in wos} source_type_labels = {"production": "生产入库", "manual": "手动录入"} for item in items: tid = item.get("productTypeId") or "" sid = item.get("specId") or "" slot_id = item.get("storageSlotId") or "" item["productTypeName"] = type_map.get(tid, "") if tid else "" stored_spec = (item.get("specText") or item.get("spec_text") or "").strip() if stored_spec: item["specName"] = stored_spec elif sid: item["specName"] = spec_map.get(sid, "") else: item["specName"] = (item.get("specName") or "").strip() item["storageSlotName"] = slot_map.get(slot_id, "") if slot_id else "" # 来源信息 src_type = item.get("sourceType", "") item["sourceTypeLabel"] = source_type_labels.get(src_type, src_type or "") tid = item.get("productionTaskId") or "" if tid and tid in task_map: t_obj = task_map[tid] item["sourceTaskNo"] = t_obj.task_no or "" wo_id = task_to_wo.get(tid) if wo_id and wo_id in wo_map: wo_obj = wo_map[wo_id] item["sourceWorkOrderNo"] = wo_obj.work_order_no or "" item["sourceProtocolNo"] = wo_obj.protocol_no or "" item["sourceReceptionSource"] = wo_obj.reception_source or "" else: item["sourceWorkOrderNo"] = "" item["sourceProtocolNo"] = "" item["sourceReceptionSource"] = "" else: item["sourceTaskNo"] = "" item["sourceWorkOrderNo"] = "" item["sourceProtocolNo"] = "" item["sourceReceptionSource"] = "" return items def product_to_response(db: Session, tenant_no: str, obj: BizCellProduct) -> Dict[str, Any]: enriched = _enrich_product_dicts(db, tenant_no, [_row_to_dict(obj)]) return enriched[0] if enriched else _row_to_dict(obj) def list_cell_products( db: Session, *, tenant_no: str, page: int = 1, page_size: int = 20, keyword: Optional[str] = None, product_code: Optional[str] = None, product_name: Optional[str] = None, product_type_id: Optional[str] = None, product_status: Optional[str] = None, passage: Optional[str] = None, source_type: Optional[str] = None, user_info=None, ) -> Dict[str, Any]: kw = (keyword or "").strip() matched_slot_ids: List[str] = ( search_slot_ids_by_keyword(db, tenant_no, kw) if kw else [] ) def _filtered(): q0 = db.query(BizCellProduct).filter( BizCellProduct.del_flag == 0, BizCellProduct.tenant_no == tenant_no, ) if kw: like = f"%{kw}%" keyword_conds = [ BizCellProduct.product_code.ilike(like), BizCellProduct.product_name.ilike(like), ] if matched_slot_ids: keyword_conds.append( BizCellProduct.storage_slot_id.in_(matched_slot_ids) ) q0 = q0.filter(or_(*keyword_conds)) pc = (product_code or "").strip() if pc: q0 = q0.filter(BizCellProduct.product_code.like(f"%{pc}%")) pn = (product_name or "").strip() if pn: q0 = q0.filter(BizCellProduct.product_name.like(f"%{pn}%")) if product_type_id: q0 = q0.filter(BizCellProduct.product_type_id == product_type_id) if product_status: q0 = q0.filter(BizCellProduct.product_status == product_status) if passage: q0 = q0.filter(BizCellProduct.passage == passage) if source_type: q0 = q0.filter(BizCellProduct.source_type == source_type) if user_info: from app.core.data_scope import apply_data_scope q0 = apply_data_scope(q0, BizCellProduct, user_info, db, tenant_no) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCellProduct.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by( BizCellProduct.create_time.desc(), BizCellProduct.product_code.asc(), ) .offset((page - 1) * page_size) .limit(page_size) .all() ) items = _enrich_product_dicts( db, tenant_no, [_row_to_dict(x) for x in rows] ) return {"items": items, "total": int(total)} def get_cell_product(db: Session, id: str, tenant_no: str) -> Optional[BizCellProduct]: return ( db.query(BizCellProduct) .filter( and_( BizCellProduct.id == id, BizCellProduct.tenant_no == tenant_no, BizCellProduct.del_flag == 0, ) ) .first() ) def create_cell_product( db: Session, *, tenant_no: str, user_id: Optional[str] = None, create_dept_id: Optional[str] = None, data: Any = None, ) -> BizCellProduct: code = (data.product_code or "").strip() if not code: raise ValueError("产品编码不能为空") existing = db.query(BizCellProduct).filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code == code, BizCellProduct.del_flag == 0, ).first() if existing: raise ValueError(f"产品编码 {code} 已存在") obj = BizCellProduct( id=generate_uuid(), product_code=code, product_name=(data.product_name or "").strip(), product_type_id=(data.product_type_id or "").strip() or None, spec_id=(data.spec_id or "").strip() or None, spec_text=(getattr(data, "spec_name", None) or "").strip() or None, batch_no=(data.batch_no or "").strip() or None, storage_slot_id=(data.storage_slot_id or "").strip() or None, passage=((data.passage or "").strip() or None) if getattr(data, "passage", None) else None, freeze_date=data.freeze_date if hasattr(data, "freeze_date") else None, expected_cell_count=data.expected_cell_count if hasattr(data, "expected_cell_count") else None, viability=data.viability if hasattr(data, "viability") else None, product_status=data.product_status or "frozen", source_type=data.source_type or "manual", quantity=data.quantity, unit=(data.unit or "").strip() or None, expiry_date=data.expiry_date, product_images=(data.product_images or "").strip() or None, tenant_no=tenant_no, order_num=0, remark=(data.remark or "").strip() or None, create_by=user_id, create_dept_id=create_dept_id or "", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.flush() slot_id = (obj.storage_slot_id or "").strip() if slot_id and (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES: occupancy_svc.assert_slot_vacant(db, slot_id, tenant_no=tenant_no, exclude_product_id=obj.id) occupancy_svc._add_stored_occupancy_no_commit( db, slot_id=slot_id, item_type="cell_product", item_id=obj.id, item_code=obj.product_code or "", item_name=obj.product_name or "", item_source=obj.source_type or "manual", tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id or "", ) db.commit() db.refresh(obj) return obj def update_cell_product( db: Session, id: str, tenant_no: str, user_id: Optional[str] = None, data: Any = None, ) -> BizCellProduct: obj = get_cell_product(db, id, tenant_no) if not obj: raise ValueError("细胞产品不存在") old_slot_id = (obj.storage_slot_id or "").strip() slot_change_requested = data.storage_slot_id is not None if getattr(data, "product_code", None) is not None: code = (data.product_code or "").strip() if code and code != (obj.product_code or "").strip(): if product_code_exists(db, tenant_no, code, exclude_id=obj.id): raise ValueError(f"产品编码 {code} 已存在") obj.product_code = code if data.product_name is not None: obj.product_name = (data.product_name or "").strip() if data.product_type_id is not None: obj.product_type_id = (data.product_type_id or "").strip() or None if data.spec_id is not None: obj.spec_id = (data.spec_id or "").strip() or None if getattr(data, "spec_name", None) is not None: obj.spec_text = (data.spec_name or "").strip() or None if data.batch_no is not None: obj.batch_no = (data.batch_no or "").strip() or None if data.storage_slot_id is not None: obj.storage_slot_id = (data.storage_slot_id or "").strip() or None if hasattr(data, "passage") and data.passage is not None: obj.passage = (data.passage or "").strip() or None if hasattr(data, "freeze_date") and data.freeze_date is not None: obj.freeze_date = data.freeze_date if hasattr(data, "expected_cell_count") and data.expected_cell_count is not None: obj.expected_cell_count = data.expected_cell_count if hasattr(data, "viability") and data.viability is not None: obj.viability = data.viability if data.product_status is not None and data.product_status != obj.product_status: _apply_cell_product_status_change( db, obj, data.product_status, tenant_no, user_id, datetime.now(), ) if data.quantity is not None: obj.quantity = data.quantity if data.unit is not None: obj.unit = (data.unit or "").strip() or None if data.expiry_date is not None: obj.expiry_date = data.expiry_date if data.product_images is not None: obj.product_images = (data.product_images or "").strip() or None if data.remark is not None: obj.remark = (data.remark or "").strip() or None new_slot_id = (obj.storage_slot_id or "").strip() in_store = (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES if slot_change_requested and old_slot_id != new_slot_id: stored_occ = db.query(BizStorageSlotOccupancy).filter( BizStorageSlotOccupancy.item_type == "cell_product", BizStorageSlotOccupancy.item_id == obj.id, BizStorageSlotOccupancy.occupancy_status == "stored", BizStorageSlotOccupancy.del_flag == 0, ).first() if stored_occ and (stored_occ.slot_id or "").strip() != new_slot_id: occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id) stored_occ = None if new_slot_id and in_store and not stored_occ: occupancy_svc.assert_slot_vacant( db, new_slot_id, tenant_no=tenant_no, exclude_product_id=obj.id, ) occupancy_svc._add_stored_occupancy_no_commit( db, slot_id=new_slot_id, item_type="cell_product", item_id=obj.id, item_code=obj.product_code or "", item_name=obj.product_name or "", item_source=obj.source_type or "manual", tenant_no=tenant_no, user_id=user_id, create_dept_id=obj.create_dept_id or "", ) elif not new_slot_id and stored_occ: occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id) obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def delete_cell_product( db: Session, id: str, tenant_no: str, user_id: Optional[str] = None, ) -> None: obj = get_cell_product(db, id, tenant_no) if not obj: raise ValueError("细胞产品不存在") occupancy_svc.remove_by_item_no_commit(db, "cell_product", obj.id, user_id=user_id) obj.del_flag = 1 obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() def _next_code_start_seq(db: Session, tenant_no: str, prefix: str) -> int: last = ( db.query(BizCellProduct) .filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code.like(f"{prefix}%"), BizCellProduct.del_flag == 0, ) .order_by(BizCellProduct.product_code.desc()) .first() ) seq = 1 if last and last.product_code: try: base = normalize_product_code_base(last.product_code) seq = int(base[-6:]) + 1 except (ValueError, IndexError): seq = 1 return seq def get_next_code(db: Session, tenant_no: str) -> str: today_str = datetime.now().strftime("%Y%m%d") prefix = f"CP{today_str}" seq = _next_code_start_seq(db, tenant_no, prefix) return f"{prefix}{seq:06d}" _PRODUCT_CODE_SUFFIX_RE = re.compile(r"^(.+)-(\d{2})$") _PASSAGE_PRODUCT_CODE_RE = re.compile(r"^(.+)-(P\d+)-(\d{2})$", re.IGNORECASE) def normalize_passage_product_base(code: str) -> str: """去掉末尾 -P{n}-XX 或 -XX,得到代次产品编码的基础前缀。""" c = (code or "").strip() if not c: return "" m = _PASSAGE_PRODUCT_CODE_RE.match(c) if m: return m.group(1).strip() return normalize_product_code_base(c) def get_max_passage_product_seq( db: Session, tenant_no: str, base_code: str, passage: str, ) -> int: """查询 `{base_code}-{passage}-XX` 形式产品编码在库中的最大两位序号(无匹配则 0)。""" base = normalize_passage_product_base(base_code) passage = (passage or "").strip() if not base or not passage: return 0 prefix = f"{base}-{passage}-" suffix_re = re.compile(rf"^{re.escape(prefix)}(\d{{2}})$") rows = ( db.query(BizCellProduct.product_code) .filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code.like(f"{prefix}%"), BizCellProduct.del_flag == 0, ) .all() ) max_seq = 0 for (code,) in rows: m = suffix_re.match((code or "").strip()) if not m: continue try: max_seq = max(max_seq, int(m.group(1))) except ValueError: continue return max_seq def get_max_passage_product_seq_map( db: Session, tenant_no: str, base_code: str, passages: List[str], ) -> Dict[str, int]: """批量查询各代次在 `{base_code}-{passage}-XX` 规则下的最大序号。""" result: Dict[str, int] = {} seen: set[str] = set() for raw in passages or []: passage = (raw or "").strip() if not passage or passage in seen: continue seen.add(passage) result[passage] = get_max_passage_product_seq(db, tenant_no, base_code, passage) return result def get_max_product_suffix_seq( db: Session, tenant_no: str, base_code: str, ) -> int: """查询 `{base_code}-XX` 形式产品编码在库中的最大两位序号(无匹配则 0)。""" base = normalize_product_code_base((base_code or "").strip()) if not base: return 0 prefix = f"{base}-" suffix_re = re.compile(rf"^{re.escape(base)}-(\d{{2}})$") rows = ( db.query(BizCellProduct.product_code) .filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code.like(f"{prefix}%"), BizCellProduct.del_flag == 0, ) .all() ) max_seq = 0 for (code,) in rows: m = suffix_re.match((code or "").strip()) if not m: continue try: max_seq = max(max_seq, int(m.group(1))) except ValueError: continue return max_seq def product_code_exists( db: Session, tenant_no: str, product_code: str, exclude_id: Optional[str] = None, ) -> bool: code = (product_code or "").strip() if not code: return False q = db.query(BizCellProduct.id).filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code == code, BizCellProduct.del_flag == 0, ) if exclude_id: q = q.filter(BizCellProduct.id != exclude_id) return q.first() is not None def normalize_product_code_base(code: str) -> str: """去掉末尾 -01/-02 等序号,得到多份入库时的基础编码。""" c = (code or "").strip() if not c: return "" m = _PRODUCT_CODE_SUFFIX_RE.match(c) if m: return m.group(1).strip() return c def build_split_product_codes(base: str, count: int) -> List[str]: """按基础编码生成 基础-01、基础-02 …""" if count < 1: return [] return [f"{base}-{i:02d}" for i in range(1, count + 1)] def reserve_next_codes(db: Session, tenant_no: str, count: int) -> List[str]: if count < 1: return [] count = min(count, 50) today_str = datetime.now().strftime("%Y%m%d") prefix = f"CP{today_str}" start = _next_code_start_seq(db, tenant_no, prefix) return [f"{prefix}{start + i:06d}" for i in range(count)] def _validate_batch_items( db: Session, tenant_no: str, items: List[Any], *, require_storage_slot: bool = True, ) -> Tuple[List[str], List[Optional[str]]]: if not items: raise ValueError("至少需要一份产品明细") if len(items) > 50: raise ValueError("单次入库份数不能超过 50") codes: List[str] = [] slot_ids: List[Optional[str]] = [] for it in items: code = (getattr(it, "product_code", None) or "").strip() if not code: raise ValueError("产品编码不能为空") if code in codes: raise ValueError(f"产品编码 {code} 重复") codes.append(code) sid = (getattr(it, "storage_slot_id", None) or "").strip() or None if require_storage_slot and not sid: raise ValueError("每份产品须选择储物格位置") slot_ids.append(sid) occupancy_svc.assert_batch_slots_unique(slot_ids) existing_codes = ( db.query(BizCellProduct.product_code) .filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.product_code.in_(codes), BizCellProduct.del_flag == 0, ) .all() ) if existing_codes: dup = existing_codes[0][0] raise ValueError(f"产品编码 {dup} 已存在") return codes, slot_ids def create_cell_products_by_quantity( db: Session, *, tenant_no: str, user_id: Optional[str] = None, create_dept_id: Optional[str] = None, data: Any = None, count: int, ) -> List[BizCellProduct]: """按入库份数拆分:基础编码-01、基础编码-02 …(须通过 items 为每份指定独立储物格)""" count = max(1, min(int(count), 50)) if count > 1: raise ValueError("入库多份时请通过入库明细为每份产品分别选择储物格") base = normalize_product_code_base(getattr(data, "product_code", None) or "") if not base: raise ValueError("产品编码不能为空") codes = build_split_product_codes(base, count) slot_id = (getattr(data, "storage_slot_id", None) or "").strip() or None from types import SimpleNamespace items = [ SimpleNamespace(product_code=c, storage_slot_id=slot_id) for c in codes ] return create_cell_products_batch( db, tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id, data=data, items=items, require_storage_slot=False, ) def create_cell_products_batch( db: Session, *, tenant_no: str, user_id: Optional[str] = None, create_dept_id: Optional[str] = None, data: Any = None, items: List[Any], require_storage_slot: bool = True, ) -> List[BizCellProduct]: codes, slot_ids = _validate_batch_items( db, tenant_no, items, require_storage_slot=require_storage_slot, ) for slot_id in slot_ids: if slot_id: occupancy_svc.assert_slot_vacant(db, slot_id, tenant_no=tenant_no) now = datetime.now() created: List[BizCellProduct] = [] for it, slot_id in zip(items, slot_ids): code = (getattr(it, "product_code", None) or "").strip() item_passage = (getattr(it, "passage", None) or "").strip() or None if not item_passage and getattr(data, "passage", None): item_passage = (data.passage or "").strip() or None item_spec = (getattr(it, "spec_name", None) or "").strip() or None if not item_spec and getattr(data, "spec_name", None): item_spec = (data.spec_name or "").strip() or None item_expected = getattr(it, "expected_cell_count", None) if item_expected is None: item_expected = data.expected_cell_count if hasattr(data, "expected_cell_count") else None item_viability = getattr(it, "viability", None) if item_viability is None: item_viability = data.viability if hasattr(data, "viability") else None obj = BizCellProduct( id=generate_uuid(), product_code=code, product_name=(data.product_name or "").strip(), product_type_id=(data.product_type_id or "").strip() or None, spec_id=(data.spec_id or "").strip() or None, spec_text=item_spec, batch_no=(data.batch_no or "").strip() or None, storage_slot_id=slot_id, passage=item_passage, freeze_date=data.freeze_date if hasattr(data, "freeze_date") else None, expected_cell_count=item_expected, viability=item_viability, product_status=data.product_status or "frozen", source_type=data.source_type or "manual", quantity=1, unit=(data.unit or "").strip() or None, expiry_date=data.expiry_date, product_images=(data.product_images or "").strip() or None, tenant_no=tenant_no, order_num=0, remark=(data.remark or "").strip() or None, create_by=user_id, create_dept_id=create_dept_id or "", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(obj) created.append(obj) db.flush() for obj in created: slot_id = (obj.storage_slot_id or "").strip() if slot_id and (obj.product_status or "") in _CELL_PRODUCT_STORE_STATUSES: occupancy_svc._add_stored_occupancy_no_commit( db, slot_id=slot_id, item_type="cell_product", item_id=obj.id, item_code=obj.product_code, item_name=obj.product_name, item_source=obj.source_type or "manual", tenant_no=tenant_no, user_id=user_id, create_dept_id=create_dept_id or "", stored_at=now, ) db.commit() for obj in created: db.refresh(obj) return created def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size))) # --------------------------------------------------------------------------- # 来源链路 API # --------------------------------------------------------------------------- _MAX_CHAIN_DEPTH = 10 _SOURCE_TYPE_LABELS = {"production": "生产入库", "manual": "手动录入"} def _product_node(obj: BizCellProduct, is_current: bool = False) -> Dict[str, Any]: return { "id": obj.id, "productCode": obj.product_code or "", "productName": obj.product_name or "", "nodeType": "cell_product", "productStatus": obj.product_status or "", "passage": obj.passage or "", "isCurrent": is_current, } def _nest_ancestor_parent(chain: List[Dict[str, Any]]) -> Any: """多进一时 parent 为数组;单链时为对象;无上游为 null。""" if not chain: return None if len(chain) == 1: return chain[0] return chain _WORK_ORDER_TYPE_LABELS = { "inbound": "入库生产", "temporary": "临时生产", "outbound": "出库生产", } def _edge_meta(task: Any, wo: Any = None) -> Dict[str, Any]: edge: Dict[str, Any] = { "workOrderNo": "", "workOrderType": "", "workOrderTypeLabel": "", "taskNo": "", "taskName": "", "taskId": "", } if task: edge["taskNo"] = task.task_no or "" edge["taskName"] = task.task_name or "" edge["taskId"] = task.id or "" if wo: edge["workOrderNo"] = wo.work_order_no or "" wo_type = getattr(wo, "work_order_type", None) or "inbound" edge["workOrderType"] = wo_type edge["workOrderTypeLabel"] = _WORK_ORDER_TYPE_LABELS.get(wo_type, wo_type) return edge _SOURCE_LABELS = {"cell_storage": "细胞存储", "research_donation": "科研捐赠"} def _sample_customer_dict(db: Session, protocol_id: str, tenant_no: str) -> Dict[str, Any]: if not protocol_id: return {} from app.models.models import BizProtocolPartyCustomer pc = ( db.query(BizProtocolPartyCustomer) .filter( BizProtocolPartyCustomer.collection_protocol_id == protocol_id, BizProtocolPartyCustomer.tenant_no == tenant_no, BizProtocolPartyCustomer.del_flag == 0, ) .first() ) if not pc: return {} return { "motherName": pc.mother_name or "", "motherPhone": pc.mother_phone or "", "motherIdCard": pc.mother_id_card or "", "newbornName": pc.newborn_name or "", } def _donation_sample_customer_dict(db: Session, donation_id: str, tenant_no: str) -> Dict[str, Any]: if not donation_id: return {} from app.models.research_donation import BizResearchDonationParty party = ( db.query(BizResearchDonationParty) .filter( BizResearchDonationParty.research_donation_id == donation_id, BizResearchDonationParty.tenant_no == tenant_no, BizResearchDonationParty.del_flag == 0, ) .first() ) if not party: return {} return { "motherName": party.mother_name or "", "motherPhone": party.mother_phone or "", "motherIdCard": party.mother_id_card or "", "newbornName": party.newborn_name or "", } def _sample_type_name_from_id(db: Session, sample_type_id: Optional[str]) -> str: if not sample_type_id: return "" from app.models.models import BizCellSampleType row = ( db.query(BizCellSampleType.type_name) .filter(BizCellSampleType.id == sample_type_id, BizCellSampleType.del_flag == 0) .first() ) return (row[0] if row else "") or "" def _find_protocol_sample( db: Session, tenant_no: str, input_id: str, input_code: str, ) -> Optional[Any]: from app.models.models import BizProtocolSample sample = None if input_id: sample = ( db.query(BizProtocolSample) .filter( BizProtocolSample.id == input_id, BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, ) .first() ) if not sample and input_code: sample = ( db.query(BizProtocolSample) .filter( BizProtocolSample.sample_code == input_code, BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, ) .first() ) return sample def _find_donation_sample( db: Session, tenant_no: str, input_id: str, input_code: str, wo: Any = None, ) -> Optional[Any]: from app.models.research_donation import BizResearchDonationSample from app.crud.sample_reception import _donation_sample_id_col sample = None if input_id: sample = ( db.query(BizResearchDonationSample) .filter( _donation_sample_id_col(BizResearchDonationSample.id) == input_id, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, ) .first() ) if not sample and input_code: sample = ( db.query(BizResearchDonationSample) .filter( BizResearchDonationSample.sample_code == input_code, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, ) .first() ) if not sample and wo and (getattr(wo, "reception_source", None) or "") == "research_donation": wo_sample_id = (getattr(wo, "protocol_sample_id", None) or "").strip() if wo_sample_id: sample = ( db.query(BizResearchDonationSample) .filter( _donation_sample_id_col(BizResearchDonationSample.id) == wo_sample_id, BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, ) .first() ) return sample def _protocol_sample_chain_node( db: Session, tenant_no: str, sample: Any, *, wo: Any, edge: Dict[str, Any], sample_type_name: str, ) -> Dict[str, Any]: reception_source = getattr(wo, "reception_source", None) if wo else "" reception_source = reception_source or "cell_storage" protocol_id = sample.collection_protocol_id or "" return { "node": { "id": sample.id, "productCode": sample.sample_code or "", "productName": "", "nodeType": "sample", "sampleCode": sample.sample_code or "", "sampleTypeName": sample_type_name, "protocolNo": wo.protocol_no if wo else "", "collectionProtocolId": protocol_id, "receptionSource": reception_source, "receptionSourceLabel": _SOURCE_LABELS.get(reception_source, reception_source), "customer": _sample_customer_dict(db, protocol_id, tenant_no), "isCurrent": False, **sample_special_to_dict(sample), }, "edge": edge, "parent": None, } def _donation_sample_chain_node( db: Session, tenant_no: str, sample: Any, *, wo: Any, edge: Dict[str, Any], sample_type_name: str, ) -> Dict[str, Any]: donation_id = sample.research_donation_id or "" type_name = sample_type_name or _sample_type_name_from_id(db, sample.sample_type_id) return { "node": { "id": sample.id, "productCode": sample.sample_code or "", "productName": "", "nodeType": "sample", "sampleCode": sample.sample_code or "", "sampleTypeName": type_name, "protocolNo": wo.protocol_no if wo else "", "collectionProtocolId": donation_id, "receptionSource": "research_donation", "receptionSourceLabel": _SOURCE_LABELS["research_donation"], "customer": _donation_sample_customer_dict(db, donation_id, tenant_no), "isCurrent": False, **sample_special_to_dict(sample), }, "edge": edge, "parent": None, } def _ancestor_sample_chain_link( db: Session, tenant_no: str, sample_input: Any, wo: Any, edge: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """解析任务样本投入物,返回祖先链单段(样本终止节点)。""" input_id = (sample_input.input_id or "").strip() input_code = (sample_input.input_code or "").strip() if not input_id and not input_code: if wo and (getattr(wo, "reception_source", None) or "") == "research_donation": input_code = (getattr(wo, "sample_code", None) or "").strip() input_id = (getattr(wo, "protocol_sample_id", None) or "").strip() if not input_id and not input_code: return None input_name = (sample_input.input_name or "").strip() reception_source = (getattr(wo, "reception_source", None) or "") if wo else "" donation_sample = None if reception_source == "research_donation": donation_sample = _find_donation_sample(db, tenant_no, input_id, input_code, wo) if donation_sample: return _donation_sample_chain_node( db, tenant_no, donation_sample, wo=wo, edge=edge, sample_type_name=input_name, ) protocol_sample = _find_protocol_sample(db, tenant_no, input_id, input_code) if protocol_sample: return _protocol_sample_chain_node( db, tenant_no, protocol_sample, wo=wo, edge=edge, sample_type_name=input_name or _sample_type_name_from_id(db, protocol_sample.sample_type_id), ) if reception_source != "research_donation": donation_sample = _find_donation_sample(db, tenant_no, input_id, input_code, wo) if donation_sample: return _donation_sample_chain_node( db, tenant_no, donation_sample, wo=wo, edge=edge, sample_type_name=input_name, ) return None def _find_cell_product_by_input(db: Session, inp: Any) -> Optional[BizCellProduct]: if not (inp.input_id or inp.input_code): return None if inp.input_id: product = ( db.query(BizCellProduct) .filter(BizCellProduct.id == inp.input_id, BizCellProduct.del_flag == 0) .first() ) if product: return product if inp.input_code: return ( db.query(BizCellProduct) .filter(BizCellProduct.product_code == inp.input_code, BizCellProduct.del_flag == 0) .first() ) return None def _ancestor_link_from_cell_product_input( db: Session, tenant_no: str, inp: Any, edge: Dict[str, Any], depth: int, ) -> Optional[Dict[str, Any]]: parent_product = _find_cell_product_by_input(db, inp) if not parent_product: return None parent_chain = _build_ancestor_chain(db, parent_product, depth + 1, tenant_no) return { "node": _product_node(parent_product), "edge": edge, "parent": _nest_ancestor_parent(parent_chain), } def _build_ancestor_chain( db: Session, product: BizCellProduct, depth: int, tenant_no: str ) -> List[Dict[str, Any]]: if depth >= _MAX_CHAIN_DEPTH: return [] task_id = product.production_task_id if not task_id: return [] from app.models.models import BizProductionTask, BizProductionWorkOrder from app.models.models import BizProductionTaskInput task = ( db.query(BizProductionTask) .filter(BizProductionTask.id == task_id, BizProductionTask.del_flag == 0) .first() ) if not task: return [] wo = None if task.work_order_id: wo = ( db.query(BizProductionWorkOrder) .filter(BizProductionWorkOrder.id == task.work_order_id, BizProductionWorkOrder.del_flag == 0) .first() ) edge = _edge_meta(task, wo) inputs = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.task_id == task_id, BizProductionTaskInput.input_type.in_(("cell_product", "sample")), BizProductionTaskInput.del_flag == 0, ) .order_by( BizProductionTaskInput.order_num.asc(), BizProductionTaskInput.create_time.asc(), ) .all() ) if not inputs: return [] links: List[Dict[str, Any]] = [] seen_node_ids: set = set() for inp in inputs: link: Optional[Dict[str, Any]] = None if inp.input_type == "cell_product": link = _ancestor_link_from_cell_product_input( db, tenant_no, inp, edge, depth + 1 ) elif inp.input_type == "sample": link = _ancestor_sample_chain_link(db, tenant_no, inp, wo, edge) if not link: continue node_id = (link.get("node") or {}).get("id") or "" if node_id and node_id in seen_node_ids: continue if node_id: seen_node_ids.add(node_id) links.append(link) return links def _build_descendant_chain( db: Session, product_id: str, product_code: str, depth: int, tenant_no: str ) -> List[Dict[str, Any]]: if depth >= _MAX_CHAIN_DEPTH: return [] from app.models.models import BizProductionTaskInput, BizProductionTaskOutput, BizProductionTask, BizProductionWorkOrder # 找到引用此产品作为投入物的所有任务(id 或编码) input_match = [BizProductionTaskInput.input_id == product_id] if product_code: input_match.append(BizProductionTaskInput.input_code == product_code) input_rows = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.input_type == "cell_product", BizProductionTaskInput.del_flag == 0, or_(*input_match), ) .all() ) if not input_rows: return [] task_ids = list({r.task_id for r in input_rows}) tasks = ( db.query(BizProductionTask) .filter(BizProductionTask.id.in_(task_ids), BizProductionTask.del_flag == 0) .all() ) task_by_id = {t.id: t for t in tasks} wo_ids = list({t.work_order_id for t in tasks if t.work_order_id}) wo_by_id: Dict[str, Any] = {} if wo_ids: wos = ( db.query(BizProductionWorkOrder) .filter(BizProductionWorkOrder.id.in_(wo_ids), BizProductionWorkOrder.del_flag == 0) .all() ) wo_by_id = {w.id: w for w in wos} # 对每个任务,找产出 result: List[Dict[str, Any]] = [] output_rows = ( db.query(BizProductionTaskOutput) .filter( BizProductionTaskOutput.task_id.in_(task_ids), BizProductionTaskOutput.del_flag == 0, ) .all() ) child_product_ids = list({o.cell_product_id for o in output_rows if o.cell_product_id}) child_products = {} if child_product_ids: cps = ( db.query(BizCellProduct) .filter(BizCellProduct.id.in_(child_product_ids), BizCellProduct.del_flag == 0) .all() ) child_products = {cp.id: cp for cp in cps} seen_child_ids: set = set() for o in output_rows: cp = child_products.get(o.cell_product_id) if not cp or cp.id in seen_child_ids: continue seen_child_ids.add(cp.id) t = task_by_id.get(o.task_id) wo_obj = wo_by_id.get(t.work_order_id) if t and t.work_order_id else None edge = _edge_meta(t, wo_obj) result.append({ "node": _product_node(cp), "edge": edge, "children": _build_descendant_chain( db, cp.id, cp.product_code or "", depth + 1, tenant_no ), }) return result def _lineage_node_key(node: Dict[str, Any]) -> str: nt = (node.get("nodeType") or "").strip() nid = (node.get("id") or "").strip() if nt == "sample": code = (node.get("sampleCode") or "").strip() return f"sample:{(nid or code)}" return f"{nt}:{nid}" def _walk_ancestor_link( link: Any, out: List[Dict[str, Any]], seen: set, ) -> None: if not link: return if isinstance(link, list): for item in link: _walk_ancestor_link(item, out, seen) return if not isinstance(link, dict): return parent = link.get("parent") _walk_ancestor_link(parent, out, seen) node = link.get("node") if not isinstance(node, dict): return key = _lineage_node_key(node) if not key or key in seen: return seen.add(key) out.append(dict(node)) def _flatten_lineage_nodes(chain: Dict[str, Any]) -> List[Dict[str, Any]]: """将 get_product_chain 结果展平为上游→下游节点列表(含 current,不含 descendants)。""" out: List[Dict[str, Any]] = [] seen: set = set() for link in chain.get("ancestors") or []: _walk_ancestor_link(link, out, seen) current = chain.get("current") if isinstance(current, dict): key = _lineage_node_key(current) if key and key not in seen: seen.add(key) node = dict(current) node["isCurrent"] = True out.append(node) elif key: for n in out: if _lineage_node_key(n) == key: n["isCurrent"] = True break return out def get_product_chain(db: Session, product_id: str, tenant_no: str) -> Optional[Dict[str, Any]]: product = get_cell_product(db, product_id, tenant_no) if not product: return None return { "current": _product_node(product, is_current=True), "ancestors": _build_ancestor_chain(db, product, 0, tenant_no), "descendants": _build_descendant_chain( db, product.id, product.product_code or "", 0, tenant_no ), } def _wo_status_label(status: str) -> str: from app.crud.qc_work_order import _WORK_ORDER_STATUS_LABELS return _WORK_ORDER_STATUS_LABELS.get(status or "", status or "") def _slim_task_for_lineage( task: Dict[str, Any], results: List[Dict[str, Any]] ) -> Dict[str, Any]: return { "id": task.get("id") or "", "taskNo": task.get("taskNo") or "", "itemName": task.get("itemName") or "", "itemCode": task.get("itemCode") or "", "taskType": task.get("taskType") or "", "taskStatus": task.get("taskStatus") or "", "taskStatusLabel": task.get("taskStatusLabel") or "", "conclusion": task.get("conclusion") or "", "conclusionLabel": task.get("conclusionLabel") or "", "remark": task.get("remark") or "", "completedTime": task.get("completedTime") or "", "companyName": task.get("companyName") or "", "results": [ { "paramName": r.get("paramName") or "", "paramValue": r.get("paramValue") or "", "paramUnit": r.get("paramUnit") or "", } for r in (results or []) ], "attachmentFiles": task.get("attachmentFiles") or [], } def _load_lineage_qc_for_nodes( db: Session, nodes: List[Dict[str, Any]], tenant_no: str ) -> List[Dict[str, Any]]: from app.models.models import BizQcWorkOrder from app.crud import qc_task as task_svc from app.crud import qc_task_result as result_svc from app.crud.qc_work_order import list_work_orders_by_sample product_ids = [ (n.get("id") or "").strip() for n in nodes if (n.get("nodeType") or "") == "cell_product" and (n.get("id") or "").strip() ] wo_by_product: Dict[str, List[BizQcWorkOrder]] = {pid: [] for pid in product_ids} if product_ids: rows = ( db.query(BizQcWorkOrder) .filter( BizQcWorkOrder.cell_product_id.in_(product_ids), BizQcWorkOrder.tenant_no == tenant_no, BizQcWorkOrder.del_flag == 0, ) .order_by(BizQcWorkOrder.create_time.desc()) .all() ) for wo in rows: pid = (wo.cell_product_id or "").strip() if pid in wo_by_product: wo_by_product[pid].append(wo) sample_wo_cache: Dict[Tuple[str, str], List[Dict[str, Any]]] = {} node_work_orders_orm: List[List[BizQcWorkOrder]] = [] for n in nodes: if (n.get("nodeType") or "") == "cell_product": node_work_orders_orm.append(wo_by_product.get((n.get("id") or "").strip(), [])) else: node_work_orders_orm.append([]) sample_wo_dicts: Dict[int, List[Dict[str, Any]]] = {} for idx, n in enumerate(nodes): if (n.get("nodeType") or "") != "sample": continue code = (n.get("sampleCode") or "").strip() rs = (n.get("receptionSource") or "").strip() if not code or not rs: sample_wo_dicts[idx] = [] continue key = (code, rs) if key not in sample_wo_cache: sample_wo_cache[key] = list_work_orders_by_sample( db, sample_code=code, reception_source=rs, tenant_no=tenant_no, user_info=None, ) sample_wo_dicts[idx] = sample_wo_cache[key] pending_wo_ids: List[str] = [] for wos in node_work_orders_orm: for wo in wos: pending_wo_ids.append(wo.id) for wos in sample_wo_dicts.values(): for wo in wos: wid = (wo.get("id") or "").strip() if wid: pending_wo_ids.append(wid) unique_wo_ids = list({(x or "").strip() for x in pending_wo_ids if (x or "").strip()}) tasks_by_wo: Dict[str, List[Dict[str, Any]]] = {} for i in range(0, len(unique_wo_ids), 100): chunk = unique_wo_ids[i : i + 100] chunk_tasks = task_svc.list_tasks_by_work_order_ids(db, chunk, tenant_no) for wo_id, tasks in chunk_tasks.items(): tasks_by_wo.setdefault(wo_id, []).extend(tasks) all_task_ids = [ (t.get("id") or "").strip() for tasks in tasks_by_wo.values() for t in tasks if (t.get("id") or "").strip() ] results_by_task = result_svc.list_saved_results_by_task_ids(db, all_task_ids, tenant_no) def pack_orm_wos(wos: List[BizQcWorkOrder]) -> List[Dict[str, Any]]: packed = [] for wo in wos: tasks = tasks_by_wo.get(wo.id, []) packed.append( { "id": wo.id, "workOrderNo": wo.work_order_no or "", "workOrderType": wo.work_order_type or "", "workOrderStatus": wo.work_order_status or "", "workOrderStatusLabel": _wo_status_label(wo.work_order_status or ""), "tasks": [ _slim_task_for_lineage( t, results_by_task.get(t.get("id") or "", []) ) for t in tasks ], } ) return packed def pack_dict_wos(wos: List[Dict[str, Any]]) -> List[Dict[str, Any]]: packed = [] for wo in wos: wid = (wo.get("id") or "").strip() tasks = tasks_by_wo.get(wid, []) packed.append( { "id": wid, "workOrderNo": wo.get("workOrderNo") or "", "workOrderType": wo.get("workOrderType") or "", "workOrderStatus": wo.get("workOrderStatus") or "", "workOrderStatusLabel": wo.get("workOrderStatusLabel") or _wo_status_label(wo.get("workOrderStatus") or ""), "tasks": [ _slim_task_for_lineage( t, results_by_task.get(t.get("id") or "", []) ) for t in tasks ], } ) return packed out_nodes: List[Dict[str, Any]] = [] for idx, n in enumerate(nodes): nt = n.get("nodeType") or "" if nt == "sample": label = (n.get("sampleCode") or n.get("id") or "").strip() work_orders = pack_dict_wos(sample_wo_dicts.get(idx, [])) else: label = (n.get("productCode") or n.get("id") or "").strip() work_orders = pack_orm_wos(node_work_orders_orm[idx]) out_nodes.append( { "nodeType": nt, "id": n.get("id") or "", "label": label, "productName": n.get("productName") or "", "passage": n.get("passage") or None, "receptionSource": n.get("receptionSource") or None, "receptionSourceLabel": n.get("receptionSourceLabel") or None, "isCurrent": bool(n.get("isCurrent")), "workOrders": work_orders, } ) return out_nodes def get_product_lineage_qc( db: Session, product_id: str, tenant_no: str ) -> Optional[Dict[str, Any]]: chain = get_product_chain(db, product_id, tenant_no) if not chain: return None current = chain.get("current") or {} flat = _flatten_lineage_nodes(chain) nodes = _load_lineage_qc_for_nodes(db, flat, tenant_no) return { "productId": current.get("id") or product_id, "productCode": current.get("productCode") or "", "productName": current.get("productName") or "", "nodes": nodes, }