skill-template/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

TASK_LOGS_TABLE = "task_logs"
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,
    sort_order INTEGER,
    visible INTEGER NOT NULL DEFAULT 1,
    readonly INTEGER NOT NULL DEFAULT 0
)
"""

_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,
    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)
)
"""

TASK_LOGS_TABLE_METADATA = {
    "table_name": TASK_LOGS_TABLE,
    "display_name": "任务日志",
    "description": "记录技能任务的执行状态和结果（系统自动写入，宿主数据管理只读展示）",
    "visible": 1,
    "readonly": 1,
}

TASK_LOGS_COLUMN_METADATA: tuple[dict[str, object], ...] = (
    {"column_name": "id", "display_name": "编号", "description": "任务日志唯一编号", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "task_type", "display_name": "任务类型", "description": "任务的业务类型", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "target_id", "display_name": "目标编号", "description": "任务目标对象编号", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "input_id", "display_name": "输入编号", "description": "输入对象编号", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "input_title", "display_name": "输入标题", "description": "输入对象标题", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "status", "display_name": "状态", "description": "任务当前状态", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "error_msg", "display_name": "错误信息", "description": "任务失败时的错误说明", "visible": 1, "searchable": 1, "editable": 0},
    {"column_name": "result_summary", "display_name": "结果摘要", "description": "任务执行结果摘要", "visible": 1, "searchable": 1, "editable": 0},
    {
        "column_name": "created_at",
        "display_name": "创建时间",
        "description": "记录创建时间（Unix 秒级时间戳，数据库自动生成）",
        "visible": 1,
        "searchable": 0,
        "editable": 0,
        "display_type": DATETIME_UNIX_SECONDS,
    },
    {
        "column_name": "updated_at",
        "display_name": "更新时间",
        "description": "记录最后更新时间（Unix 秒级时间戳，自动维护）",
        "visible": 1,
        "searchable": 0,
        "editable": 0,
        "display_type": DATETIME_UNIX_SECONDS,
    },
)


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


def upsert_table_metadata(cursor, *, table_name: str, display_name: str, description: str | None = None, visible: int = 1, readonly: int = 0) -> None:
    cursor.execute(
        """
        INSERT INTO _jiangchang_tables (table_name, display_name, description, visible, readonly)
        VALUES (?, ?, ?, ?, ?)
        ON CONFLICT(table_name) DO UPDATE SET
            display_name = excluded.display_name,
            description = excluded.description,
            visible = excluded.visible,
            readonly = excluded.readonly
        """,
        (table_name, display_name, description, visible, readonly),
    )


def upsert_column_metadata(
    cursor,
    *,
    table_name: str,
    column_name: str,
    display_name: str,
    description: 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。
    cursor.execute(
        """
        INSERT INTO _jiangchang_columns (
            table_name, column_name, display_name, description,
            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,
            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,
            visible,
            searchable,
            editable,
            display_type,
            options_json,
        ),
    )


def seed_task_logs_display_metadata(cursor) -> None:
    meta = TASK_LOGS_TABLE_METADATA
    upsert_table_metadata(
        cursor,
        table_name=str(meta["table_name"]),
        display_name=str(meta["display_name"]),
        description=str(meta["description"]) if meta.get("description") else None,
        visible=int(meta["visible"]),
        readonly=int(meta["readonly"]),
    )
    seed_column_metadata_batch(
        cursor,
        table_name=TASK_LOGS_TABLE,
        columns=TASK_LOGS_COLUMN_METADATA,
        valid_column_names=[str(column["column_name"]) for column in TASK_LOGS_COLUMN_METADATA],
    )


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,
            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)
    seed_task_logs_display_metadata(cursor)


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]