zlwuandCursor
e2a50b33fe
提供第 1–10 条阴/阳→阴性/阳性及空值默认回填的预览执行脚本与迁移 SQL。 Co-authored-by: Cursor <cursoragent@cursor.com>
584 lines
20 KiB
Python
584 lines
20 KiB
Python
#!/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())
|