815 lines
28 KiB
Python
815 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.utils import generate_uuid
|
|
from app.models.models import (
|
|
BizCollectionProtocol,
|
|
BizProductionTask,
|
|
BizProductionWorkOrder,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolSample,
|
|
BizProtocolSampleTransport,
|
|
BizQcTask,
|
|
BizQcWorkOrder,
|
|
BizSampleReception,
|
|
BizStorageReleaseReview,
|
|
SysUser,
|
|
)
|
|
from app.models.research_donation import (
|
|
BizResearchDonation,
|
|
BizResearchDonationParty,
|
|
BizResearchDonationSample,
|
|
BizResearchDonationSampleTransport,
|
|
)
|
|
|
|
_RECEPTION_SOURCE_LABELS = {
|
|
"cell_storage": "细胞存储",
|
|
"research_donation": "科研捐赠",
|
|
}
|
|
_REVIEW_CONCLUSION_LABELS = {"tg": "通过", "btg": "不通过"}
|
|
_STORAGE_STATUS_LABELS = {"drk": "待入库", "yrk": "已入库"}
|
|
_PROTOCOL_STATUS_LABELS = {
|
|
"dsx": "待生效",
|
|
"sxz": "生效中",
|
|
"ygq": "已过期",
|
|
"yzf": "已作废",
|
|
"zky": "转科研",
|
|
}
|
|
_DONATION_STATUS_LABELS = {
|
|
"dsx": "待生效",
|
|
"sxz": "生效中",
|
|
"ygq": "已过期",
|
|
"yzf": "已作废",
|
|
"zckc": "转细胞存储",
|
|
}
|
|
_QC_WORK_ORDER_STATUS_LABELS = {"dpz": "待配置", "ypz": "已配置", "ywc": "已完成"}
|
|
_PRODUCTION_WORK_ORDER_STATUS_LABELS = {"dpz": "待配置", "jxz": "进行中", "ywc": "已完成"}
|
|
|
|
|
|
def _fmt_dt(t: Optional[datetime]) -> str:
|
|
return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""
|
|
|
|
|
|
def total_pages(total: int, page_size: int) -> int:
|
|
if total <= 0:
|
|
return 0
|
|
return int(math.ceil(total / float(page_size)))
|
|
|
|
|
|
def _row_to_dict(r: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
return {
|
|
"id": r.id,
|
|
"sampleCode": r.sample_code or "",
|
|
"receptionSource": r.reception_source or "",
|
|
"receptionSourceLabel": _RECEPTION_SOURCE_LABELS.get(r.reception_source or "", r.reception_source or ""),
|
|
"protocolId": r.protocol_id or "",
|
|
"receptionId": r.reception_id or "",
|
|
"protocolNo": r.protocol_no or "",
|
|
"sampleName": r.sample_name or "",
|
|
"reviewConclusion": r.review_conclusion or "",
|
|
"reviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(r.review_conclusion or "", r.review_conclusion or ""),
|
|
"reviewRemark": r.review_remark or "",
|
|
"reviewAttachments": r.review_attachments or "",
|
|
"reviewBy": r.review_by or "",
|
|
"reviewByName": r.review_by_name or "",
|
|
"reviewTime": _fmt_dt(r.review_time),
|
|
"storageStatus": r.storage_status or "drk",
|
|
"storageStatusLabel": _STORAGE_STATUS_LABELS.get(r.storage_status or "drk", r.storage_status or "drk"),
|
|
"storageBy": r.storage_by or "",
|
|
"storageByName": r.storage_by_name or "",
|
|
"storageTime": _fmt_dt(r.storage_time),
|
|
"remark": r.remark or "",
|
|
"createTime": _fmt_dt(r.create_time),
|
|
"updateTime": _fmt_dt(r.update_time),
|
|
}
|
|
|
|
|
|
def create_review_from_reception(
|
|
db: Session,
|
|
*,
|
|
reception: BizSampleReception,
|
|
protocol_no: str,
|
|
sample_name: str,
|
|
tenant_no: str,
|
|
user_id: Optional[str] = None,
|
|
create_dept_id: str = "",
|
|
) -> BizStorageReleaseReview:
|
|
source = getattr(reception, "reception_source", "cell_storage") or "cell_storage"
|
|
if source == "research_donation":
|
|
protocol_id = reception.research_donation_id or ""
|
|
else:
|
|
protocol_id = reception.collection_protocol_id or ""
|
|
|
|
obj = BizStorageReleaseReview(
|
|
id=generate_uuid(),
|
|
sample_code=reception.sample_code or "",
|
|
reception_source=source,
|
|
protocol_id=protocol_id,
|
|
reception_id=reception.id,
|
|
protocol_no=protocol_no or None,
|
|
sample_name=sample_name or None,
|
|
storage_status="drk",
|
|
tenant_no=tenant_no,
|
|
create_by=user_id,
|
|
create_dept_id=create_dept_id or "",
|
|
update_by=user_id,
|
|
create_time=datetime.now(),
|
|
update_time=datetime.now(),
|
|
del_flag=0,
|
|
)
|
|
db.add(obj)
|
|
db.flush()
|
|
return obj
|
|
|
|
|
|
def list_reviews(
|
|
db: Session,
|
|
*,
|
|
tenant_no: str,
|
|
page: int = 1,
|
|
page_size: int = 10,
|
|
reception_source: Optional[str] = None,
|
|
review_conclusion: Optional[str] = None,
|
|
storage_status: Optional[str] = None,
|
|
keyword: Optional[str] = None,
|
|
user_info=None,
|
|
) -> Dict[str, Any]:
|
|
q0 = db.query(BizStorageReleaseReview).filter(
|
|
BizStorageReleaseReview.tenant_no == tenant_no,
|
|
BizStorageReleaseReview.del_flag == 0,
|
|
)
|
|
rs = (reception_source or "").strip()
|
|
if rs:
|
|
q0 = q0.filter(BizStorageReleaseReview.reception_source == rs)
|
|
rc = (review_conclusion or "").strip()
|
|
if rc:
|
|
q0 = q0.filter(BizStorageReleaseReview.review_conclusion == rc)
|
|
ss = (storage_status or "").strip()
|
|
if ss:
|
|
q0 = q0.filter(BizStorageReleaseReview.storage_status == ss)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
q0 = q0.filter(
|
|
or_(
|
|
BizStorageReleaseReview.protocol_no.like(like),
|
|
BizStorageReleaseReview.sample_code.like(like),
|
|
BizStorageReleaseReview.sample_name.like(like),
|
|
)
|
|
)
|
|
if user_info:
|
|
from app.core.data_scope import apply_data_scope
|
|
q0 = apply_data_scope(q0, BizStorageReleaseReview, user_info, db, tenant_no)
|
|
total = q0.order_by(None).with_entities(func.count(BizStorageReleaseReview.id)).scalar() or 0
|
|
page = max(1, page)
|
|
page_size = min(max(1, page_size), 100)
|
|
rows = (
|
|
q0.order_by(BizStorageReleaseReview.create_time.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
.all()
|
|
)
|
|
items = _enrich_review_list_items(db, [_row_to_dict(r) for r in rows])
|
|
return {
|
|
"list": items,
|
|
"total": int(total),
|
|
"page": page,
|
|
"pageSize": page_size,
|
|
"totalPages": total_pages(int(total), page_size),
|
|
}
|
|
|
|
|
|
def _enrich_review_list_items(db: Session, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
if not items:
|
|
return items
|
|
from app.crud.qc_work_order import _enrich_work_order_list_items
|
|
|
|
items = _enrich_work_order_list_items(db, items)
|
|
|
|
cell_pids = [
|
|
x["protocolId"]
|
|
for x in items
|
|
if (x.get("receptionSource") or "cell_storage") != "research_donation" and x.get("protocolId")
|
|
]
|
|
donation_pids = [
|
|
x["protocolId"]
|
|
for x in items
|
|
if (x.get("receptionSource") or "") == "research_donation" and x.get("protocolId")
|
|
]
|
|
|
|
proto_map: Dict[str, BizCollectionProtocol] = {}
|
|
if cell_pids:
|
|
proto_map = {
|
|
p.id: p
|
|
for p in db.query(BizCollectionProtocol)
|
|
.filter(BizCollectionProtocol.id.in_(cell_pids), BizCollectionProtocol.del_flag == 0)
|
|
.all()
|
|
}
|
|
party_map: Dict[str, str] = {}
|
|
if cell_pids:
|
|
for party in (
|
|
db.query(BizProtocolPartyCustomer)
|
|
.filter(
|
|
BizProtocolPartyCustomer.collection_protocol_id.in_(cell_pids),
|
|
BizProtocolPartyCustomer.del_flag == 0,
|
|
)
|
|
.all()
|
|
):
|
|
party_map[party.collection_protocol_id] = party.mother_name or ""
|
|
|
|
donation_map: Dict[str, BizResearchDonation] = {}
|
|
if donation_pids:
|
|
donation_map = {
|
|
d.id: d
|
|
for d in db.query(BizResearchDonation)
|
|
.filter(BizResearchDonation.id.in_(donation_pids), BizResearchDonation.del_flag == 0)
|
|
.all()
|
|
}
|
|
dparty_map: Dict[str, str] = {}
|
|
if donation_pids:
|
|
for party in (
|
|
db.query(BizResearchDonationParty)
|
|
.filter(
|
|
BizResearchDonationParty.research_donation_id.in_(donation_pids),
|
|
BizResearchDonationParty.del_flag == 0,
|
|
)
|
|
.all()
|
|
):
|
|
dparty_map[party.research_donation_id] = party.mother_name or ""
|
|
|
|
for item in items:
|
|
source = item.get("receptionSource") or "cell_storage"
|
|
pid = item.get("protocolId") or ""
|
|
item.setdefault("motherName", "")
|
|
item.setdefault("protocolStatus", "")
|
|
item.setdefault("protocolStatusLabel", "")
|
|
if source == "research_donation":
|
|
donation = donation_map.get(pid)
|
|
if donation:
|
|
st = donation.donation_status or ""
|
|
item["protocolStatus"] = st
|
|
item["protocolStatusLabel"] = _DONATION_STATUS_LABELS.get(st, st)
|
|
item["motherName"] = dparty_map.get(pid, "")
|
|
else:
|
|
proto = proto_map.get(pid)
|
|
if proto:
|
|
st = getattr(proto, "protocol_status", "") or ""
|
|
item["protocolStatus"] = st
|
|
item["protocolStatusLabel"] = _PROTOCOL_STATUS_LABELS.get(st, st)
|
|
item["motherName"] = party_map.get(pid, "")
|
|
return items
|
|
|
|
|
|
def _get_user_name(db: Session, user_id: str) -> str:
|
|
u = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
return (u.nick_name or "") if u else ""
|
|
|
|
|
|
def get_review_detail(db: Session, review_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
|
|
rev = (
|
|
db.query(BizStorageReleaseReview)
|
|
.filter(
|
|
BizStorageReleaseReview.id == review_id,
|
|
BizStorageReleaseReview.tenant_no == tenant_no,
|
|
BizStorageReleaseReview.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not rev:
|
|
return None
|
|
|
|
result: Dict[str, Any] = {"review": _row_to_dict(rev)}
|
|
source = rev.reception_source or "cell_storage"
|
|
|
|
if source == "cell_storage":
|
|
protocol_info = _build_cell_storage_protocol_info(db, rev)
|
|
result["protocolInfo"] = protocol_info
|
|
transport_info = _build_cell_storage_transport_info(db, rev)
|
|
result["transportInfo"] = transport_info
|
|
else:
|
|
protocol_info = _build_research_donation_protocol_info(db, rev)
|
|
result["protocolInfo"] = protocol_info
|
|
transport_info = _build_research_donation_transport_info(db, rev)
|
|
result["transportInfo"] = transport_info
|
|
|
|
from app.crud import sample_reception as sample_reception_svc
|
|
|
|
reception_detail = sample_reception_svc.get_reception_by_id(
|
|
db, rev.reception_id or "", tenant_no
|
|
)
|
|
if reception_detail:
|
|
result["reception"] = reception_detail
|
|
result["receptionInfo"] = {
|
|
"receptionTime": reception_detail.get("receptionTime") or "",
|
|
"handlerName": reception_detail.get("handlerName") or "",
|
|
"receiveTemperature": reception_detail.get("receiveTemperature") or "",
|
|
"volumeMl": reception_detail.get("volumeMl"),
|
|
"netWeightG": reception_detail.get("netWeightG"),
|
|
"acceptConclusion": reception_detail.get("acceptConclusion") or "",
|
|
"acceptConclusionLabel": reception_detail.get("acceptConclusionLabel") or "",
|
|
"remark": reception_detail.get("remark") or "",
|
|
"checkResults": reception_detail.get("checkResults") or [],
|
|
"receiveAttachments": reception_detail.get("receiveAttachments") or "",
|
|
"receiveAttachmentFiles": reception_detail.get("receiveAttachmentFiles") or [],
|
|
"transportType": reception_detail.get("transportType") or "",
|
|
"expressCompanyName": reception_detail.get("expressCompanyName") or "",
|
|
"waybillNo": reception_detail.get("waybillNo") or "",
|
|
"carrierName": reception_detail.get("carrierName") or "",
|
|
"carrierPhone": reception_detail.get("carrierPhone") or "",
|
|
"etaTime": reception_detail.get("etaTime") or "",
|
|
}
|
|
else:
|
|
result["receptionInfo"] = _build_reception_info(db, rev)
|
|
|
|
qc_info = _build_qc_info(db, rev, tenant_no)
|
|
result["qcInfo"] = qc_info
|
|
|
|
production_info = _build_production_info(db, rev, tenant_no)
|
|
result["productionInfo"] = production_info
|
|
|
|
return result
|
|
|
|
|
|
def _build_cell_storage_protocol_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
proto = (
|
|
db.query(BizCollectionProtocol)
|
|
.filter(
|
|
BizCollectionProtocol.id == (rev.protocol_id or ""),
|
|
BizCollectionProtocol.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not proto:
|
|
return {}
|
|
sample = (
|
|
db.query(BizProtocolSample)
|
|
.filter(
|
|
BizProtocolSample.collection_protocol_id == proto.id,
|
|
BizProtocolSample.sample_code == (rev.sample_code or ""),
|
|
BizProtocolSample.tenant_no == rev.tenant_no,
|
|
BizProtocolSample.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
party = (
|
|
db.query(BizProtocolPartyCustomer)
|
|
.filter(
|
|
BizProtocolPartyCustomer.collection_protocol_id == proto.id,
|
|
BizProtocolPartyCustomer.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
info: Dict[str, Any] = {
|
|
"protocolNo": proto.protocol_no or "",
|
|
"protocolStatus": getattr(proto, "protocol_status", "") or "",
|
|
"motherName": "",
|
|
"motherPhone": "",
|
|
"fatherName": "",
|
|
"fatherPhone": "",
|
|
"newbornName": "",
|
|
"sampleCode": rev.sample_code or "",
|
|
"sampleTypeName": "",
|
|
"collectAmount": None,
|
|
"storageYears": None,
|
|
}
|
|
if sample:
|
|
from app.crud.qc_work_order import _load_sample_type_name_map
|
|
type_map = _load_sample_type_name_map(db, [sample.sample_type_id]) if sample.sample_type_id else {}
|
|
info["sampleTypeName"] = type_map.get(sample.sample_type_id or "", "")
|
|
info["collectAmount"] = float(sample.collect_amount) if sample.collect_amount else None
|
|
info["storageYears"] = sample.storage_years
|
|
if party:
|
|
info["motherName"] = party.mother_name or ""
|
|
info["motherPhone"] = party.mother_phone or ""
|
|
info["fatherName"] = party.father_name or ""
|
|
info["fatherPhone"] = party.father_phone or ""
|
|
info["newbornName"] = party.newborn_name or ""
|
|
return info
|
|
|
|
|
|
def _build_research_donation_protocol_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
donation = (
|
|
db.query(BizResearchDonation)
|
|
.filter(
|
|
BizResearchDonation.id == (rev.protocol_id or ""),
|
|
BizResearchDonation.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not donation:
|
|
return {}
|
|
dsample = (
|
|
db.query(BizResearchDonationSample)
|
|
.filter(
|
|
BizResearchDonationSample.research_donation_id == donation.id,
|
|
BizResearchDonationSample.sample_code == (rev.sample_code or ""),
|
|
BizResearchDonationSample.tenant_no == rev.tenant_no,
|
|
BizResearchDonationSample.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
party = (
|
|
db.query(BizResearchDonationParty)
|
|
.filter(
|
|
BizResearchDonationParty.research_donation_id == donation.id,
|
|
BizResearchDonationParty.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
info: Dict[str, Any] = {
|
|
"protocolNo": donation.donation_no or "",
|
|
"protocolStatus": donation.donation_status or "",
|
|
"motherName": "",
|
|
"motherPhone": "",
|
|
"fatherName": "",
|
|
"fatherPhone": "",
|
|
"newbornName": "",
|
|
"sampleCode": rev.sample_code or "",
|
|
"sampleTypeName": "",
|
|
"collectAmount": None,
|
|
"storageYears": None,
|
|
}
|
|
if dsample:
|
|
from app.crud.qc_work_order import _load_sample_type_name_map
|
|
type_map = _load_sample_type_name_map(db, [dsample.sample_type_id]) if dsample.sample_type_id else {}
|
|
info["sampleTypeName"] = type_map.get(dsample.sample_type_id or "", "")
|
|
info["collectAmount"] = float(dsample.collect_amount) if dsample.collect_amount else None
|
|
info["storageYears"] = dsample.storage_years
|
|
if party:
|
|
info["motherName"] = party.mother_name or ""
|
|
info["motherPhone"] = party.mother_phone or ""
|
|
info["fatherName"] = party.father_name or ""
|
|
info["fatherPhone"] = party.father_phone or ""
|
|
info["newbornName"] = party.newborn_name or ""
|
|
return info
|
|
|
|
|
|
def _build_cell_storage_transport_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
sample = (
|
|
db.query(BizProtocolSample)
|
|
.filter(
|
|
BizProtocolSample.sample_code == (rev.sample_code or ""),
|
|
BizProtocolSample.collection_protocol_id == (rev.protocol_id or ""),
|
|
BizProtocolSample.tenant_no == rev.tenant_no,
|
|
BizProtocolSample.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not sample:
|
|
return {}
|
|
transport = (
|
|
db.query(BizProtocolSampleTransport)
|
|
.filter(
|
|
BizProtocolSampleTransport.sample_id == sample.id,
|
|
BizProtocolSampleTransport.collection_protocol_id == (rev.protocol_id or ""),
|
|
BizProtocolSampleTransport.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not transport:
|
|
return {}
|
|
return {
|
|
"transportType": transport.transport_type or "",
|
|
"waybillNo": transport.waybill_no or "",
|
|
"expressFee": float(transport.express_fee) if transport.express_fee else None,
|
|
"etaTime": _fmt_dt(transport.eta_time),
|
|
"carrierName": transport.carrier_name or "",
|
|
"carrierPhone": transport.carrier_phone or "",
|
|
}
|
|
|
|
|
|
def _build_research_donation_transport_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
dsample = (
|
|
db.query(BizResearchDonationSample)
|
|
.filter(
|
|
BizResearchDonationSample.sample_code == (rev.sample_code or ""),
|
|
BizResearchDonationSample.research_donation_id == (rev.protocol_id or ""),
|
|
BizResearchDonationSample.tenant_no == rev.tenant_no,
|
|
BizResearchDonationSample.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not dsample:
|
|
return {}
|
|
transport = (
|
|
db.query(BizResearchDonationSampleTransport)
|
|
.filter(
|
|
BizResearchDonationSampleTransport.sample_id == dsample.id,
|
|
BizResearchDonationSampleTransport.research_donation_id == (rev.protocol_id or ""),
|
|
BizResearchDonationSampleTransport.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not transport:
|
|
return {}
|
|
return {
|
|
"transportType": transport.transport_type or "",
|
|
"waybillNo": transport.waybill_no or "",
|
|
"expressFee": float(transport.express_fee) if transport.express_fee else None,
|
|
"etaTime": _fmt_dt(transport.eta_time),
|
|
"carrierName": transport.carrier_name or "",
|
|
"carrierPhone": transport.carrier_phone or "",
|
|
}
|
|
|
|
|
|
def _build_reception_info(db: Session, rev: BizStorageReleaseReview) -> Dict[str, Any]:
|
|
rec = (
|
|
db.query(BizSampleReception)
|
|
.filter(
|
|
BizSampleReception.id == (rev.reception_id or ""),
|
|
BizSampleReception.tenant_no == rev.tenant_no,
|
|
BizSampleReception.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not rec:
|
|
return {}
|
|
return {
|
|
"receptionTime": _fmt_dt(rec.reception_time),
|
|
"handlerName": rec.handler_name or "",
|
|
"receiveTemperature": rec.receive_temperature or "",
|
|
"volumeMl": float(rec.volume_ml) if rec.volume_ml else None,
|
|
"netWeightG": float(rec.net_weight_g) if rec.net_weight_g else None,
|
|
"acceptConclusion": rec.accept_conclusion or "",
|
|
}
|
|
|
|
|
|
def _build_qc_info(db: Session, rev: BizStorageReleaseReview, tenant_no: str) -> Dict[str, Any]:
|
|
wo = (
|
|
db.query(BizQcWorkOrder)
|
|
.filter(
|
|
BizQcWorkOrder.sample_code == (rev.sample_code or ""),
|
|
BizQcWorkOrder.tenant_no == tenant_no,
|
|
BizQcWorkOrder.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not wo:
|
|
return {}
|
|
tasks = (
|
|
db.query(BizQcTask)
|
|
.filter(
|
|
BizQcTask.work_order_id == wo.id,
|
|
BizQcTask.tenant_no == tenant_no,
|
|
BizQcTask.del_flag == 0,
|
|
)
|
|
.all()
|
|
)
|
|
task_list = []
|
|
_task_type_labels = {"sysz": "实验室自检", "dsfjc": "第三方检测"}
|
|
_task_status_labels = {"dzx": "待执行", "jxz": "进行中", "ywc": "已完成"}
|
|
_conclusion_labels = {"tg": "通过", "btg": "不通过"}
|
|
for t in tasks:
|
|
task_list.append({
|
|
"id": t.id,
|
|
"taskType": t.task_type or "",
|
|
"taskTypeLabel": _task_type_labels.get(t.task_type or "", t.task_type or ""),
|
|
"taskStatus": t.task_status or "",
|
|
"taskStatusLabel": _task_status_labels.get(t.task_status or "", t.task_status or ""),
|
|
"conclusion": getattr(t, "conclusion", "") or "",
|
|
"conclusionLabel": _conclusion_labels.get(getattr(t, "conclusion", "") or "", getattr(t, "conclusion", "") or ""),
|
|
})
|
|
wo_status = wo.work_order_status or ""
|
|
return {
|
|
"workOrderNo": wo.work_order_no or "",
|
|
"workOrderStatus": wo_status,
|
|
"workOrderStatusLabel": _QC_WORK_ORDER_STATUS_LABELS.get(wo_status, wo_status),
|
|
"tasks": task_list,
|
|
}
|
|
|
|
|
|
def _build_production_info(db: Session, rev: BizStorageReleaseReview, tenant_no: str) -> Dict[str, Any]:
|
|
pwo = (
|
|
db.query(BizProductionWorkOrder)
|
|
.filter(
|
|
BizProductionWorkOrder.reception_id == (rev.reception_id or ""),
|
|
BizProductionWorkOrder.tenant_no == tenant_no,
|
|
BizProductionWorkOrder.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not pwo:
|
|
return {}
|
|
tasks = (
|
|
db.query(BizProductionTask)
|
|
.filter(
|
|
BizProductionTask.work_order_id == pwo.id,
|
|
BizProductionTask.tenant_no == tenant_no,
|
|
BizProductionTask.del_flag == 0,
|
|
)
|
|
.all()
|
|
)
|
|
task_list = []
|
|
_task_status_labels = {"dpz": "待配置", "dzx": "待执行", "jxz": "执行中", "ywc": "已完成"}
|
|
for t in tasks:
|
|
task_list.append({
|
|
"id": t.id,
|
|
"taskNo": t.task_no or "",
|
|
"taskName": t.task_name or "",
|
|
"taskStatus": t.task_status or "",
|
|
"taskStatusLabel": _task_status_labels.get(t.task_status or "", t.task_status or ""),
|
|
"startedTime": _fmt_dt(t.started_time),
|
|
"completedTime": _fmt_dt(t.completed_time),
|
|
})
|
|
pwo_status = pwo.work_order_status or ""
|
|
return {
|
|
"workOrderNo": pwo.work_order_no or "",
|
|
"workOrderStatus": pwo_status,
|
|
"workOrderStatusLabel": _PRODUCTION_WORK_ORDER_STATUS_LABELS.get(pwo_status, pwo_status),
|
|
"tasks": task_list,
|
|
}
|
|
|
|
|
|
def submit_review(
|
|
db: Session,
|
|
review_id: str,
|
|
*,
|
|
review_conclusion: str,
|
|
review_remark: Optional[str] = None,
|
|
review_attachments: Optional[str] = None,
|
|
tenant_no: str,
|
|
user_id: str,
|
|
) -> BizStorageReleaseReview:
|
|
rev = (
|
|
db.query(BizStorageReleaseReview)
|
|
.filter(
|
|
BizStorageReleaseReview.id == review_id,
|
|
BizStorageReleaseReview.tenant_no == tenant_no,
|
|
BizStorageReleaseReview.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not rev:
|
|
raise ValueError("审核记录不存在")
|
|
user_name = _get_user_name(db, user_id)
|
|
rev.review_conclusion = review_conclusion
|
|
rev.review_remark = (review_remark or "").strip() or None
|
|
rev.review_attachments = (review_attachments or "").strip() or None
|
|
rev.review_by = user_id
|
|
rev.review_by_name = user_name
|
|
rev.review_time = datetime.now()
|
|
rev.update_by = user_id
|
|
rev.update_time = datetime.now()
|
|
db.add(rev)
|
|
db.flush()
|
|
return rev
|
|
|
|
|
|
def batch_submit_review(
|
|
db: Session,
|
|
ids: List[str],
|
|
*,
|
|
review_conclusion: str,
|
|
review_remark: Optional[str] = None,
|
|
review_attachments: Optional[str] = None,
|
|
tenant_no: str,
|
|
user_id: str,
|
|
) -> Dict[str, Any]:
|
|
success = 0
|
|
fail = 0
|
|
for rid in ids:
|
|
try:
|
|
submit_review(
|
|
db,
|
|
rid,
|
|
review_conclusion=review_conclusion,
|
|
review_remark=review_remark,
|
|
review_attachments=review_attachments,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
)
|
|
success += 1
|
|
except Exception:
|
|
fail += 1
|
|
return {"success": success, "fail": fail}
|
|
|
|
|
|
def store_sample(
|
|
db: Session,
|
|
review_id: str,
|
|
*,
|
|
tenant_no: str,
|
|
user_id: str,
|
|
) -> BizStorageReleaseReview:
|
|
rev = (
|
|
db.query(BizStorageReleaseReview)
|
|
.filter(
|
|
BizStorageReleaseReview.id == review_id,
|
|
BizStorageReleaseReview.tenant_no == tenant_no,
|
|
BizStorageReleaseReview.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not rev:
|
|
raise ValueError("审核记录不存在")
|
|
if rev.review_conclusion != "tg":
|
|
raise ValueError("只有审核通过的记录才能执行入库操作")
|
|
if rev.storage_status != "drk":
|
|
raise ValueError("当前状态不允许入库操作")
|
|
user_name = _get_user_name(db, user_id)
|
|
rev.storage_status = "yrk"
|
|
rev.storage_by = user_id
|
|
rev.storage_by_name = user_name
|
|
rev.storage_time = datetime.now()
|
|
rev.update_by = user_id
|
|
rev.update_time = datetime.now()
|
|
db.add(rev)
|
|
|
|
if (rev.reception_source or "cell_storage") == "cell_storage":
|
|
sample_code = (rev.sample_code or "").strip()
|
|
if sample_code:
|
|
ps = (
|
|
db.query(BizProtocolSample)
|
|
.filter(
|
|
BizProtocolSample.sample_code == sample_code,
|
|
BizProtocolSample.tenant_no == tenant_no,
|
|
BizProtocolSample.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if ps:
|
|
ps.sample_status = "yrk"
|
|
ps.update_time = datetime.now()
|
|
db.add(ps)
|
|
|
|
db.flush()
|
|
return rev
|
|
|
|
|
|
def batch_store_samples(
|
|
db: Session,
|
|
ids: List[str],
|
|
*,
|
|
tenant_no: str,
|
|
user_id: str,
|
|
) -> Dict[str, Any]:
|
|
success = 0
|
|
fail = 0
|
|
for rid in ids:
|
|
try:
|
|
store_sample(db, rid, tenant_no=tenant_no, user_id=user_id)
|
|
success += 1
|
|
except Exception:
|
|
fail += 1
|
|
return {"success": success, "fail": fail}
|
|
|
|
|
|
def get_storage_release_map_for_samples(
|
|
db: Session,
|
|
sample_ids: List[str],
|
|
tenant_no: str,
|
|
*,
|
|
sample_code_by_id: Optional[Dict[str, str]] = None,
|
|
) -> Dict[str, Dict[str, Any]]:
|
|
"""按样本批量返回存储放行审核摘要。
|
|
|
|
当传入 sample_code_by_id 时，按租户 + 样本编码关联存储放行审核记录，
|
|
返回以 sample_id 为 key 的审核信息映射。
|
|
"""
|
|
if not tenant_no or not sample_code_by_id:
|
|
return {}
|
|
|
|
codes = list({(c or "").strip() for c in sample_code_by_id.values() if (c or "").strip()})
|
|
if not codes:
|
|
return {}
|
|
|
|
rows = (
|
|
db.query(BizStorageReleaseReview)
|
|
.filter(
|
|
BizStorageReleaseReview.tenant_no == tenant_no,
|
|
BizStorageReleaseReview.del_flag == 0,
|
|
BizStorageReleaseReview.sample_code.in_(codes),
|
|
)
|
|
.all()
|
|
)
|
|
code_to_review: Dict[str, BizStorageReleaseReview] = {}
|
|
for r in rows:
|
|
code = (r.sample_code or "").strip()
|
|
if code and code not in code_to_review:
|
|
code_to_review[code] = r
|
|
|
|
out: Dict[str, Dict[str, Any]] = {}
|
|
for sid, raw_code in sample_code_by_id.items():
|
|
code = (raw_code or "").strip()
|
|
if not code:
|
|
continue
|
|
rev = code_to_review.get(code)
|
|
if not rev or not (rev.review_conclusion or "").strip():
|
|
continue
|
|
rc = rev.review_conclusion or ""
|
|
out[sid] = {
|
|
"reviewConclusion": rc,
|
|
"reviewConclusionLabel": _REVIEW_CONCLUSION_LABELS.get(rc, rc),
|
|
"reviewByName": rev.review_by_name or "",
|
|
"reviewTime": _fmt_dt(rev.review_time),
|
|
"reviewRemark": rev.review_remark or "",
|
|
"reviewAttachments": rev.review_attachments or "",
|
|
}
|
|
return out
|