chore(qc): 补充阴阳性参数变更后历史结果刷新工具
提供第 1–10 条阴/阳→阴性/阳性及空值默认回填的预览执行脚本与迁移 SQL。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
3d9a71da46
commit
e2a50b33fe
2 files changed
+931
No files matched your search
@@ -0,0 +1,583 @@
|
||||
#!/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())
|
||||
Reference in new issue
Block a user