"""样本编码全链路纠错:预览影响面 + 事务级联更新。""" from __future__ import annotations import json from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.models.models import ( BizCellProduct, BizProductionTaskInput, BizProductionWorkOrder, BizProtocolPaymentSchedule, BizProtocolPaymentTransfer, BizProtocolSample, BizProtocolSamplePayment, BizProtocolSampleTransport, BizQcTask, BizQcWorkOrder, BizSampleCode, BizSampleCodeCorrectionLog, BizSampleCustomerService, BizSampleFreightItem, BizSampleInsurance, BizSampleProductionReception, BizSampleReception, BizSampleStorageCert, BizSampleStorageCertQc, BizStorageReleaseReview, ) from app.models.research_donation import ( BizResearchDonationSample, BizResearchDonationSampleTransport, ) SOURCE_CELL = "cell_storage" SOURCE_DONATION = "research_donation" VALID_SOURCES = {SOURCE_CELL, SOURCE_DONATION} # (key, label, model, code_attr, has_reception_source, cell_only) _TABLE_SPECS: List[Tuple[str, str, Any, str, bool, bool]] = [ ("biz_protocol_sample_payment", "按样本缴费", BizProtocolSamplePayment, "sample_code", False, True), ("biz_protocol_payment_schedule", "缴费期次", BizProtocolPaymentSchedule, "sample_code", False, True), ("biz_protocol_payment_transfer", "缴费转账", BizProtocolPaymentTransfer, "sample_code", False, True), ("biz_protocol_sample_transport", "协议样本运输", BizProtocolSampleTransport, "sample_code", False, True), ( "biz_research_donation_sample_transport", "捐赠样本运输", BizResearchDonationSampleTransport, "sample_code", False, False, ), ("biz_sample_freight_item", "运费明细", BizSampleFreightItem, "sample_code", False, False), ("biz_sample_reception", "样本接收", BizSampleReception, "sample_code", True, False), ("biz_qc_work_order", "质检工单", BizQcWorkOrder, "sample_code", True, False), ("biz_qc_task", "质检任务", BizQcTask, "sample_code", False, False), ("biz_storage_release_review", "入库放行", BizStorageReleaseReview, "sample_code", True, False), ("biz_sample_storage_cert_qc", "储存证质检", BizSampleStorageCertQc, "sample_code", False, True), ( "biz_sample_production_reception", "生产部门接收", BizSampleProductionReception, "sample_code", True, False, ), ("biz_production_work_order", "生产工单", BizProductionWorkOrder, "sample_code", True, False), ("biz_sample_storage_cert", "储存证", BizSampleStorageCert, "sample_code", False, True), ("biz_sample_insurance", "样本保险", BizSampleInsurance, "sample_code", False, True), ("biz_sample_customer_service", "客户服务扩展", BizSampleCustomerService, "sample_code", False, True), ] _UNIQUE_TABLES = { "biz_sample_storage_cert", "biz_sample_insurance", "biz_sample_customer_service", "biz_sample_storage_cert_qc", } class SampleCodeCorrectionError(ValueError): """业务校验失败(可直接返回给前端)。""" def _norm(code: Optional[str]) -> str: return (code or "").strip() def _now() -> datetime: return datetime.now() def _load_source_sample( db: Session, *, tenant_no: str, reception_source: str, sample_code: str, ): code = _norm(sample_code) if reception_source == SOURCE_CELL: return ( db.query(BizProtocolSample) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code == code, ) .first() ) return ( db.query(BizResearchDonationSample) .filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_code == code, ) .first() ) def _code_used_by_other_sample( db: Session, *, tenant_no: str, sample_code: str, exclude_sample_id: Optional[str], ) -> bool: code = _norm(sample_code) q1 = db.query(BizProtocolSample.id).filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code == code, ) q2 = db.query(BizResearchDonationSample.id).filter( BizResearchDonationSample.tenant_no == tenant_no, BizResearchDonationSample.del_flag == 0, BizResearchDonationSample.sample_code == code, ) if exclude_sample_id: q1 = q1.filter(BizProtocolSample.id != exclude_sample_id) q2 = q2.filter(BizResearchDonationSample.id != exclude_sample_id) return q1.first() is not None or q2.first() is not None def _count_by_code( db: Session, model, *, tenant_no: str, code: str, code_attr: str = "sample_code", reception_source: Optional[str] = None, has_reception_source: bool = False, ) -> int: col = getattr(model, code_attr) q = db.query(model).filter( model.tenant_no == tenant_no, model.del_flag == 0, col == code, ) if has_reception_source and reception_source: q = q.filter(model.reception_source == reception_source) return int(q.count()) def _update_by_code( db: Session, model, *, tenant_no: str, old_code: str, new_code: str, code_attr: str = "sample_code", reception_source: Optional[str] = None, has_reception_source: bool = False, user_id: Optional[str] = None, ) -> int: col = getattr(model, code_attr) q = db.query(model).filter( model.tenant_no == tenant_no, model.del_flag == 0, col == old_code, ) if has_reception_source and reception_source: q = q.filter(model.reception_source == reception_source) rows = q.all() now = _now() for row in rows: setattr(row, code_attr, new_code) if hasattr(row, "update_by"): row.update_by = user_id if hasattr(row, "update_time"): row.update_time = now return len(rows) def _count_task_inputs(db: Session, tenant_no: str, code: str) -> int: return int( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.tenant_no == tenant_no, BizProductionTaskInput.del_flag == 0, BizProductionTaskInput.input_type == "sample", BizProductionTaskInput.input_code == code, ) .count() ) def _update_task_inputs( db: Session, *, tenant_no: str, old_code: str, new_code: str, user_id: Optional[str], ) -> int: rows = ( db.query(BizProductionTaskInput) .filter( BizProductionTaskInput.tenant_no == tenant_no, BizProductionTaskInput.del_flag == 0, BizProductionTaskInput.input_type == "sample", BizProductionTaskInput.input_code == old_code, ) .all() ) now = _now() for row in rows: row.input_code = new_code row.update_by = user_id row.update_time = now return len(rows) def _count_same_product_code(db: Session, tenant_no: str, code: str) -> int: return int( db.query(BizCellProduct) .filter( BizCellProduct.tenant_no == tenant_no, BizCellProduct.del_flag == 0, BizCellProduct.product_code == code, ) .count() ) def _unique_conflict( db: Session, model, *, tenant_no: str, new_code: str, ) -> bool: return ( db.query(model.id) .filter( model.tenant_no == tenant_no, model.del_flag == 0, model.sample_code == new_code, ) .first() is not None ) def _spec_applies(reception_source: str, cell_only: bool, table_key: str) -> bool: if reception_source == SOURCE_CELL: if table_key == "biz_research_donation_sample_transport": return False return True # donation if cell_only: return False if table_key.startswith("biz_protocol_"): return False return True def preview_correction( db: Session, *, tenant_no: str, reception_source: str, old_sample_code: str, new_sample_code: str, new_sample_code_id: Optional[str] = None, ) -> Dict[str, Any]: source = _norm(reception_source) or SOURCE_CELL old_code = _norm(old_sample_code) new_code = _norm(new_sample_code) blockers: List[str] = [] if source not in VALID_SOURCES: blockers.append("receptionSource 必须是 cell_storage 或 research_donation") if not old_code: blockers.append("旧样本编码不能为空") if not new_code: blockers.append("新样本编码不能为空") if old_code and new_code and old_code == new_code: blockers.append("新样本编码不能与旧编码相同") sample = None if not blockers: sample = _load_source_sample( db, tenant_no=tenant_no, reception_source=source, sample_code=old_code ) if sample is None: blockers.append(f"未找到来源「{source}」下样本编码「{old_code}」") elif _code_used_by_other_sample( db, tenant_no=tenant_no, sample_code=new_code, exclude_sample_id=sample.id, ): blockers.append(f"样本编码「{new_code}」已被其它协议/捐赠样本占用") impacts: List[Dict[str, Any]] = [] if sample is not None and not blockers: sot_key = ( "biz_protocol_sample" if source == SOURCE_CELL else "biz_research_donation_sample" ) impacts.append( { "table": sot_key, "label": "样本赋值源表", "count": 1, } ) pool_count = 0 if source == SOURCE_CELL: pool_id = _norm(new_sample_code_id) or _norm(getattr(sample, "sample_code_id", None)) if pool_id: pool_row = ( db.query(BizSampleCode) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.id == pool_id, ) .first() ) if pool_row and _norm(pool_row.sample_code) == old_code: pool_count = 1 elif not pool_row and _norm(new_sample_code_id): blockers.append(f"指定的新编码池 ID「{new_sample_code_id}」不存在") existing_new_pool = ( db.query(BizSampleCode) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.sample_code == new_code, ) .first() ) if existing_new_pool and pool_count == 0: impacts.append( { "table": "biz_sample_code", "label": "编码池(切换占用)", "count": 1, } ) elif pool_count: impacts.append( { "table": "biz_sample_code", "label": "编码池(就地改码)", "count": pool_count, } ) for key, label, model, attr, has_rs, cell_only in _TABLE_SPECS: if not _spec_applies(source, cell_only, key): continue cnt = _count_by_code( db, model, tenant_no=tenant_no, code=old_code, code_attr=attr, reception_source=source if has_rs else None, has_reception_source=has_rs, ) if cnt: impacts.append({"table": key, "label": label, "count": cnt}) if ( key in _UNIQUE_TABLES and source == SOURCE_CELL and cnt > 0 and _unique_conflict(db, model, tenant_no=tenant_no, new_code=new_code) ): blockers.append( f"表 {key} 已存在样本编码「{new_code}」记录,无法直接改码" ) ti = _count_task_inputs(db, tenant_no, old_code) if ti: impacts.append( { "table": "biz_production_task_input", "label": "生产任务投入物(样本)", "count": ti, } ) same_product_old = ( _count_same_product_code(db, tenant_no, old_code) if old_code else 0 ) same_product_new = ( _count_same_product_code(db, tenant_no, new_code) if new_code else 0 ) total = sum(int(i.get("count") or 0) for i in impacts) return { "ok": len(blockers) == 0, "blockers": blockers, "receptionSource": source, "oldSampleCode": old_code, "newSampleCode": new_code, "sampleId": getattr(sample, "id", None) if sample else None, "impacts": impacts, "totalAffectedRows": total, "hints": { "sameProductCodeAsOld": same_product_old, "sameProductCodeAsNew": same_product_new, "labStorageCodeUnchanged": True, "cellProductCodeNotAutoUpdated": True, }, } def _resolve_pool( db: Session, *, tenant_no: str, sample: BizProtocolSample, old_code: str, new_code: str, new_sample_code_id: Optional[str], user_id: Optional[str], ) -> Dict[str, Any]: """返回 pool 处理结果摘要。""" now = _now() result = {"action": "none", "poolId": None} preferred_id = _norm(new_sample_code_id) existing_new = ( db.query(BizSampleCode) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.sample_code == new_code, ) .first() ) if preferred_id: preferred = ( db.query(BizSampleCode) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.id == preferred_id, ) .first() ) if preferred is None: raise SampleCodeCorrectionError(f"指定的新编码池 ID「{preferred_id}」不存在") if _norm(preferred.sample_code) not in {new_code, old_code}: # 允许指向已是新码的池行,或指向即将就地改码的旧池行 if _norm(preferred.sample_code) != new_code: preferred.sample_code = new_code preferred.update_by = user_id preferred.update_time = now result["action"] = "rename_preferred" existing_new = preferred old_pool_id = _norm(getattr(sample, "sample_code_id", None)) old_pool = None if old_pool_id: old_pool = ( db.query(BizSampleCode) .filter( BizSampleCode.tenant_no == tenant_no, BizSampleCode.del_flag == 0, BizSampleCode.id == old_pool_id, ) .first() ) if existing_new is not None: sample.sample_code_id = existing_new.id sample.sample_code = new_code existing_new.issued = 1 existing_new.update_by = user_id existing_new.update_time = now result["action"] = result["action"] if result["action"] != "none" else "reassign" result["poolId"] = existing_new.id if ( old_pool and old_pool.id != existing_new.id and _norm(old_pool.sample_code) == old_code ): still_used = ( db.query(BizProtocolSample.id) .filter( BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, BizProtocolSample.sample_code_id == old_pool.id, BizProtocolSample.id != sample.id, ) .first() ) if still_used is None: old_pool.issued = 0 old_pool.update_by = user_id old_pool.update_time = now return result if old_pool and _norm(old_pool.sample_code) == old_code: old_pool.sample_code = new_code old_pool.issued = 1 old_pool.update_by = user_id old_pool.update_time = now sample.sample_code = new_code sample.sample_code_id = old_pool.id result["action"] = "rename_inplace" result["poolId"] = old_pool.id return result sample.sample_code = new_code result["action"] = "sample_text_only" return result def apply_correction( db: Session, *, tenant_no: str, reception_source: str, old_sample_code: str, new_sample_code: str, new_sample_code_id: Optional[str] = None, reason: Optional[str] = None, user_id: Optional[str] = None, ) -> Dict[str, Any]: preview = preview_correction( db, tenant_no=tenant_no, reception_source=reception_source, old_sample_code=old_sample_code, new_sample_code=new_sample_code, new_sample_code_id=new_sample_code_id, ) if not preview.get("ok"): raise SampleCodeCorrectionError( ";".join(preview.get("blockers") or ["校验未通过"]) ) source = preview["receptionSource"] old_code = preview["oldSampleCode"] new_code = preview["newSampleCode"] sample = _load_source_sample( db, tenant_no=tenant_no, reception_source=source, sample_code=old_code ) if sample is None: raise SampleCodeCorrectionError(f"未找到样本编码「{old_code}」") now = _now() impact_counts: Dict[str, int] = {} if source == SOURCE_CELL: pool_info = _resolve_pool( db, tenant_no=tenant_no, sample=sample, old_code=old_code, new_code=new_code, new_sample_code_id=new_sample_code_id, user_id=user_id, ) impact_counts["biz_sample_code"] = 1 if pool_info.get("action") != "none" else 0 sample.update_by = user_id sample.update_time = now impact_counts["biz_protocol_sample"] = 1 else: sample.sample_code = new_code sample.update_by = user_id sample.update_time = now impact_counts["biz_research_donation_sample"] = 1 # UNIQUE 表先更新,避免其它逻辑读到半新半旧 for key, _label, model, attr, has_rs, cell_only in _TABLE_SPECS: if key not in _UNIQUE_TABLES: continue if not _spec_applies(source, cell_only, key): continue n = _update_by_code( db, model, tenant_no=tenant_no, old_code=old_code, new_code=new_code, code_attr=attr, reception_source=source if has_rs else None, has_reception_source=has_rs, user_id=user_id, ) if n: impact_counts[key] = n for key, _label, model, attr, has_rs, cell_only in _TABLE_SPECS: if key in _UNIQUE_TABLES: continue if not _spec_applies(source, cell_only, key): continue n = _update_by_code( db, model, tenant_no=tenant_no, old_code=old_code, new_code=new_code, code_attr=attr, reception_source=source if has_rs else None, has_reception_source=has_rs, user_id=user_id, ) if n: impact_counts[key] = n ti = _update_task_inputs( db, tenant_no=tenant_no, old_code=old_code, new_code=new_code, user_id=user_id, ) if ti: impact_counts["biz_production_task_input"] = ti log = BizSampleCodeCorrectionLog( id=generate_uuid(), reception_source=source, old_sample_code=old_code, new_sample_code=new_code, impact_json=json.dumps(impact_counts, ensure_ascii=False), reason=_norm(reason) or None, tenant_no=tenant_no, order_num=0, create_by=user_id, create_dept_id="", create_time=now, update_by=user_id, update_time=now, del_flag=0, ) db.add(log) db.commit() return { "ok": True, "logId": log.id, "receptionSource": source, "oldSampleCode": old_code, "newSampleCode": new_code, "impactCounts": impact_counts, "totalAffectedRows": sum(impact_counts.values()), "hints": preview.get("hints") or {}, }