query-payslip-default/scripts/db/display_metadata.py

"""匠厂数据管理展示元数据（_jiangchang_tables / _jiangchang_columns）。

SQLite 无原生表/字段 COMMENT；中文展示名称通过这两张元数据表提供给宿主「数据管理」页面。
字段在界面上的顺序由 CREATE TABLE / PRAGMA table_info(cid) 决定，**不**使用 display_order 排序。
"""

from __future__ import annotations

from typing import Iterable

# 宿主会隐藏这些表，不作为业务表展示（见 jiangchang electron/data-management/types.ts）
INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns"})
LEGACY_METADATA_TABLES = frozenset({"_schema_meta"})
METADATA_TABLES = INTERNAL_METADATA_TABLES | LEGACY_METADATA_TABLES

DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})

_JIANGCHANG_TABLES_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_tables (
    table_name TEXT PRIMARY KEY,
    display_name TEXT NOT NULL,
    description TEXT,
    grid_help TEXT,
    sort_order INTEGER,
    visible INTEGER NOT NULL DEFAULT 1,
    readonly INTEGER NOT NULL DEFAULT 0,
    default_sort_json TEXT,
    default_visible_columns_json TEXT
)
"""

_JIANGCHANG_COLUMNS_DDL = """
CREATE TABLE IF NOT EXISTS _jiangchang_columns (
    table_name TEXT NOT NULL,
    column_name TEXT NOT NULL,
    display_name TEXT NOT NULL,
    description TEXT,
    grid_help TEXT,
    display_order INTEGER,
    visible INTEGER NOT NULL DEFAULT 1,
    searchable INTEGER NOT NULL DEFAULT 1,
    editable INTEGER NOT NULL DEFAULT 1,
    display_type TEXT,
    options_json TEXT,
    PRIMARY KEY (table_name, column_name)
)
"""

def _pragma_column_names(cursor, table_name: str) -> set[str]:
    cursor.execute(f"PRAGMA table_info({table_name!r})")
    return {str(row[1]) for row in cursor.fetchall()}


def _ensure_jiangchang_tables_columns(cursor) -> None:
    """旧库补列：CREATE IF NOT EXISTS 不会给已有表加新字段。"""
    cols = _pragma_column_names(cursor, "_jiangchang_tables")
    if "default_sort_json" not in cols:
        cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN default_sort_json TEXT")
    if "default_visible_columns_json" not in cols:
        cursor.execute(
            "ALTER TABLE _jiangchang_tables ADD COLUMN default_visible_columns_json TEXT"
        )
    if "grid_help" not in cols:
        cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN grid_help TEXT")


def _ensure_jiangchang_columns_columns(cursor) -> None:
    cols = _pragma_column_names(cursor, "_jiangchang_columns")
    if "grid_help" not in cols:
        cursor.execute("ALTER TABLE _jiangchang_columns ADD COLUMN grid_help TEXT")


def init_display_metadata_tables(cursor) -> None:
    cursor.execute(_JIANGCHANG_TABLES_DDL)
    cursor.execute(_JIANGCHANG_COLUMNS_DDL)
    _ensure_jiangchang_tables_columns(cursor)
    _ensure_jiangchang_columns_columns(cursor)


def upsert_table_metadata(
    cursor,
    *,
    table_name: str,
    display_name: str,
    description: str | None = None,
    grid_help: str | None = None,
    visible: int = 1,
    readonly: int = 0,
    default_sort_json: str | None = None,
    default_visible_columns_json: str | None = None,
) -> None:
    """写入表级元数据。

    default_sort_json: 宿主数据管理默认排序，JSON 数组，例如
    '[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]'
    为空则宿主回退到 updated_at → created_at → 主键。

    default_visible_columns_json: 宿主数据管理默认展示列，JSON 字符串数组，例如
    '["stat_type","period","stage"]'
    为空则全显；用户改过「列」后以本机偏好为准（软默认，区别于 visible=0 硬隐藏）。

    grid_help: 数据管理表格上方说明；空则不出横幅。与表单用的 description 独立，禁止互拷。
    """
    cursor.execute(
        """
        INSERT INTO _jiangchang_tables (
            table_name, display_name, description, grid_help, visible, readonly,
            default_sort_json, default_visible_columns_json
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(table_name) DO UPDATE SET
            display_name = excluded.display_name,
            description = excluded.description,
            grid_help = excluded.grid_help,
            visible = excluded.visible,
            readonly = excluded.readonly,
            default_sort_json = excluded.default_sort_json,
            default_visible_columns_json = excluded.default_visible_columns_json
        """,
        (
            table_name,
            display_name,
            description,
            grid_help,
            visible,
            readonly,
            default_sort_json,
            default_visible_columns_json,
        ),
    )


def upsert_column_metadata(
    cursor,
    *,
    table_name: str,
    column_name: str,
    display_name: str,
    description: str | None = None,
    grid_help: str | None = None,
    visible: int = 1,
    searchable: int = 1,
    editable: int = 1,
    display_type: str | None = None,
    options_json: str | None = None,
) -> None:
    # display_order 保留为 NULL：宿主按 PRAGMA table_info cid 展示字段，不读 display_order。
    # grid_help: 表头「?」；空则不出。与表单用的 description 独立，禁止互拷。
    cursor.execute(
        """
        INSERT INTO _jiangchang_columns (
            table_name, column_name, display_name, description, grid_help,
            display_order, visible, searchable, editable, display_type, options_json
        )
        VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?)
        ON CONFLICT(table_name, column_name) DO UPDATE SET
            display_name = excluded.display_name,
            description = excluded.description,
            grid_help = excluded.grid_help,
            display_order = NULL,
            visible = excluded.visible,
            searchable = excluded.searchable,
            editable = excluded.editable,
            display_type = excluded.display_type,
            options_json = excluded.options_json
        """,
        (
            table_name,
            column_name,
            display_name,
            description,
            grid_help,
            visible,
            searchable,
            editable,
            display_type,
            options_json,
        ),
    )


def seed_column_metadata_batch(
    cursor,
    *,
    table_name: str,
    columns: Iterable[dict[str, object]],
    valid_column_names: Iterable[str] | None = None,
) -> None:
    """批量 upsert 字段元数据；传入 valid_column_names 时会在 upsert 后清理陈旧字段元数据。"""
    for column in columns:
        upsert_column_metadata(
            cursor,
            table_name=table_name,
            column_name=str(column["column_name"]),
            display_name=str(column["display_name"]),
            description=str(column["description"]) if column.get("description") else None,
            grid_help=str(column["grid_help"]) if column.get("grid_help") else None,
            visible=int(column.get("visible", 1)),
            searchable=int(column.get("searchable", 1)),
            editable=int(column.get("editable", 1)),
            display_type=str(column["display_type"]) if column.get("display_type") else None,
            options_json=str(column["options_json"]) if column.get("options_json") else None,
        )
    if valid_column_names is not None:
        prune_stale_column_metadata(cursor, table_name, valid_column_names)


def prune_stale_column_metadata(cursor, table_name: str, valid_column_names: Iterable[str]) -> None:
    """删除表中已不存在物理字段的元数据行（幂等初始化后不得残留旧字段）。"""
    valid = set(valid_column_names)
    cursor.execute(
        "SELECT column_name FROM _jiangchang_columns WHERE table_name = ?",
        (table_name,),
    )
    for (column_name,) in cursor.fetchall():
        if str(column_name) not in valid:
            cursor.execute(
                "DELETE FROM _jiangchang_columns WHERE table_name = ? AND column_name = ?",
                (table_name, column_name),
            )


def get_table_readonly(cursor, table_name: str) -> int | None:
    cursor.execute(
        "SELECT readonly FROM _jiangchang_tables WHERE table_name = ?",
        (table_name,),
    )
    row = cursor.fetchone()
    if not row:
        return None
    return int(row[0] or 0)


def get_column_editable(cursor, table_name: str, column_name: str) -> int | None:
    cursor.execute(
        """
        SELECT editable FROM _jiangchang_columns
        WHERE table_name = ? AND column_name = ?
        """,
        (table_name, column_name),
    )
    row = cursor.fetchone()
    if not row:
        return None
    return int(row[0] or 0)


def list_editable_business_columns(cursor, table_name: str) -> list[str]:
    """返回 editable=1 的业务字段名（不含宿主侧主键/generated 等二次过滤）。"""
    physical = list_physical_column_names(cursor, table_name)
    cursor.execute(
        """
        SELECT column_name, editable
        FROM _jiangchang_columns
        WHERE table_name = ?
        """,
        (table_name,),
    )
    editable_set = {
        str(column_name)
        for column_name, editable in cursor.fetchall()
        if editable is not None and int(editable) == 1
    }
    return [name for name in physical if name in editable_set]


def table_allows_manual_create(cursor, table_name: str) -> bool:
    """表级 readonly=0 且至少有一个 editable=1 字段时，宿主才应提供空白新增表单。"""
    readonly = get_table_readonly(cursor, table_name)
    if readonly is None or int(readonly) != 0:
        return False
    return bool(list_editable_business_columns(cursor, table_name))


def init_display_metadata(cursor) -> None:
    init_display_metadata_tables(cursor)


def drop_deprecated_user_table(
    cursor,
    table_name: str,
    *,
    triggers: Iterable[str] | None = None,
) -> None:
    """永久退役用户可见表：删 trigger、DROP 物理表、清理展示元数据（幂等）。

    用于废弃 task_logs 等旧表。禁止只删元数据而保留物理表。
    见 references/SCHEMA.md「表增删改与退役」。
    """
    for trigger_name in triggers or ():
        cursor.execute(f"DROP TRIGGER IF EXISTS {trigger_name}")
    default_updated_at = f"{table_name}_set_updated_at"
    cursor.execute(f"DROP TRIGGER IF EXISTS {default_updated_at}")
    cursor.execute(f"DROP TABLE IF EXISTS {_quote_identifier(table_name)}")
    cursor.execute(
        "DELETE FROM _jiangchang_columns WHERE table_name = ?",
        (table_name,),
    )
    cursor.execute(
        "DELETE FROM _jiangchang_tables WHERE table_name = ?",
        (table_name,),
    )


def iter_user_tables(cursor) -> Iterable[str]:
    cursor.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
    )
    for (name,) in cursor.fetchall():
        if name in METADATA_TABLES:
            continue
        yield str(name)


def _quote_identifier(name: str) -> str:
    return '"' + name.replace('"', '""') + '"'


def list_physical_column_names(cursor, table_name: str) -> list[str]:
    cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
    rows = cursor.fetchall()
    rows.sort(key=lambda row: int(row[0]))
    return [str(row[1]) for row in rows]