Files
zkxw/backend/scripts/refresh_qc_task_result_after_param_change.py
T
zlwuandCursor e2a50b33fe chore(qc): 补充阴阳性参数变更后历史结果刷新工具
提供第 1–10 条阴/阳→阴性/阳性及空值默认回填的预览执行脚本与迁移 SQL。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-11-09 11:37:41 +08:00

584 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""质检参数模板变更后,刷新历史任务结果(只改已有结果行,不给缺行任务插数)。
对应 2026-11-09 配置变更中的第 1–10 条:
1. `ZKXW-QC-000-1`:历史值「阴/阳」→「阴性/阳性」
2–10. 文本改阴阳性单选:已有结果行若值为空 → 写成模板默认「阴性」;
已有「阴/阳」同样归一;定量等异形值(如 0.01)保留不动。
不处理第 11 条 HLA(单选改文本),也不为「从未录入」的任务 INSERT 结果行。
等价 SQL(推荐在客户端直接跑预览/事务)见:
docs/sql/migrations/2026-11-09-refresh-qc-task-result-yin-yang-1-10.sql
用法(读取 backend/.env 的 DATABASE_URL,默认 dry-run):
# 推荐:按第 1–10 思路预览
python backend/scripts/refresh_qc_task_result_after_param_change.py --preset items-1-10
# 测试库实际写入
python backend/scripts/refresh_qc_task_result_after_param_change.py --preset items-1-10 --apply
# 生产库需显式放行库名
python backend/scripts/refresh_qc_task_result_after_param_change.py --preset items-1-10 --apply --allow-db zkxw
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.parse
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple
from sqlalchemy import create_engine, text
from sqlalchemy.engine import Engine
# 第 1–10 条(阴阳性类);不含第 11 条 HLA、不含仅重存项
ITEM_CODES_1_TO_10 = (
"ZKXW-QC-000-1", # 1 文案:阴/阳 → 阴性/阳性
"ZKXW-QC-003-18", # 2
"ZKXW-QC-003-19", # 3
"ZKXW-QC-003-17", # 4
"ZKXW-QC-001-12", # 5
"ZKXW-QC-001-11", # 6
"ZKXW-QC-001-10", # 7 可能有定量历史,异形值跳过
"ZKXW-QC-001-06", # 8
"ZKXW-QC-001-18", # 9
"ZKXW-QC-002-02", # 10
)
# 完整变更清单(含 HLA,供手工 --item-codes 对照;preset 不用)
ALL_CHANGED_ITEM_CODES = ITEM_CODES_1_TO_10 + ("ZKXW-QC-002-01",)
LABEL_MAP = {"阴": "阴性", "阳": "阳性"}
# 默认拒绝直连的库名(生产库 Docker 名);测试库 zkxw-sc 可写
DEFAULT_BLOCKED_DBS = {"zkxw"}
PRESETS = {
"items-1-10": {
"item_codes": ITEM_CODES_1_TO_10,
"normalize_labels": True,
"fill_empty_default": True,
"sync_param_type": True,
"description": "第1–10条:阴/阳归一 + 空结果行补默认阴性 + 兼容时同步类型快照;不插缺行",
}
}
def _load_database_url() -> str:
env_url = (os.environ.get("DATABASE_URL") or "").strip()
if env_url:
return env_url
root = Path(__file__).resolve().parents[2]
env_path = root / "backend" / ".env"
if not env_path.is_file():
env_path = Path(__file__).resolve().parents[1] / ".env"
if not env_path.is_file():
raise SystemExit("未找到 DATABASE_URL 或 backend/.env")
for line in env_path.read_text(encoding="utf-8").splitlines():
if line.startswith("DATABASE_URL="):
return line.split("=", 1)[1].strip().strip('"').strip("'")
raise SystemExit(f"{env_path} 中无 DATABASE_URL")
def _db_name(url: str) -> str:
return (urllib.parse.urlparse(url).path or "").lstrip("/")
def _parse_options(raw: Any) -> List[str]:
if not raw:
return []
if isinstance(raw, list):
return [str(x).strip() for x in raw if str(x).strip()]
try:
arr = json.loads(str(raw))
except json.JSONDecodeError:
return []
if not isinstance(arr, list):
return []
return [str(x).strip() for x in arr if str(x).strip()]
def _fetch_params(engine: Engine, codes: Sequence[str]) -> List[Dict[str, Any]]:
sql = text(
"""
SELECT
p.id AS param_id,
p.item_source,
p.item_id,
COALESCE(l.item_code, t.item_code) AS item_code,
COALESCE(l.item_name, t.item_name) AS item_name,
p.param_name,
p.param_code,
p.param_unit,
p.param_type,
p.options,
p.default_value,
p.tenant_no
FROM biz_qc_item_param p
LEFT JOIN biz_qc_lab_test_item l
ON p.item_source = 'lab' AND p.item_id = l.id AND l.del_flag = 0
LEFT JOIN biz_qc_third_party_item t
ON p.item_source = 'third_party' AND p.item_id = t.id AND t.del_flag = 0
WHERE p.del_flag = 0
AND COALESCE(l.item_code, t.item_code) IN :codes
ORDER BY COALESCE(l.item_code, t.item_code), p.order_num
"""
)
with engine.connect() as conn:
return [dict(r) for r in conn.execute(sql, {"codes": tuple(codes)}).mappings()]
def _scan_existing_results(
engine: Engine, param_ids: Sequence[str]
) -> List[Dict[str, Any]]:
if not param_ids:
return []
sql = text(
"""
SELECT
r.id AS result_id,
r.task_id,
r.param_id,
r.param_name,
r.param_code,
r.param_type,
r.param_value,
r.tenant_no
FROM biz_qc_task_result r
WHERE r.del_flag = 0
AND r.param_id IN :param_ids
"""
)
with engine.connect() as conn:
return [dict(r) for r in conn.execute(sql, {"param_ids": tuple(param_ids)}).mappings()]
def _count_tasks_missing_result(
engine: Engine, params: Sequence[Dict[str, Any]]
) -> Dict[str, int]:
"""只读统计:任务已关联项目但无对应结果行(本脚本不会 INSERT)。"""
out: Dict[str, int] = {}
lab = [p for p in params if p["item_source"] == "lab"]
third = [p for p in params if p["item_source"] == "third_party"]
with engine.connect() as conn:
if lab:
rows = conn.execute(
text(
"""
SELECT lti.item_code, COUNT(*) AS c
FROM biz_qc_task task
JOIN biz_qc_work_order_lab_item wi
ON wi.id = task.source_line_id AND wi.del_flag = 0
JOIN biz_qc_lab_test_item lti
ON lti.id = wi.lab_test_item_id AND lti.del_flag = 0
JOIN biz_qc_item_param p
ON p.item_source = 'lab' AND p.item_id = lti.id AND p.del_flag = 0
WHERE task.del_flag = 0
AND task.task_type = 'sysz'
AND p.id IN :param_ids
AND NOT EXISTS (
SELECT 1 FROM biz_qc_task_result r
WHERE r.task_id = task.id AND r.del_flag = 0
AND (
r.param_id = p.id
OR (p.param_code <> '' AND r.param_code = p.param_code)
)
)
GROUP BY lti.item_code
"""
),
{"param_ids": tuple(p["param_id"] for p in lab)},
).mappings()
for r in rows:
out[str(r["item_code"])] = int(r["c"])
if third:
rows = conn.execute(
text(
"""
SELECT tpi.item_code, COUNT(*) AS c
FROM biz_qc_task task
JOIN biz_qc_work_order_third_item wi
ON wi.id = task.source_line_id AND wi.del_flag = 0
JOIN biz_qc_third_party_item tpi
ON tpi.id = wi.third_party_item_id AND tpi.del_flag = 0
JOIN biz_qc_item_param p
ON p.item_source = 'third_party' AND p.item_id = tpi.id
AND p.del_flag = 0
WHERE task.del_flag = 0
AND task.task_type = 'dsfjc'
AND p.id IN :param_ids
AND NOT EXISTS (
SELECT 1 FROM biz_qc_task_result r
WHERE r.task_id = task.id AND r.del_flag = 0
AND (
r.param_id = p.id
OR (p.param_code <> '' AND r.param_code = p.param_code)
)
)
GROUP BY tpi.item_code
"""
),
{"param_ids": tuple(p["param_id"] for p in third)},
).mappings()
for r in rows:
code = str(r["item_code"])
out[code] = out.get(code, 0) + int(r["c"])
return out
def _inventory(
params: List[Dict[str, Any]], rows: List[Dict[str, Any]]
) -> Dict[str, Dict[str, int]]:
by_param = {p["param_id"]: p for p in params}
inv: Dict[str, Dict[str, int]] = defaultdict(
lambda: {
"result_rows": 0,
"empty": 0,
"old_label": 0,
"standard_label": 0,
"other_value": 0,
}
)
for row in rows:
param = by_param.get(row["param_id"])
if not param:
continue
code = param["item_code"]
inv[code]["result_rows"] += 1
value = (row.get("param_value") or "").strip()
if not value:
inv[code]["empty"] += 1
elif value in LABEL_MAP:
inv[code]["old_label"] += 1
elif value in ("阴性", "阳性"):
inv[code]["standard_label"] += 1
else:
inv[code]["other_value"] += 1
return dict(inv)
def _plan_updates(
params: List[Dict[str, Any]],
rows: List[Dict[str, Any]],
*,
normalize_labels: bool,
fill_empty_default: bool,
sync_param_type: bool,
) -> Tuple[List[Dict[str, Any]], Dict[str, int]]:
by_param = {p["param_id"]: p for p in params}
planned: List[Dict[str, Any]] = []
stats = {
"scanned": len(rows),
"normalize_labels": 0,
"fill_empty_default": 0,
"sync_param_type": 0,
"skip_incompatible_value": 0,
"unchanged": 0,
}
for row in rows:
param = by_param.get(row["param_id"])
if not param:
continue
old_value = (row.get("param_value") or "").strip()
new_value = old_value
old_type = (row.get("param_type") or "text").strip() or "text"
new_type = (param.get("param_type") or "text").strip() or "text"
opts = _parse_options(param.get("options"))
default = (param.get("default_value") or "").strip()
reasons: List[str] = []
if normalize_labels and old_value in LABEL_MAP:
new_value = LABEL_MAP[old_value]
reasons.append("normalize_labels")
stats["normalize_labels"] += 1
if fill_empty_default and not new_value and new_type == "select" and default:
new_value = default
reasons.append("fill_empty_default")
stats["fill_empty_default"] += 1
type_to_write = old_type
if sync_param_type and old_type != new_type:
# 值空或已在新选项内才同步类型;定量等异形值跳过
if (not new_value) or (new_type != "select") or (new_value in opts):
type_to_write = new_type
reasons.append("sync_param_type")
stats["sync_param_type"] += 1
else:
stats["skip_incompatible_value"] += 1
if not reasons or (new_value == old_value and type_to_write == old_type):
if not reasons:
stats["unchanged"] += 1
elif new_value == old_value and type_to_write == old_type:
stats["unchanged"] += 1
continue
planned.append(
{
"result_id": row["result_id"],
"task_id": row["task_id"],
"item_code": param["item_code"],
"param_name": param["param_name"],
"old_value": old_value,
"new_value": new_value,
"old_type": old_type,
"new_type": type_to_write,
"reasons": reasons,
}
)
return planned, stats
def _apply_updates(
engine: Engine,
planned: List[Dict[str, Any]],
*,
update_by: str,
) -> int:
if not planned:
return 0
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
sql = text(
"""
UPDATE biz_qc_task_result
SET param_value = :param_value,
param_type = :param_type,
update_by = :update_by,
update_time = :update_time
WHERE id = :id AND del_flag = 0
"""
)
with engine.begin() as conn:
for item in planned:
conn.execute(
sql,
{
"id": item["result_id"],
"param_value": item["new_value"] or None,
"param_type": item["new_type"],
"update_by": update_by,
"update_time": now,
},
)
return len(planned)
def _print_by_item(planned: List[Dict[str, Any]]) -> None:
grouped: Dict[str, Counter] = defaultdict(Counter)
for item in planned:
for reason in item["reasons"]:
grouped[item["item_code"]][reason] += 1
grouped[item["item_code"]]["rows"] += 1
if not grouped:
print(" (none)")
return
for code in sorted(grouped):
c = grouped[code]
print(
f" {code}: rows={c['rows']} "
f"normalize={c['normalize_labels']} "
f"fill_empty={c['fill_empty_default']} "
f"sync_type={c['sync_param_type']}"
)
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = argparse.ArgumentParser(
description="刷新质检历史结果(默认 dry-run)。推荐 --preset items-1-10"
)
parser.add_argument(
"--preset",
choices=sorted(PRESETS.keys()),
help="使用约定刷新策略(推荐 items-1-10)",
)
parser.add_argument(
"--apply",
action="store_true",
help="实际写入;默认只预览",
)
parser.add_argument(
"--normalize-labels",
action=argparse.BooleanOptionalAction,
default=None,
help="阴/阳 → 阴性/阳性;preset 默认开",
)
parser.add_argument(
"--fill-empty-default",
action=argparse.BooleanOptionalAction,
default=None,
help="已有空结果行写成模板默认值;preset items-1-10 默认开;绝不 INSERT 缺行",
)
parser.add_argument(
"--sync-param-type",
action=argparse.BooleanOptionalAction,
default=None,
help="值兼容时同步 param_type 快照;preset 默认开",
)
parser.add_argument(
"--item-codes",
nargs="*",
default=None,
help="手工指定项目编码;与 --preset 同时给时以本参数为准",
)
parser.add_argument(
"--allow-db",
action="append",
default=[],
help="额外允许写入的数据库名;生产库名默认拒绝",
)
parser.add_argument(
"--update-by",
default="d3000000000000000000000000000001",
help="写入 update_by",
)
parser.add_argument(
"--limit-print",
type=int,
default=50,
help="明细最多打印条数",
)
parser.add_argument(
"--skip-missing-stats",
action="store_true",
help="跳过「缺结果行任务」只读统计",
)
args = parser.parse_args(argv)
if args.preset:
preset = PRESETS[args.preset]
item_codes = list(args.item_codes) if args.item_codes is not None else list(
preset["item_codes"]
)
normalize_labels = (
preset["normalize_labels"]
if args.normalize_labels is None
else args.normalize_labels
)
fill_empty_default = (
preset["fill_empty_default"]
if args.fill_empty_default is None
else args.fill_empty_default
)
sync_param_type = (
preset["sync_param_type"]
if args.sync_param_type is None
else args.sync_param_type
)
print(f"preset={args.preset}: {preset['description']}")
else:
item_codes = (
list(args.item_codes)
if args.item_codes is not None
else list(ITEM_CODES_1_TO_10)
)
# 无 preset 时保持保守:只开文案归一与类型同步,空值需显式打开
normalize_labels = True if args.normalize_labels is None else args.normalize_labels
fill_empty_default = (
False if args.fill_empty_default is None else args.fill_empty_default
)
sync_param_type = True if args.sync_param_type is None else args.sync_param_type
print(
"preset=(none); 默认仅 normalize+sync_type。"
"完整第1–10策略请加 --preset items-1-10"
)
url = _load_database_url()
db = _db_name(url)
print(f"database={db}")
print(
f"flags: normalize_labels={normalize_labels} "
f"fill_empty_default={fill_empty_default} sync_param_type={sync_param_type}"
)
print(f"item_codes({len(item_codes)}): {', '.join(item_codes)}")
blocked = DEFAULT_BLOCKED_DBS - set(args.allow_db)
if args.apply and db in blocked:
raise SystemExit(
f"拒绝写入疑似生产库 `{db}`。"
f"测试库确认后重试;确需写入请加 --allow-db {db}"
)
engine = create_engine(url, pool_pre_ping=True)
params = _fetch_params(engine, item_codes)
if not params:
print("未找到匹配的参数模板,退出")
return 1
found_codes = {p["item_code"] for p in params}
missing_codes = [c for c in item_codes if c not in found_codes]
print(f"matched_params={len(params)}")
for p in params:
print(
f" {p['item_code']} | {p['param_name']} | type={p['param_type']} "
f"| options={p['options']} | default={p['default_value']}"
)
if missing_codes:
print(f"WARNING missing item templates: {', '.join(missing_codes)}")
rows = _scan_existing_results(engine, [p["param_id"] for p in params])
inv = _inventory(params, rows)
print("--- inventory (existing result rows only) ---")
if not inv:
print(" (no existing result rows for these params)")
for code in item_codes:
if code not in inv:
print(f" {code}: result_rows=0")
continue
s = inv[code]
print(
f" {code}: rows={s['result_rows']} empty={s['empty']} "
f"old_label(阴/阳)={s['old_label']} "
f"standard(阴性/阳性)={s['standard_label']} "
f"other={s['other_value']}"
)
if not args.skip_missing_stats:
missing = _count_tasks_missing_result(engine, params)
print("--- tasks with NO result row (info only; will NOT insert) ---")
if not missing:
print(" (none)")
else:
for code in sorted(missing):
print(f" {code}: {missing[code]}")
planned, stats = _plan_updates(
params,
rows,
normalize_labels=normalize_labels,
fill_empty_default=fill_empty_default,
sync_param_type=sync_param_type,
)
print("--- plan stats ---")
for k, v in stats.items():
print(f" {k}: {v}")
print("--- plan by item ---")
_print_by_item(planned)
print(f"planned_updates: {len(planned)}")
for item in planned[: args.limit_print]:
print(
f" {item['item_code']} task={item['task_id']} "
f"value {item['old_value']!r}->{item['new_value']!r} "
f"type {item['old_type']}->{item['new_type']} "
f"reasons={item['reasons']}"
)
if len(planned) > args.limit_print:
print(f" ... and {len(planned) - args.limit_print} more")
if not args.apply:
print("dry-run only;加 --apply 才会写入")
return 0
n = _apply_updates(engine, planned, update_by=args.update_by)
print(f"applied={n}")
return 0
if __name__ == "__main__":
sys.exit(main())