Compare commits

..
Author SHA1 Message Date
zlwu 5fe1193a2c feat(lab-sample): 实验室样本处理三表台账与部门字段 ACL
按造血/单个核/间充质聚合展示,部门员工只改本部门、主管可改全部,并支持筛选与 Excel 导出;同步纠正采集接受人串权与严格菜单守卫。
2026-11-12 12:48:05 +08:00
zlwu 58b252ce6b merge: 合入 #98 实验室样本处理数据台账 2026-11-12 12:12:55 +08:00
zlwuandCursor acb81cadd5 feat(qc): 新增实验室样本处理数据台账(4 Tab + 补录 + 导出)
在质检管理下增加按来源×样本类型展示的台账页,聚合接收/生产/质检参数,
补录按实验室入库编码关联,支持 Excel 导出与中文时间格式。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-12 12:11:57 +08:00
30 changed files with 4168 additions and 5 deletions

No files matched your search

+4
View File
@@ -45,11 +45,14 @@ from app.api.sample_storage_cert_qc import router as sample_storage_cert_qc_rout
from app.api.customer_service_dict import router as customer_service_dict_router from app.api.customer_service_dict import router as customer_service_dict_router
from app.api.sample_cert_insurance import router as sample_cert_insurance_router from app.api.sample_cert_insurance import router as sample_cert_insurance_router
from app.api.sample_code_correction import router as sample_code_correction_router from app.api.sample_code_correction import router as sample_code_correction_router
from app.api.lab_sample_process import router as lab_sample_process_router
from app.api.lab_sample_processing import router as lab_sample_processing_router
from app.api.project_management import router as project_management_router from app.api.project_management import router as project_management_router
__all__ = [ __all__ = [
"project_management_router", "project_management_router",
"lab_sample_process_router",
"public_router", "public_router",
"menu_router", "menu_router",
"admin_router", "admin_router",
@@ -96,4 +99,5 @@ __all__ = [
"customer_service_dict_router", "customer_service_dict_router",
"sample_cert_insurance_router", "sample_cert_insurance_router",
"sample_code_correction_router", "sample_code_correction_router",
"lab_sample_processing_router",
] ]
+138
View File
@@ -0,0 +1,138 @@
"""实验室样本处理数据台账 API"""
from typing import Any, Dict, Optional
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.menu_guard import require_menu_code
from app.core.response import error_response, success_response
from app.crud import lab_sample_process as svc
from app.models.models import SysUser
from app.services import lab_sample_process as ledger
router = APIRouter(prefix="/web/admin", tags=["质检管理"])
LIST_MENU = "biz:labSampleProcess:list"
SUPPLEMENT_MENU = "biz:labSampleProcess:supplement"
def _tenant_no(db: Session, user: dict) -> str:
user_id = user.get("user_id")
if not user_id:
raise PermissionError("用户身份信息缺失,无法确定租户")
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
if not sys_user or not sys_user.tenant_no:
raise PermissionError("用户租户信息缺失")
return sys_user.tenant_no
def _user_id(user: dict) -> str:
uid = user.get("user_id")
if not uid:
raise PermissionError("用户身份信息缺失")
return str(uid)
def _dept_id(user: dict) -> str:
return str(user.get("dept_id") or user.get("create_dept_id") or "")
class SupplementBody(BaseModel):
model_config = ConfigDict(populate_by_name=True)
lab_storage_code: str = Field(..., alias="labStorageCode")
reception_source: str = Field(..., alias="receptionSource")
sample_type_name: str = Field(..., alias="sampleTypeName")
payload: Dict[str, Any] = Field(default_factory=dict)
remark: Optional[str] = None
@router.get("/lab-sample-process/list")
def list_lab_sample_process(
tabKey: str = Query(..., description="cell_hsc|donation_hsc|cell_msc|donation_msc"),
page: int = Query(1, ge=1),
pageSize: int = Query(10, ge=1, le=200),
labStorageCode: Optional[str] = Query(None),
motherName: Optional[str] = Query(None),
createTimeStart: Optional[str] = Query(None),
createTimeEnd: Optional[str] = Query(None),
db: Session = Depends(get_db),
user=Depends(require_menu_code(LIST_MENU)),
):
try:
tenant_no = _tenant_no(db, user)
items, total = svc.list_ledger(
db,
tab_key=tabKey,
page=page,
page_size=pageSize,
tenant_no=tenant_no,
lab_storage_code=labStorageCode,
mother_name=motherName,
create_time_start=createTimeStart,
create_time_end=createTimeEnd,
user_info=user,
)
return success_response(
data={
"list": items,
"total": total,
"page": page,
"pageSize": pageSize,
"tabKey": tabKey,
"tabMeta": ledger.TAB_META.get(tabKey),
"supplementKeys": ledger.SUPPLEMENT_KEYS.get(tabKey, []),
}
)
except PermissionError as e:
return error_response(str(e), code=403)
except ValueError as e:
return error_response(str(e))
except Exception as e:
return error_response(f"查询失败:{e}")
@router.get("/lab-sample-process/supplement")
def get_supplement(
labStorageCode: str = Query(...),
db: Session = Depends(get_db),
user=Depends(require_menu_code(LIST_MENU)),
):
try:
tenant_no = _tenant_no(db, user)
data = svc.get_supplement(db, tenant_no=tenant_no, lab_storage_code=labStorageCode)
return success_response(data=data or {"labStorageCode": labStorageCode, "payload": {}})
except PermissionError as e:
return error_response(str(e), code=403)
except Exception as e:
return error_response(f"查询补录失败:{e}")
@router.put("/lab-sample-process/supplement")
def save_supplement(
body: SupplementBody,
db: Session = Depends(get_db),
user=Depends(require_menu_code(SUPPLEMENT_MENU)),
):
try:
tenant_no = _tenant_no(db, user)
data = svc.upsert_supplement(
db,
tenant_no=tenant_no,
user_id=_user_id(user),
dept_id=_dept_id(user),
lab_storage_code=body.lab_storage_code,
reception_source=body.reception_source,
sample_type_name=body.sample_type_name,
payload=body.payload or {},
remark=body.remark,
)
return success_response(data=data, message="补录已保存")
except PermissionError as e:
return error_response(str(e), code=403)
except ValueError as e:
return error_response(str(e))
except Exception as e:
return error_response(f"保存补录失败:{e}")
+193
View File
@@ -0,0 +1,193 @@
"""实验室样本处理数据 API"""
from __future__ import annotations
import csv
import io
from typing import Any, Dict, Optional
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.menu_guard import require_strict_any_menu_code
from app.core.response import error_response, success_response
from app.models.models import SysUser
from app.services import lab_sample_processing as svc
from app.services.lab_sample_processing_fields import (
DATASETS,
DATASET_LABELS,
PERM_LIST,
PERM_MANAGER,
PERM_MARKET_EDIT,
PERM_PRODUCTION_EDIT,
PERM_QC_EDIT,
schema_payload,
)
router = APIRouter(prefix="/web/admin/lab-sample-processing", tags=["实验室样本处理数据"])
# 列表/导出:有列表、主管或任一部门编辑权即可查看全部字段
READ_MENUS = (
PERM_LIST,
PERM_MANAGER,
PERM_MARKET_EDIT,
PERM_PRODUCTION_EDIT,
PERM_QC_EDIT,
)
# 更新:至少要有主管或某一部门编辑权(字段级 ACL 在服务内再收紧)
WRITE_MENUS = (
PERM_MANAGER,
PERM_MARKET_EDIT,
PERM_PRODUCTION_EDIT,
PERM_QC_EDIT,
)
def _tenant_no(db: Session, user: dict) -> str:
user_id = user.get("user_id")
if not user_id:
raise PermissionError("用户身份信息缺失")
row = db.query(SysUser).filter(SysUser.id == user_id, SysUser.del_flag == 0).first()
if not row or not row.tenant_no:
raise PermissionError("用户租户信息缺失")
return row.tenant_no
def _ensure_dataset(dataset: str) -> Optional[str]:
ds = (dataset or "").strip().lower()
return ds if ds in DATASETS else None
class UpdateBody(BaseModel):
model_config = ConfigDict(populate_by_name=True)
payload: Dict[str, Any] = Field(default_factory=dict)
@router.get("/meta/{dataset}")
async def get_meta(
dataset: str,
user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
db: Session = Depends(get_db),
):
ds = _ensure_dataset(dataset)
if not ds:
return error_response(400, "未知数据集")
acl = svc.user_acl(db, user)
return success_response(
{
"dataset": ds,
"label": DATASET_LABELS.get(ds),
"fields": schema_payload(ds),
"acl": acl,
}
)
@router.get("/{dataset}/list")
async def list_dataset(
dataset: str,
keyword: Optional[str] = Query(None),
dateFrom: Optional[str] = Query(None),
dateTo: Optional[str] = Query(None),
page: int = Query(1, ge=1),
pageSize: int = Query(20, ge=1, le=200),
user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
db: Session = Depends(get_db),
):
ds = _ensure_dataset(dataset)
if not ds:
return error_response(400, "未知数据集")
tenant_no = _tenant_no(db, user)
rows, total = svc.list_rows(
db,
tenant_no=tenant_no,
dataset=ds,
keyword=keyword,
date_from=dateFrom,
date_to=dateTo,
page=page,
page_size=pageSize,
)
acl = svc.user_acl(db, user)
return success_response(
{
"list": rows,
"total": total,
"page": page,
"pageSize": pageSize,
"fields": schema_payload(ds),
"acl": acl,
}
)
@router.put("/{dataset}/{sample_code}")
async def update_dataset_row(
dataset: str,
sample_code: str,
body: UpdateBody,
user: dict = Depends(require_strict_any_menu_code(*WRITE_MENUS)),
db: Session = Depends(get_db),
):
ds = _ensure_dataset(dataset)
if not ds:
return error_response(400, "未知数据集")
tenant_no = _tenant_no(db, user)
acl = svc.user_acl(db, user)
editable = set(acl.get("editableDepts") or [])
if not editable:
return error_response(403, "无编辑权限")
try:
row = svc.update_row(
db,
tenant_no=tenant_no,
dataset=ds,
sample_code=sample_code,
payload=body.payload or {},
editable_depts=editable,
operator_id=str(user.get("user_id") or ""),
)
except PermissionError as exc:
return error_response(403, str(exc))
except LookupError as exc:
return error_response(404, str(exc))
except ValueError as exc:
return error_response(400, str(exc))
return success_response(row)
@router.get("/{dataset}/export")
async def export_dataset(
dataset: str,
keyword: Optional[str] = Query(None),
dateFrom: Optional[str] = Query(None),
dateTo: Optional[str] = Query(None),
user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
db: Session = Depends(get_db),
):
ds = _ensure_dataset(dataset)
if not ds:
return error_response(400, "未知数据集")
tenant_no = _tenant_no(db, user)
headers, body = svc.export_rows(
db,
tenant_no=tenant_no,
dataset=ds,
keyword=keyword,
date_from=dateFrom,
date_to=dateTo,
)
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(headers)
for row in body:
writer.writerow(["" if v is None else v for v in row])
raw = ("\ufeff" + buf.getvalue()).encode("utf-8")
filename = f"lab-sample-{ds}.csv"
return StreamingResponse(
io.BytesIO(raw),
media_type="text/csv; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
+753
View File
@@ -0,0 +1,753 @@
"""实验室样本处理台账列表与补录 CRUD。"""
from __future__ import annotations
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy import and_
from sqlalchemy.orm import Session
from app.core.data_scope import apply_data_scope
from app.core.utils import generate_uuid
from app.crud import sample_production_reception as prod_svc
from app.models.models import (
BizCellSampleType,
BizCollectionHospital,
BizCollectionProtocol,
BizLabSampleProcessSupplement,
BizProtocolCollectionInfo,
BizProtocolPartyCustomer,
BizProtocolSample,
BizSampleProductionReception,
BizSampleReception,
BizSampleReceptionCheckResult,
BizStorageReleaseReview,
)
from app.models.research_donation import (
BizResearchDonation,
BizResearchDonationCollection,
BizResearchDonationParty,
BizResearchDonationSample,
)
from app.services import lab_sample_process as ledger
def _fmt_raw(dt: Optional[datetime]) -> str:
return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else ""
def _normalize_bound(value: Optional[str], *, end_of_day: bool = False) -> Optional[datetime]:
text = (value or "").strip()
if not text:
return None
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
try:
dt = datetime.strptime(text[:19] if " " in text else text[:10], fmt)
if end_of_day and fmt == "%Y-%m-%d":
return dt.replace(hour=23, minute=59, second=59)
return dt
except ValueError:
continue
return None
def _type_ids_by_name(db: Session, tenant_no: str, type_name: str) -> List[str]:
rows = (
db.query(BizCellSampleType.id)
.filter(
BizCellSampleType.tenant_no == tenant_no,
BizCellSampleType.del_flag == 0,
BizCellSampleType.type_name == type_name,
)
.all()
)
return [r.id for r in rows if r.id]
def _get_supplement_map(
db: Session, tenant_no: str, codes: List[str]
) -> Dict[str, Dict[str, Any]]:
codes = [c for c in {(x or "").strip() for x in codes} if c]
if not codes:
return {}
rows = (
db.query(BizLabSampleProcessSupplement)
.filter(
BizLabSampleProcessSupplement.tenant_no == tenant_no,
BizLabSampleProcessSupplement.del_flag == 0,
BizLabSampleProcessSupplement.lab_storage_code.in_(codes),
)
.all()
)
out: Dict[str, Dict[str, Any]] = {}
for row in rows:
out[row.lab_storage_code] = {
"id": row.id,
"payload": ledger.parse_payload(row.payload),
"remark": row.remark or "",
"updateTime": _fmt_raw(row.update_time),
}
return out
def get_supplement(
db: Session, *, tenant_no: str, lab_storage_code: str
) -> Optional[Dict[str, Any]]:
code = (lab_storage_code or "").strip()
if not code:
return None
row = (
db.query(BizLabSampleProcessSupplement)
.filter(
BizLabSampleProcessSupplement.tenant_no == tenant_no,
BizLabSampleProcessSupplement.lab_storage_code == code,
BizLabSampleProcessSupplement.del_flag == 0,
)
.first()
)
if not row:
return None
return {
"id": row.id,
"labStorageCode": row.lab_storage_code,
"receptionSource": row.reception_source,
"sampleTypeName": row.sample_type_name or "",
"payload": ledger.parse_payload(row.payload),
"remark": row.remark or "",
"createTime": _fmt_raw(row.create_time),
"updateTime": _fmt_raw(row.update_time),
}
def upsert_supplement(
db: Session,
*,
tenant_no: str,
user_id: str,
dept_id: str,
lab_storage_code: str,
reception_source: str,
sample_type_name: str,
payload: Dict[str, Any],
remark: Optional[str] = None,
) -> Dict[str, Any]:
code = (lab_storage_code or "").strip()
if not code:
raise ValueError("实验室入库编码不能为空")
src = (reception_source or "").strip() or "cell_storage"
if src not in ("cell_storage", "research_donation"):
raise ValueError("来源类型无效")
type_name = (sample_type_name or "").strip()
clean_payload = {
str(k): ("" if v is None else str(v).strip())
for k, v in (payload or {}).items()
if str(k).strip()
}
row = (
db.query(BizLabSampleProcessSupplement)
.filter(
BizLabSampleProcessSupplement.tenant_no == tenant_no,
BizLabSampleProcessSupplement.lab_storage_code == code,
BizLabSampleProcessSupplement.del_flag == 0,
)
.first()
)
now = datetime.now()
if row:
row.reception_source = src
row.sample_type_name = type_name or row.sample_type_name
row.payload = ledger.dumps_payload(clean_payload)
if remark is not None:
row.remark = (remark or "").strip() or None
row.update_by = user_id
row.update_time = now
else:
row = BizLabSampleProcessSupplement(
id=generate_uuid(),
lab_storage_code=code,
reception_source=src,
sample_type_name=type_name or None,
payload=ledger.dumps_payload(clean_payload),
tenant_no=tenant_no,
remark=(remark or "").strip() or None,
create_by=user_id,
create_dept_id=dept_id or "",
create_time=now,
update_by=user_id,
update_time=now,
del_flag=0,
)
db.add(row)
db.commit()
db.refresh(row)
return get_supplement(db, tenant_no=tenant_no, lab_storage_code=code) or {}
def _check_flag_map(db: Session, tenant_no: str, reception_ids: List[str]) -> Dict[str, Dict[str, bool]]:
ids = [i for i in reception_ids if i]
if not ids:
return {}
rows = (
db.query(BizSampleReceptionCheckResult)
.filter(
BizSampleReceptionCheckResult.tenant_no == tenant_no,
BizSampleReceptionCheckResult.del_flag == 0,
BizSampleReceptionCheckResult.reception_id.in_(ids),
)
.all()
)
out: Dict[str, Dict[str, bool]] = {}
for r in rows:
bucket = out.setdefault(r.reception_id, {})
name = (r.item_name or "").strip()
if name:
bucket[name] = bool(r.checked)
return out
def _storage_status_map(db: Session, tenant_no: str, sample_codes: List[str]) -> Dict[str, str]:
codes = [c for c in sample_codes if c]
if not codes:
return {}
rows = (
db.query(BizStorageReleaseReview.sample_code, BizStorageReleaseReview.storage_status)
.filter(
BizStorageReleaseReview.tenant_no == tenant_no,
BizStorageReleaseReview.del_flag == 0,
BizStorageReleaseReview.sample_code.in_(codes),
)
.all()
)
labels = {"drk": "待入库", "yrk": "已入库"}
return {r.sample_code: labels.get(r.storage_status or "", r.storage_status or "") for r in rows}
def _hospital_name_map(db: Session, hospital_ids: List[str]) -> Dict[str, str]:
ids = [i for i in hospital_ids if i]
if not ids:
return {}
rows = (
db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name)
.filter(BizCollectionHospital.id.in_(ids), BizCollectionHospital.del_flag == 0)
.all()
)
return {r.id: r.hospital_name or "" for r in rows}
def _build_cell_system(
*,
tab_key: str,
rec: BizSampleReception,
sample: BizProtocolSample,
party: Optional[BizProtocolPartyCustomer],
coll: Optional[BizProtocolCollectionInfo],
protocol: Optional[BizCollectionProtocol],
prod: Optional[Dict[str, Any]],
hospital_name: str,
check_flags: Dict[str, bool],
storage_status: str,
qc_mother: Dict[str, str],
qc_process: Dict[str, str],
) -> Dict[str, Any]:
prod = prod or {}
maternal_time = coll.maternal_blood_collect_time if coll else None
clear_flag = None
for name, checked in check_flags.items():
if "标识清晰" in name or "样本信息正确" in name:
clear_flag = checked
break
base: Dict[str, Any] = {
"receptionId": rec.id,
"sampleCode": (rec.sample_code or sample.sample_code or "").strip(),
"labStorageCode": (rec.lab_storage_code or "").strip(),
"motherName": (party.mother_name if party else "") or "",
"collectionTime": ledger.format_cn_datetime(coll.sample_collect_time if coll else None),
"collectionTimeRaw": _fmt_raw(coll.sample_collect_time if coll else None),
"hospitalName": hospital_name,
"receptionTime": ledger.format_cn_datetime(rec.reception_time),
"receptionTimeRaw": _fmt_raw(rec.reception_time),
"receiveTemperature": rec.receive_temperature or "",
"handlerName": rec.handler_name or "",
"remarkReception": rec.remark or "",
"createTime": ledger.format_cn_datetime(rec.create_time),
"createTimeRaw": _fmt_raw(rec.create_time),
"operatorName": prod.get("handlerName") or "",
"remarkProcess": prod.get("remark") or "",
"storageStatus": storage_status,
}
if tab_key == "cell_hsc":
base.update(
{
"cellCode": (rec.lab_storage_code or "").strip(),
"hasMaternalBlood": "是" if maternal_time else "否",
"remeasureMaternalBlood": "",
"consentForm": ledger.yes_no_from_urls(
protocol.consent_form_urls if protocol else None
),
"transportPackingList": "",
"cordBloodReceiveConfirm": "",
"healthSurvey": ledger.yes_no_from_urls(
protocol.health_survey_form_urls if protocol else None
),
"labelClear": ledger.yes_no_from_bool(clear_flag) if clear_flag is not None else "",
"receiveCheckText": "",
"originalBloodWeightG": prod.get("originalBloodWeightG"),
"hesVolumeMl": prod.get("hesVolumeMl"),
"bloodType": prod.get("bloodType") or "",
"infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel")
or ledger.infectious_label(prod.get("infectiousFourTestResult")),
"infectiousFourExternal": "",
"preEnrichNucleatedCount": "",
"enrichedBloodWeightG": prod.get("enrichedBloodWeightG"),
"postEnrichNucleatedCount": "",
"cryopreservationVolumeMl": prod.get("cryopreservationVolumeMl"),
"cd34Count1e4": "",
}
)
else: # cell_msc
base.update(
{
"labFlowCode": (rec.lab_storage_code or "").strip(),
"cordCollectTime": ledger.format_cn_datetime(
coll.sample_collect_time if coll else None
),
"transportTempC": rec.receive_temperature or "",
"cordReceiveTime": ledger.format_cn_datetime(rec.reception_time),
"inoculationAmount": prod.get("inoculationAmount") or "",
"cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")),
"storageTotalTubes": prod.get("storageTotalTubes"),
"availableCellCount1e7": prod.get("availableCellCount1e7"),
"remark": rec.remark or prod.get("remark") or "",
"storageDateP2": "",
}
)
base.update(qc_process)
base.update(qc_mother)
return base
def _build_donation_system(
*,
tab_key: str,
rec: BizSampleReception,
sample: BizResearchDonationSample,
party: Optional[BizResearchDonationParty],
coll: Optional[BizResearchDonationCollection],
prod: Optional[Dict[str, Any]],
storage_status: str,
qc_mother: Dict[str, str],
qc_process: Dict[str, str],
) -> Dict[str, Any]:
prod = prod or {}
base: Dict[str, Any] = {
"receptionId": rec.id,
"sampleCode": (rec.sample_code or sample.sample_code or "").strip(),
"labStorageCode": (rec.lab_storage_code or "").strip(),
"motherName": (party.mother_name if party else "") or "",
"receptionTime": ledger.format_cn_datetime(rec.reception_time),
"receptionTimeRaw": _fmt_raw(rec.reception_time),
"receiveTemperature": rec.receive_temperature or "",
"handlerName": rec.handler_name or "",
"remarkReception": rec.remark or "",
"createTime": ledger.format_cn_datetime(rec.create_time),
"createTimeRaw": _fmt_raw(rec.create_time),
"operatorName": prod.get("handlerName") or "",
"remarkProcess": prod.get("remark") or "",
"storageStatus": storage_status,
}
if tab_key == "donation_hsc":
base.update(
{
"cellCode": (rec.lab_storage_code or "").strip(),
"labMaterialCode": "",
"inboundCode": (rec.lab_storage_code or "").strip(),
"discardReason": "",
"receiveCheckText": "",
"originalBloodVolumeMl": prod.get("originalBloodVolumeMl"),
"nucleatedCellCount1e8": prod.get("nucleatedCellCount1e8"),
"preOpNucleatedCount": "",
"infectiousFourTestLabel": prod.get("infectiousFourTestResultLabel")
or ledger.infectious_label(prod.get("infectiousFourTestResult")),
}
)
# 科研捐赠母血 5 项(不含核酸)
for k in (
"motherHbsag",
"motherAntiHcv",
"motherAntiTp",
"motherAntiHiv",
"motherCmvIgm",
):
if qc_mother.get(k):
base[k] = qc_mother[k]
else:
base.update(
{
"labFlowCode": (rec.lab_storage_code or "").strip(),
"cordCollectTime": ledger.format_cn_datetime(
coll.sample_collect_time if coll else None
),
"transportTempC": rec.receive_temperature or "",
"cordReceiveTime": ledger.format_cn_datetime(rec.reception_time),
"inoculationAmount": prod.get("inoculationAmount") or "",
"cryopreservationTime": ledger.format_cn_datetime(prod.get("cryopreservationTime")),
"storageTotalTubes": prod.get("storageTotalTubes"),
"availableCellCount1e7": prod.get("availableCellCount1e7"),
"remark": rec.remark or prod.get("remark") or "",
"storageDateP2": "",
}
)
base.update(qc_process)
base.update(qc_mother)
return base
def list_ledger(
db: Session,
*,
tab_key: str,
page: int = 1,
page_size: int = 10,
tenant_no: str,
lab_storage_code: Optional[str] = None,
mother_name: Optional[str] = None,
create_time_start: Optional[str] = None,
create_time_end: Optional[str] = None,
user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
meta = ledger.TAB_META.get(tab_key)
if not meta:
raise ValueError("无效的 Tab")
source = meta["source"]
type_name = meta["sampleTypeName"]
type_ids = _type_ids_by_name(db, tenant_no, type_name)
if not type_ids:
return [], 0
q = db.query(BizSampleReception).filter(
BizSampleReception.tenant_no == tenant_no,
BizSampleReception.del_flag == 0,
BizSampleReception.reception_source == source,
BizSampleReception.lab_storage_code.isnot(None),
BizSampleReception.lab_storage_code != "",
BizSampleReception.accept_conclusion == "js",
)
code_kw = (lab_storage_code or "").strip()
if code_kw:
q = q.filter(BizSampleReception.lab_storage_code.like(f"%{code_kw}%"))
start = _normalize_bound(create_time_start)
end = _normalize_bound(create_time_end, end_of_day=True)
if start:
q = q.filter(BizSampleReception.create_time >= start)
if end:
q = q.filter(BizSampleReception.create_time <= end)
if user_info:
q = apply_data_scope(q, BizSampleReception, user_info, db, tenant_no)
mother_kw = (mother_name or "").strip()
# 按来源限制样本类型,并可选母亲姓名
if source == "cell_storage":
sample_q = (
db.query(BizProtocolSample.id)
.join(
BizCollectionProtocol,
and_(
BizCollectionProtocol.id == BizProtocolSample.collection_protocol_id,
BizCollectionProtocol.del_flag == 0,
BizCollectionProtocol.tenant_no == tenant_no,
),
)
.filter(
BizProtocolSample.tenant_no == tenant_no,
BizProtocolSample.del_flag == 0,
BizProtocolSample.sample_type_id.in_(type_ids),
)
)
if mother_kw:
sample_q = sample_q.join(
BizProtocolPartyCustomer,
and_(
BizProtocolPartyCustomer.collection_protocol_id
== BizProtocolSample.collection_protocol_id,
BizProtocolPartyCustomer.del_flag == 0,
BizProtocolPartyCustomer.tenant_no == tenant_no,
),
).filter(BizProtocolPartyCustomer.mother_name.like(f"%{mother_kw}%"))
sample_ids = [r.id for r in sample_q.all()]
if not sample_ids:
return [], 0
q = q.filter(BizSampleReception.protocol_sample_id.in_(sample_ids))
else:
sample_q = (
db.query(BizResearchDonationSample.id)
.join(
BizResearchDonation,
and_(
BizResearchDonation.id == BizResearchDonationSample.research_donation_id,
BizResearchDonation.del_flag == 0,
BizResearchDonation.tenant_no == tenant_no,
),
)
.filter(
BizResearchDonationSample.tenant_no == tenant_no,
BizResearchDonationSample.del_flag == 0,
BizResearchDonationSample.sample_type_id.in_(type_ids),
)
)
if mother_kw:
sample_q = sample_q.join(
BizResearchDonationParty,
and_(
BizResearchDonationParty.research_donation_id
== BizResearchDonationSample.research_donation_id,
BizResearchDonationParty.del_flag == 0,
BizResearchDonationParty.tenant_no == tenant_no,
),
).filter(BizResearchDonationParty.mother_name.like(f"%{mother_kw}%"))
sample_ids = [r.id for r in sample_q.all()]
if not sample_ids:
return [], 0
q = q.filter(BizSampleReception.donation_sample_id.in_(sample_ids))
total = q.count()
page = max(1, int(page or 1))
page_size = min(max(1, int(page_size or 10)), 200)
receptions = (
q.order_by(BizSampleReception.create_time.desc(), BizSampleReception.id.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
if not receptions:
return [], int(total)
reception_ids = [r.id for r in receptions]
codes = [(r.lab_storage_code or "").strip() for r in receptions]
sample_codes = [(r.sample_code or "").strip() for r in receptions]
prod_map = _prod_map(db, tenant_no, reception_ids)
check_map = _check_flag_map(db, tenant_no, reception_ids)
storage_map = _storage_status_map(db, tenant_no, [c for c in sample_codes if c])
sup_map = _get_supplement_map(db, tenant_no, codes)
items: List[Dict[str, Any]] = []
if source == "cell_storage":
items = _assemble_cell_rows(
db,
tenant_no=tenant_no,
tab_key=tab_key,
receptions=receptions,
prod_map=prod_map,
check_map=check_map,
storage_map=storage_map,
sup_map=sup_map,
)
else:
items = _assemble_donation_rows(
db,
tenant_no=tenant_no,
tab_key=tab_key,
receptions=receptions,
prod_map=prod_map,
storage_map=storage_map,
sup_map=sup_map,
)
return items, int(total)
def _prod_map(db: Session, tenant_no: str, reception_ids: List[str]) -> Dict[str, Dict[str, Any]]:
rows = (
db.query(BizSampleProductionReception)
.filter(
BizSampleProductionReception.tenant_no == tenant_no,
BizSampleProductionReception.del_flag == 0,
BizSampleProductionReception.reception_id.in_(reception_ids),
)
.all()
)
return {r.reception_id: prod_svc._row_to_dict(r) for r in rows}
def _assemble_cell_rows(
db: Session,
*,
tenant_no: str,
tab_key: str,
receptions: List[BizSampleReception],
prod_map: Dict[str, Dict[str, Any]],
check_map: Dict[str, Dict[str, bool]],
storage_map: Dict[str, str],
sup_map: Dict[str, Dict[str, Any]],
) -> List[Dict[str, Any]]:
sample_ids = [r.protocol_sample_id for r in receptions if r.protocol_sample_id]
samples = (
db.query(BizProtocolSample)
.filter(
BizProtocolSample.tenant_no == tenant_no,
BizProtocolSample.del_flag == 0,
BizProtocolSample.id.in_(sample_ids),
)
.all()
if sample_ids
else []
)
sample_map = {s.id: s for s in samples}
protocol_ids = list({s.collection_protocol_id for s in samples if s.collection_protocol_id})
parties = (
db.query(BizProtocolPartyCustomer)
.filter(
BizProtocolPartyCustomer.tenant_no == tenant_no,
BizProtocolPartyCustomer.del_flag == 0,
BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
)
.all()
if protocol_ids
else []
)
party_map = {p.collection_protocol_id: p for p in parties}
colls = (
db.query(BizProtocolCollectionInfo)
.filter(
BizProtocolCollectionInfo.tenant_no == tenant_no,
BizProtocolCollectionInfo.del_flag == 0,
BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
)
.all()
if protocol_ids
else []
)
coll_map = {c.collection_protocol_id: c for c in colls}
protocols = (
db.query(BizCollectionProtocol)
.filter(
BizCollectionProtocol.tenant_no == tenant_no,
BizCollectionProtocol.del_flag == 0,
BizCollectionProtocol.id.in_(protocol_ids),
)
.all()
if protocol_ids
else []
)
protocol_map = {p.id: p for p in protocols}
hospital_ids = [c.hospital_id for c in colls if c.hospital_id]
hospital_map = _hospital_name_map(db, hospital_ids)
out: List[Dict[str, Any]] = []
for rec in receptions:
sample = sample_map.get(rec.protocol_sample_id or "")
if not sample:
continue
pid = sample.collection_protocol_id
sample_code = (rec.sample_code or sample.sample_code or "").strip()
tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code)
qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True)
qc_process = ledger.extract_process_tests(tasks)
system = _build_cell_system(
tab_key=tab_key,
rec=rec,
sample=sample,
party=party_map.get(pid),
coll=coll_map.get(pid),
protocol=protocol_map.get(pid),
prod=prod_map.get(rec.id),
hospital_name=hospital_map.get((coll_map.get(pid).hospital_id if coll_map.get(pid) else "") or "", ""),
check_flags=check_map.get(rec.id, {}),
storage_status=storage_map.get(sample_code, ""),
qc_mother=qc_mother,
qc_process=qc_process,
)
lab_code = system.get("labStorageCode") or ""
merged = ledger.merge_row(
tab_key=tab_key,
system_values=system,
supplement=(sup_map.get(lab_code) or {}).get("payload"),
)
merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or ""
merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload"))
out.append(merged)
return out
def _assemble_donation_rows(
db: Session,
*,
tenant_no: str,
tab_key: str,
receptions: List[BizSampleReception],
prod_map: Dict[str, Dict[str, Any]],
storage_map: Dict[str, str],
sup_map: Dict[str, Dict[str, Any]],
) -> List[Dict[str, Any]]:
sample_ids = [r.donation_sample_id for r in receptions if r.donation_sample_id]
samples = (
db.query(BizResearchDonationSample)
.filter(
BizResearchDonationSample.tenant_no == tenant_no,
BizResearchDonationSample.del_flag == 0,
BizResearchDonationSample.id.in_(sample_ids),
)
.all()
if sample_ids
else []
)
sample_map = {s.id: s for s in samples}
donation_ids = list({s.research_donation_id for s in samples if s.research_donation_id})
parties = (
db.query(BizResearchDonationParty)
.filter(
BizResearchDonationParty.tenant_no == tenant_no,
BizResearchDonationParty.del_flag == 0,
BizResearchDonationParty.research_donation_id.in_(donation_ids),
)
.all()
if donation_ids
else []
)
party_map = {p.research_donation_id: p for p in parties}
colls = (
db.query(BizResearchDonationCollection)
.filter(
BizResearchDonationCollection.tenant_no == tenant_no,
BizResearchDonationCollection.del_flag == 0,
BizResearchDonationCollection.research_donation_id.in_(donation_ids),
)
.all()
if donation_ids
else []
)
coll_map = {c.research_donation_id: c for c in colls}
out: List[Dict[str, Any]] = []
for rec in receptions:
sample = sample_map.get(rec.donation_sample_id or "")
if not sample:
continue
did = sample.research_donation_id
sample_code = (rec.sample_code or sample.sample_code or "").strip()
tasks = ledger.load_qc_tasks_for_sample(db, tenant_no, sample_code)
qc_mother = ledger.extract_mother_blood(tasks, include_nucleic=True)
qc_process = ledger.extract_process_tests(tasks)
system = _build_donation_system(
tab_key=tab_key,
rec=rec,
sample=sample,
party=party_map.get(did),
coll=coll_map.get(did),
prod=prod_map.get(rec.id),
storage_status=storage_map.get(sample_code, ""),
qc_mother=qc_mother,
qc_process=qc_process,
)
lab_code = system.get("labStorageCode") or ""
merged = ledger.merge_row(
tab_key=tab_key,
system_values=system,
supplement=(sup_map.get(lab_code) or {}).get("payload"),
)
merged["supplementId"] = (sup_map.get(lab_code) or {}).get("id") or ""
merged["hasSupplement"] = bool((sup_map.get(lab_code) or {}).get("payload"))
out.append(merged)
return out
+3 -1
View File
@@ -13,7 +13,7 @@ from app.core.database import SessionLocal
from app.core.api_request_log_middleware import ApiRequestLogMiddleware from app.core.api_request_log_middleware import ApiRequestLogMiddleware
from app.core.api_crypto_middleware import ApiCryptoMiddleware from app.core.api_crypto_middleware import ApiCryptoMiddleware
from app.models import models # noqa: F401 from app.models import models # noqa: F401
from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router, market_mobile_router, sample_storage_cert_qc_router, customer_service_dict_router, sample_cert_insurance_router, sample_code_correction_router, project_management_router from app.api import public_router, menu_router, admin_router, workflow_router, basic_info_router, collection_protocol_router, protocol_sample_payment_router, protocol_refund_record_router, presign_order_router, sample_freight_router, leave_router, operations_router, ethnic_router, sample_reception_router, sample_production_reception_router, qc_router, research_donation_router, production_router, pilot_service_router, insurance_company_router, material_router, storage_release_review_router, sample_sms_notification_router, storage_entry_sms_notification_router, qc_sms_notification_router, protocol_sample_collection_router, research_donation_sample_collection_router, pilot_application_router, pilot_checkup_router, pilot_delivery_router, pilot_infusion_confirm_router, pilot_infusion_schedule_mgmt_router, outbound_release_review_router, clinical_payment_router, env_detect_router, cleanroom_garment_router, market_router, pre_transit_sample_router, oa_approval_router, mini_auth_router, mini_menu_router, market_mobile_router, sample_storage_cert_qc_router, customer_service_dict_router, sample_cert_insurance_router, sample_code_correction_router, lab_sample_process_router, lab_sample_processing_router, project_management_router
@asynccontextmanager @asynccontextmanager
@@ -157,6 +157,8 @@ app.include_router(sample_storage_cert_qc_router, prefix=settings.API_V1_PREFIX)
app.include_router(customer_service_dict_router, prefix=settings.API_V1_PREFIX) app.include_router(customer_service_dict_router, prefix=settings.API_V1_PREFIX)
app.include_router(sample_cert_insurance_router, prefix=settings.API_V1_PREFIX) app.include_router(sample_cert_insurance_router, prefix=settings.API_V1_PREFIX)
app.include_router(sample_code_correction_router, prefix=settings.API_V1_PREFIX) app.include_router(sample_code_correction_router, prefix=settings.API_V1_PREFIX)
app.include_router(lab_sample_process_router, prefix=settings.API_V1_PREFIX)
app.include_router(lab_sample_processing_router, prefix=settings.API_V1_PREFIX)
app.include_router(project_management_router, prefix=settings.API_V1_PREFIX) app.include_router(project_management_router, prefix=settings.API_V1_PREFIX)
+43
View File
@@ -1969,6 +1969,26 @@ class BizSampleReception(Base):
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
class BizLabSampleProcessingExtra(Base):
"""实验室样本处理数据补充字段 biz_lab_sample_processing_extra"""
__tablename__ = "biz_lab_sample_processing_extra"
id = Column(String(64), primary_key=True, comment="主键ID")
dataset = Column(String(16), nullable=False, comment="hsc|ubmc|msc")
sample_code = Column(String(32), nullable=False, comment="样本编码")
payload_json = Column(Text, nullable=True, comment="补充字段 JSON")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
class BizSampleProductionReception(Base): class BizSampleProductionReception(Base):
"""生产部门样本接收登记 biz_sample_production_reception""" """生产部门样本接收登记 biz_sample_production_reception"""
@@ -2007,6 +2027,29 @@ class BizSampleProductionReception(Base):
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记") del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
class BizLabSampleProcessSupplement(Base):
"""实验室样本处理台账补录 biz_lab_sample_process_supplement(键:tenant_no + lab_storage_code)"""
__tablename__ = "biz_lab_sample_process_supplement"
id = Column(String(64), primary_key=True, comment="主键ID")
lab_storage_code = Column(String(32), nullable=False, comment="实验室入库编码")
reception_source = Column(
String(20), nullable=False, default="cell_storage", comment="cell_storage|research_donation"
)
sample_type_name = Column(String(100), nullable=True, comment="样本类型名称快照")
payload = Column(Text, nullable=False, comment="补录字段 JSON")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")
class BizSampleSmsNotification(Base): class BizSampleSmsNotification(Base):
__tablename__ = "biz_sample_sms_notification" __tablename__ = "biz_sample_sms_notification"
+374
View File
@@ -0,0 +1,374 @@
"""实验室样本处理台账:字段合并、质检参数匹配、中文时间格式。"""
from __future__ import annotations
import json
from datetime import datetime
from typing import Any, Dict, Iterable, List, Optional, Set
from sqlalchemy.orm import Session
from app.models.models import (
BizQcTask,
BizQcWorkOrder,
BizQcWorkOrderInspectionBatch,
)
from app.models.qc_param import BizQcTaskResult
from app.services.sample_storage_cert_qc_prefill import (
_MOTHER_SEROLOGY_RULES,
_process_qual_from_tasks,
_serology_from_tasks,
_stamp,
batch_kind_from_name,
normalize_negative_positive,
qual_yin_yang,
)
SAMPLE_TYPE_PERINATAL_HSC = "围产期造血干细胞"
SAMPLE_TYPE_UC_MSC = "脐带间充质干细胞"
TAB_KEYS = (
"cell_hsc",
"donation_hsc",
"cell_msc",
"donation_msc",
)
TAB_META = {
"cell_hsc": {
"source": "cell_storage",
"sampleTypeName": SAMPLE_TYPE_PERINATAL_HSC,
"label": "细胞存储·围产期造血干细胞",
},
"donation_hsc": {
"source": "research_donation",
"sampleTypeName": SAMPLE_TYPE_PERINATAL_HSC,
"label": "科研捐赠·围产期造血干细胞",
},
"cell_msc": {
"source": "cell_storage",
"sampleTypeName": SAMPLE_TYPE_UC_MSC,
"label": "细胞存储·脐带间充质干细胞",
},
"donation_msc": {
"source": "research_donation",
"sampleTypeName": SAMPLE_TYPE_UC_MSC,
"label": "科研捐赠·脐带间充质干细胞",
},
}
# 各 Tab 可补录字段(系统空或口径需覆盖时写入)
SUPPLEMENT_KEYS: Dict[str, List[str]] = {
"cell_hsc": [
"receiveCheckText",
"hasMaternalBlood",
"remeasureMaternalBlood",
"transportPackingList",
"cordBloodReceiveConfirm",
"infectiousFourExternal",
"preEnrichNucleatedCount",
"postEnrichNucleatedCount",
"cd34Count1e4",
"remarkProcess",
],
"donation_hsc": [
"labMaterialCode",
"storageStatus",
"discardReason",
"receiveCheckText",
"preOpNucleatedCount",
"infectiousFourTestLabel",
"motherHbsag",
"motherAntiHcv",
"motherAntiTp",
"motherAntiHiv",
"motherCmvIgm",
"remarkProcess",
],
"cell_msc": [
"storageStatus",
"endotoxinTest",
"bacteriaTest",
"fungusTest",
"mycoplasmaTest",
"motherHbsag",
"motherHbvDna",
"motherAntiHcv",
"motherHcvRna",
"motherAntiTp",
"motherCmvIgm",
"motherAntiHiv",
"motherHiv1Rna",
"storageDateP2",
"remark",
],
"donation_msc": [
"storageStatus",
"storageTotalTubes",
"availableCellCount1e7",
"endotoxinTest",
"bacteriaTest",
"fungusTest",
"mycoplasmaTest",
"motherHbsag",
"motherHbvDna",
"motherAntiHcv",
"motherHcvRna",
"motherAntiTp",
"motherCmvIgm",
"motherAntiHiv",
"motherHiv1Rna",
"storageDateP2",
"remark",
],
}
# 补录优先覆盖系统推导值的字段
SUPPLEMENT_OVERRIDE_KEYS: Dict[str, Set[str]] = {
"cell_hsc": {"receiveCheckText"},
"donation_hsc": {
"receiveCheckText",
"storageStatus",
"infectiousFourTestLabel",
"motherHbsag",
"motherAntiHcv",
"motherAntiTp",
"motherAntiHiv",
"motherCmvIgm",
},
"cell_msc": {
"storageStatus",
"endotoxinTest",
"bacteriaTest",
"fungusTest",
"mycoplasmaTest",
"motherHbsag",
"motherHbvDna",
"motherAntiHcv",
"motherHcvRna",
"motherAntiTp",
"motherCmvIgm",
"motherAntiHiv",
"motherHiv1Rna",
"storageDateP2",
},
"donation_msc": {
"storageStatus",
"storageTotalTubes",
"availableCellCount1e7",
"endotoxinTest",
"bacteriaTest",
"fungusTest",
"mycoplasmaTest",
"motherHbsag",
"motherHbvDna",
"motherAntiHcv",
"motherHcvRna",
"motherAntiTp",
"motherCmvIgm",
"motherAntiHiv",
"motherHiv1Rna",
"storageDateP2",
},
}
def format_cn_datetime(value: Any) -> str:
"""统一展示:2026年11月12日 00时00分00秒"""
if value is None or value == "":
return ""
if isinstance(value, datetime):
dt = value
else:
text = str(value).strip().replace("T", " ")
if not text:
return ""
dt = None
for candidate, fmt in (
(text[:19], "%Y-%m-%d %H:%M:%S"),
(text[:16], "%Y-%m-%d %H:%M"),
(text[:10], "%Y-%m-%d"),
):
try:
dt = datetime.strptime(candidate, fmt)
break
except ValueError:
continue
if dt is None:
return text
return (
f"{dt.year:04d}年{dt.month:02d}月{dt.day:02d}日 "
f"{dt.hour:02d}时{dt.minute:02d}分{dt.second:02d}秒"
)
def parse_payload(raw: Any) -> Dict[str, Any]:
if isinstance(raw, dict):
return dict(raw)
if not raw:
return {}
try:
data = json.loads(raw)
except (TypeError, ValueError, json.JSONDecodeError):
return {}
return data if isinstance(data, dict) else {}
def dumps_payload(payload: Dict[str, Any]) -> str:
return json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":"))
def yes_no_from_urls(urls: Optional[str]) -> str:
return "是" if (urls or "").strip() else "否"
def yes_no_from_bool(flag: Optional[bool]) -> str:
if flag is None:
return ""
return "是" if flag else "否"
def merge_row(
*,
tab_key: str,
system_values: Dict[str, Any],
supplement: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
"""系统值优先;override 键补录优先;空用补录填。"""
out = dict(system_values or {})
sup = supplement or {}
overrides = SUPPLEMENT_OVERRIDE_KEYS.get(tab_key, set())
for key, val in sup.items():
text = "" if val is None else str(val).strip()
if not text:
continue
cur = out.get(key)
cur_empty = cur is None or (isinstance(cur, str) and not cur.strip())
if key in overrides or cur_empty:
out[key] = text
return out
def load_qc_tasks_for_sample(db: Session, tenant_no: str, sample_code: str) -> List[Dict[str, Any]]:
code = (sample_code or "").strip()
if not code:
return []
rows = (
db.query(
BizQcTask.id,
BizQcTask.item_name,
BizQcTask.completed_time,
BizQcTask.update_time,
BizQcWorkOrderInspectionBatch.batch_name,
BizQcTaskResult.param_name,
BizQcTaskResult.param_value,
)
.join(BizQcWorkOrder, BizQcWorkOrder.id == BizQcTask.work_order_id)
.outerjoin(
BizQcWorkOrderInspectionBatch,
BizQcWorkOrderInspectionBatch.id == BizQcTask.batch_id,
)
.outerjoin(
BizQcTaskResult,
(BizQcTaskResult.task_id == BizQcTask.id) & (BizQcTaskResult.del_flag == 0),
)
.filter(
BizQcWorkOrder.tenant_no == tenant_no,
BizQcWorkOrder.sample_code == code,
BizQcWorkOrder.del_flag == 0,
BizQcTask.tenant_no == tenant_no,
BizQcTask.del_flag == 0,
BizQcTask.task_status == "ywc",
)
.all()
)
grouped: Dict[str, Dict[str, Any]] = {}
for row in rows:
task = grouped.setdefault(
row.id,
{
"item_name": row.item_name or "",
"batch_kind": batch_kind_from_name(row.batch_name),
"stamp": _stamp(row.completed_time, row.update_time),
"params": [],
"values": [],
},
)
if (row.param_name or "").strip() or (row.param_value or "").strip():
task["params"].append((row.param_name or "", row.param_value or ""))
if (row.param_value or "").strip():
task["values"].append(row.param_value or "")
return list(grouped.values())
def extract_mother_blood(tasks: List[Dict[str, Any]], *, include_nucleic: bool = True) -> Dict[str, str]:
rules = _MOTHER_SEROLOGY_RULES
if not include_nucleic:
rules = tuple(
r
for r in rules
if r[0] not in {"mother_hbv_dna", "mother_hcv_rna", "mother_hiv1_rna"}
)
raw = _serology_from_tasks(tasks, rules)
mapping = {
"mother_hbsag": "motherHbsag",
"mother_hbv_dna": "motherHbvDna",
"mother_anti_hcv": "motherAntiHcv",
"mother_hcv_rna": "motherHcvRna",
"mother_anti_tp": "motherAntiTp",
"mother_cmv_igm": "motherCmvIgm",
"mother_anti_hiv": "motherAntiHiv",
"mother_hiv1_rna": "motherHiv1Rna",
}
out: Dict[str, str] = {}
for src, dst in mapping.items():
val = normalize_negative_positive(raw.get(src) or "") or (raw.get(src) or "")
if val:
# 台账 Excel 多用「阴」
if val in ("阴性", "阴"):
out[dst] = "阴"
elif val in ("阳性", "阳"):
out[dst] = "阳"
else:
out[dst] = val
return out
def extract_process_tests(tasks: List[Dict[str, Any]]) -> Dict[str, str]:
raw = _process_qual_from_tasks(tasks)
bacteria_values: List[str] = []
for task in tasks:
name = task.get("item_name") or ""
params = task.get("params") or []
hit = "细菌" in name or any("细菌" in (pn or "") for pn, _ in params)
if not hit:
continue
for _, pv in params:
bacteria_values.append(pv)
bacteria_values.extend(task.get("values") or [])
bacteria = qual_yin_yang(bacteria_values)
out = {
"endotoxinTest": _to_yin(raw.get("endotoxin_test")),
"fungusTest": _to_yin(raw.get("fungus_test")),
"mycoplasmaTest": _to_yin(raw.get("mycoplasma_test")),
}
if bacteria:
out["bacteriaTest"] = _to_yin(bacteria)
return {k: v for k, v in out.items() if v}
def _to_yin(value: Optional[str]) -> str:
text = (value or "").strip()
if text in ("阴性", "阴"):
return "阴"
if text in ("阳性", "阳"):
return "阳"
return text
def infectious_label(code: Optional[str]) -> str:
c = (code or "").strip()
if c == "zc":
return "正常"
if c == "bzc":
return "不正常"
return c
@@ -0,0 +1,644 @@
"""实验室样本处理数据 — 聚合读取、Excel 导入与部门字段回写。"""
from __future__ import annotations
import json
import uuid
from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.models.models import (
BizCellSampleType,
BizCollectionHospital,
BizProtocolCollectionInfo,
BizProtocolPartyCustomer,
BizProtocolSample,
BizSampleProductionReception,
BizSampleReception,
)
from app.services.lab_sample_processing_fields import (
DATASET_HSC,
DATASET_MSC,
DATASET_UBMC,
DEPT_EDIT_PERMS,
PERM_LIST,
PERM_MANAGER,
SOURCE_EXTRA,
SOURCE_PRODUCTION,
SOURCE_PROTOCOL,
SOURCE_RECEPTION,
editable_depts_for_user,
field_map,
fields_for,
resolve_dataset_by_type_name,
)
def _new_id() -> str:
return uuid.uuid4().hex.upper()
def _fmt_dt(v: Any) -> Optional[str]:
if v is None:
return None
if isinstance(v, datetime):
return v.strftime("%Y-%m-%d %H:%M:%S")
return str(v)
def _fmt_num(v: Any) -> Any:
if v is None:
return None
if isinstance(v, Decimal):
return float(v)
return v
def _parse_extra(raw: Optional[str]) -> Dict[str, Any]:
if not raw:
return {}
try:
data = json.loads(raw)
except Exception:
return {}
return data if isinstance(data, dict) else {}
def user_acl(db: Session, user: dict) -> Dict[str, Any]:
from app.core.menu_guard import user_has_menu_perm
is_manager = user_has_menu_perm(db, user, PERM_MANAGER)
dept_flags = {
dept: user_has_menu_perm(db, user, code) for dept, code in DEPT_EDIT_PERMS.items()
}
editable = editable_depts_for_user(is_manager=is_manager, dept_flags=dept_flags)
return {
"canList": user_has_menu_perm(db, user, PERM_LIST) or is_manager or any(dept_flags.values()),
"isManager": is_manager,
"editableDepts": sorted(editable),
"deptFlags": dept_flags,
}
def _type_name_filters(dataset: str):
if dataset == DATASET_HSC:
return "%造血%"
if dataset == DATASET_MSC:
return "%间充质%"
if dataset == DATASET_UBMC:
return "%单个核%"
return None
def _code_prefix_filters(dataset: str) -> List[str]:
if dataset == DATASET_HSC:
return ["B%"]
if dataset == DATASET_MSC:
return ["BC%", "UC%"]
if dataset == DATASET_UBMC:
return ["UBMC%"]
return []
def _match_keyword(row: Dict[str, Any], kw: str) -> bool:
if not kw:
return True
text = f"{row.get('sampleCode') or ''} {row.get('motherName') or ''}"
return kw.lower() in text.lower()
def _match_date_range(row: Dict[str, Any], date_from: Optional[str], date_to: Optional[str]) -> bool:
if not date_from and not date_to:
return True
for key in ("receptionTime", "collectTime", "cryoTime", "storageDateP2"):
val = row.get(key)
if not val:
continue
day = str(val)[:10].replace(".", "-").replace("/", "-")
# normalize compact dates like 2025-9-8
parts = day.split("-")
if len(parts) == 3:
try:
day = f"{int(parts[0]):04d}-{int(parts[1]):02d}-{int(parts[2]):02d}"
except Exception:
pass
if date_from and day < date_from:
continue
if date_to and day > date_to:
continue
return True
return False
def list_rows(
db: Session,
*,
tenant_no: str,
dataset: str,
keyword: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
page: int = 1,
page_size: int = 20,
) -> Tuple[List[Dict[str, Any]], int]:
from app.models.models import BizLabSampleProcessingExtra
type_like = _type_name_filters(dataset)
code_likes = _code_prefix_filters(dataset)
q = (
db.query(
BizProtocolSample,
BizCellSampleType,
BizProtocolPartyCustomer,
BizProtocolCollectionInfo,
BizCollectionHospital,
)
.outerjoin(
BizCellSampleType,
(BizCellSampleType.id == BizProtocolSample.sample_type_id)
& (BizCellSampleType.del_flag == 0),
)
.outerjoin(
BizProtocolPartyCustomer,
(BizProtocolPartyCustomer.collection_protocol_id == BizProtocolSample.collection_protocol_id)
& (BizProtocolPartyCustomer.tenant_no == tenant_no)
& (BizProtocolPartyCustomer.del_flag == 0),
)
.outerjoin(
BizProtocolCollectionInfo,
(BizProtocolCollectionInfo.collection_protocol_id == BizProtocolSample.collection_protocol_id)
& (BizProtocolCollectionInfo.tenant_no == tenant_no)
& (BizProtocolCollectionInfo.del_flag == 0),
)
.outerjoin(
BizCollectionHospital,
(BizCollectionHospital.id == BizProtocolCollectionInfo.hospital_id)
& (BizCollectionHospital.del_flag == 0),
)
.filter(
BizProtocolSample.tenant_no == tenant_no,
BizProtocolSample.del_flag == 0,
BizProtocolSample.sample_code.isnot(None),
BizProtocolSample.sample_code != "",
)
)
type_or_code = []
if type_like:
type_or_code.append(BizCellSampleType.type_name.like(type_like))
for pref in code_likes:
type_or_code.append(BizProtocolSample.sample_code.like(pref))
if type_or_code:
q = q.filter(or_(*type_or_code))
rows = q.order_by(BizProtocolSample.create_time.desc()).all()
filtered = []
for sample, stype, party, coll, hospital in rows:
type_name = (stype.type_name if stype else "") or ""
code = sample.sample_code or ""
if resolve_dataset_by_type_name(type_name, code) != dataset:
continue
filtered.append((sample, stype, party, coll, hospital))
extras = {
e.sample_code: e
for e in db.query(BizLabSampleProcessingExtra)
.filter(
BizLabSampleProcessingExtra.tenant_no == tenant_no,
BizLabSampleProcessingExtra.del_flag == 0,
BizLabSampleProcessingExtra.dataset == dataset,
)
.all()
}
codes = list({*(s.sample_code for s, *_ in filtered if s.sample_code), *extras.keys()})
receptions: Dict[str, BizSampleReception] = {}
productions: Dict[str, BizSampleProductionReception] = {}
if codes:
for r in (
db.query(BizSampleReception)
.filter(
BizSampleReception.tenant_no == tenant_no,
BizSampleReception.del_flag == 0,
BizSampleReception.sample_code.in_(codes),
)
.order_by(BizSampleReception.reception_time.desc())
.all()
):
receptions.setdefault(r.sample_code, r)
if receptions:
rids = [r.id for r in receptions.values()]
for p in (
db.query(BizSampleProductionReception)
.filter(
BizSampleProductionReception.tenant_no == tenant_no,
BizSampleProductionReception.del_flag == 0,
BizSampleProductionReception.reception_id.in_(rids),
)
.order_by(BizSampleProductionReception.process_time.desc())
.all()
):
productions.setdefault(p.reception_id, p)
built_map: Dict[str, Dict[str, Any]] = {}
for sample, stype, party, coll, hospital in filtered:
code = sample.sample_code
reception = receptions.get(code)
production = productions.get(reception.id) if reception else None
extra = _parse_extra(extras[code].payload_json if code in extras else None)
built_map[code] = _build_row(
dataset=dataset,
sample=sample,
type_name=(stype.type_name if stype else None),
party=party,
coll=coll,
hospital=hospital,
reception=reception,
production=production,
extra=extra,
)
# Excel/补充表独有样本也要展示
for code, extra_row in extras.items():
if code in built_map:
continue
extra = _parse_extra(extra_row.payload_json)
reception = receptions.get(code)
production = productions.get(reception.id) if reception else None
built_map[code] = _build_row(
dataset=dataset,
sample=None,
type_name=None,
party=None,
coll=None,
hospital=None,
reception=reception,
production=production,
extra=extra,
)
kw = (keyword or "").strip()
built = []
for row in built_map.values():
if not _match_keyword(row, kw):
continue
if not _match_date_range(row, date_from, date_to):
continue
built.append(row)
# 稳定排序:优先按样本编码
built.sort(key=lambda r: str(r.get("sampleCode") or ""))
total = len(built)
page = max(1, int(page or 1))
page_size = max(1, min(5000, int(page_size or 20)))
start = (page - 1) * page_size
return built[start : start + page_size], total
def _build_row(
*,
dataset: str,
sample: Optional[BizProtocolSample],
type_name: Optional[str],
party: Optional[BizProtocolPartyCustomer],
coll: Optional[BizProtocolCollectionInfo],
hospital: Optional[BizCollectionHospital],
reception: Optional[BizSampleReception],
production: Optional[BizSampleProductionReception],
extra: Dict[str, Any],
) -> Dict[str, Any]:
protocol_vals = {
"sample_code": sample.sample_code if sample else extra.get("sampleCode"),
"mother_name": party.mother_name if party else None,
"sample_collect_time": coll.sample_collect_time if coll else None,
"hospital_name": hospital.hospital_name if hospital else None,
}
reception_vals = {}
if reception:
reception_vals = {
"reception_time": reception.reception_time,
"receive_temperature": reception.receive_temperature,
"handler_name": reception.handler_name,
"remark": reception.remark,
"lab_storage_code": reception.lab_storage_code,
}
production_vals = {}
if production:
production_vals = {
"original_blood_weight_g": production.original_blood_weight_g,
"hes_volume_ml": production.hes_volume_ml,
"blood_type": production.blood_type,
"infectious_four_test_result": production.infectious_four_test_result,
"enriched_blood_weight_g": production.enriched_blood_weight_g,
"cryopreservation_volume_ml": production.cryopreservation_volume_ml,
"handler_name": production.handler_name,
"remark": production.remark,
"original_blood_volume_ml": production.original_blood_volume_ml,
"nucleated_cell_count_1e8": production.nucleated_cell_count_1e8,
"inoculation_amount": production.inoculation_amount,
"cryopreservation_time": production.cryopreservation_time,
"storage_total_tubes": production.storage_total_tubes,
"available_cell_count_1e7": production.available_cell_count_1e7,
}
sample_code = (
(sample.sample_code if sample else None)
or extra.get("sampleCode")
or ""
)
data: Dict[str, Any] = {
"id": (sample.id if sample else f"extra:{dataset}:{sample_code}"),
"dataset": dataset,
"sampleCode": sample_code,
"sampleTypeName": type_name,
"protocolSampleId": sample.id if sample else None,
"collectionProtocolId": sample.collection_protocol_id if sample else None,
"receptionId": reception.id if reception else None,
"productionReceptionId": production.id if production else None,
"fromExtraOnly": sample is None,
}
for f in fields_for(dataset):
val = None
if f.source == SOURCE_PROTOCOL:
val = protocol_vals.get(f.attr)
elif f.source == SOURCE_RECEPTION:
val = reception_vals.get(f.attr)
elif f.source == SOURCE_PRODUCTION:
val = production_vals.get(f.attr)
elif f.source == SOURCE_EXTRA:
val = extra.get(f.key)
# 导入/补充表优先:有值则覆盖源表,保证 Excel 录入可见
if f.key in extra and extra.get(f.key) not in (None, ""):
val = extra.get(f.key)
if isinstance(val, datetime):
val = _fmt_dt(val)
elif isinstance(val, Decimal):
val = _fmt_num(val)
data[f.key] = val
if not data.get("sampleCode"):
data["sampleCode"] = sample_code
return data
def upsert_extra_payload(
db: Session,
*,
tenant_no: str,
dataset: str,
sample_code: str,
payload: Dict[str, Any],
operator_id: str = "import",
replace: bool = False,
) -> None:
from app.models.models import BizLabSampleProcessingExtra
code = (sample_code or "").strip()
if not code:
raise ValueError("样本编码不能为空")
now = datetime.now()
row = (
db.query(BizLabSampleProcessingExtra)
.filter(
BizLabSampleProcessingExtra.tenant_no == tenant_no,
BizLabSampleProcessingExtra.dataset == dataset,
BizLabSampleProcessingExtra.sample_code == code,
BizLabSampleProcessingExtra.del_flag == 0,
)
.first()
)
current = {} if replace else _parse_extra(row.payload_json if row else None)
for k, v in (payload or {}).items():
if v in (None, ""):
continue
current[k] = v
current["sampleCode"] = code
if row is None:
db.add(
BizLabSampleProcessingExtra(
id=_new_id(),
dataset=dataset,
sample_code=code,
payload_json=json.dumps(current, ensure_ascii=False),
tenant_no=tenant_no,
create_by=operator_id,
create_dept_id="",
create_time=now,
update_by=operator_id,
update_time=now,
del_flag=0,
)
)
else:
row.payload_json = json.dumps(current, ensure_ascii=False)
row.update_by = operator_id
row.update_time = now
def update_row(
db: Session,
*,
tenant_no: str,
dataset: str,
sample_code: str,
payload: Dict[str, Any],
editable_depts: set,
operator_id: str,
) -> Dict[str, Any]:
code = (sample_code or "").strip()
if not code:
raise ValueError("样本编码不能为空")
fmap = field_map(dataset)
if not fmap:
raise ValueError("未知数据集")
denied = []
accepted: Dict[str, Any] = {}
for key, value in (payload or {}).items():
f = fmap.get(key)
if not f:
continue
if f.dept not in editable_depts:
denied.append(key)
continue
accepted[key] = value
if denied:
raise PermissionError(f"无权编辑字段: {', '.join(denied)}")
if not accepted:
raise ValueError("没有可保存的字段")
sample = (
db.query(BizProtocolSample)
.filter(
BizProtocolSample.tenant_no == tenant_no,
BizProtocolSample.sample_code == code,
BizProtocolSample.del_flag == 0,
)
.first()
)
reception = (
db.query(BizSampleReception)
.filter(
BizSampleReception.tenant_no == tenant_no,
BizSampleReception.sample_code == code,
BizSampleReception.del_flag == 0,
)
.order_by(BizSampleReception.reception_time.desc())
.first()
)
production = None
if reception:
production = (
db.query(BizSampleProductionReception)
.filter(
BizSampleProductionReception.tenant_no == tenant_no,
BizSampleProductionReception.reception_id == reception.id,
BizSampleProductionReception.del_flag == 0,
)
.order_by(BizSampleProductionReception.process_time.desc())
.first()
)
party = None
coll = None
if sample:
party = (
db.query(BizProtocolPartyCustomer)
.filter(
BizProtocolPartyCustomer.collection_protocol_id == sample.collection_protocol_id,
BizProtocolPartyCustomer.tenant_no == tenant_no,
BizProtocolPartyCustomer.del_flag == 0,
)
.first()
)
coll = (
db.query(BizProtocolCollectionInfo)
.filter(
BizProtocolCollectionInfo.collection_protocol_id == sample.collection_protocol_id,
BizProtocolCollectionInfo.tenant_no == tenant_no,
BizProtocolCollectionInfo.del_flag == 0,
)
.first()
)
now = datetime.now()
extra_payload: Dict[str, Any] = {}
def _set_dt(obj, attr, value):
if value in (None, ""):
setattr(obj, attr, None)
return
if isinstance(value, datetime):
setattr(obj, attr, value)
return
text = str(value).strip().replace("/", "-").replace(".", "-")
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
try:
setattr(obj, attr, datetime.strptime(text, fmt))
return
except ValueError:
continue
setattr(obj, attr, value)
for key, value in accepted.items():
f = fmap[key]
wrote = False
if f.source == SOURCE_PROTOCOL and sample is not None:
if f.attr == "mother_name" and party is not None:
party.mother_name = None if value in (None, "") else str(value)
party.update_by = operator_id
party.update_time = now
wrote = True
elif f.attr == "sample_collect_time" and coll is not None:
_set_dt(coll, "sample_collect_time", value)
coll.update_by = operator_id
coll.update_time = now
wrote = True
elif f.source == SOURCE_RECEPTION and reception is not None:
if f.attr == "reception_time":
_set_dt(reception, f.attr, value)
elif f.attr in {"receive_temperature", "handler_name", "remark", "lab_storage_code"}:
setattr(reception, f.attr, None if value in (None, "") else value)
reception.update_by = operator_id
reception.update_time = now
wrote = True
elif f.source == SOURCE_PRODUCTION and production is not None:
numeric_attrs = {
"original_blood_weight_g",
"hes_volume_ml",
"enriched_blood_weight_g",
"cryopreservation_volume_ml",
"original_blood_volume_ml",
"nucleated_cell_count_1e8",
"available_cell_count_1e7",
}
if f.attr == "cryopreservation_time":
_set_dt(production, f.attr, value)
elif f.attr == "storage_total_tubes":
production.storage_total_tubes = None if value in (None, "") else int(value)
elif f.attr in numeric_attrs:
production_val = None if value in (None, "") else Decimal(str(value))
setattr(production, f.attr, production_val)
elif f.attr in {
"blood_type",
"infectious_four_test_result",
"handler_name",
"remark",
"inoculation_amount",
}:
setattr(production, f.attr, None if value in (None, "") else value)
production.update_by = operator_id
production.update_time = now
wrote = True
# 补充表始终保留一份,保证导入台账与无权源表场景可编辑
extra_payload[key] = value
_ = wrote
upsert_extra_payload(
db,
tenant_no=tenant_no,
dataset=dataset,
sample_code=code,
payload=extra_payload,
operator_id=operator_id,
replace=False,
)
db.commit()
rows, _ = list_rows(db, tenant_no=tenant_no, dataset=dataset, keyword=code, page=1, page_size=20)
for row in rows:
if row.get("sampleCode") == code:
return row
raise LookupError("保存成功但读取失败")
def export_rows(
db: Session,
*,
tenant_no: str,
dataset: str,
keyword: Optional[str] = None,
date_from: Optional[str] = None,
date_to: Optional[str] = None,
) -> Tuple[List[str], List[List[Any]]]:
rows, _ = list_rows(
db,
tenant_no=tenant_no,
dataset=dataset,
keyword=keyword,
date_from=date_from,
date_to=date_to,
page=1,
page_size=5000,
)
headers = [f.label for f in fields_for(dataset)]
keys = [f.key for f in fields_for(dataset)]
body = [[row.get(k) for k in keys] for row in rows]
return headers, body
@@ -0,0 +1,176 @@
"""实验室样本处理数据 — 字段目录与部门归属。"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
DATASET_HSC = "hsc"
DATASET_UBMC = "ubmc"
DATASET_MSC = "msc"
DATASETS = (DATASET_HSC, DATASET_UBMC, DATASET_MSC)
DEPT_MARKET = "market"
DEPT_PRODUCTION = "production"
DEPT_QC = "qc"
DEPTS = (DEPT_MARKET, DEPT_PRODUCTION, DEPT_QC)
PERM_LIST = "lab:sample-processing:list"
PERM_MARKET_EDIT = "lab:sample-processing:market:edit"
PERM_PRODUCTION_EDIT = "lab:sample-processing:production:edit"
PERM_QC_EDIT = "lab:sample-processing:qc:edit"
PERM_MANAGER = "lab:sample-processing:manager"
DEPT_EDIT_PERMS = {
DEPT_MARKET: PERM_MARKET_EDIT,
DEPT_PRODUCTION: PERM_PRODUCTION_EDIT,
DEPT_QC: PERM_QC_EDIT,
}
SOURCE_PROTOCOL = "protocol"
SOURCE_RECEPTION = "reception"
SOURCE_PRODUCTION = "production_reception"
SOURCE_EXTRA = "extra"
@dataclass(frozen=True)
class FieldDef:
key: str
label: str
dept: str
source: str
attr: Optional[str] = None # ORM attr on source object when applicable
def _f(key: str, label: str, dept: str, source: str, attr: Optional[str] = None) -> FieldDef:
return FieldDef(key=key, label=label, dept=dept, source=source, attr=attr or key)
HSC_FIELDS: Tuple[FieldDef, ...] = (
_f("sampleCode", "细胞编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
_f("motherName", "准妈妈姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
_f("collectTime", "采集时间", DEPT_MARKET, SOURCE_PROTOCOL, "sample_collect_time"),
_f("collectHospital", "采集医院", DEPT_MARKET, SOURCE_PROTOCOL, "hospital_name"),
_f("receptionTime", "接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
_f("receiveTemperature", "样品温度到达/℃", DEPT_PRODUCTION, SOURCE_RECEPTION, "receive_temperature"),
_f("receiveCheck", "收样检查", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("hasMaternalBlood", "是否有母血", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("remeasureMaternalBlood", "是否重采母血", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("hasConsent", "知情同意书或协议书", DEPT_MARKET, SOURCE_EXTRA),
_f("hasPackingList", "脐血运输装箱清单", DEPT_MARKET, SOURCE_EXTRA),
_f("hasReceiveConfirm", "脐带血接收确认单", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("hasHealthSurvey", "孕妇健康调查表", DEPT_MARKET, SOURCE_EXTRA),
_f("labelClear", "标识清晰", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
_f("receptionRemark", "备注", DEPT_PRODUCTION, SOURCE_RECEPTION, "remark"),
_f("originalBloodWeightG", "原血重量/g", DEPT_PRODUCTION, SOURCE_PRODUCTION, "original_blood_weight_g"),
_f("hesVolumeMl", "HES用量/ml", DEPT_PRODUCTION, SOURCE_PRODUCTION, "hes_volume_ml"),
_f("bloodType", "血型", DEPT_PRODUCTION, SOURCE_PRODUCTION, "blood_type"),
_f("infectiousFourCard", "传染四项检测(检测卡)", DEPT_QC, SOURCE_PRODUCTION, "infectious_four_test_result"),
_f("infectiousFourExternal", "传染四项检测(外检)", DEPT_QC, SOURCE_EXTRA),
_f("preEnrichTnc", "富集前有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("enrichedBloodWeightG", "富集后血量/g", DEPT_PRODUCTION, SOURCE_PRODUCTION, "enriched_blood_weight_g"),
_f("postEnrichTnc", "富集后有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("cryoVolumeMl", "冻存液用量/ml", DEPT_PRODUCTION, SOURCE_PRODUCTION, "cryopreservation_volume_ml"),
_f("cd34Count", "CD34+细胞计数/*10E4", DEPT_QC, SOURCE_EXTRA),
_f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
_f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)
UBMC_FIELDS: Tuple[FieldDef, ...] = (
_f("sampleCode", "细胞编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
_f("motherName", "准妈妈姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
_f("labMaterialCode", "实验室物料编码", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("labStorageCode", "入库编码", DEPT_PRODUCTION, SOURCE_RECEPTION, "lab_storage_code"),
_f("storageStatus", "储存状态", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("discardReason", "废弃原因", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("receptionTime", "接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
_f("receiveTemperature", "样品温度到达/℃", DEPT_PRODUCTION, SOURCE_RECEPTION, "receive_temperature"),
_f("receiveCheck", "收样检查", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
_f("receptionRemark", "备注", DEPT_PRODUCTION, SOURCE_RECEPTION, "remark"),
_f("originalBloodVolumeMl", "原血体积/mL", DEPT_PRODUCTION, SOURCE_PRODUCTION, "original_blood_volume_ml"),
_f("nucleatedCellCount1e8", "有核细胞数/108", DEPT_PRODUCTION, SOURCE_PRODUCTION, "nucleated_cell_count_1e8"),
_f("preOpTnc", "操作前有核细胞数/109/L", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("infectiousFourCard", "传染四项检测(检测卡)", DEPT_QC, SOURCE_PRODUCTION, "infectious_four_test_result"),
_f("hbsag", "乙型肝炎病毒表面抗原(HBsAg)", DEPT_QC, SOURCE_EXTRA),
_f("antiHcv", "丙型肝炎病毒抗体(抗HCV)", DEPT_QC, SOURCE_EXTRA),
_f("antiTp", "梅毒螺旋体特异性抗体(抗TP)", DEPT_QC, SOURCE_EXTRA),
_f("antiHiv", "人类免疫缺陷病毒抗体(抗HIV-1/2)", DEPT_QC, SOURCE_EXTRA),
_f("cmvIgm", "巨细胞病毒IgM 抗体 (CMV-IgM)", DEPT_QC, SOURCE_EXTRA),
_f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
_f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)
MSC_FIELDS: Tuple[FieldDef, ...] = (
_f("sampleCode", "实验室流转编码", DEPT_MARKET, SOURCE_PROTOCOL, "sample_code"),
_f("motherName", "母亲姓名", DEPT_MARKET, SOURCE_PROTOCOL, "mother_name"),
_f("collectTime", "脐带采集时间", DEPT_MARKET, SOURCE_PROTOCOL, "sample_collect_time"),
_f("transportTempC", "运输温度/℃", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("receptionTime", "脐带接收时间", DEPT_PRODUCTION, SOURCE_RECEPTION, "reception_time"),
_f("receptionHandler", "接收登记人", DEPT_PRODUCTION, SOURCE_RECEPTION, "handler_name"),
_f("inoculationAmount", "接种量", DEPT_PRODUCTION, SOURCE_PRODUCTION, "inoculation_amount"),
_f("storageStatus", "储存状态", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("cryoTime", "冻存时间", DEPT_PRODUCTION, SOURCE_PRODUCTION, "cryopreservation_time"),
_f("storageTotalTubes", "储存总量(管)", DEPT_PRODUCTION, SOURCE_PRODUCTION, "storage_total_tubes"),
_f("availableCellCount1e7", "可提供细胞数量/*10E7", DEPT_PRODUCTION, SOURCE_PRODUCTION, "available_cell_count_1e7"),
_f("endotoxinTest", "内毒素检测", DEPT_QC, SOURCE_EXTRA),
_f("bacteriaTest", "细菌检测", DEPT_QC, SOURCE_EXTRA),
_f("fungusTest", "真菌检测", DEPT_QC, SOURCE_EXTRA),
_f("mycoplasmaTest", "支原体检测", DEPT_QC, SOURCE_EXTRA),
_f("operatorName", "操作人", DEPT_PRODUCTION, SOURCE_PRODUCTION, "handler_name"),
_f("motherHbsag", "乙型肝炎病毒表面抗原(HBsAg)", DEPT_QC, SOURCE_EXTRA),
_f("motherHbvDna", "乙型肝炎病毒核酸(HBV DNA)", DEPT_QC, SOURCE_EXTRA),
_f("motherAntiHcv", "丙型肝炎病毒抗体(抗HCV)", DEPT_QC, SOURCE_EXTRA),
_f("motherHcvRna", "丙型肝炎病毒核酸(HCV RNA)", DEPT_QC, SOURCE_EXTRA),
_f("motherAntiTp", "梅毒螺旋体特异性抗体(抗TP)", DEPT_QC, SOURCE_EXTRA),
_f("motherCmvIgm", "巨细胞病毒lgM 抗体 (CMV lgM)", DEPT_QC, SOURCE_EXTRA),
_f("motherAntiHiv", "人类免疫缺陷病毒抗体(抗HIV-1/2)", DEPT_QC, SOURCE_EXTRA),
_f("motherHiv1Rna", "人类免疫缺陷病毒核酸(1型) (HIV-1RNA)", DEPT_QC, SOURCE_EXTRA),
_f("storageDateP2", "储存日期(P2冻存日期)", DEPT_PRODUCTION, SOURCE_EXTRA),
_f("processRemark", "备注", DEPT_PRODUCTION, SOURCE_PRODUCTION, "remark"),
)
FIELDS_BY_DATASET: Dict[str, Tuple[FieldDef, ...]] = {
DATASET_HSC: HSC_FIELDS,
DATASET_UBMC: UBMC_FIELDS,
DATASET_MSC: MSC_FIELDS,
}
DATASET_LABELS = {
DATASET_HSC: "造血干细胞数据",
DATASET_UBMC: "单个核数据",
DATASET_MSC: "脐带间充质干细胞数据",
}
def fields_for(dataset: str) -> Tuple[FieldDef, ...]:
return FIELDS_BY_DATASET.get(dataset) or ()
def field_map(dataset: str) -> Dict[str, FieldDef]:
return {f.key: f for f in fields_for(dataset)}
def editable_depts_for_user(*, is_manager: bool, dept_flags: Dict[str, bool]) -> set:
if is_manager:
return set(DEPTS)
return {d for d, ok in dept_flags.items() if ok}
def resolve_dataset_by_type_name(name: Optional[str], sample_code: Optional[str] = None) -> Optional[str]:
text = (name or "").strip()
code = (sample_code or "").strip().upper()
if "单个核" in text or "脐血单个核" in text or code.startswith("UBMC"):
return DATASET_UBMC
if "间充质" in text or code.startswith("BC") or code.startswith("UC"):
return DATASET_MSC
if "造血" in text or code.startswith("B"):
return DATASET_HSC
return None
def schema_payload(dataset: str) -> List[Dict[str, str]]:
return [
{"key": f.key, "label": f.label, "dept": f.dept, "source": f.source}
for f in fields_for(dataset)
]
@@ -0,0 +1,324 @@
"""将会议 Excel 三张表导入实验室样本处理补充台账。
用法(工程/worktree 根或 backend 下):
ENV_FILE=.../backend/.env python backend/scripts/import_lab_sample_processing_xlsx.py --xlsx PATH
"""
from __future__ import annotations
import argparse
import re
import sys
import zipfile
import xml.etree.ElementTree as ET
from pathlib import Path
# 允许直接脚本运行
ROOT = Path(__file__).resolve().parents[2]
BACKEND = ROOT / "backend"
if str(BACKEND) not in sys.path:
sys.path.insert(0, str(BACKEND))
from sqlalchemy.orm import Session # noqa: E402
from app.core.config import settings # noqa: E402
from app.core.database import SessionLocal # noqa: E402
from app.services.lab_sample_processing import upsert_extra_payload # noqa: E402
from app.services.lab_sample_processing_fields import ( # noqa: E402
DATASET_HSC,
DATASET_MSC,
DATASET_UBMC,
fields_for,
)
NS = {
"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
}
SHEET_DATASET = {
"造血干细胞数据": DATASET_HSC,
"单个核数据": DATASET_UBMC,
"脐带间充质干细胞数据": DATASET_MSC,
}
LABEL_ALIASES = {
"细胞编码": "sampleCode",
"实验室流转编码": "sampleCode",
"准妈妈姓名": "motherName",
"母亲姓名": "motherName",
"采集时间": "collectTime",
"脐带采集时间": "collectTime",
"采集医院": "collectHospital",
"接收时间": "receptionTime",
"脐带接收时间": "receptionTime",
"样品温度到达/℃": "receiveTemperature",
"运输温度/℃": "transportTempC",
"运输温度\n/℃": "transportTempC",
"收样检查": "receiveCheck",
"是否有母血": "hasMaternalBlood",
"是否重采母血": "remeasureMaternalBlood",
"知情同意书或协议书": "hasConsent",
"脐血运输装箱清单": "hasPackingList",
"脐带血接收确认单": "hasReceiveConfirm",
"孕妇健康调查表": "hasHealthSurvey",
"标识清晰": "labelClear",
"接收登记人": "receptionHandler",
"原血重量/g": "originalBloodWeightG",
"HES用量/ml": "hesVolumeMl",
"血型": "bloodType",
"传染四项检测(检测卡)": "infectiousFourCard",
"传染四项检测(外检)": "infectiousFourExternal",
"富集前有核细胞数/109/L": "preEnrichTnc",
"富集后血量/g": "enrichedBloodWeightG",
"富集后有核细胞数/109/L": "postEnrichTnc",
"冻存液用量/ml": "cryoVolumeMl",
"CD34+细胞计数/*10E4": "cd34Count",
"操作人": "operatorName",
"实验室物料编码": "labMaterialCode",
"入库编码": "labStorageCode",
"储存状态": "storageStatus",
"废弃原因": "discardReason",
"原血体积/mL": "originalBloodVolumeMl",
"有核细胞数/108": "nucleatedCellCount1e8",
"操作前有核细胞数/109/L": "preOpTnc",
"乙型肝炎病毒表面抗原(HBsAg)": "hbsag",
"丙型肝炎病毒抗体(抗HCV)": "antiHcv",
"梅毒螺旋体特异性抗体(抗TP)": "antiTp",
"人类免疫缺陷病毒抗体(抗HIV-1/2)": "antiHiv",
"巨细胞病毒IgM 抗体 (CMV-IgM)": "cmvIgm",
"接种量": "inoculationAmount",
"冻存时间": "cryoTime",
"储存总量(管)": "storageTotalTubes",
"储存总量\n(管)": "storageTotalTubes",
"可提供细胞数量/*10E7": "availableCellCount1e7",
"内毒素检测": "endotoxinTest",
"细菌检测": "bacteriaTest",
"真菌检测": "fungusTest",
"支原体检测": "mycoplasmaTest",
"乙型肝炎病毒表面抗原(HBsAg):": "motherHbsag",
"乙型肝炎病毒核酸(HBV DNA)": "motherHbvDna",
"丙型肝炎病毒核酸(HCV RNA)": "motherHcvRna",
"巨细胞病毒lgM 抗体 (CMV lgM)": "motherCmvIgm",
"人类免疫缺陷病毒核酸(1型) (HIV-1RNA)": "motherHiv1Rna",
"储存日期\n(P2冻存日期)": "storageDateP2",
"储存日期(P2冻存日期)": "storageDateP2",
}
def _norm_label(s: str) -> str:
return re.sub(r"\s+", "", (s or "").replace("\r", "").strip())
def _col_row(ref: str):
m = re.match(r"([A-Z]+)(\d+)", ref)
col = m.group(1)
row = int(m.group(2))
n = 0
for ch in col:
n = n * 26 + ord(ch) - 64
return n, row
def read_sheet_matrix(z: zipfile.ZipFile, target: str):
ss = []
if "xl/sharedStrings.xml" in z.namelist():
root = ET.fromstring(z.read("xl/sharedStrings.xml"))
for si in root.findall("m:si", NS):
texts = [
t.text or ""
for t in si.iter("{http://schemas.openxmlformats.org/spreadsheetml/2006/main}t")
]
ss.append("".join(texts))
root = ET.fromstring(z.read(target))
rows = {}
for c in root.findall(".//m:sheetData/m:row/m:c", NS):
ref = c.attrib.get("r")
t = c.attrib.get("t")
v = c.find("m:v", NS)
if v is None or ref is None:
continue
val = v.text or ""
if t == "s":
try:
val = ss[int(val)]
except Exception:
pass
cn, rn = _col_row(ref)
rows.setdefault(rn, {})[cn] = val
if not rows:
return []
max_r = max(rows)
max_c = max(max(cols) for cols in rows.values())
matrix = []
for rn in range(1, max_r + 1):
line = [str(rows.get(rn, {}).get(c, "") or "").strip() for c in range(1, max_c + 1)]
matrix.append(line)
return matrix
def build_header_map(header_rows: list, dataset: str):
"""合并多行表头,按列映射到字段 key。"""
width = max(len(r) for r in header_rows)
merged = [""] * width
for r in header_rows:
for i in range(width):
cell = r[i] if i < len(r) else ""
if cell:
merged[i] = cell if not merged[i] else merged[i]
# MSC: 第二行填充分组子列
if len(header_rows) >= 2:
for i in range(width):
top = header_rows[0][i] if i < len(header_rows[0]) else ""
sub = header_rows[1][i] if i < len(header_rows[1]) else ""
if sub:
merged[i] = sub
elif top and not merged[i]:
merged[i] = top
label_to_key = {_norm_label(f.label): f.key for f in fields_for(dataset)}
for alias, key in LABEL_ALIASES.items():
label_to_key[_norm_label(alias)] = key
# 备注列:第一处 receptionRemark,第二处 processRemark
remark_keys = []
col_map = {}
for idx, label in enumerate(merged):
n = _norm_label(label)
if not n or n in {"序号"}:
continue
if n == "备注":
remark_keys.append(idx)
continue
# 母血相关与单项同名字段在 MSC 中优先 mother*
key = label_to_key.get(n)
if not key and n.endswith(":"):
key = label_to_key.get(n[:-1])
if key:
col_map[idx] = key
if remark_keys:
col_map[remark_keys[0]] = "receptionRemark"
if len(remark_keys) > 1:
col_map[remark_keys[1]] = "processRemark"
elif "processRemark" in {f.key for f in fields_for(dataset)} and dataset != DATASET_UBMC:
# 单备注时,造血第二业务备注也可能只有一列;保留 receptionRemark
pass
# UBMC 仅一列备注时落到 receptionRemark;有操作备注需求时同步 processRemark
return col_map, merged
def parse_workbook(path: Path):
out = {}
with zipfile.ZipFile(path) as z:
wb = ET.fromstring(z.read("xl/workbook.xml"))
sheets = []
for sh in wb.findall("m:sheets/m:sheet", NS):
sheets.append(
(
sh.attrib.get("name"),
sh.attrib.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"),
)
)
rels = ET.fromstring(z.read("xl/_rels/workbook.xml.rels"))
rid_to_target = {rel.attrib["Id"]: rel.attrib["Target"] for rel in rels}
for name, rid in sheets:
dataset = SHEET_DATASET.get(name)
if not dataset:
continue
target = rid_to_target[rid]
if not target.startswith("xl/"):
target = "xl/" + target
matrix = read_sheet_matrix(z, target)
if not matrix:
continue
# MSC 两行表头;其他一行(UBMC 可能首列分组)
if dataset == DATASET_MSC:
header_rows = matrix[:2]
data_rows = matrix[2:]
else:
header_rows = matrix[:1]
data_rows = matrix[1:]
col_map, merged = build_header_map(header_rows, dataset)
records = []
for row in data_rows:
payload = {}
for idx, key in col_map.items():
val = row[idx] if idx < len(row) else ""
if val != "":
payload[key] = val
code = str(payload.get("sampleCode") or "").strip()
if not code:
continue
# UBMC 同名母血检测字段:Excel 用的是非 mother 前缀,已映射
# MSC 母血字段别名
if dataset == DATASET_MSC:
if "antiHcv" in payload and "motherAntiHcv" not in payload:
payload["motherAntiHcv"] = payload.pop("antiHcv")
if "antiTp" in payload and "motherAntiTp" not in payload:
payload["motherAntiTp"] = payload.pop("antiTp")
if "antiHiv" in payload and "motherAntiHiv" not in payload:
payload["motherAntiHiv"] = payload.pop("antiHiv")
if "hbsag" in payload and "motherHbsag" not in payload:
payload["motherHbsag"] = payload.pop("hbsag")
payload["sampleCode"] = code
records.append(payload)
out[dataset] = {"headers": merged, "records": records, "col_map": col_map}
return out
def import_all(db: Session, tenant_no: str, parsed: dict) -> dict:
summary = {}
for dataset, pack in parsed.items():
n = 0
for payload in pack["records"]:
upsert_extra_payload(
db,
tenant_no=tenant_no,
dataset=dataset,
sample_code=payload["sampleCode"],
payload=payload,
operator_id="excel-import",
replace=True,
)
n += 1
summary[dataset] = n
db.commit()
return summary
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--xlsx", required=True)
parser.add_argument("--tenant", default=None)
args = parser.parse_args()
xlsx = Path(args.xlsx)
if not xlsx.exists():
raise SystemExit(f"xlsx not found: {xlsx}")
parsed = parse_workbook(xlsx)
if not parsed:
raise SystemExit("no sheets parsed")
db = SessionLocal()
try:
tenant_no = args.tenant
if not tenant_no:
# 取库中任一有效租户
from app.models.models import SysUser
u = (
db.query(SysUser)
.filter(SysUser.del_flag == 0, SysUser.status == "1", SysUser.tenant_no.isnot(None))
.first()
)
tenant_no = (u.tenant_no if u else None) or "zkxw"
summary = import_all(db, tenant_no, parsed)
print("tenant", tenant_no)
for ds, n in summary.items():
print(f"imported {ds}={n}")
print("ok")
finally:
db.close()
if __name__ == "__main__":
main()
+42
View File
@@ -0,0 +1,42 @@
"""实验室样本处理台账:时间格式与字段合并。"""
from __future__ import annotations
import unittest
from datetime import datetime
from app.services.lab_sample_process import (
format_cn_datetime,
infectious_label,
merge_row,
)
class LabSampleProcessServiceTests(unittest.TestCase):
def test_format_cn_datetime(self):
self.assertEqual(
format_cn_datetime(datetime(2026, 11, 12, 0, 0, 0)),
"2026年11月12日 00时00分00秒",
)
self.assertEqual(
format_cn_datetime("2026-11-12 08:30:01"),
"2026年11月12日 08时30分01秒",
)
self.assertEqual(format_cn_datetime(""), "")
def test_infectious_label(self):
self.assertEqual(infectious_label("zc"), "正常")
self.assertEqual(infectious_label("bzc"), "不正常")
def test_merge_row_system_first_and_override(self):
merged = merge_row(
tab_key="donation_hsc",
system_values={"receiveCheckText": "系统汇总", "motherName": "张三"},
supplement={"receiveCheckText": "补录文案", "labMaterialCode": "M001"},
)
self.assertEqual(merged["receiveCheckText"], "补录文案")
self.assertEqual(merged["labMaterialCode"], "M001")
self.assertEqual(merged["motherName"], "张三")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,53 @@
import unittest
from app.services.lab_sample_processing_fields import (
DATASET_HSC,
DATASET_MSC,
DATASET_UBMC,
DEPT_MARKET,
DEPT_PRODUCTION,
DEPT_QC,
editable_depts_for_user,
field_map,
resolve_dataset_by_type_name,
schema_payload,
)
class LabSampleProcessingFieldsTest(unittest.TestCase):
def test_resolve_dataset(self):
self.assertEqual(resolve_dataset_by_type_name("围产期造血干细胞", "B0000001"), DATASET_HSC)
self.assertEqual(resolve_dataset_by_type_name("脐带间充质干细胞", "BC0000003"), DATASET_MSC)
self.assertEqual(resolve_dataset_by_type_name("", "UBMC2025001"), DATASET_UBMC)
self.assertEqual(resolve_dataset_by_type_name("", "UC2026001"), DATASET_MSC)
def test_schema_not_empty(self):
self.assertGreater(len(schema_payload(DATASET_HSC)), 10)
self.assertGreater(len(schema_payload(DATASET_UBMC)), 10)
self.assertGreater(len(schema_payload(DATASET_MSC)), 10)
def test_dept_acl(self):
self.assertEqual(
editable_depts_for_user(
is_manager=True,
dept_flags={DEPT_MARKET: False, DEPT_PRODUCTION: False, DEPT_QC: False},
),
{DEPT_MARKET, DEPT_PRODUCTION, DEPT_QC},
)
self.assertEqual(
editable_depts_for_user(
is_manager=False,
dept_flags={DEPT_MARKET: False, DEPT_PRODUCTION: True, DEPT_QC: False},
),
{DEPT_PRODUCTION},
)
def test_hsc_market_and_qc_fields(self):
fmap = field_map(DATASET_HSC)
self.assertEqual(fmap["motherName"].dept, DEPT_MARKET)
self.assertEqual(fmap["cd34Count"].dept, DEPT_QC)
self.assertEqual(fmap["originalBloodWeightG"].dept, DEPT_PRODUCTION)
if __name__ == "__main__":
unittest.main()
+7 -4
View File
@@ -79,10 +79,13 @@
"evidence": "用户确认既有 Gitea/Drone Tag 打包已完成,本次不复查或修改流水线。" "evidence": "用户确认既有 Gitea/Drone Tag 打包已完成,本次不复查或修改流水线。"
}, },
"permissions": { "permissions": {
"push_task_branch": false, "push_task_branch": true,
"create_pr": false, "create_pr": true,
"write_issue": false, "write_issue": true,
"merge": false, "close_issue": true,
"merge": true,
"manage_claims": true,
"update_issue_labels": true,
"push_tag": false, "push_tag": false,
"deploy": false "deploy": false
}, },
@@ -0,0 +1,148 @@
-- Issue #98:实验室样本处理数据台账(补录表 + 质检管理菜单)
-- MySQL 5.7,可重复执行。
CREATE TABLE IF NOT EXISTS biz_lab_sample_process_supplement (
id VARCHAR(64) NOT NULL,
lab_storage_code VARCHAR(32) NOT NULL COMMENT '实验室入库编码(唯一业务键)',
reception_source VARCHAR(20) NOT NULL DEFAULT 'cell_storage' COMMENT 'cell_storage|research_donation',
sample_type_name VARCHAR(100) NULL COMMENT '样本类型名称快照',
payload TEXT NOT NULL COMMENT '补录字段 JSON',
tenant_no VARCHAR(64) NOT NULL,
order_num INT NULL,
remark VARCHAR(512) NULL,
create_by VARCHAR(64) NULL,
create_dept_id VARCHAR(64) NOT NULL DEFAULT '',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
update_by VARCHAR(64) NULL,
update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
del_flag SMALLINT NOT NULL DEFAULT 0,
PRIMARY KEY (id),
KEY ix_tenant_lab_code (tenant_no, lab_storage_code, del_flag),
KEY ix_tenant_source_type (tenant_no, reception_source, sample_type_name, del_flag)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 菜单:挂在「质检管理」下(以样本接收菜单同父节点为准)
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
INSERT(INSERT(MD5(CONCAT('lab-sample-process:', parent.tenant_no, ':', parent.id)),
13, 1, '3'), 17, 1, '8'),
'实验室样本处理数据', parent.parent_id, 'C',
'lab-sample-process',
'quality-inspection/lab-sample-process/index',
'biz:labSampleProcess:list',
1, '1', '1', parent.tenant_no, 35,
'质检管理-实验室样本处理数据台账',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu AS parent
WHERE parent.del_flag = 0 AND parent.status = '1'
AND parent.perms = 'biz:sampleReception:list'
AND parent.parent_id IS NOT NULL
AND parent.parent_id != ''
AND NOT EXISTS (
SELECT 1 FROM sys_menu AS existing
WHERE existing.tenant_no = parent.tenant_no
AND existing.perms = 'biz:labSampleProcess:list'
AND existing.del_flag = 0
);
-- 补录按钮
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
INSERT(INSERT(MD5(CONCAT('lab-sample-process-sup:', m.tenant_no, ':', m.id)),
13, 1, '3'), 17, 1, '8'),
'补录', m.id, 'F', '', NULL,
'biz:labSampleProcess:supplement',
1, '1', '1', m.tenant_no, 10,
'实验室样本处理数据-补录',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu AS m
WHERE m.del_flag = 0 AND m.status = '1'
AND m.perms = 'biz:labSampleProcess:list'
AND NOT EXISTS (
SELECT 1 FROM sys_menu AS existing
WHERE existing.parent_id = m.id
AND existing.tenant_no = m.tenant_no
AND existing.perms = 'biz:labSampleProcess:supplement'
AND existing.del_flag = 0
);
-- 导出按钮
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
INSERT(INSERT(MD5(CONCAT('lab-sample-process-exp:', m.tenant_no, ':', m.id)),
13, 1, '3'), 17, 1, '8'),
'导出', m.id, 'F', '', NULL,
'biz:labSampleProcess:export',
1, '1', '1', m.tenant_no, 20,
'实验室样本处理数据-导出',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu AS m
WHERE m.del_flag = 0 AND m.status = '1'
AND m.perms = 'biz:labSampleProcess:list'
AND NOT EXISTS (
SELECT 1 FROM sys_menu AS existing
WHERE existing.parent_id = m.id
AND existing.tenant_no = m.tenant_no
AND existing.perms = 'biz:labSampleProcess:export'
AND existing.del_flag = 0
);
-- 授权给拥有样本接收列表权限的角色
INSERT INTO sys_role_menu (
id, role_id, menu_id, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
INSERT(INSERT(MD5(CONCAT('lab-sample-process-rm:', r.tenant_no, ':', r.id, ':', m.id)),
13, 1, '3'), 17, 1, '8'),
r.id,
m.id,
r.tenant_no,
NULL,
'migration: grant lab sample process from sample reception roles',
'migration',
'',
NOW(),
'migration',
NOW(),
0
FROM sys_role AS r
JOIN sys_role_menu AS src
ON src.role_id = r.id AND src.tenant_no = r.tenant_no AND src.del_flag = 0
JOIN sys_menu AS srcm
ON srcm.id = src.menu_id AND srcm.tenant_no = src.tenant_no AND srcm.del_flag = 0
AND srcm.perms = 'biz:sampleReception:list'
JOIN sys_menu AS m
ON m.tenant_no = r.tenant_no
AND m.perms IN (
'biz:labSampleProcess:list',
'biz:labSampleProcess:supplement',
'biz:labSampleProcess:export'
)
AND m.del_flag = 0
AND m.status = '1'
WHERE r.del_flag = 0
AND NOT EXISTS (
SELECT 1 FROM sys_role_menu AS existing
WHERE existing.role_id = r.id
AND existing.menu_id = m.id
AND existing.tenant_no = r.tenant_no
AND existing.del_flag = 0
);
SELECT tenant_no, parent_id, id, menu_name, path, component, perms
FROM sys_menu
WHERE perms LIKE 'biz:labSampleProcess:%' AND del_flag = 0
ORDER BY tenant_no, order_num, perms;
@@ -0,0 +1,12 @@
-- 纠正实验室样本处理部门 ACL:
-- 「采集接受人」仅挂在「质检专员」职位上,不应再授予市场字段编辑权,
-- 否则质检人员经职位角色并集会同时可改市场+质检字段。
UPDATE sys_role_menu rm
JOIN sys_role r ON r.id = rm.role_id AND r.del_flag = 0 AND r.role_name = '采集接受人'
JOIN sys_menu m ON m.id = rm.menu_id AND m.del_flag = 0
AND m.perms = 'lab:sample-processing:market:edit'
SET rm.del_flag = 1,
rm.update_by = 'migration',
rm.update_time = NOW()
WHERE rm.del_flag = 0;
@@ -0,0 +1,120 @@
-- 实验室样本处理数据:补充表 + 一级模块菜单/按钮权限(不自动授权角色)
-- MySQL 5.7,可重复执行
CREATE TABLE IF NOT EXISTS biz_lab_sample_processing_extra (
id VARCHAR(64) NOT NULL,
dataset VARCHAR(16) NOT NULL COMMENT 'hsc|ubmc|msc',
sample_code VARCHAR(32) NOT NULL COMMENT '样本编码',
payload_json TEXT NULL COMMENT '补充字段 JSON',
tenant_no VARCHAR(64) NOT NULL,
order_num INT NULL,
remark VARCHAR(512) NULL,
create_by VARCHAR(64) NULL,
create_dept_id VARCHAR(64) NOT NULL DEFAULT '',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
update_by VARCHAR(64) NULL,
update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
del_flag SMALLINT NOT NULL DEFAULT 0,
PRIMARY KEY (id),
KEY ix_lab_sample_extra_tenant_ds_code (tenant_no, dataset, sample_code, del_flag)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- 挂到与「质检」同一 Web 虚拟根下(复用 menu:quality 的 parent_id)
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
MD5(CONCAT(quality.tenant_no, ':menu:lab-sample-processing')),
'实验室样本处理数据', quality.parent_id, 'M', '/lab-sample-processing', '/index/index',
'menu:lab-sample-processing', 1, '1', '1', quality.tenant_no,
COALESCE(quality.order_num, 50) + 1,
'20261104 会议:造血/单个核/间充质台账',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu quality
WHERE quality.perms = 'menu:quality' AND quality.del_flag = 0 AND quality.status = '1'
AND NOT EXISTS (
SELECT 1 FROM sys_menu e
WHERE e.tenant_no = quality.tenant_no AND e.perms = 'menu:lab-sample-processing' AND e.del_flag = 0
);
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
MD5(CONCAT(p.tenant_no, ':lab:sample-processing:list')),
'造血干细胞数据', p.id, 'C', 'lab-sample-processing/hsc',
'/lab-sample-processing/hsc/index', 'lab:sample-processing:list',
1, '1', '1', p.tenant_no, 10, '造血干细胞数据列表',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu p
WHERE p.perms = 'menu:lab-sample-processing' AND p.del_flag = 0 AND p.status = '1'
AND NOT EXISTS (
SELECT 1 FROM sys_menu e
WHERE e.tenant_no = p.tenant_no AND e.path = 'lab-sample-processing/hsc' AND e.del_flag = 0
);
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
MD5(CONCAT(p.tenant_no, ':lab:sample-processing:ubmc')),
'单个核数据', p.id, 'C', 'lab-sample-processing/ubmc',
'/lab-sample-processing/ubmc/index', 'lab:sample-processing:list',
1, '1', '1', p.tenant_no, 20, '单个核数据列表',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu p
WHERE p.perms = 'menu:lab-sample-processing' AND p.del_flag = 0 AND p.status = '1'
AND NOT EXISTS (
SELECT 1 FROM sys_menu e
WHERE e.tenant_no = p.tenant_no AND e.path = 'lab-sample-processing/ubmc' AND e.del_flag = 0
);
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
MD5(CONCAT(p.tenant_no, ':lab:sample-processing:msc')),
'脐带间充质干细胞数据', p.id, 'C', 'lab-sample-processing/msc',
'/lab-sample-processing/msc/index', 'lab:sample-processing:list',
1, '1', '1', p.tenant_no, 30, '脐带间充质干细胞数据列表',
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu p
WHERE p.perms = 'menu:lab-sample-processing' AND p.del_flag = 0 AND p.status = '1'
AND NOT EXISTS (
SELECT 1 FROM sys_menu e
WHERE e.tenant_no = p.tenant_no AND e.path = 'lab-sample-processing/msc' AND e.del_flag = 0
);
-- 编辑按钮挂在列表菜单下(三页共用同一 list 权限父节点:造血页)
INSERT INTO sys_menu (
id, menu_name, parent_id, menu_type, path, component, perms,
is_frame, visible, status, tenant_no, order_num, remark,
create_by, create_dept_id, create_time, update_by, update_time, del_flag
)
SELECT
MD5(CONCAT(p.tenant_no, ':', btn.perms)),
btn.menu_name, p.id, 'F', '', NULL, btn.perms,
1, '0', '1', p.tenant_no, btn.order_num, btn.remark,
'migration', '', NOW(), 'migration', NOW(), 0
FROM sys_menu p
JOIN (
SELECT '市场字段编辑' AS menu_name, 'lab:sample-processing:market:edit' AS perms, 1 AS order_num, '仅可编辑市场部门字段' AS remark
UNION ALL SELECT '生产字段编辑', 'lab:sample-processing:production:edit', 2, '仅可编辑生产部门字段'
UNION ALL SELECT '质检字段编辑', 'lab:sample-processing:qc:edit', 3, '仅可编辑质检部门字段'
UNION ALL SELECT '部门主管', 'lab:sample-processing:manager', 4, '可编辑全部部门字段'
) btn
WHERE p.perms = 'lab:sample-processing:list'
AND p.path = 'lab-sample-processing/hsc'
AND p.del_flag = 0 AND p.status = '1'
AND NOT EXISTS (
SELECT 1 FROM sys_menu e
WHERE e.tenant_no = p.tenant_no AND e.perms = btn.perms AND e.del_flag = 0
);
@@ -0,0 +1,2 @@
schema: spec-driven
created: 2026-11-12
@@ -0,0 +1,23 @@
## Context
附件三张表字段覆盖市场采集信息、实验室接收/处理与质检结果。系统内对应源主要为:
- 市场:`biz_protocol_party_customer`、`biz_protocol_collection_info`、`biz_collection_hospital`、`biz_protocol_sample`
- 生产/接收:`biz_sample_reception`、`biz_sample_production_reception`
- 质检:生产接收传染四项、储存证质检 payload、检查项结果;缺口进补充表
## Decisions
1. **数据集**:`hsc`(造血)、`ubmc`(单个核)、`msc`(脐带间充质),以样本类型名 / 编码前缀识别。
2. **权限码**:
- 列表:`lab:sample-processing:list`(三子页共用读)
- 部门编辑:`lab:sample-processing:market:edit` / `production:edit` / `qc:edit`
- 主管:`lab:sample-processing:manager`(可编辑全部部门字段)
3. **读写**:列表聚合只读源表 + 补充 JSON;更新时按字段 `source` 回写源表或补充表,服务端强制部门 ACL。
4. **UI**:P01 列表 + 筛选 + 导出;行内编辑弹窗按部门分区,无权字段只读。
5. **Figma**:MCP 授权不可用,按现有 Fresh Mint 列表页实现;合入后用户做视觉验收。
## Risks
- 历史 Excel 手工列与系统字段不完全一一对应,未映射列落补充表,避免阻断上线。
- 菜单种子不自动授权角色,需管理员分配。
@@ -0,0 +1,27 @@
## Why
2026-11-04 需求会议要求按线上三张实验室样本处理表(造血干细胞 / 单个核 / 脐带间充质)在系统中单独建模块,按市场、生产、质检拆分可编辑字段,支持筛选与导出。现有业务数据分散在协议、样本接收、生产接收与质检等表,缺少统一台账视图。
## What Changes
- 新增 Web 一级模块「实验室样本处理数据」,下挂三个子模块对应三张表。
- 按样本编码从现有业务表聚合展示全字段;部门员工可读全部、只写本部门字段;主管可写全部部门字段。
- 支持关键词/时间筛选与 Excel 导出。
- 无法直接回写源表的字段写入补充表 `biz_lab_sample_processing_extra`。
## Capabilities
### New Capabilities
- `lab-sample-processing-data`: 实验室样本处理数据聚合台账、部门字段 ACL、筛选导出
### Modified Capabilities
- (无)
## Impact
- **backend**:聚合服务、API、补充表、菜单权限
- **web-admin**:三子模块列表页、编辑弹窗、导出
- **docs/sql**:结构 + 菜单种子
- **不改** mobile-h5 / mobile-oa-h5
@@ -0,0 +1,33 @@
## ADDED Requirements
### Requirement: 三子模块台账
系统 SHALL 提供「实验室样本处理数据」模块,包含造血干细胞、单个核、脐带间充质三个子模块;每个子模块按样本展示该表全部约定字段。
#### Scenario: 按数据集查询
- **WHEN** 用户打开造血干细胞子模块并具备列表权限
- **THEN** 系统仅返回造血干细胞数据集样本行,并包含市场/生产/质检字段聚合结果
### Requirement: 部门字段编辑权限
系统 SHALL 允许具备列表权限的用户查看全部字段;普通员工仅可编辑本部门字段;具备主管权限的用户可编辑全部部门字段。服务端 MUST 拒绝无权字段写入。
#### Scenario: 生产员工改市场字段被拒
- **WHEN** 仅有生产编辑权限的用户提交市场部门字段变更
- **THEN** 系统拒绝该请求且不持久化无权字段
#### Scenario: 主管可改全部
- **WHEN** 具备主管权限的用户提交跨部门字段变更
- **THEN** 系统接受并回写对应数据源或补充表
### Requirement: 筛选与导出
系统 SHALL 支持按样本编码、姓名关键词及接收/采集时间范围筛选,并导出当前筛选结果为 Excel。
#### Scenario: 导出当前筛选
- **WHEN** 用户在筛选条件下点击导出
- **THEN** 下载的表格列与当前子模块字段目录一致,行集与列表筛选一致
@@ -0,0 +1,21 @@
## 1. 规格与数据
- [x] 1.1 建立 OpenSpec 变更与字段/权限设计
- [x] 1.2 定义三数据集字段目录与部门归属、回写源
## 2. 后端
- [x] 2.1 补充表模型与迁移 SQL、菜单种子
- [x] 2.2 聚合列表 / 详情 / 更新 / 导出 API 与部门 ACL
- [x] 2.3 单元测试:数据集识别、字段 ACL、无权拒绝
## 3. 前端
- [x] 3.1 共享列表页组件(筛选、编辑、导出)
- [x] 3.2 三子模块路由与菜单码
- [x] 3.3 API 客户端
## 4. 验证
- [x] 4.1 受影响单元测试
- [ ] 4.2 字段一致性/菜单说明回写工单(凭据就绪后)
+55
View File
@@ -0,0 +1,55 @@
import request from '@/utils/http'
export type LabSampleProcessTabKey =
| 'cell_hsc'
| 'donation_hsc'
| 'cell_msc'
| 'donation_msc'
export interface LabSampleProcessListParams {
tabKey: LabSampleProcessTabKey
page?: number
pageSize?: number
labStorageCode?: string
motherName?: string
createTimeStart?: string
createTimeEnd?: string
}
export function fetchLabSampleProcessList(params: LabSampleProcessListParams) {
return request.get<{
list: Record<string, unknown>[]
total: number
page: number
pageSize: number
tabKey: string
supplementKeys: string[]
}>({
url: '/web/admin/lab-sample-process/list',
params
})
}
export function fetchLabSampleProcessSupplement(labStorageCode: string) {
return request.get<{
labStorageCode?: string
payload?: Record<string, string>
remark?: string
}>({
url: '/web/admin/lab-sample-process/supplement',
params: { labStorageCode }
})
}
export function saveLabSampleProcessSupplement(body: {
labStorageCode: string
receptionSource: string
sampleTypeName: string
payload: Record<string, string>
remark?: string
}) {
return request.put({
url: '/web/admin/lab-sample-process/supplement',
data: body
})
}
@@ -0,0 +1,74 @@
import request from '@/utils/http'
export type LabSampleDataset = 'hsc' | 'ubmc' | 'msc'
export interface LabSampleFieldMeta {
key: string
label: string
dept: 'market' | 'production' | 'qc'
source: string
}
export interface LabSampleAcl {
canList?: boolean
isManager?: boolean
editableDepts?: string[]
deptFlags?: Record<string, boolean>
}
export function fetchLabSampleMeta(dataset: LabSampleDataset) {
return request.get<{
dataset: LabSampleDataset
label: string
fields: LabSampleFieldMeta[]
acl: LabSampleAcl
}>({
url: `/web/admin/lab-sample-processing/meta/${dataset}`
})
}
export function fetchLabSampleList(
dataset: LabSampleDataset,
params: {
keyword?: string
dateFrom?: string
dateTo?: string
page?: number
pageSize?: number
}
) {
return request.get<{
list: Record<string, any>[]
total: number
page: number
pageSize: number
fields: LabSampleFieldMeta[]
acl: LabSampleAcl
}>({
url: `/web/admin/lab-sample-processing/${dataset}/list`,
params
})
}
export function updateLabSampleRow(
dataset: LabSampleDataset,
sampleCode: string,
payload: Record<string, any>
) {
return request.put<Record<string, any>>({
url: `/web/admin/lab-sample-processing/${dataset}/${encodeURIComponent(sampleCode)}`,
data: { payload }
})
}
export function exportLabSampleUrl(
dataset: LabSampleDataset,
params: { keyword?: string; dateFrom?: string; dateTo?: string }
) {
const q = new URLSearchParams()
if (params.keyword) q.set('keyword', params.keyword)
if (params.dateFrom) q.set('dateFrom', params.dateFrom)
if (params.dateTo) q.set('dateTo', params.dateTo)
const qs = q.toString()
return `/api/web/admin/lab-sample-processing/${dataset}/export${qs ? `?${qs}` : ''}`
}
+64
View File
@@ -1123,6 +1123,17 @@ export const sqxpAdminRoutes: AppRouteRecord[] = [
menuCode: 'biz:sampleReception:list' menuCode: 'biz:sampleReception:list'
} }
}, },
{
id: '70000000000000000000000000000015',
name: 'QualityInspectionLabSampleProcess',
path: 'lab-sample-process',
component: '/quality-inspection/lab-sample-process/index',
meta: {
title: '实验室样本处理数据',
keepAlive: true,
menuCode: 'biz:labSampleProcess:list'
}
},
{ {
id: '70000000000000000000000000000007', id: '70000000000000000000000000000007',
name: 'QualityInspectionQcWorkOrder', name: 'QualityInspectionQcWorkOrder',
@@ -1302,6 +1313,59 @@ export const sqxpAdminRoutes: AppRouteRecord[] = [
] ]
}, },
// ═══════════════════════════════════════════════════════════════════════════
// 3b. 实验室样本处理数据 (menu:lab-sample-processing)
// ═══════════════════════════════════════════════════════════════════════════
{
id: 'TOP0000000000000000000000000000L1',
name: 'LabSampleProcessing',
path: '/lab-sample-processing',
component: '/index/index',
meta: {
title: '实验室样本处理数据',
icon: 'ri:test-tube-line',
keepAlive: false,
isTopMenu: true,
menuCode: 'menu:lab-sample-processing',
roles: ['admin']
},
children: [
{
id: 'LAB00000000000000000000000000001',
name: 'LabSampleProcessingHsc',
path: 'hsc',
component: '/lab-sample-processing/hsc/index',
meta: {
title: '造血干细胞数据',
keepAlive: true,
menuCode: 'lab:sample-processing:list'
}
},
{
id: 'LAB00000000000000000000000000002',
name: 'LabSampleProcessingUbmc',
path: 'ubmc',
component: '/lab-sample-processing/ubmc/index',
meta: {
title: '单个核数据',
keepAlive: true,
menuCode: 'lab:sample-processing:list'
}
},
{
id: 'LAB00000000000000000000000000003',
name: 'LabSampleProcessingMsc',
path: 'msc',
component: '/lab-sample-processing/msc/index',
meta: {
title: '脐带间充质干细胞数据',
keepAlive: true,
menuCode: 'lab:sample-processing:list'
}
}
]
},
// ═══════════════════════════════════════════════════════════════════════════ // ═══════════════════════════════════════════════════════════════════════════
// 4. 运营支撑 (menu:operations) // 4. 运营支撑 (menu:operations)
// 二级菜单:系统基础信息、运营支撑 // 二级菜单:系统基础信息、运营支撑
@@ -0,0 +1,138 @@
import type { LabSampleProcessTabKey } from '@/api/lab-sample-process'
export type LabColumn = {
prop: string
label: string
minWidth?: number
}
export const LAB_PROCESS_TABS: Array<{
key: LabSampleProcessTabKey
label: string
source: 'cell_storage' | 'research_donation'
sampleTypeName: string
}> = [
{
key: 'cell_hsc',
label: '细胞存储·围产期造血',
source: 'cell_storage',
sampleTypeName: '围产期造血干细胞'
},
{
key: 'donation_hsc',
label: '科研捐赠·围产期造血',
source: 'research_donation',
sampleTypeName: '围产期造血干细胞'
},
{
key: 'cell_msc',
label: '细胞存储·脐带间充质',
source: 'cell_storage',
sampleTypeName: '脐带间充质干细胞'
},
{
key: 'donation_msc',
label: '科研捐赠·脐带间充质',
source: 'research_donation',
sampleTypeName: '脐带间充质干细胞'
}
]
const CELL_HSC_COLUMNS: LabColumn[] = [
{ prop: 'cellCode', label: '细胞编码', minWidth: 140 },
{ prop: 'motherName', label: '准妈妈姓名', minWidth: 110 },
{ prop: 'collectionTime', label: '采集时间', minWidth: 200 },
{ prop: 'hospitalName', label: '采集医院', minWidth: 140 },
{ prop: 'receptionTime', label: '接收时间', minWidth: 200 },
{ prop: 'receiveTemperature', label: '样品温度到达/℃', minWidth: 130 },
{ prop: 'receiveCheckText', label: '收样检查', minWidth: 120 },
{ prop: 'hasMaternalBlood', label: '是否有母血', minWidth: 100 },
{ prop: 'remeasureMaternalBlood', label: '是否重采母血', minWidth: 110 },
{ prop: 'consentForm', label: '知情同意书或协议书', minWidth: 140 },
{ prop: 'transportPackingList', label: '脐血运输装箱清单', minWidth: 140 },
{ prop: 'cordBloodReceiveConfirm', label: '脐带血接收确认单', minWidth: 140 },
{ prop: 'healthSurvey', label: '孕妇健康调查表', minWidth: 120 },
{ prop: 'labelClear', label: '标识清晰', minWidth: 100 },
{ prop: 'handlerName', label: '接收登记人', minWidth: 110 },
{ prop: 'remarkReception', label: '备注', minWidth: 120 },
{ prop: 'originalBloodWeightG', label: '原血重量/g', minWidth: 110 },
{ prop: 'hesVolumeMl', label: 'HES用量/ml', minWidth: 110 },
{ prop: 'bloodType', label: '血型', minWidth: 80 },
{ prop: 'infectiousFourTestLabel', label: '传染四项检测(检测卡)', minWidth: 150 },
{ prop: 'infectiousFourExternal', label: '传染四项检测(外检)', minWidth: 150 },
{ prop: 'preEnrichNucleatedCount', label: '富集前有核细胞数/109/L', minWidth: 170 },
{ prop: 'enrichedBloodWeightG', label: '富集后血量/g', minWidth: 120 },
{ prop: 'postEnrichNucleatedCount', label: '富集后有核细胞数/109/L', minWidth: 170 },
{ prop: 'cryopreservationVolumeMl', label: '冻存液用量/ml', minWidth: 120 },
{ prop: 'cd34Count1e4', label: 'CD34+细胞计数/*10E4', minWidth: 160 },
{ prop: 'operatorName', label: '操作人', minWidth: 100 },
{ prop: 'remarkProcess', label: '备注', minWidth: 120 },
{ prop: 'createTime', label: '创建时间', minWidth: 200 }
]
const DONATION_HSC_COLUMNS: LabColumn[] = [
{ prop: 'cellCode', label: '细胞编码', minWidth: 140 },
{ prop: 'motherName', label: '准妈妈姓名', minWidth: 110 },
{ prop: 'labMaterialCode', label: '实验室物料编码', minWidth: 130 },
{ prop: 'inboundCode', label: '入库编码', minWidth: 140 },
{ prop: 'storageStatus', label: '储存状态', minWidth: 100 },
{ prop: 'discardReason', label: '废弃原因', minWidth: 120 },
{ prop: 'receptionTime', label: '接收时间', minWidth: 200 },
{ prop: 'receiveTemperature', label: '样品温度到达/℃', minWidth: 130 },
{ prop: 'receiveCheckText', label: '收样检查', minWidth: 120 },
{ prop: 'handlerName', label: '接收登记人', minWidth: 110 },
{ prop: 'remarkReception', label: '备注', minWidth: 120 },
{ prop: 'originalBloodVolumeMl', label: '原血体积/mL', minWidth: 120 },
{ prop: 'nucleatedCellCount1e8', label: '有核细胞数/108', minWidth: 130 },
{ prop: 'preOpNucleatedCount', label: '操作前有核细胞数/109/L', minWidth: 170 },
{ prop: 'infectiousFourTestLabel', label: '传染四项检测(检测卡)', minWidth: 150 },
{ prop: 'motherHbsag', label: '乙型肝炎病毒表面抗原(HBsAg)', minWidth: 180 },
{ prop: 'motherAntiHcv', label: '丙型肝炎病毒抗体(抗HCV)', minWidth: 170 },
{ prop: 'motherAntiTp', label: '梅毒螺旋体特异性抗体(抗TP)', minWidth: 180 },
{ prop: 'motherAntiHiv', label: '人类免疫缺陷病毒抗体(抗HIV-1/2)', minWidth: 200 },
{ prop: 'motherCmvIgm', label: '巨细胞病毒IgM抗体(CMV-IgM)', minWidth: 180 },
{ prop: 'operatorName', label: '操作人', minWidth: 100 },
{ prop: 'remarkProcess', label: '备注', minWidth: 120 },
{ prop: 'createTime', label: '创建时间', minWidth: 200 }
]
const MSC_COLUMNS: LabColumn[] = [
{ prop: 'labFlowCode', label: '实验室流转编码', minWidth: 140 },
{ prop: 'motherName', label: '母亲姓名', minWidth: 110 },
{ prop: 'cordCollectTime', label: '脐带采集时间', minWidth: 200 },
{ prop: 'transportTempC', label: '运输温度/℃', minWidth: 110 },
{ prop: 'cordReceiveTime', label: '脐带接收时间', minWidth: 200 },
{ prop: 'handlerName', label: '接收登记人', minWidth: 110 },
{ prop: 'inoculationAmount', label: '接种量', minWidth: 100 },
{ prop: 'storageStatus', label: '储存状态', minWidth: 100 },
{ prop: 'cryopreservationTime', label: '冻存时间', minWidth: 200 },
{ prop: 'storageTotalTubes', label: '储存总量(管)', minWidth: 120 },
{ prop: 'availableCellCount1e7', label: '可提供细胞数量/*10E7', minWidth: 160 },
{ prop: 'endotoxinTest', label: '内毒素检测', minWidth: 100 },
{ prop: 'bacteriaTest', label: '细菌检测', minWidth: 100 },
{ prop: 'fungusTest', label: '真菌检测', minWidth: 100 },
{ prop: 'mycoplasmaTest', label: '支原体检测', minWidth: 100 },
{ prop: 'operatorName', label: '操作人', minWidth: 100 },
{ prop: 'motherHbsag', label: '乙型肝炎病毒表面抗原(HBsAg)', minWidth: 180 },
{ prop: 'motherHbvDna', label: '乙型肝炎病毒核酸(HBV DNA)', minWidth: 180 },
{ prop: 'motherAntiHcv', label: '丙型肝炎病毒抗体(抗HCV)', minWidth: 170 },
{ prop: 'motherHcvRna', label: '丙型肝炎病毒核酸(HCV RNA)', minWidth: 180 },
{ prop: 'motherAntiTp', label: '梅毒螺旋体特异性抗体(抗TP)', minWidth: 180 },
{ prop: 'motherCmvIgm', label: '巨细胞病毒IgM抗体(CMV IgM)', minWidth: 180 },
{ prop: 'motherAntiHiv', label: '人类免疫缺陷病毒抗体(抗HIV-1/2)', minWidth: 200 },
{ prop: 'motherHiv1Rna', label: '人类免疫缺陷病毒核酸(1型)(HIV-1RNA)', minWidth: 210 },
{ prop: 'storageDateP2', label: '储存日期(P2冻存日期)', minWidth: 160 },
{ prop: 'remark', label: '备注', minWidth: 120 },
{ prop: 'createTime', label: '创建时间', minWidth: 200 }
]
export const LAB_PROCESS_COLUMNS: Record<LabSampleProcessTabKey, LabColumn[]> = {
cell_hsc: CELL_HSC_COLUMNS,
donation_hsc: DONATION_HSC_COLUMNS,
cell_msc: MSC_COLUMNS,
donation_msc: MSC_COLUMNS
}
export const SUPPLEMENT_FIELD_LABELS: Record<string, string> = Object.fromEntries(
[...CELL_HSC_COLUMNS, ...DONATION_HSC_COLUMNS, ...MSC_COLUMNS].map((c) => [c.prop, c.label])
)
@@ -0,0 +1,329 @@
<!-- 实验室样本处理数据 — 共享列表页 -->
<template>
<div class="page-container">
<section class="panel">
<div class="panel-top">
<h2 class="title-bar-heading">{{ pageTitle }}</h2>
</div>
<div class="filter-block">
<el-form :model="searchForm" class="filter-form" @submit.prevent>
<div class="filter-grid">
<el-form-item label="关键词">
<el-input
v-model="searchForm.keyword"
placeholder="样本编码/姓名"
clearable
/>
</el-form-item>
<el-form-item label="时间范围">
<el-date-picker
v-model="dateRange"
type="daterange"
range-separator="至"
start-placeholder="开始日期"
end-placeholder="结束日期"
value-format="YYYY-MM-DD"
style="width: 100%"
/>
</el-form-item>
</div>
<div class="filter-actions-row">
<div class="filter-actions-end">
<el-button class="btn-ghost" @click="handleReset">
<el-icon><RefreshLeft /></el-icon>重置
</el-button>
<el-button class="btn-main" type="primary" @click="handleSearch">
<el-icon><Search /></el-icon>搜索
</el-button>
</div>
</div>
</el-form>
</div>
<div class="list-toolbar">
<div class="list-toolbar-row">
<div class="list-toolbar-selected">
共 <span class="sel-num">{{ pagination.total }}</span> 条
</div>
<el-button class="btn-main" type="primary" :loading="exporting" @click="handleExport">
<el-icon><Download /></el-icon>
导出
</el-button>
</div>
</div>
<div class="list-body list-wrap">
<el-table
class="list-table"
:data="tableData"
v-loading="loading"
stripe
size="small"
row-key="id"
>
<el-table-column
type="index"
label="序号"
min-width="68"
align="center"
header-align="center"
:index="(index) => (pagination.page - 1) * pagination.pageSize + index + 1"
/>
<el-table-column
v-for="col in visibleColumns"
:key="col.key"
:prop="col.key"
:label="col.label"
min-width="140"
align="center"
header-align="center"
show-overflow-tooltip
/>
<el-table-column
label="操作"
fixed="right"
min-width="260"
align="center"
header-align="center"
>
<template #default="{ row }">
<el-button
v-if="(acl.editableDepts || []).length"
class="btn-ghost"
link
type="primary"
@click="openEdit(row)"
>
<el-icon><Edit /></el-icon>
编辑
</el-button>
<span v-else class="text-muted">只读</span>
</template>
</el-table-column>
</el-table>
</div>
<div class="pagination list-pager-bar">
<el-pagination
v-model:current-page="pagination.page"
v-model:page-size="pagination.pageSize"
:total="pagination.total"
:page-sizes="[10, 20, 50, 100]"
layout="total, sizes, prev, pager, next, jumper"
@size-change="loadData"
@current-change="loadData"
/>
</div>
</section>
<el-dialog
v-model="editVisible"
class="dialog-form-layout"
:title="`编辑 — ${editRow?.sampleCode || ''}`"
width="860px"
align-center
destroy-on-close
append-to-body
>
<el-form label-position="top" class="form-grid-2" v-loading="saving">
<template v-for="group in editGroups" :key="group.dept">
<div class="form-col-span-2 dept-title">{{ group.label }}</div>
<el-form-item v-for="f in group.fields" :key="f.key" :label="f.label">
<el-input
v-model="editForm[f.key]"
:disabled="!canEditDept(f.dept)"
clearable
/>
</el-form-item>
</template>
</el-form>
<template #footer>
<el-button class="btn-ghost" @click="editVisible = false">
<el-icon><Close /></el-icon>取消
</el-button>
<el-button class="btn-main" type="primary" :loading="saving" @click="submitEdit">
<el-icon><Check /></el-icon>确定
</el-button>
</template>
</el-dialog>
</div>
</template>
<script setup lang="ts">
import { computed, onMounted, reactive, ref } from 'vue'
import { Check, Close, Download, Edit, RefreshLeft, Search } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import {
fetchLabSampleList,
updateLabSampleRow,
type LabSampleAcl,
type LabSampleDataset,
type LabSampleFieldMeta
} from '@/api/lab-sample-processing'
const props = defineProps<{
dataset: LabSampleDataset
title: string
}>()
const pageTitle = computed(() => props.title)
const loading = ref(false)
const saving = ref(false)
const exporting = ref(false)
const tableData = ref<Record<string, any>[]>([])
const fields = ref<LabSampleFieldMeta[]>([])
const acl = ref<LabSampleAcl>({})
const searchForm = reactive({ keyword: '' })
const dateRange = ref<string[] | null>(null)
const pagination = reactive({ page: 1, pageSize: 20, total: 0 })
const editVisible = ref(false)
const editRow = ref<Record<string, any> | null>(null)
const editForm = reactive<Record<string, any>>({})
const deptLabel: Record<string, string> = {
market: '市场字段',
production: '生产字段',
qc: '质检字段'
}
const visibleColumns = computed(() => fields.value)
const editGroups = computed(() => {
const groups = ['market', 'production', 'qc'].map((dept) => ({
dept,
label: deptLabel[dept] || dept,
fields: fields.value.filter((f) => f.dept === dept)
}))
return groups.filter((g) => g.fields.length > 0)
})
function canEditDept(dept: string) {
const list = acl.value.editableDepts || []
return list.includes(dept)
}
async function loadData() {
loading.value = true
try {
const res = await fetchLabSampleList(props.dataset, {
keyword: searchForm.keyword || undefined,
dateFrom: dateRange.value?.[0],
dateTo: dateRange.value?.[1],
page: pagination.page,
pageSize: pagination.pageSize
})
tableData.value = res?.list || []
pagination.total = res?.total || 0
fields.value = res?.fields || []
acl.value = res?.acl || {}
} catch (e: any) {
ElMessage.error(e?.message || '加载失败')
} finally {
loading.value = false
}
}
function handleSearch() {
pagination.page = 1
void loadData()
}
function handleReset() {
searchForm.keyword = ''
dateRange.value = null
handleSearch()
}
function openEdit(row: Record<string, any>) {
editRow.value = row
Object.keys(editForm).forEach((k) => delete editForm[k])
for (const f of fields.value) {
editForm[f.key] = row[f.key] ?? ''
}
editVisible.value = true
}
async function submitEdit() {
if (!editRow.value?.sampleCode) return
const editableKeys = fields.value
.filter((f) => canEditDept(f.dept))
.map((f) => f.key)
if (!editableKeys.length) {
ElMessage.warning('当前账号无可编辑字段')
return
}
const payload: Record<string, any> = {}
for (const key of editableKeys) {
payload[key] = editForm[key]
}
saving.value = true
try {
await updateLabSampleRow(props.dataset, editRow.value.sampleCode, payload)
ElMessage.success('保存成功')
editVisible.value = false
await loadData()
} catch (e: any) {
ElMessage.error(e?.message || '保存失败')
} finally {
saving.value = false
}
}
async function handleExport() {
exporting.value = true
try {
const res = await fetchLabSampleList(props.dataset, {
keyword: searchForm.keyword || undefined,
dateFrom: dateRange.value?.[0],
dateTo: dateRange.value?.[1],
page: 1,
pageSize: 5000
})
const cols = res?.fields?.length ? res.fields : fields.value
const rows = res?.list || []
const ExcelJS = (await import('exceljs')).default
const workbook = new ExcelJS.Workbook()
const sheet = workbook.addWorksheet(props.title || props.dataset)
sheet.addRow(cols.map((c) => c.label))
for (const row of rows) {
sheet.addRow(cols.map((c) => row[c.key] ?? ''))
}
const header = sheet.getRow(1)
header.font = { bold: true }
const buffer = await workbook.xlsx.writeBuffer()
const blob = new Blob([buffer], {
type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
})
const a = document.createElement('a')
a.href = URL.createObjectURL(blob)
a.download = `${props.title || props.dataset}.xlsx`
a.click()
URL.revokeObjectURL(a.href)
ElMessage.success(`已导出 ${rows.length} 条`)
} catch (e: any) {
ElMessage.error(e?.message || '导出失败')
} finally {
exporting.value = false
}
}
onMounted(() => {
void loadData()
})
</script>
<style scoped lang="scss">
.dept-title {
font-weight: 600;
color: var(--el-color-primary);
margin: 4px 0 8px;
}
.text-muted {
color: var(--el-text-color-secondary);
font-size: 12px;
}
</style>
@@ -0,0 +1,7 @@
<template>
<LabSampleProcessingPage dataset="hsc" title="造血干细胞数据" />
</template>
<script setup lang="ts">
import LabSampleProcessingPage from '../LabSampleProcessingPage.vue'
</script>
@@ -0,0 +1,7 @@
<template>
<LabSampleProcessingPage dataset="msc" title="脐带间充质干细胞数据" />
</template>
<script setup lang="ts">
import LabSampleProcessingPage from '../LabSampleProcessingPage.vue'
</script>
@@ -0,0 +1,7 @@
<template>
<LabSampleProcessingPage dataset="ubmc" title="单个核数据" />
</template>
<script setup lang="ts">
import LabSampleProcessingPage from '../LabSampleProcessingPage.vue'
</script>
@@ -0,0 +1,347 @@
<!-- 质检管理 — 实验室样本处理数据 -->
<!-- eslint-disable vue/no-v-model-argument -->
<template>
<div class="page-container">
<section class="panel">
<div class="panel-top">
<h2 class="title-bar-heading">实验室样本处理数据</h2>
</div>
<el-tabs v-model="activeTab" class="lab-process-tabs" @tab-change="handleTabChange">
<el-tab-pane
v-for="tab in LAB_PROCESS_TABS"
:key="tab.key"
:label="tab.label"
:name="tab.key"
/>
</el-tabs>
<div class="filter-block">
<el-form :model="searchForm" class="filter-form" @submit.prevent>
<div class="filter-grid">
<el-form-item label="实验室流转编码/细胞编码">
<el-input
v-model="searchForm.labStorageCode"
placeholder="实验室入库编码"
clearable
/>
</el-form-item>
<el-form-item label="母亲姓名">
<el-input v-model="searchForm.motherName" placeholder="母亲/准妈妈姓名" clearable />
</el-form-item>
<el-form-item label="创建时间">
<el-date-picker
v-model="searchForm.createTimeRange"
type="datetimerange"
value-format="YYYY-MM-DD HH:mm:ss"
start-placeholder="开始时间"
end-placeholder="结束时间"
clearable
/>
</el-form-item>
</div>
<div class="filter-actions-row">
<div class="filter-actions-end">
<el-button class="btn-ghost" @click="handleReset"
><el-icon><RefreshLeft /></el-icon>重置</el-button
>
<el-button class="btn-main" type="primary" @click="handleSearch"
><el-icon><Search /></el-icon>搜索</el-button
>
</div>
</div>
</el-form>
</div>
<div class="list-toolbar">
<div class="list-toolbar-row">
<div class="list-toolbar-selected">
共 <span class="sel-num">{{ pagination.total }}</span> 条
</div>
<el-button class="btn-main" type="primary" @click="handleExport">
<el-icon><Download /></el-icon>
导出 Excel
</el-button>
</div>
</div>
<div class="list-body list-wrap">
<el-table
class="list-table"
:data="tableData"
v-loading="loading"
stripe
row-key="receptionId"
>
<el-table-column
type="index"
label="序号"
min-width="68"
align="center"
header-align="center"
:index="(index) => (pagination.page - 1) * pagination.pageSize + index + 1"
/>
<el-table-column
v-for="col in currentColumns"
:key="col.prop"
:prop="col.prop"
:label="col.label"
:min-width="col.minWidth || 120"
align="center"
header-align="center"
show-overflow-tooltip
>
<template #default="{ row }">{{ displayCell(row[col.prop]) }}</template>
</el-table-column>
<el-table-column
label="操作"
min-width="260"
fixed="right"
align="center"
header-align="center"
>
<template #default="{ row }">
<div style="display: flex; justify-content: center; gap: 8px">
<el-button class="btn-op-item" @click="openSupplement(row)">
<el-icon><Edit /></el-icon>补录
</el-button>
</div>
</template>
</el-table-column>
</el-table>
<div class="pagination list-pager-bar">
<el-pagination
v-model:current-page="pagination.page"
v-model:page-size="pagination.pageSize"
:total="pagination.total"
:page-sizes="[10, 20, 50, 100]"
layout="total, sizes, prev, pager, next, jumper"
background
@size-change="loadData"
@current-change="loadData"
/>
</div>
</div>
</section>
<el-dialog
v-model="supplementVisible"
title="补录"
width="720px"
class="dialog-form-layout"
align-center
append-to-body
modal-class="protocol-dialog-mask"
destroy-on-close
:close-on-click-modal="false"
>
<el-form label-position="top" class="biz-dialog-form">
<p class="form-tip"
>入库编码:{{ supplementForm.labStorageCode || '—' }}(补录数据跟随实验室入库编码)</p
>
<div class="form-grid-2">
<el-form-item
v-for="key in supplementKeys"
:key="key"
:label="SUPPLEMENT_FIELD_LABELS[key] || key"
>
<el-input v-model="supplementForm.payload[key]" clearable />
</el-form-item>
<el-form-item label="备注" class="form-col-span-2">
<el-input v-model="supplementForm.remark" type="textarea" :rows="2" />
</el-form-item>
</div>
</el-form>
<template #footer>
<el-button class="btn-ghost" @click="supplementVisible = false"
><el-icon><Close /></el-icon>取消</el-button
>
<el-button class="btn-main" type="primary" :loading="saving" @click="submitSupplement"
><el-icon><Check /></el-icon>确定</el-button
>
</template>
</el-dialog>
</div>
</template>
<script setup lang="ts">
import { Check, Close, Download, Edit, RefreshLeft, Search } from '@element-plus/icons-vue'
import { ElMessage } from 'element-plus'
import * as XLSX from 'xlsx'
import {
fetchLabSampleProcessList,
fetchLabSampleProcessSupplement,
saveLabSampleProcessSupplement,
type LabSampleProcessTabKey
} from '@/api/lab-sample-process'
import {
LAB_PROCESS_COLUMNS,
LAB_PROCESS_TABS,
SUPPLEMENT_FIELD_LABELS
} from '@/utils/lab-sample-process-columns'
defineOptions({ name: 'QualityInspectionLabSampleProcess' })
const activeTab = ref<LabSampleProcessTabKey>('cell_hsc')
const loading = ref(false)
const saving = ref(false)
const tableData = ref<Record<string, unknown>[]>([])
const supplementKeys = ref<string[]>([])
const searchForm = reactive({
labStorageCode: '',
motherName: '',
createTimeRange: null as [string, string] | null
})
const pagination = reactive({
page: 1,
pageSize: 10,
total: 0
})
const currentColumns = computed(() => LAB_PROCESS_COLUMNS[activeTab.value] || [])
const currentTabMeta = computed(
() => LAB_PROCESS_TABS.find((t) => t.key === activeTab.value) || LAB_PROCESS_TABS[0]
)
const supplementVisible = ref(false)
const supplementForm = reactive({
labStorageCode: '',
receptionSource: 'cell_storage',
sampleTypeName: '',
remark: '',
payload: {} as Record<string, string>
})
function displayCell(value: unknown) {
if (value === null || value === undefined || value === '') return '—'
return String(value)
}
async function loadData() {
loading.value = true
try {
const range = searchForm.createTimeRange
const res = await fetchLabSampleProcessList({
tabKey: activeTab.value,
page: pagination.page,
pageSize: pagination.pageSize,
labStorageCode: searchForm.labStorageCode || undefined,
motherName: searchForm.motherName || undefined,
createTimeStart: range?.[0],
createTimeEnd: range?.[1]
})
tableData.value = (res?.list || []) as Record<string, unknown>[]
pagination.total = Number(res?.total || 0)
supplementKeys.value = (res?.supplementKeys || []) as string[]
} catch (e: any) {
ElMessage.error(e?.message || '加载失败')
} finally {
loading.value = false
}
}
function handleSearch() {
pagination.page = 1
loadData()
}
function handleReset() {
searchForm.labStorageCode = ''
searchForm.motherName = ''
searchForm.createTimeRange = null
handleSearch()
}
function handleTabChange() {
pagination.page = 1
loadData()
}
async function openSupplement(row: Record<string, unknown>) {
const code = String(row.labStorageCode || row.cellCode || row.labFlowCode || '').trim()
if (!code) {
ElMessage.warning('缺少实验室入库编码,无法补录')
return
}
const meta = currentTabMeta.value
supplementForm.labStorageCode = code
supplementForm.receptionSource = meta.source
supplementForm.sampleTypeName = meta.sampleTypeName
supplementForm.remark = ''
const keys = supplementKeys.value.length
? supplementKeys.value
: currentColumns.value.map((c) => c.prop)
const payload: Record<string, string> = {}
keys.forEach((k) => {
payload[k] = ''
})
try {
const data = await fetchLabSampleProcessSupplement(code)
const existing = (data?.payload || {}) as Record<string, string>
keys.forEach((k) => {
payload[k] = existing[k] != null ? String(existing[k]) : ''
})
supplementForm.remark = data?.remark || ''
} catch {
// 无补录记录时用空表单
}
supplementForm.payload = payload
supplementVisible.value = true
}
async function submitSupplement() {
saving.value = true
try {
await saveLabSampleProcessSupplement({
labStorageCode: supplementForm.labStorageCode,
receptionSource: supplementForm.receptionSource,
sampleTypeName: supplementForm.sampleTypeName,
payload: { ...supplementForm.payload },
remark: supplementForm.remark
})
ElMessage.success('补录已保存')
supplementVisible.value = false
loadData()
} catch (e: any) {
ElMessage.error(e?.message || '保存失败')
} finally {
saving.value = false
}
}
function handleExport() {
const cols = currentColumns.value
const headers = ['序号', ...cols.map((c) => c.label)]
const rows = tableData.value.map((row, idx) => {
const line: Record<string, string | number> = {
序号: (pagination.page - 1) * pagination.pageSize + idx + 1
}
cols.forEach((c) => {
const v = row[c.prop]
line[c.label] = v === null || v === undefined || v === '' ? '' : String(v)
})
return line
})
const sheet = XLSX.utils.json_to_sheet(rows, { header: headers })
const workbook = XLSX.utils.book_new()
XLSX.utils.book_append_sheet(workbook, sheet, currentTabMeta.value.label.slice(0, 31))
const stamp = new Date().toISOString().slice(0, 19).replace(/[:T]/g, '')
XLSX.writeFile(workbook, `实验室样本处理数据_${currentTabMeta.value.key}_${stamp}.xlsx`)
}
onMounted(() => {
loadData()
})
</script>
<style scoped lang="scss">
.lab-process-tabs {
margin: 0 0 12px;
}
.form-tip {
margin: 0 0 12px;
color: var(--mint-text-muted, #6b7280);
font-size: 13px;
}
</style>