ajy/openspec/aijiuyiBackend/scripts/generate_seed_sql.py

# -*- coding: utf-8 -*-
"""
从《智能穴位识别数据模型》xlsx 生成种子 SQL，并写回 V1__init.sql 中
「-- >>> GENERATED_SEED_START」与「-- <<< GENERATED_SEED_END」之间的内容。

读取工作表：
  - 第 1 张：穴位数据模型（筛选「艾灸椅产品穴位」为「是」，且身体部位∈{背部,颈部}，
    部位分区∈{背部,骶区,颈后部,腰部}；列名支持「身体部位」或「身体补位」）
  - 第 2 张：症状模型 -> symptom
  - 第 3 张：部位定位模型 -> body_region

并追加：admin_role、admin_user（默认 admin/admin 的 BCrypt）、admin_user_role、device_model。
需在环境中安装：pip install openpyxl
"""
import glob
import os

try:
    import openpyxl
except ImportError:
    raise SystemExit("pip install openpyxl")

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
os.chdir(BASE)

V1_SQL = os.path.join(BASE, "src", "main", "resources", "db", "migration", "V1__init.sql")
START_TAG = "-- >>> GENERATED_SEED_START"
END_TAG = "-- <<< GENERATED_SEED_END"


def esc(s):
    if s is None:
        return "NULL"
    s = str(s).replace("\\", "\\\\").replace("'", "''").replace("\r", " ").replace("\n", " ")
    return "'" + s + "'"


