from __future__ import annotations import re from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SOURCE = ROOT / "docs" / "00-新数据库基线设计.md" SPECIAL_SOURCE = ROOT / "docs" / "02-数据库建表设计.md" OUTPUT = ROOT / "alembic" / "baseline_generated.sql" # 文档以两种写法声明唯一约束： # 1) 显式联合键：| product_id | ... | 唯一键 `(product_id, trade_date)` | # 2) 联合标记： | customer_id | ... | 联合唯一键、索引 |（同表内多字段共同构成一个键） # 单列唯一键只在规则列出现"唯一"且未参与上述两种写法时生成。 COMBINED_KEY_RE = re.compile(r"唯一键[^()]*\(([^)]+)\)") COMBINED_FLAG = "联合唯一键" MAX_INDEX_NAME = 64 TYPE_MAP = {"BIGINT UNSIGNED": "BIGINT UNSIGNED", "INT": "INT", "FLOAT": "FLOAT", "TINYINT(1)": "TINYINT(1)", "JSON": "JSON", "DATE": "DATE", "DATETIME": "DATETIME", "TEXT": "TEXT", "MEDIUMTEXT": "MEDIUMTEXT"} def infer_type(raw: str) -> str: raw = raw.strip().replace("`", "") if raw in TYPE_MAP: return TYPE_MAP[raw] return raw if re.match(r"^(?:VAR)?CHAR\(|DECIMAL\(|CHAR\(", raw) else "TEXT" def constraints(raw: str, field: str) -> list[str]: result = ["NOT NULL"] if "可空" not in raw and field not in {"id", "episode_uuid", "profile_uuid"} else [] if "默认" in raw: match = re.search(r"默认\s*[`'“]?([^`'”]+)", raw) if match: value = match.group(1).strip().split()[0] if "CURRENT_TIMESTAMP" in value: default = "CURRENT_TIMESTAMP" elif re.match(r"^-?\d+(?:\.\d+)?$", value): default = value else: default = "'" + value.strip("'") + "'" result.append("DEFAULT " + default) if "主键" in raw: result.append("PRIMARY KEY") return result def parse_detailed_tables(text: str) -> list[str]: tables: list[str] = [] matches = list(re.finditer(r"(?m)^#### `([^`]+)`[^\n]*\n", text)) for index, match in enumerate(matches): name = match.group(1) heading = re.search(r"(?m)^#{3,5} ", text[match.end():]) end = match.end() + heading.start() if heading else len(text) body = text[match.end():end] rows = re.findall(r"^\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|", body, re.M) fields = [] for field, typ, rule in rows: if field.strip() in {"字段", "表名"} or set(field.strip()) == {"-"}: continue field = field.strip().strip("`") if field.startswith("sys_") and ("、" in typ or "字段" not in typ): continue if "," in field or field.startswith("id、"): continue fields.append((field, infer_type(typ), rule)) if not fields: continue columns = [] combined_keys: list[tuple[str, ...]] = [] combined_flags: list[str] = [] single_uniques: list[str] = [] indexes = [] for field, typ, rule in fields: col = f"`{field}` {typ}" cs = constraints(rule, field) if "CURRENT_TIMESTAMP" in rule and "DATETIME" in typ: cs.append("DEFAULT CURRENT_TIMESTAMP") columns.append(col + (" " + " ".join(cs) if cs else " NULL")) matched_key = False for raw in COMBINED_KEY_RE.findall(rule): key_columns = tuple( dict.fromkeys(c.strip().strip("`") for c in raw.split(",") if c.strip()) ) if len(key_columns) > 1: if key_columns not in combined_keys: combined_keys.append(key_columns) elif key_columns: single_uniques.append(key_columns[0]) matched_key = True if COMBINED_FLAG in rule: combined_flags.append(field) elif "唯一" in rule and not matched_key: single_uniques.append(field) if "索引" in rule: indexes.append(field) if not any("PRIMARY KEY" in c for c in columns): for n, c in enumerate(columns): if c.startswith("`id`") or c.startswith("`customer_id`") and name == "fin_customer_profile": columns[n] += " PRIMARY KEY" break if combined_flags: flag_columns = tuple(dict.fromkeys(combined_flags)) if len(flag_columns) > 1 and flag_columns not in combined_keys: combined_keys.append(flag_columns) for key_columns in combined_keys: key_name = f"uk_{name}_" + "_".join(key_columns) if len(key_name) > MAX_INDEX_NAME: key_name = key_name[:MAX_INDEX_NAME] rendered = ", ".join(f"`{column}`" for column in key_columns) columns.append(f"UNIQUE KEY `{key_name}` ({rendered})") for field in dict.fromkeys(single_uniques): if not any(f"`{field}`" in c and "PRIMARY KEY" in c for c in columns): columns.append(f"UNIQUE KEY `uk_{name}_{field}` (`{field}`)") for field in indexes: columns.append(f"KEY `idx_{name}_{field}` (`{field}`)") tables.append("CREATE TABLE `{}` (\n {}\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;".format(name, ",\n ".join(columns))) return tables def main() -> None: text = SOURCE.read_text(encoding="utf-8") manual = [ "CREATE TABLE `sys_role` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `role_code` VARCHAR(32) NOT NULL UNIQUE, `role_name` VARCHAR(64) NOT NULL, `status` VARCHAR(16) NOT NULL, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `sys_permission` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `permission_code` VARCHAR(64) NOT NULL UNIQUE, `resource` VARCHAR(64) NOT NULL, `action` VARCHAR(32) NOT NULL, `data_scope` VARCHAR(32) NOT NULL, `field_policy` JSON NULL, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, KEY `idx_sys_permission_resource_action` (`resource`,`action`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `sys_user_role` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `user_id` BIGINT UNSIGNED NOT NULL, `role_id` BIGINT UNSIGNED NOT NULL, `assigned_at` DATETIME NOT NULL, `expires_at` DATETIME NULL, UNIQUE KEY `uk_sys_user_role` (`user_id`,`role_id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `sys_role_permission` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `role_id` BIGINT UNSIGNED NOT NULL, `permission_id` BIGINT UNSIGNED NOT NULL, `created_at` DATETIME NOT NULL, UNIQUE KEY `uk_sys_role_permission` (`role_id`,`permission_id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `episodes` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `episode_uuid` CHAR(36) NULL UNIQUE, `customer_id` BIGINT UNSIGNED NOT NULL, `session_id` VARCHAR(64) NOT NULL, `start_message_no` INT NULL, `end_message_no` INT NULL, `portals_involved` JSON NOT NULL, `summary` TEXT NOT NULL, `extraction_status` VARCHAR(16) NOT NULL DEFAULT '待提取', `content_hash` CHAR(64) NULL UNIQUE, `retry_count` INT NOT NULL DEFAULT 0, `started_at` DATETIME NULL, `ended_at` DATETIME NULL, `handoff_to_employee_id` BIGINT UNSIGNED NULL, `start_at` DATETIME NULL, `end_at` DATETIME NULL, `promoted_to_ltm` TINYINT(1) NOT NULL DEFAULT 0, `created_at` DATETIME NOT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `memory_unit` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `memory_uuid` CHAR(36) NOT NULL UNIQUE, `customer_id` BIGINT UNSIGNED NOT NULL, `memory_key` VARCHAR(128) NOT NULL, `content` TEXT NOT NULL, `structured_value` JSON NULL, `memory_type` VARCHAR(24) NOT NULL, `source_type` VARCHAR(24) NOT NULL, `source_confidence` DECIMAL(5,4) NOT NULL, `confidence` DECIMAL(5,4) NOT NULL, `evidence_count` INT NOT NULL DEFAULT 0, `conflict_count` INT NOT NULL DEFAULT 0, `recall_count` INT NOT NULL DEFAULT 0, `status` VARCHAR(16) NOT NULL, `valid_from` DATETIME NOT NULL, `valid_until` DATETIME NULL, `last_evidenced_at` DATETIME NULL, `last_recall_at` DATETIME NULL, `promoted_at` DATETIME NULL, `version` BIGINT UNSIGNED NOT NULL DEFAULT 1, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, KEY `idx_memory_unit_customer` (`customer_id`,`status`,`confidence`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `memory_evidence` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `memory_id` BIGINT UNSIGNED NOT NULL, `evidence_type` VARCHAR(24) NOT NULL, `source_table` VARCHAR(64) NULL, `source_record_id` VARCHAR(64) NULL, `source_episode_id` BIGINT UNSIGNED NULL, `evidence_excerpt` TEXT NULL, `evidence_snapshot` JSON NULL, `weight` DECIMAL(5,4) NOT NULL, `idempotency_key` VARCHAR(128) NOT NULL UNIQUE, `occurred_at` DATETIME NOT NULL, `created_at` DATETIME NOT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `memory_conflict` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `left_memory_id` BIGINT UNSIGNED NOT NULL, `right_memory_id` BIGINT UNSIGNED NOT NULL, `conflict_type` VARCHAR(24) NOT NULL, `severity` VARCHAR(8) NOT NULL, `status` VARCHAR(16) NOT NULL, `resolution` TEXT NULL, `winner_memory_id` BIGINT UNSIGNED NULL, `resolved_by` BIGINT UNSIGNED NULL, `resolved_at` DATETIME NULL, `created_at` DATETIME NOT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `profile_snapshots` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `profile_uuid` CHAR(36) NULL UNIQUE, `customer_id` BIGINT UNSIGNED NOT NULL, `version` BIGINT UNSIGNED NOT NULL, `snapshot` JSON NOT NULL, `generation_basis` JSON NULL, `snapshot_hash` CHAR(64) NULL, `is_current` TINYINT(1) NOT NULL DEFAULT 0, `current_customer_id` BIGINT UNSIGNED NULL, `generated_at` DATETIME NULL, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, UNIQUE KEY `uk_profile_snapshot_version` (`customer_id`,`version`), UNIQUE KEY `uk_profile_snapshot_current` (`current_customer_id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `memory_sync_outbox` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `event_uuid` CHAR(36) NOT NULL, `aggregate_type` VARCHAR(24) NOT NULL, `aggregate_uuid` CHAR(36) NOT NULL, `aggregate_version` BIGINT UNSIGNED NOT NULL, `target_store` VARCHAR(16) NOT NULL, `operation` VARCHAR(16) NOT NULL, `payload` JSON NOT NULL, `status` VARCHAR(16) NOT NULL, `retry_count` INT NOT NULL DEFAULT 0, `next_retry_at` DATETIME NULL, `last_error` TEXT NULL, `created_at` DATETIME NOT NULL, `processed_at` DATETIME NULL, UNIQUE KEY `uk_memory_sync_event` (`event_uuid`,`target_store`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `interaction_audit` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `actor_type` VARCHAR(16) NOT NULL, `actor_id` BIGINT UNSIGNED NULL, `target_customer_id` BIGINT UNSIGNED NULL, `session_id` VARCHAR(64) NULL, `portal` VARCHAR(32) NULL, `action_type` VARCHAR(64) NOT NULL, `detail` JSON NOT NULL, `created_at` DATETIME NOT NULL, KEY `idx_audit_actor` (`actor_type`,`actor_id`,`created_at`), KEY `idx_audit_customer` (`target_customer_id`,`created_at`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `client_facing_content` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `customer_id` BIGINT UNSIGNED NOT NULL, `content_type` VARCHAR(32) NOT NULL, `draft_content` JSON NOT NULL, `generated_by_portal` VARCHAR(32) NOT NULL, `review_status` VARCHAR(16) NOT NULL, `reviewer_user_id` BIGINT UNSIGNED NULL, `reviewed_at` DATETIME NULL, `published_at` DATETIME NULL, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, KEY `idx_content_customer_status` (`customer_id`,`review_status`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `conversation_message` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `session_id` VARCHAR(64) NOT NULL, `message_no` INT NULL, `customer_id` BIGINT UNSIGNED NULL, `portal` VARCHAR(32) NOT NULL, `role` VARCHAR(16) NOT NULL, `content` MEDIUMTEXT NOT NULL, `tool_calls` JSON NULL, `trace_id` VARCHAR(64) NULL, `intent` VARCHAR(32) NULL, `confidence` DECIMAL(5,4) NULL, `source_references` JSON NULL, `created_at` DATETIME NOT NULL, UNIQUE KEY `uk_message_session_no` (`session_id`,`message_no`), KEY `idx_message_customer_created` (`customer_id`,`created_at`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", "CREATE TABLE `fin_knowledge_meta` (`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, `knowledge_type` VARCHAR(32) NOT NULL, `title` VARCHAR(256) NOT NULL, `source_file` VARCHAR(256) NULL, `minio_path` VARCHAR(512) NULL, `milvus_collection` VARCHAR(64) NOT NULL, `version` VARCHAR(16) NULL, `effective_date` DATE NULL, `expire_date` DATE NULL, `content_text` MEDIUMTEXT NOT NULL, `tags` JSON NULL, `reviewer_id` BIGINT UNSIGNED NULL, `review_status` VARCHAR(16) NOT NULL DEFAULT 'pending', `status` VARCHAR(16) NOT NULL, `created_at` DATETIME NOT NULL, `updated_at` DATETIME NOT NULL, KEY `idx_knowledge_status` (`status`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;", ] tables = manual + parse_detailed_tables(text) special = SPECIAL_SOURCE.read_text(encoding="utf-8") special_sql = re.findall(r"(CREATE TABLE (?:IF NOT EXISTS )?`?([a-zA-Z0-9_]+)`?[\s\S]*?;)", special) for statement, name in special_sql[:6]: tables.append(statement) unique: dict[str, str] = {} for statement in tables: match = re.search(r"CREATE TABLE\s+`?([A-Za-z0-9_]+)`?", statement) if match: unique.setdefault(match.group(1), statement) OUTPUT.write_text("SET FOREIGN_KEY_CHECKS=0;\n" + "\n\n".join(unique.values()) + "\nSET FOREIGN_KEY_CHECKS=1;\n", encoding="utf-8") print(f"generated={len(unique)} output={OUTPUT}") if __name__ == "__main__": main()