feat(entity-dim): 账套维度收口批 — subjects/analysis_results 加列 + 分析结论账套化 + 模板实例化构造点补账套
ci/woodpecker/push/main Pipeline failed
ci/woodpecker/push/main Pipeline failed
- ① kpi_templates instantiate 构造点显式传会话账套(原缺 entity_id ⇒ 静默落 e1),kpi_code 判重按账套收紧
- ② kpi_subject_map 3 个读取点加护栏注释(不加列、不改语义)
- ③ subjects 分层:读口 union(NULL=全局标准准则科目 / N=账套私有),UPSERT 唯一键改 (entity_id, subject_code)
- ④ analysis_results 账套化:读/写/删三端点按会话账套判定归属(列未落地时 kpi_code JOIN 止血),
无归属行进 unattributed 登记、删除一律 404
- 模型层按现网 DDL 补齐:KPIDefinition(template_id/is_system + uk_entity_kpi)、AnalysisResult.entity_id、Subject.entity_id
- 新增 migrate_{subjects,analysis_results}_entity_id_20260923.py(--check/--apply/--down)、
22 条对账用例 test_entity_dimension_convergence_20260923.py
DDL 已由 ops 受控通道落地(subjects 21:31:56 / analysis_results 21:31:59 CST,审批 1e9b4d9a/5ffb1640)
本提交为切片提交,未 push(发布须另取放行)
This commit is contained in:
@@ -200,6 +200,11 @@ def main():
|
||||
code_name, names, codes = load_voucher_index(db)
|
||||
subject_dict = load_subject_dict(db)
|
||||
kpis = {k.id: k for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == ENTITY_ID).all()}
|
||||
# ⚠️ kpi_subject_map 护栏(2026-09-23 账套维度收口批 ②,零逻辑改动):
|
||||
# 该表无账套列 —— **账套维度经 `kpi_id` 继承**。此处全表取映射后由 `kpis`(已按
|
||||
# `entity_id == ENTITY_ID` 过滤)逐行滤掉非本账套 KPI ⇒ 现状安全。
|
||||
# **禁止新增按 `subject_code` 反查的读口,禁止按 `subject_code` 批量写的入口**:
|
||||
# 同一科目码可同时属多个账套(标准准则科目全局共享),按科目反查/批量写立刻跨账套。
|
||||
rows = db.query(KPISubjectMap).order_by(KPISubjectMap.kpi_id, KPISubjectMap.id).all()
|
||||
grouped = defaultdict(list)
|
||||
for r in rows:
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""`analysis_results` 账套化 DDL — 2026-09-23
|
||||
|
||||
任务:`0-P1-project-20260923-entity-dimension-convergence-fullstack` §三④
|
||||
(裁定件 `project-entity-dimension-ruling-20260923` §3.2「读侧先止血、写侧随 DDL」)
|
||||
|
||||
背景
|
||||
----
|
||||
`analysis_results` 只有 2 行(`id=1 period=2026-06 kpi_code=TEST001` /
|
||||
`id=2 period=2026-07 kpi_code=NULL`,均 2026-07-29 创建),**无账套列**;
|
||||
`GET/POST/DELETE /api/cma/analysis/result*` 三端点原无 `get_entity_id`
|
||||
⇒ 双 token 返回体**逐字节相同**(跨账套读,页面 `AnalysisConfidence.vue` 已上线)。
|
||||
|
||||
本脚本只做 DDL(加列 + 索引),不做任何数据订正:
|
||||
· 存量 2 行**保持 `entity_id = NULL`** —— 归属存疑,按裁定件交 DAMA 取证 / 财务确认后另办;
|
||||
· 代码侧口径(`app/api/analysis_results.py`):`NULL` 行 = 无归属,进 `unattributed` 登记池,
|
||||
不参与任何账套过滤;列落地后读侧自动从 `kpi_code JOIN` 止血口径切到列口径。
|
||||
|
||||
用法
|
||||
----
|
||||
venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --check
|
||||
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
|
||||
--apply --allow-prod
|
||||
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
|
||||
--down --allow-prod --yes
|
||||
venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --show-create
|
||||
|
||||
生产保护(生产资产派单纪律):目标库名 == 'cma' 时 `--apply/--down` 一律拒绝;
|
||||
须 `--allow-prod` **且** `CMA_DDL_APPROVED=1`(= 已拿到 ops 审批单)才放行;
|
||||
`--check/--show-create` 为只读,不受限但会打标。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from sqlalchemy import create_engine, inspect, text # noqa: E402
|
||||
|
||||
TABLE = "analysis_results"
|
||||
COLUMN = "entity_id"
|
||||
NEW_INDEX = "ix_analysis_results_entity_id"
|
||||
PROD_DB_NAMES = ("cma",)
|
||||
DDL_ADD_COLUMN = (
|
||||
f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL "
|
||||
"COMMENT '企业ID(NULL=历史未归属行,不参与账套过滤)' AFTER id"
|
||||
)
|
||||
DDL_DROP_COLUMN = f"ALTER TABLE {TABLE} DROP COLUMN {COLUMN}"
|
||||
|
||||
|
||||
def is_sqlite(engine) -> bool:
|
||||
"""沙箱演练(SQLite)与生产(MySQL)的索引/列 DDL 方言差异"""
|
||||
return engine.dialect.name == "sqlite"
|
||||
|
||||
|
||||
def stmt_add_column(engine) -> str:
|
||||
"""生产形态 = MySQL(带 COMMENT / AFTER);沙箱 = SQLite 精简形态(语法不支持二者)"""
|
||||
if is_sqlite(engine):
|
||||
return f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL"
|
||||
return DDL_ADD_COLUMN
|
||||
|
||||
|
||||
def stmt_add_index(engine, name: str, cols: list) -> str:
|
||||
cols_s = ", ".join(cols)
|
||||
if is_sqlite(engine):
|
||||
return f"CREATE INDEX {name} ON {TABLE} ({cols_s})"
|
||||
return f"ALTER TABLE {TABLE} ADD INDEX {name} ({cols_s})"
|
||||
|
||||
|
||||
def stmt_drop_index(engine, name: str) -> str:
|
||||
return f"DROP INDEX {name}" if is_sqlite(engine) else f"ALTER TABLE {TABLE} DROP INDEX {name}"
|
||||
|
||||
|
||||
def db_name_of(engine) -> str:
|
||||
try:
|
||||
return engine.url.database or ""
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def state(engine) -> dict:
|
||||
insp = inspect(engine)
|
||||
tables = set(insp.get_table_names())
|
||||
out = {"table_exists": TABLE in tables, "has_column": False, "indexes": [], "rows": []}
|
||||
if not out["table_exists"]:
|
||||
return out
|
||||
out["has_column"] = COLUMN in {c["name"] for c in insp.get_columns(TABLE)}
|
||||
out["indexes"] = sorted({i["name"] for i in insp.get_indexes(TABLE)})
|
||||
with engine.connect() as conn:
|
||||
out["total"] = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
|
||||
if out["has_column"]:
|
||||
rows = conn.execute(text(
|
||||
f"SELECT id, period, kpi_code, IFNULL({COLUMN},'NULL') FROM {TABLE} ORDER BY id")).fetchall()
|
||||
else:
|
||||
rows = conn.execute(text(
|
||||
f"SELECT id, period, kpi_code, 'NO_COLUMN' FROM {TABLE} ORDER BY id")).fetchall()
|
||||
out["rows"] = [[r[0], r[1], r[2], str(r[3])] for r in rows]
|
||||
return out
|
||||
|
||||
|
||||
def print_check(st: dict) -> None:
|
||||
print("== 只读预检(inspector / SELECT,零写入)==")
|
||||
print(f" table_exists={st['table_exists']} has_column={st['has_column']} total={st.get('total')}")
|
||||
print(f" indexes={st['indexes']}")
|
||||
for r in st.get("rows", []):
|
||||
print(f" id={r[0]} period={r[1]} kpi_code={r[2]} entity_id={r[3]}")
|
||||
|
||||
|
||||
def do_apply(engine) -> int:
|
||||
st = state(engine)
|
||||
if not st["table_exists"]:
|
||||
print("!!! 目标库没有 analysis_results 表,拒绝 --apply")
|
||||
return 3
|
||||
done = []
|
||||
with engine.begin() as conn:
|
||||
if not st["has_column"]:
|
||||
add_col = stmt_add_column(engine)
|
||||
conn.execute(text(add_col))
|
||||
done.append(add_col)
|
||||
cur = state(engine)
|
||||
if NEW_INDEX not in cur["indexes"]:
|
||||
add_idx = stmt_add_index(engine, NEW_INDEX, [COLUMN])
|
||||
conn.execute(text(add_idx))
|
||||
done.append(add_idx)
|
||||
print("== APPLY 完成,执行语句:")
|
||||
for s in done:
|
||||
print(" " + s)
|
||||
if not done:
|
||||
print(" (无语句执行:已是目标形态,幂等)")
|
||||
print_check(state(engine))
|
||||
return 0
|
||||
|
||||
|
||||
def do_down(engine, yes: bool) -> int:
|
||||
st = state(engine)
|
||||
if not st["has_column"]:
|
||||
print("== DOWN 跳过:entity_id 列不存在(已是回滚态)")
|
||||
return 0
|
||||
assigned = [r for r in st.get("rows", []) if r[3] not in ("NULL", "NO_COLUMN")]
|
||||
if assigned and not yes:
|
||||
print(f"!!! 已有 {len(assigned)} 行带账套归属 {assigned}:回滚会丢失归属,须显式 --yes")
|
||||
return 3
|
||||
with engine.begin() as conn:
|
||||
if NEW_INDEX in st["indexes"]:
|
||||
conn.execute(text(stmt_drop_index(engine, NEW_INDEX)))
|
||||
conn.execute(text(DDL_DROP_COLUMN))
|
||||
print("== DOWN 完成:删除 entity_id 索引 + 列")
|
||||
print_check(state(engine))
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="analysis_results 账套化 DDL")
|
||||
g = ap.add_mutually_exclusive_group()
|
||||
g.add_argument("--check", action="store_true", help="只读预检(默认)")
|
||||
g.add_argument("--apply", action="store_true", help="执行 DDL(幂等)")
|
||||
g.add_argument("--down", action="store_true", help="回滚:删索引 + 删列")
|
||||
ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE")
|
||||
ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL(默认 backend/.env)")
|
||||
ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关(须 CMA_DDL_APPROVED=1)")
|
||||
ap.add_argument("--yes", action="store_true", help="--down 有账套归属行时的确认开关")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.url:
|
||||
engine = create_engine(args.url, echo=False)
|
||||
else:
|
||||
from app.database import get_engine
|
||||
engine = get_engine()
|
||||
|
||||
dbname = db_name_of(engine)
|
||||
print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | host={engine.url.host or 'socket'}")
|
||||
is_prod = dbname in PROD_DB_NAMES
|
||||
|
||||
if (args.apply or args.down) and is_prod:
|
||||
if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
|
||||
print(f"!!! 拒绝执行:目标库是生产库 {dbname}。")
|
||||
print("!!! 生产 DDL 须 ops 审批后另派执行(本单边界:不得直连生产写入)。")
|
||||
print("!!! 放行方式:--allow-prod 且 export CMA_DDL_APPROVED=1(持有审批单)。")
|
||||
return 3
|
||||
|
||||
if args.apply:
|
||||
return do_apply(engine)
|
||||
if args.down:
|
||||
return do_down(engine, args.yes)
|
||||
|
||||
st = state(engine)
|
||||
print_check(st)
|
||||
if is_prod:
|
||||
print(" [注意] 目标是生产库:以上仅为只读预检,未做任何写入。")
|
||||
if args.show_create:
|
||||
if engine.dialect.name != "mysql":
|
||||
print("== SHOW CREATE TABLE 跳过(非 MySQL 目标)==")
|
||||
else:
|
||||
with engine.connect() as conn:
|
||||
print(conn.execute(text(f"SHOW CREATE TABLE {TABLE}")).fetchone()[1])
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,297 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""`subjects` 账套分层 DDL + 存量回填 — 2026-09-23
|
||||
|
||||
任务:`0-P1-project-20260923-entity-dimension-convergence-fullstack` §三③
|
||||
(裁定件 `project-entity-dimension-ruling-20260923` §3.1「方案 A:分层」)
|
||||
|
||||
背景
|
||||
----
|
||||
`subjects` 207 行混装且 `subject_code` **全局唯一**(`uk_subject_code`):
|
||||
· 43 行标准准则科目(1001 库存现金 …,`remark IS NULL`)= 天然全局
|
||||
· 164 行 `remark='zxt-sync'` 酣客(智享通)企业明细 = 账套私有
|
||||
双 token 实测两账套拿到**同一份 207 行** ⇒ 博海页面出现酣客的
|
||||
`100106 酒窖备用金` / `100107 现金备用卡(唐)` / `100108 友赞商户`(结构性串配置)。
|
||||
|
||||
分层口径(`knowledge_articles` 同款已验模式)
|
||||
------------------------------------------
|
||||
`entity_id IS NULL` = 全局标准准则科目(两账套共享)
|
||||
`entity_id = N` = 账套 N 私有明细
|
||||
唯一键 `(subject_code)` → `(entity_id, subject_code)`
|
||||
⚠️ MySQL 把 NULL 视为互不相同的索引值 ⇒ 「全局行同码重复」索引拦不住,
|
||||
由写入口(`POST /api/cma/bot/subjects` UPSERT 判重)兜住,不可依赖索引。
|
||||
|
||||
动作(四段可独立跑,互不隐含)
|
||||
----------------------------
|
||||
--check 只读预检(默认):列/索引现状 + 行数分布 + 唯一键重复检查,零写入
|
||||
--apply DDL:ADD COLUMN entity_id INT NULL + INDEX + UNIQUE(entity_id,subject_code)
|
||||
− DROP INDEX uk_subject_code(幂等:已存在的步骤自动跳过)
|
||||
--backfill 存量回填(**须 --evidence <json>**,与 DDL 分开放行):
|
||||
`remark='zxt-sync'` 的 164 行 → `--target-entity`(默认 1);其余保持 NULL
|
||||
--down 回滚:重建 `uk_subject_code`(先断言无同码重复,fail-closed)→ 删新唯一键 → 删列
|
||||
--show-create 打印 SHOW CREATE TABLE subjects 原文(验收证据)
|
||||
|
||||
生产保护(生产资产派单纪律)
|
||||
--------------------------
|
||||
目标库名 == 'cma' 时 `--apply/--backfill/--down` 一律拒绝;须 `--allow-prod` **且**
|
||||
环境变量 `CMA_DDL_APPROVED=1`(= 已拿到 ops 审批单)才放行。`--check/--show-create` 只读不受限。
|
||||
|
||||
用法
|
||||
----
|
||||
venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py --check
|
||||
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py \\
|
||||
--apply --allow-prod
|
||||
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py \\
|
||||
--backfill --allow-prod --evidence /path/to/evidence.json --target-entity 1
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from sqlalchemy import create_engine, inspect, text # noqa: E402
|
||||
|
||||
TABLE = "subjects"
|
||||
COLUMN = "entity_id"
|
||||
NEW_UNIQUE = "uk_entity_subject"
|
||||
OLD_UNIQUE = "uk_subject_code"
|
||||
NEW_INDEX = "ix_subjects_entity_id"
|
||||
PROD_DB_NAMES = ("cma",)
|
||||
DDL_ADD_COLUMN = (
|
||||
f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL "
|
||||
"COMMENT '企业ID(NULL=全局标准准则科目 / N=账套私有明细)' AFTER id"
|
||||
)
|
||||
DDL_ADD_UNIQUE = f"ALTER TABLE {TABLE} ADD UNIQUE INDEX {NEW_UNIQUE} ({COLUMN}, subject_code)"
|
||||
DDL_DROP_OLD_UNIQUE = f"ALTER TABLE {TABLE} DROP INDEX {OLD_UNIQUE}"
|
||||
DDL_ADD_OLD_UNIQUE = f"ALTER TABLE {TABLE} ADD UNIQUE INDEX {OLD_UNIQUE} (subject_code)"
|
||||
DDL_DROP_COLUMN = f"ALTER TABLE {TABLE} DROP COLUMN {COLUMN}"
|
||||
|
||||
|
||||
def is_sqlite(engine) -> bool:
|
||||
"""沙箱演练(SQLite)与生产(MySQL)的 DDL 方言差异:索引/列定义语句形态不同"""
|
||||
return engine.dialect.name == "sqlite"
|
||||
|
||||
|
||||
def stmt_add_column(engine) -> str:
|
||||
"""生产形态 = MySQL(带 COMMENT / AFTER);沙箱 = SQLite 精简形态(语法不支持二者)"""
|
||||
if is_sqlite(engine):
|
||||
return f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL"
|
||||
return DDL_ADD_COLUMN
|
||||
|
||||
|
||||
def stmt_add_index(engine, name: str, cols: list, unique: bool = False) -> str:
|
||||
cols_s = ", ".join(cols)
|
||||
if is_sqlite(engine):
|
||||
return f"CREATE {'UNIQUE ' if unique else ''}INDEX {name} ON {TABLE} ({cols_s})"
|
||||
return f"ALTER TABLE {TABLE} ADD {'UNIQUE ' if unique else ''}INDEX {name} ({cols_s})"
|
||||
|
||||
|
||||
def stmt_drop_index(engine, name: str) -> str:
|
||||
return f"DROP INDEX {name}" if is_sqlite(engine) else f"ALTER TABLE {TABLE} DROP INDEX {name}"
|
||||
|
||||
|
||||
def db_name_of(engine) -> str:
|
||||
try:
|
||||
return engine.url.database or ""
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def state(engine) -> dict:
|
||||
insp = inspect(engine)
|
||||
tables = set(insp.get_table_names())
|
||||
out = {"table_exists": TABLE in tables, "has_column": False,
|
||||
"indexes": [], "unique_keys": [], "counts": {}}
|
||||
if not out["table_exists"]:
|
||||
return out
|
||||
out["has_column"] = COLUMN in {c["name"] for c in insp.get_columns(TABLE)}
|
||||
out["indexes"] = sorted({i["name"] for i in insp.get_indexes(TABLE)})
|
||||
out["unique_keys"] = sorted({uc["name"] for uc in insp.get_unique_constraints(TABLE)})
|
||||
with engine.connect() as conn:
|
||||
out["counts"]["total"] = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
|
||||
if out["has_column"]:
|
||||
rows = conn.execute(text(
|
||||
f"SELECT IFNULL({COLUMN},'NULL'), COUNT(*) FROM {TABLE} GROUP BY 1 ORDER BY 1")).fetchall()
|
||||
out["counts"]["by_entity"] = [[str(r[0]), int(r[1])] for r in rows]
|
||||
rows = conn.execute(text(
|
||||
f"SELECT IFNULL(remark,'(null)'), COUNT(*) FROM {TABLE} GROUP BY 1 ORDER BY 2 DESC")).fetchall()
|
||||
out["counts"]["by_remark"] = [[str(r[0]), int(r[1])] for r in rows]
|
||||
dup_code = conn.execute(text(
|
||||
f"SELECT COUNT(*) FROM (SELECT subject_code FROM {TABLE} GROUP BY subject_code "
|
||||
f"HAVING COUNT(*) > 1) t")).scalar()
|
||||
out["duplicate_subject_code_groups"] = int(dup_code)
|
||||
if out["has_column"]:
|
||||
dup_pair = conn.execute(text(
|
||||
f"SELECT COUNT(*) FROM (SELECT {COLUMN}, subject_code FROM {TABLE} "
|
||||
f"GROUP BY {COLUMN}, subject_code HAVING COUNT(*) > 1) t")).scalar()
|
||||
out["duplicate_entity_code_groups"] = int(dup_pair)
|
||||
return out
|
||||
|
||||
|
||||
def print_check(st: dict) -> None:
|
||||
print("== 只读预检(inspector / SELECT,零写入)==")
|
||||
print(f" table_exists={st['table_exists']} has_column={st['has_column']}")
|
||||
print(f" indexes={st['indexes']}")
|
||||
print(f" unique_keys={st['unique_keys']}")
|
||||
print(f" counts={st['counts']}")
|
||||
print(f" duplicate_subject_code_groups={st.get('duplicate_subject_code_groups')}")
|
||||
if "duplicate_entity_code_groups" in st:
|
||||
print(f" duplicate_entity_code_groups={st['duplicate_entity_code_groups']}")
|
||||
|
||||
|
||||
def do_apply(engine) -> int:
|
||||
st = state(engine)
|
||||
if not st["table_exists"]:
|
||||
print("!!! 目标库没有 subjects 表,拒绝 --apply")
|
||||
return 3
|
||||
if st.get("duplicate_subject_code_groups"):
|
||||
print("!!! subjects.subject_code 已有重复组,先处置数据再加/换唯一键")
|
||||
return 3
|
||||
done = []
|
||||
with engine.begin() as conn:
|
||||
if not st["has_column"]:
|
||||
add_col = stmt_add_column(engine)
|
||||
conn.execute(text(add_col))
|
||||
done.append(add_col)
|
||||
cur = state(engine)
|
||||
present = set(cur["indexes"]) | set(cur["unique_keys"])
|
||||
add_idx = stmt_add_index(engine, NEW_INDEX, [COLUMN])
|
||||
if NEW_INDEX not in present:
|
||||
conn.execute(text(add_idx))
|
||||
done.append(add_idx)
|
||||
add_uk = stmt_add_index(engine, NEW_UNIQUE, [COLUMN, "subject_code"], unique=True)
|
||||
if NEW_UNIQUE not in present:
|
||||
conn.execute(text(add_uk))
|
||||
done.append(add_uk)
|
||||
if OLD_UNIQUE in present:
|
||||
drop_old = stmt_drop_index(engine, OLD_UNIQUE)
|
||||
conn.execute(text(drop_old))
|
||||
done.append(drop_old)
|
||||
print("== APPLY 完成,执行语句:")
|
||||
for s in done:
|
||||
print(" " + s)
|
||||
if not done:
|
||||
print(" (无语句执行:已是目标形态,幂等)")
|
||||
print_check(state(engine))
|
||||
return 0
|
||||
|
||||
|
||||
def do_backfill(engine, evidence_path: str, target: int) -> int:
|
||||
if not evidence_path or not os.path.exists(evidence_path):
|
||||
print("!!! --backfill 必须提供 --evidence <json>(DAMA 只读取证 + 财务确认件锚点),拒绝执行")
|
||||
return 3
|
||||
ev = json.load(open(evidence_path, encoding="utf-8"))
|
||||
need = ("dama_evidence", "finance_confirmation", "counts")
|
||||
missing = [k for k in need if not ev.get(k)]
|
||||
if missing:
|
||||
print(f"!!! 取证件缺字段 {missing},拒绝执行(fail-closed)")
|
||||
return 3
|
||||
st = state(engine)
|
||||
if not st["has_column"]:
|
||||
print("!!! entity_id 列未落地,先跑 --apply")
|
||||
return 3
|
||||
by_remark = dict(st["counts"]["by_remark"])
|
||||
live_private = int(by_remark.get("zxt-sync", 0))
|
||||
live_global = int(by_remark.get("(null)", 0))
|
||||
exp = ev["counts"]
|
||||
if int(exp.get("e1", -1)) != live_private or int(exp.get("global_null", -1)) != live_global:
|
||||
print(f"!!! 取证件行数锚点与现网不符:evidence e1={exp.get('e1')}/global_null="
|
||||
f"{exp.get('global_null')} vs live {live_private}/{live_global},拒绝执行")
|
||||
return 3
|
||||
with engine.begin() as conn:
|
||||
n1 = conn.execute(text(
|
||||
f"UPDATE {TABLE} SET {COLUMN} = :e WHERE remark = 'zxt-sync'"), {"e": target}).rowcount
|
||||
n0 = conn.execute(text(
|
||||
f"UPDATE {TABLE} SET {COLUMN} = NULL WHERE remark IS NULL AND {COLUMN} IS NOT NULL")).rowcount
|
||||
print(f"== BACKFILL 完成:remark='zxt-sync' → entity_id={target} 命中 {n1} 行;"
|
||||
f"标准准则科目置 NULL {n0} 行(evidence: {evidence_path})")
|
||||
print_check(state(engine))
|
||||
return 0
|
||||
|
||||
|
||||
def do_down(engine, yes: bool) -> int:
|
||||
st = state(engine)
|
||||
if not st["has_column"]:
|
||||
print("== DOWN 跳过:entity_id 列不存在(已是回滚态)")
|
||||
return 0
|
||||
if st.get("duplicate_subject_code_groups"):
|
||||
print(f"!!! 已存在同码多行({st['duplicate_subject_code_groups']} 组)⇒ 无法重建全局唯一键 "
|
||||
f"uk_subject_code,拒绝 --down(fail-closed:先人工处置同码行)")
|
||||
return 3
|
||||
if st["counts"]["by_entity"] and len(st["counts"]["by_entity"]) > 1 and not yes:
|
||||
print(f"!!! 列内已有分账套数据 {st['counts']['by_entity']}:回滚会丢失账套归属,须显式 --yes")
|
||||
return 3
|
||||
with engine.begin() as conn:
|
||||
present = set(st["indexes"]) | set(st["unique_keys"])
|
||||
if NEW_UNIQUE in present:
|
||||
conn.execute(text(stmt_drop_index(engine, NEW_UNIQUE)))
|
||||
cur = state(engine)
|
||||
present = set(cur["indexes"]) | set(cur["unique_keys"])
|
||||
if NEW_INDEX in present:
|
||||
conn.execute(text(stmt_drop_index(engine, NEW_INDEX)))
|
||||
if OLD_UNIQUE not in present:
|
||||
conn.execute(text(stmt_add_index(engine, OLD_UNIQUE, ["subject_code"], unique=True)))
|
||||
conn.execute(text(DDL_DROP_COLUMN))
|
||||
print("== DOWN 完成:重建 uk_subject_code + 删除 entity_id 列")
|
||||
print_check(state(engine))
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="subjects 账套分层 DDL/回填")
|
||||
g = ap.add_mutually_exclusive_group()
|
||||
g.add_argument("--check", action="store_true", help="只读预检(默认)")
|
||||
g.add_argument("--apply", action="store_true", help="执行 DDL(幂等)")
|
||||
g.add_argument("--backfill", action="store_true", help="存量回填(须 --evidence)")
|
||||
g.add_argument("--down", action="store_true", help="回滚")
|
||||
ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE")
|
||||
ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL(默认 backend/.env)")
|
||||
ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关(须 CMA_DDL_APPROVED=1)")
|
||||
ap.add_argument("--evidence", default=None, help="--backfill 的取证件 JSON")
|
||||
ap.add_argument("--target-entity", type=int, default=1, help="--backfill 目标账套(默认 1=酣客)")
|
||||
ap.add_argument("--yes", action="store_true", help="--down 有分账套数据时的确认开关")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.url:
|
||||
engine = create_engine(args.url, echo=False)
|
||||
else:
|
||||
from app.database import get_engine
|
||||
engine = get_engine()
|
||||
|
||||
dbname = db_name_of(engine)
|
||||
print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | host={engine.url.host or 'socket'}")
|
||||
is_prod = dbname in PROD_DB_NAMES
|
||||
|
||||
if (args.apply or args.backfill or args.down) and is_prod:
|
||||
if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
|
||||
print(f"!!! 拒绝执行:目标库是生产库 {dbname}。")
|
||||
print("!!! 生产 DDL/回填须 ops 审批后另派执行(本单边界:不得直连生产写入)。")
|
||||
print("!!! 放行方式:--allow-prod 且 export CMA_DDL_APPROVED=1(持有审批单)。")
|
||||
return 3
|
||||
|
||||
if args.apply:
|
||||
return do_apply(engine)
|
||||
if args.backfill:
|
||||
return do_backfill(engine, args.evidence, args.target_entity)
|
||||
if args.down:
|
||||
return do_down(engine, args.yes)
|
||||
|
||||
st = state(engine)
|
||||
print_check(st)
|
||||
if is_prod:
|
||||
print(" [注意] 目标是生产库:以上仅为只读预检,未做任何写入。")
|
||||
if args.show_create:
|
||||
if engine.dialect.name != "mysql":
|
||||
print("== SHOW CREATE TABLE 跳过(非 MySQL 目标)==")
|
||||
else:
|
||||
with engine.connect() as conn:
|
||||
print(conn.execute(text(f"SHOW CREATE TABLE {TABLE}")).fetchone()[1])
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user