model-platform/docs/database/build_full_schema.py

"""Build the single-file model_operations schema from versioned modules."""

from __future__ import annotations

from pathlib import Path


BASE = Path(__file__).parent
SOURCE_DIR = BASE / "model_operations-V0.1"
OUTPUT = BASE / "model_operations-完整建表-V0.1.sql"
SOURCES = [
    "10_reference_and_models.sql",
    "20_monitoring_source.sql",
    "30_governance_and_reviews.sql",
    "40_reports_and_prompts.sql",
    "50_workflows_documents_usage.sql",
]


def statements(text: str) -> list[str]:
    """Split SQL safely enough for quotes/backticks and strip line comments."""
    output: list[str] = []
    current: list[str] = []
    quote: str | None = None
    line_comment = False
    index = 0
    while index < len(text):
        char = text[index]
        next_char = text[index + 1] if index + 1 < len(text) else ""
        if line_comment:
            if char == "\n":
                line_comment = False
                current.append(" ")
            index += 1
            continue
        if quote is None and char == "-" and next_char == "-":
            line_comment = True
            index += 2
            continue
        if quote:
            current.append(char)
            if char == quote:
                if quote == "'" and next_char == "'":
                    current.append(next_char)
                    index += 2
                    continue
                quote = None
            index += 1
            continue
        if char in ("'", "`"):
            quote = char
            current.append(char)
        elif char == ";":
            statement = " ".join("".join(current).split())
            if statement:
                output.append(statement + ";")
            current = []
        else:
            current.append(char)
        index += 1
    tail = " ".join("".join(current).split())
    if tail:
        output.append(tail + ";")
    return output


def format_create_table(statement: str) -> str:
    """Format one CREATE TABLE compactly while keeping DBeaver-friendly lines."""
    open_at = statement.find("(")
    if open_at < 0:
        return statement
    quote: str | None = None
    depth = 0
    close_at = -1
    for index in range(open_at, len(statement)):
        char = statement[index]
        next_char = statement[index + 1] if index + 1 < len(statement) else ""
        if quote:
            if char == quote:
                if quote == "'" and next_char == "'":
                    continue
                quote = None
            continue
        if char in ("'", "`"):
            quote = char
        elif char == "(":
            depth += 1
        elif char == ")":
            depth -= 1
            if depth == 0:
                close_at = index
                break
    if close_at < 0:
        return statement
    body = statement[open_at + 1 : close_at]
    parts: list[str] = []
    current: list[str] = []
    quote = None
    depth = 0
    for index, char in enumerate(body):
        next_char = body[index + 1] if index + 1 < len(body) else ""
        if quote:
            current.append(char)
            if char == quote:
                if quote == "'" and next_char == "'":
                    continue
                quote = None
            continue
        if char in ("'", "`"):
            quote = char
            current.append(char)
        elif char == "(":
            depth += 1
            current.append(char)
        elif char == ")":
            depth -= 1
            current.append(char)
        elif char == "," and depth == 0:
            parts.append(" ".join("".join(current).split()))
            current = []
        else:
            current.append(char)
    tail_part = " ".join("".join(current).split())
    if tail_part:
        parts.append(tail_part)
    lines = [statement[:open_at].rstrip() + " ("]
    for index in range(0, len(parts), 2):
        chunk = ", ".join(parts[index : index + 2])
        if index + 2 < len(parts):
            chunk += ","
        lines.append("  " + chunk)
    lines.append(") " + statement[close_at + 1 :].strip())
    return "\n".join(lines)


def main() -> None:
    bundled: list[str] = [
        "-- A 卡模型智能运维平台 · model_operations 完整建表 V0.1",
        "-- 内容：仅包含新库的 24 张表，不含建库、视图、种子数据或 model_platform 现有表。",
        "-- 执行前：请在 DBeaver 中选中 model_operations 作为当前活动数据库。",
        "-- DBeaver：请使用“执行 SQL 脚本”（Alt/Option + X），不要选中全文后按 Ctrl/Cmd + Enter。",
        "-- 如使用“执行 SQL 语句”，每次只能选中一条 CREATE TABLE 单独执行。",
        "",
    ]
    for source_name in SOURCES:
        source_statements = statements((SOURCE_DIR / source_name).read_text(encoding="utf-8"))
        bundled.append(f"-- ===== {source_name} =====")
        for statement in source_statements:
            if not statement.lower().startswith("create table "):
                continue
            bundled.append(format_create_table(statement))
        bundled.append("")
    OUTPUT.write_text("\n".join(bundled).rstrip() + "\n", encoding="utf-8")
    print(f"built {OUTPUT.name}: 24 tables")


if __name__ == "__main__":
    main()