116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""数据治理 migration: 入库必检约束 + 元数据字段补充"""
|
||
import pymysql
|
||
import os
|
||
import logging
|
||
|
||
# 2026-09-12 库访问治理（ops）：凭据从 backend/.env 读取，无明文兜底
|
||
import os
|
||
try:
|
||
from dotenv import load_dotenv as _ld_gov
|
||
_ld_gov('/root/cma-management/backend/.env')
|
||
except Exception:
|
||
pass
|
||
if not os.getenv('CMA_DB_PASS'):
|
||
raise SystemExit('ERROR: 缺少 CMA_DB_PASS（2026-09-12 凭据治理已移除明文兜底）')
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger("data-governance")
|
||
|
||
DB_USER = os.getenv("CMA_DB_USER", "cma_user")
|
||
DB_PASS = os.environ["CMA_DB_PASS"]
|
||
DB_HOST = os.getenv("CMA_DB_HOST", "127.0.0.1")
|
||
DB_PORT = int(os.getenv("CMA_DB_PORT", "3306"))
|
||
DB_NAME = os.getenv("CMA_DB_NAME", "cma")
|
||
|
||
conn = pymysql.connect(
|
||
host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS, database=DB_NAME,
|
||
charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor,
|
||
)
|
||
cur = conn.cursor()
|
||
|
||
|
||
def run():
|
||
logger.info("=== 数据治理 Migration 开始 ===")
|
||
|
||
# ── 0. 添加缺失列 ──
|
||
logger.info("[步骤0] 检查并补充缺失列...")
|
||
|
||
for col_name, col_def in [
|
||
("data_source", "ALTER TABLE kpi_definitions ADD COLUMN data_source VARCHAR(500) DEFAULT NULL COMMENT '数据来源' AFTER data_source_type"),
|
||
("data_owner", "ALTER TABLE kpi_definitions ADD COLUMN data_owner VARCHAR(100) DEFAULT NULL COMMENT '数据责任人' AFTER data_source"),
|
||
]:
|
||
cur.execute("SHOW COLUMNS FROM kpi_definitions LIKE %s", (col_name,))
|
||
if not cur.fetchone():
|
||
logger.info(f" 添加 {col_name} 列...")
|
||
cur.execute(col_def)
|
||
conn.commit()
|
||
logger.info(f" {col_name} 列已添加")
|
||
else:
|
||
logger.info(f" {col_name} 列已存在")
|
||
|
||
# ── 1. 修复空值 ──
|
||
logger.info("[步骤1] 修复空值...")
|
||
|
||
for col, default, label in [
|
||
("target_value", "0", "NULL target_value"),
|
||
("unit", "'-'", "NULL/empty unit"),
|
||
("formula", "'待补充'", "NULL/empty formula"),
|
||
("data_source", "'待补充'", "NULL/empty data_source"),
|
||
("data_owner", "'待指定'", "NULL/empty data_owner"),
|
||
]:
|
||
if col in ("target_value",):
|
||
r = cur.execute(f"SELECT COUNT(*) as cnt FROM kpi_definitions WHERE {col} IS NULL")
|
||
else:
|
||
r = cur.execute(f"SELECT COUNT(*) as cnt FROM kpi_definitions WHERE {col} IS NULL OR {col} = ''")
|
||
row = cur.fetchone()
|
||
cnt = row["cnt"] if row else 0
|
||
logger.info(f" {label}: {cnt} 条")
|
||
if cnt > 0:
|
||
if col in ("target_value",):
|
||
cur.execute(f"UPDATE kpi_definitions SET {col} = {default} WHERE {col} IS NULL")
|
||
else:
|
||
cur.execute(f"UPDATE kpi_definitions SET {col} = {default} WHERE {col} IS NULL OR {col} = ''")
|
||
logger.info(f" 已修复 {cur.rowcount} 条")
|
||
|
||
conn.commit()
|
||
|
||
# ── 2. 修改列约束为 NOT NULL ──
|
||
logger.info("[步骤2] 修改列约束...")
|
||
|
||
alters = [
|
||
("target_value", "DECIMAL(15,2) NOT NULL DEFAULT 0"),
|
||
("unit", "VARCHAR(50) NOT NULL DEFAULT '-'"),
|
||
("formula", "TEXT NOT NULL"),
|
||
("data_source", "VARCHAR(500) NOT NULL DEFAULT '待补充'"),
|
||
("data_owner", "VARCHAR(100) NOT NULL DEFAULT '待指定'"),
|
||
]
|
||
for col, col_type in alters:
|
||
col_comment = {
|
||
"target_value": "目标值", "unit": "单位", "formula": "计算公式",
|
||
"data_source": "数据来源", "data_owner": "数据责任人",
|
||
}[col]
|
||
try:
|
||
cur.execute(f"ALTER TABLE kpi_definitions MODIFY {col} {col_type} COMMENT '{col_comment}'")
|
||
logger.info(f" {col} → {col_type}")
|
||
except Exception as e:
|
||
logger.warning(f" {col} 修改失败: {e}")
|
||
|
||
conn.commit()
|
||
|
||
# ── 3. 验证 ──
|
||
logger.info("[步骤3] 验证约束...")
|
||
cur.execute("DESCRIBE kpi_definitions")
|
||
for col in cur.fetchall():
|
||
if col['Field'] in ('target_value', 'unit', 'formula', 'data_source', 'data_owner'):
|
||
logger.info(f" {col['Field']}: Null={col['Null']}, Default={col['Default']}, Type={col['Type']}")
|
||
|
||
logger.info("=== 数据治理 Migration 完成 ===")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
run()
|
||
finally:
|
||
cur.close()
|
||
conn.close()
|