#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 解析医疗-consumer / 医疗-delivery 的 Schema、RLS、RPC，生成数据库对接文档 """ import re import os from collections import defaultdict OUTPUT_PATH = r"D:\骅锋\医疗-consumer\docs\数据库对接文档.md" SCHEMA_PATH = "all_schema_merged.sql" RLS_PATH = "all_rls_merged.sql" RPC_PATH = "all_rpc_merged.sql" # 定义各端涉及的表（用于标记所属端） CONSUMER_TABLES = { "ak_users", "ml_user_profiles", "ml_user_addresses", "ml_shopping_cart", "ml_products", "ml_product_skus", "ml_product_specs", "ml_product_reviews", "ml_product_categories", "medical_mall_categories", "ml_product_labels", "ml_product_member_prices", "ml_product_protections", "ml_product_templates", "ak_shipping_templates", "ml_products_ext", "ml_orders", "ml_order_items", "ml_delivery_tasks", "ml_delivery_staff", "ml_delivery_stations", "ml_system_configs", "ml_cms_content", "ml_cms_categories", "ml_cms_banners", "ml_user_bill", "ml_user_recharge", "ml_extract", "ml_invoices", "ml_user_coupons", "ml_coupon_templates", "ml_seckill", "ml_group_buy", "ml_bargain", "ml_live", "ml_lottery", "ml_signin_logs", "ml_checkin_configs", "ml_member_cards", "ml_recharge_rules", "ak_user_groups", "ak_user_labels", "ak_user_levels", "ak_distribution_agents", "ak_distribution_divisions", "ak_distribution_config", "ak_distribution_level", "ak_promoter_relations", "ak_commission_logs", "ak_distribution_agent_applications", "ak_distribution_division_applications", "ak_diy_pages", "ml_kefu_accounts", "ml_kefu_sessions", "ml_kefu_messages", "homecare_patients", "homecare_nurses", "homecare_services", "homecare_orders", "homecare_schedules", "homecare_evaluations", } DELIVERY_TABLES = { "ak_users", "ml_delivery_tasks", "ml_delivery_staff", "ml_delivery_stations", "ml_orders", "ml_order_items", "homecare_patients", "homecare_nurses", "homecare_services", "homecare_orders", "homecare_schedules", "homecare_evaluations", "ml_system_configs", } def parse_schema(path): """提取 CREATE TABLE 语句""" with open(path, "r", encoding="utf-8") as f: content = f.read() tables = [] # 匹配 CREATE TABLE [IF NOT EXISTS] name (...) 或 CREATE TABLE public.name (...) pattern = re.compile( r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(?:public\.)?(\w+)\s*\((.*?)\);", re.DOTALL | re.IGNORECASE, ) for match in pattern.finditer(content): table_name = match.group(1) body = match.group(2) fields = [] constraints = [] # 简单按行拆分（注意括号内可能有嵌套，但这里足够用） lines = [line.strip() for line in body.split("\n") if line.strip()] for line in lines: line = line.rstrip(",") # 跳过注释行 if line.startswith("--") or line.startswith("/*"): continue # 约束行（PRIMARY KEY, FOREIGN KEY, UNIQUE, CHECK, CONSTRAINT） if re.match(r"^(CONSTRAINT|PRIMARY\s+KEY|FOREIGN\s+KEY|UNIQUE|CHECK)", line, re.I): constraints.append(line) continue # 字段定义行：name type [constraints] m = re.match(r"^(\w+)\s+([\w\[\]]+(?:\s*\([^)]+\))?)\s*(.*)$", line) if m: col_name = m.group(1) col_type = m.group(2).strip() col_rest = m.group(3).strip() fields.append({ "name": col_name, "type": col_type, "rest": col_rest, }) tables.append({ "name": table_name, "fields": fields, "constraints": constraints, "raw": match.group(0), }) return tables def parse_rls(path): """提取每个表的 RLS 策略和启用状态""" with open(path, "r", encoding="utf-8") as f: content = f.read() table_rls = defaultdict(lambda: {"enabled": False, "policies": []}) # ALTER TABLE ... ENABLE ROW LEVEL SECURITY for m in re.finditer( r"ALTER\s+TABLE\s+(?:public\.)?(\w+)\s+ENABLE\s+ROW\s+LEVEL\s+SECURITY", content, re.I, ): table_rls[m.group(1)]["enabled"] = True # CREATE POLICY ... ON table_name for m in re.finditer( r"CREATE\s+POLICY\s+(\w+)\s+ON\s+(?:public\.)?(\w+)\s+.*?((?:FOR\s+\w+\s+)?(?:TO\s+[\w,\s]+\s+)?(?:USING\s*\([^)]*\)\s*)?(?:WITH\s+CHECK\s*\([^)]*\)\s*)?);", content, re.DOTALL | re.IGNORECASE, ): policy_name = m.group(1) table_name = m.group(2) policy_sql = m.group(0) table_rls[table_name]["policies"].append({ "name": policy_name, "sql": policy_sql, }) return table_rls def parse_rpc(path): """提取 RPC 函数名、参数、返回类型""" with open(path, "r", encoding="utf-8") as f: content = f.read() funcs = [] pattern = re.compile( r"CREATE\s+OR\s+REPLACE\s+FUNCTION\s+(?:public\.)?(\w+)\s*\((.*?)\)\s*RETURNS\s+(.*?)(?:LANGUAGE\s+\w+|AS\s*\$\$|SECURITY\s+DEFINER)", re.DOTALL | re.IGNORECASE, ) for m in pattern.finditer(content): funcs.append({ "name": m.group(1), "args": m.group(2).strip(), "returns": m.group(3).strip(), }) return funcs def get_table_belong(table_name): belongs = [] if table_name in CONSUMER_TABLES: belongs.append("consumer") if table_name in DELIVERY_TABLES: belongs.append("delivery") if not belongs: return "admin / 系统" return " / ".join(belongs) def main(): os.chdir(os.path.dirname(os.path.abspath(__file__))) tables = parse_schema(SCHEMA_PATH) rls_data = parse_rls(RLS_PATH) rpc_funcs = parse_rpc(RPC_PATH) # 按模块分组（简单按表名前缀） module_map = defaultdict(list) for t in tables: name = t["name"] if name.startswith("ak_users") or name.startswith("ml_user") or name.startswith("ak_user") or name in {"ak_roles", "ak_permissions", "ak_admin_roles", "ak_role_permissions"}: module_map["用户与权限"].append(t) elif name.startswith("ml_product") or name.startswith("medical_mall") or name.startswith("ak_shipping") or name.startswith("ml_products"): module_map["商品与类目"].append(t) elif name.startswith("ml_order"): module_map["订单"].append(t) elif name.startswith("ml_delivery"): module_map["配送与物流"].append(t) elif name.startswith("homecare"): module_map["居家护理"].append(t) elif name.startswith("ml_system"): module_map["系统配置"].append(t) elif name.startswith("ml_cms") or name.startswith("ak_diy"): module_map["内容与装修"].append(t) elif name.startswith("ml_kefu"): module_map["客服"].append(t) elif "finance" in name or name in {"ml_user_bill", "ml_user_recharge", "ml_extract", "ml_invoices"}: module_map["财务"].append(t) elif name.startswith("ml_") or name.startswith("ak_"): module_map["营销与分销"].append(t) else: module_map["其他"].append(t) lines = [] lines.append("# 医疗-consumer & 医疗-delivery 数据库对接文档") lines.append("") lines.append("> **生成日期**：2026-06-01") lines.append("> **适用范围**：医疗-consumer（消费者端）+ 医疗-delivery（配送端）") lines.append("> **数据库**：PostgreSQL (Supabase)") lines.append("> **规范**：所有用户侧表已启用 RLS（行级安全），全局/管理后台查询请走 RPC（SECURITY DEFINER）") lines.append("") lines.append("## 目录") lines.append("") lines.append("- [通用规范说明](#通用规范说明)") for mod in module_map: anchor = mod.replace(" ", "-").replace("与", "") lines.append(f"- [{mod}](#{anchor})") lines.append("- [完整对接 SQL 汇总](#完整对接-sql-汇总)") lines.append("- [关键 RPC 清单](#关键-rpc-清单)") lines.append("") lines.append("## 通用规范说明") lines.append("") lines.append("### 软删除标准") lines.append("本项目所有业务表默认采用**软删除**，标准字段如下：") lines.append("- `deleted_at` (timestamptz)：删除时间，NULL 表示未删除") lines.append("- `deleted_by` (uuid)：删除操作人") lines.append("- `created_at` (timestamptz)：创建时间，默认 `now()`") lines.append("- `updated_at` (timestamptz)：更新时间，默认 `now()`") lines.append("") lines.append("> **注意**：RLS 策略默认过滤 `deleted_at IS NULL` 的行，查询时无需手动加条件，但管理后台统计需视情况处理。") lines.append("") lines.append("### 角色字段权威口径") lines.append("- 统一用户主表：`public.ak_users`") lines.append("- 角色唯一权威字段：`ak_users.role`，取值：`customer`（消费者）、`merchant`（商家）、`delivery`（配送员）、`admin`（管理员）、`analytics`（数据分析师）") lines.append("- 商城用户扩展档案：`public.ml_user_profiles`，与 `ak_users` 1:1 关系（`user_id` UNIQUE）") lines.append("") lines.append("### 状态机速查") lines.append("| 状态域 | 字段名 | 关键取值 | 说明 |") lines.append("|--------|--------|----------|------|") lines.append("| 订单流程 | `order_status` | 0=待付款, 1=待发货, 2=待收货, 3=待评价, 4=已完成, 5=已取消, 6=退款中, 7=已退款 | 主状态 |") lines.append("| 支付状态 | `payment_status` | 0=未支付, 1=部分支付, 2=已支付, 3=部分退款, 4=全额退款 | 支付线 |") lines.append("| 物流状态 | `shipping_status` | 0=未发货, 1=已发货, 2=运输中, 3=已签收, 4=已送达 | 物流线 |") lines.append("| 配送任务 | `status` | 0=待分配, 1=待接单, 2=已接单, 3=取货中, 4=配送中, 5=已送达, 6=配送失败 | 配送端 |") lines.append("") # 逐模块输出 for mod, tbls in module_map.items(): anchor = mod.replace(" ", "-").replace("与", "") lines.append(f"## {mod}") lines.append("") for t in tbls: tname = t["name"] belong = get_table_belong(tname) lines.append(f"### {tname}") lines.append(f"- **所属端**：{belong}") lines.append("") lines.append("| 字段名 | 数据类型 | 约束/默认值 | 说明 |") lines.append("|--------|----------|-------------|------|") for f in t["fields"]: rest = f["rest"] # 简化 rest，突出 DEFAULT 和 NOT NULL note = rest lines.append(f"| {f['name']} | {f['type']} | {note} | |") lines.append("") if t["constraints"]: lines.append("**约束**：") for c in t["constraints"]: lines.append(f"- `{c}`") lines.append("") rls = rls_data.get(tname) if rls and rls["policies"]: lines.append("**RLS 策略**：") for p in rls["policies"]: lines.append(f"- `{p['name']}`") lines.append("") elif rls and rls["enabled"] and not rls["policies"]: lines.append("**RLS**：已启用，暂无显式策略（默认拒绝直接访问）") lines.append("") # 查找关联的 RPC related_rpc = [fn for fn in rpc_funcs if tname.replace("ml_", "").replace("ak_", "").replace("_", "") in fn["name"].replace("_", "")] if related_rpc: lines.append("**相关 RPC**：") for fn in related_rpc[:5]: # 最多显示5个，避免太长 lines.append(f"- `{fn['name']}({fn['args']}) -> {fn['returns']}`") lines.append("") lines.append("---") lines.append("") # 完整对接 SQL 汇总 lines.append("## 完整对接 SQL 汇总") lines.append("") lines.append("> 以下 SQL 按执行顺序排列：Schema → RLS → RPC。请在新项目/新环境中按顺序执行。") lines.append("") lines.append("### 1) Schema（建表）") lines.append("```sql") with open(SCHEMA_PATH, "r", encoding="utf-8") as f: lines.append(f.read()) lines.append("```") lines.append("") lines.append("### 2) RLS（行级安全）") lines.append("```sql") with open(RLS_PATH, "r", encoding="utf-8") as f: lines.append(f.read()) lines.append("```") lines.append("") lines.append("### 3) RPC（关键函数，完整见 30_rpc 目录）") lines.append("```sql") with open(RPC_PATH, "r", encoding="utf-8") as f: lines.append(f.read()) lines.append("```") lines.append("") # RPC 清单 lines.append("## 关键 RPC 清单") lines.append("") lines.append("| 函数名 | 参数 | 返回 | 用途 |") lines.append("|--------|------|------|------|") for fn in rpc_funcs: args = fn["args"].replace("\n", " ")[:80] lines.append(f"| {fn['name']} | {args} | {fn['returns']} | |") lines.append("") with open(OUTPUT_PATH, "w", encoding="utf-8") as f: f.write("\n".join(lines)) print(f"文档已生成：{OUTPUT_PATH}") print(f"共汇总表：{len(tables)} 张，RPC 函数：{len(rpc_funcs)} 个") if __name__ == "__main__": main()