"""P1 类型标签字典化：补齐 7 个缺失字典类型，幂等。 背景：ticket/notice/trait/selection_rule/seedling 五个页面的类型/状态标签此前 在前端硬编码。为统一为 sys_dict 驱动的 {value: 英码/码, label: 中文}，需要补齐 以下字典类型（sys_yes_no 已存在，直接复用）： ticket_type 工单类型 suggestion/bug/optimize/other ticket_status 工单状态 0待处理/1处理中/2已完成/3已关闭 notice_status 公告状态 0启用/1停用 trait_data_type 性状数据类型 numeric/categorical/date selection_rule_logic 选择规则逻辑 and/or selection_rule_action 选择规则动作 keep/eliminate seedling_stage 育苗阶段 1播种/2出苗/3壮苗 用法（backend/ 目录）： ENVIRONMENT=dev "C:/ai/miniconda3/envs/dpb/python.exe" scripts/seed_p1_dicts.py --check ENVIRONMENT=dev "C:/ai/miniconda3/envs/dpb/python.exe" scripts/seed_p1_dicts.py 幂等性：类型/值已存在则跳过，可重复执行。执行前将受影响字典行备份到 bre_backup_sys_dict_type / bre_backup_sys_dict_data（首次建表，之后只补增量）。 """ import argparse import os import sys import uuid BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if BACKEND_DIR not in sys.path: sys.path.insert(0, BACKEND_DIR) from sqlalchemy import create_engine, text # noqa: E402 from app.config.setting import settings # noqa: E402 # (dict_sort, dict_label, dict_value) —— dict_value 与后端枚举/库内存值一致 DICT_TYPES = { "ticket_type": { "dict_name": "工单类型", "description": "工单-类型", "rows": [ (1, "建议", "suggestion"), (2, "缺陷", "bug"), (3, "优化", "optimize"), (4, "其他", "other"), ], }, "ticket_status": { "dict_name": "工单状态", "description": "工单-状态", "rows": [ (1, "待处理", "0"), (2, "处理中", "1"), (3, "已完成", "2"), (4, "已关闭", "3"), ], }, "notice_status": { "dict_name": "公告状态", "description": "公告-启用状态", "rows": [ (1, "启用", "0"), (2, "停用", "1"), ], }, "trait_data_type": { "dict_name": "性状数据类型", "description": "性状-数据类型", "rows": [ (1, "数值型", "numeric"), (2, "分类型", "categorical"), (3, "日期型", "date"), ], }, "selection_rule_logic": { "dict_name": "选择规则逻辑", "description": "选择规则-逻辑关系", "rows": [ (1, "与", "and"), (2, "或", "or"), ], }, "selection_rule_action": { "dict_name": "选择规则动作", "description": "选择规则-动作", "rows": [ (1, "保留", "keep"), (2, "淘汰", "eliminate"), ], }, "seedling_stage": { "dict_name": "育苗阶段", "description": "育苗-阶段", "rows": [ (1, "播种", "1"), (2, "出苗", "2"), (3, "壮苗", "3"), ], }, } def backup_dict_rows(conn, types: list[str]) -> None: """把受影响字典行快照到 bre_backup_* 表（首次建表，之后只补不覆盖）。""" placeholders = ",".join(f":t{i}" for i in range(len(types))) params = {f"t{i}": t for i, t in enumerate(types)} conn.execute( text( f"CREATE TABLE IF NOT EXISTS bre_backup_sys_dict_type " f"AS SELECT * FROM sys_dict_type WHERE 1=0" ) ) conn.execute( text( f"CREATE TABLE IF NOT EXISTS bre_backup_sys_dict_data " f"AS SELECT * FROM sys_dict_data WHERE 1=0" ) ) conn.execute( text( f"INSERT INTO bre_backup_sys_dict_type " f"SELECT * FROM sys_dict_type WHERE dict_type IN ({placeholders}) " f"AND dict_type NOT IN (SELECT dict_type FROM bre_backup_sys_dict_type)" ), params, ) conn.execute( text( f"INSERT INTO bre_backup_sys_dict_data " f"SELECT * FROM sys_dict_data WHERE dict_type IN ({placeholders}) " f"AND uuid NOT IN (SELECT uuid FROM bre_backup_sys_dict_data)" ), params, ) def main() -> None: parser = argparse.ArgumentParser(description="补齐 P1 字典类型（7 个），幂等") parser.add_argument("--check", action="store_true", help="仅预览将插入的行，不落库") args = parser.parse_args() engine = create_engine(settings.DB_URI) print(f"# 连接: {settings.DB_URI} 模式: {'预览' if args.check else '执行'}") types = list(DICT_TYPES.keys()) with engine.connect() as conn: if not args.check: backup_dict_rows(conn, types) existing_types = { row[0] for row in conn.execute( text("SELECT dict_type FROM sys_dict_type WHERE dict_type IN (" + ",".join([f":t{i}" for i in range(len(types))]) + ")"), {f"t{i}": t for i, t in enumerate(types)}, ) } for dict_type, cfg in DICT_TYPES.items(): if dict_type in existing_types: type_row = conn.execute( text("SELECT id FROM sys_dict_type WHERE dict_type = :dt"), {"dt": dict_type}, ).first() dict_type_id = type_row[0] print(f"# [{dict_type}] 类型已存在 (id={dict_type_id})，仅补数据") else: dict_type_id = None print(f"# [{dict_type}] 类型缺失，将新建") existing_values = { row[0] for row in conn.execute( text("SELECT dict_value FROM sys_dict_data WHERE dict_type = :dt"), {"dt": dict_type}, ) } to_insert = [(s, l, v) for s, l, v in cfg["rows"] if v not in existing_values] if args.check: print(f" 待插入: {to_insert}") continue if dict_type_id is None: type_uuid = str(uuid.uuid4()) result = conn.execute( text( "INSERT INTO sys_dict_type " "(dict_name, dict_type, status, description, uuid, is_deleted, created_time, updated_time) " "VALUES (:name, :dt, 0, :desc, :uuid, false, now(), now()) " "RETURNING id" ), { "name": cfg["dict_name"], "dt": dict_type, "desc": cfg["description"], "uuid": type_uuid, }, ) dict_type_id = result.scalar() print(f" 已建类型: {dict_type} (id={dict_type_id})") if not to_insert: print(" 无新增数据，跳过") continue for sort, label, value in to_insert: conn.execute( text( "INSERT INTO sys_dict_data " "(status, description, dict_sort, dict_label, dict_value, css_class, list_class, " " is_default, dict_type, dict_type_id, uuid, is_deleted, created_time, updated_time) " "VALUES (0, '', :sort, :label, :value, '', NULL, false, :dt, :dtid, :uuid, false, now(), now())" ), { "sort": sort, "label": label, "value": value, "dt": dict_type, "dtid": dict_type_id, "uuid": str(uuid.uuid4()), }, ) print(f" 已插入 {len(to_insert)} 行") if not args.check: conn.commit() print("# 已提交 ✅") if __name__ == "__main__": main()