325 lines
12 KiB
Python
325 lines
12 KiB
Python
"""将会议 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()
|