# -*- 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()