def main():
    paths = glob.glob(os.path.join(BASE, "*数据模型*.xlsx"))
    if not paths:
        paths = glob.glob(os.path.join(BASE, "*.xlsx"))
        paths = [p for p in paths if "数据模型" in os.path.basename(p) or "模型" in os.path.basename(p)]
    if not paths:
        raise SystemExit("No data model xlsx found")
    path = paths[0]
    wb = openpyxl.load_workbook(path, read_only=True, data_only=True)

    lines = []
    lines.append("SET FOREIGN_KEY_CHECKS = 0;")

    # Symptoms sheet 2
    ws_sym = wb[wb.sheetnames[1]]
    rows_sym = list(ws_sym.iter_rows(values_only=True))
    lines.append("")
    lines.append("-- 症状字典（Excel「症状模型」）")
    lines.append("INSERT IGNORE INTO symptom (code, name) VALUES ")
    vals = []
    for row in rows_sym[1:]:
        if not row or row[0] is None:
            continue
        code = esc(str(row[0]).strip())
        name = esc(str(row[1]).strip() if len(row) > 1 and row[1] is not None else "")
        vals.append("({0}, {1})".format(code, name))
    lines.append(",\n".join(vals) + ";")

    # Body regions sheet 3
    ws_br = wb[wb.sheetnames[2]]
    rows_br = list(ws_br.iter_rows(values_only=True))
    lines.append("")
    lines.append("-- 身体部位与分区（Excel「部位定位模型」）")
    lines.append("INSERT IGNORE INTO body_region (body_part, region_partition) VALUES ")
    vals = []
    for row in rows_br[1:]:
        if row is None:
            continue
        bp = row[0]
        rp = row[1]
        if bp is None and rp is None:
            continue
        vals.append("({0}, {1})".format(esc(bp), esc(rp)))
    lines.append(",\n".join(vals) + ";")

    # Acupoints sheet 1
    ws = wb[wb.sheetnames[0]]
    header = next(ws.iter_rows(min_row=1, max_row=1, values_only=True))
    h = list(header)

    def col(*names):
        for n in names:
            if n in h:
                return h.index(n)
        raise SystemExit("Excel 表头缺少列: " + " / ".join(names))

    idx_chair = col("艾灸椅产品穴位")
    idx_body = col("身体部位", "身体补位")
    cols = {
        "serial_no": h.index("序号"),
        "code": h.index("穴位编号"),
        "name": h.index("穴位名称"),
        "side_type": h.index("单/双穴"),
        "meridian": h.index("所属经络"),
        "body_region": idx_body,
        "region_partition": h.index("部位分区"),
        "location_text": h.index("穴位定位"),
        "relative_position_text": h.index("相对位置"),
        "indications": h.index("主管疾病"),
        "symptom_coverage": h.index("产品症状覆盖"),
        "acupoint_method": h.index("取穴"),
        "remark": len(h) - 1,
    }

    allowed_body = {"背部", "颈部"}
    allowed_partition = {"背部", "骶区", "颈后部", "腰部"}

    lines.append("")
    lines.append(
        "-- 标准穴位：艾灸椅=是 ∧ 身体部位∈(背部,颈部) ∧ 部位分区∈(背部,骶区,颈后部,腰部)"
    )
    lines.append(
        "INSERT IGNORE INTO acupoint (serial_no, code, name, side_type, meridian, body_region, region_partition, "
        "location_text, relative_position_text, indications, symptom_coverage, acupoint_method, chair_product, remark) VALUES "
    )
    vals = []
    for row in ws.iter_rows(min_row=2, values_only=True):
        if not row or row[idx_chair] != "是":
            continue
        r = list(row) + [None] * 20

        def g(key):
            i = cols[key]
            v = r[i] if i < len(r) else None
            return v

        br = g("body_region")
        rp = g("region_partition")
        if br is None or rp is None:
            continue
        brs = str(br).strip()
        rps = str(rp).strip()
        if brs not in allowed_body or rps not in allowed_partition:
            continue

        remark = r[cols["remark"]] if cols["remark"] < len(r) else None
        vals.append(
            "({0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}, {11}, 1, {12})".format(
                esc(g("serial_no")),
                esc(g("code")),
                esc(g("name")),
                esc(g("side_type")),
                esc(g("meridian")),
                esc(g("body_region")),
                esc(g("region_partition")),
                esc(g("location_text")),
                esc(g("relative_position_text")),
                esc(g("indications")),
                esc(g("symptom_coverage")),
                esc(g("acupoint_method")),
                esc(remark),
            )
        )
    lines.append(",\n".join(vals) + ";")

    bcrypt_password = "$2b$10$yW9CwM2u/vqz4CGZsQXPyeiKYtsu4PbAj.vKstKoY4hbFT6I0LiNm"
    lines.append("")
    lines.append("-- 后台角色、默认管理员、默认设备（联调）")
    lines.append(
        "INSERT IGNORE INTO admin_role (code, name) VALUES "
        "('SUPER_ADMIN', '超级管理员'), ('ADMIN', '管理员'), ('OPERATOR', '运营');"
    )
    lines.append(
        "INSERT IGNORE INTO admin_user (username, password_hash, display_name, status) VALUES "
        "('admin', '{0}', '系统管理员', 'ACTIVE');".format(bcrypt_password)
    )
    lines.append("INSERT IGNORE INTO admin_user_role (admin_user_id, admin_role_id) VALUES (1, 1);")
    lines.append(
        "INSERT IGNORE INTO device_model (code, name, stroke_range_x, stroke_range_y, stroke_range_z, "
        "origin_offset_x, origin_offset_y, origin_offset_z, safety_distance) VALUES "
        "('CHAIR-DEFAULT', '默认艾灸椅', 800, 600, 400, 0, 0, 0, 10);"
    )
    lines.append("-- INSERT IGNORE 不更新已存在的 admin 行：将管理员 admin 密码统一为明文 admin（BCrypt）")
    lines.append(
        "UPDATE admin_user SET password_hash = '{0}' WHERE username = 'admin' AND deleted = 0;".format(
            bcrypt_password
        )
    )
    lines.append("")
    lines.append("SET FOREIGN_KEY_CHECKS = 1;")

    with open(V1_SQL, encoding="utf-8") as f:
        content = f.read()
    clines = content.splitlines()
    si = ei = None
    for i, L in enumerate(clines):
        if L.strip() == START_TAG:
            si = i
        elif si is not None and L.strip() == END_TAG:
            ei = i
            break
    if si is None or ei is None or ei <= si:
        raise SystemExit("{0} 中缺少 {1} / {2} 标记".format(V1_SQL, START_TAG, END_TAG))

    new_clines = clines[: si + 1] + lines + clines[ei:]
    with open(V1_SQL, "w", encoding="utf-8", newline="\n") as f:
        f.write("\n".join(new_clines) + "\n")

    print("Updated seed in", V1_SQL, "acupoints:", len(vals))
    wb.close()


if __name__ == "__main__":
    main()