232 lines
8.1 KiB
Python
232 lines
8.1 KiB
Python
"""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()
|