feat: 申报赛道推荐助手首版 v1.0.1(match-guide-track-default)
All checks were successful
技能自动化发布 / release (push) Successful in 6s

This commit is contained in:
chenfuqiang
2026-09-13 15:50:33 +08:00
commit 7c1f40b0e6
205 changed files with 29720 additions and 0 deletions

225
scripts/db/connection.py Normal file
View File

@@ -0,0 +1,225 @@
"""SQLite 连接与幂等初始化(申报赛道推荐助手)。
业务表 3 张:
- teacher_profiles(教师画像,可编辑)
- guide_tracks(指南赛道,可编辑)
- match_reports(推荐报告,只读,匹配自动写入)
表间禁止 SQLite 外键,仅存业务 id(见 references/SCHEMA.md)。
"""
from __future__ import annotations
import sqlite3
from db.display_metadata import (
init_display_metadata,
seed_column_metadata_batch,
upsert_table_metadata,
)
from db.timestamp_columns import init_updated_at_trigger
from util.runtime_paths import get_db_path
def get_conn() -> sqlite3.Connection:
return sqlite3.connect(get_db_path())
_TEACHER_PROFILES_DDL = """
CREATE TABLE IF NOT EXISTS teacher_profiles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
teacher_name TEXT NOT NULL UNIQUE,
department TEXT,
title TEXT,
research_tags TEXT,
recent_papers TEXT,
project_history TEXT,
sci_q1_count INTEGER,
core_paper_count INTEGER,
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
updated_at INTEGER NOT NULL DEFAULT (unixepoch())
)
"""
_GUIDE_TRACKS_DDL = """
CREATE TABLE IF NOT EXISTS guide_tracks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
year INTEGER NOT NULL,
agency TEXT,
program_name TEXT,
track_name TEXT NOT NULL,
track_tags TEXT,
priority_keywords TEXT,
min_title TEXT,
min_sci_q1 INTEGER NOT NULL DEFAULT 0,
quota INTEGER,
source_doc TEXT,
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
updated_at INTEGER NOT NULL DEFAULT (unixepoch()),
UNIQUE (year, agency, program_name, track_name)
)
"""
_MATCH_REPORTS_DDL = """
CREATE TABLE IF NOT EXISTS match_reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
batch_no TEXT NOT NULL,
teacher_name TEXT,
department TEXT,
season TEXT,
top_json TEXT,
tiers_json TEXT,
eligible_count INTEGER,
excluded_json TEXT,
report_file TEXT,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
)
"""
_MATCH_REPORTS_INDEX = """
CREATE INDEX IF NOT EXISTS idx_match_reports_batch
ON match_reports(batch_no)
"""
_TS_COLUMN_NAMES = {"created_at", "updated_at"}
def _col(column_name: str, display_name: str, description: str, **extra: object) -> dict[str, object]:
base: dict[str, object] = {
"column_name": column_name,
"display_name": display_name,
"description": description,
"visible": 1,
"searchable": 0,
"editable": 0,
}
base.update(extra)
if column_name in _TS_COLUMN_NAMES:
base["display_type"] = "datetime_unix_seconds"
return base
_TEACHER_PROFILES_COLUMNS: tuple[dict[str, object], ...] = (
_col("id", "编号", "主键", searchable=1),
_col("teacher_name", "教师姓名", "教师姓名(唯一)。", searchable=1),
_col("department", "所属学院", "二级学院/单位。", searchable=1, editable=1),
_col("title", "职称", "讲师/副教授/教授等,用于职称门槛比对。", searchable=1, editable=1, options_json='[{"value":"讲师","label":"讲师"},{"value":"副教授","label":"副教授"},{"value":"教授","label":"教授"}]'),
_col("research_tags", "研究标签", "研究方向关键词,逗号分隔(匹配主依据)。", searchable=1, editable=1),
_col("recent_papers", "近3年论文", "近 3 年论文标题,分号分隔。", editable=1),
_col("project_history", "项目经历", "在研/结题项目名称,分号分隔。", editable=1),
_col("sci_q1_count", "SCI一区篇数", "SCI 一区论文数量(硬门槛比对)。", searchable=1, editable=1, display_type="number"),
_col("core_paper_count", "核心论文篇数", "核心期刊论文数量。", searchable=1, editable=1, display_type="number"),
_col("created_at", "创建时间", "记录创建时间(Unix 秒级,自动生成)。"),
_col("updated_at", "更新时间", "记录最后更新时间(Unix 秒级,自动维护)。"),
)
_GUIDE_TRACKS_COLUMNS: tuple[dict[str, object], ...] = (
_col("id", "编号", "主键", searchable=1),
_col("year", "指南年度", "指南发布/申报年度。", searchable=1),
_col("agency", "学部/归口", "如 数理学部、科研处。", searchable=1, editable=1),
_col("program_name", "项目类型", "如 面上项目、青年项目。", searchable=1, editable=1),
_col("track_name", "赛道名称", "指南中的方向/赛道名。", searchable=1, editable=1),
_col("track_tags", "需求标签", "赛道方向关键词,逗号分隔(匹配主依据)。", searchable=1, editable=1),
_col("priority_keywords", "优先方向", "优先资助方向关键词,逗号分隔(加分项)。", editable=1),
_col("min_title", "职称门槛", "讲师及以上 / 副高及以上 / 正高及以上。", editable=1, options_json='[{"value":"讲师及以上","label":"讲师及以上"},{"value":"副高及以上","label":"副高及以上"},{"value":"正高及以上","label":"正高及以上"}]'),
_col("min_sci_q1", "SCI一区门槛", "要求的 SCI 一区最少篇数。", editable=1, display_type="number"),
_col("quota", "拟资助数", "该赛道拟资助项数(参考)。", editable=1, display_type="number"),
_col("source_doc", "指南来源", "指南文件名/文号。", editable=1),
_col("created_at", "创建时间", "记录创建时间(Unix 秒级,自动生成)。"),
_col("updated_at", "更新时间", "记录最后更新时间(Unix 秒级,自动维护)。"),
)
_MATCH_REPORTS_COLUMNS: tuple[dict[str, object], ...] = (
_col("id", "报告编号", "主键,推荐报告编号。", searchable=1),
_col("batch_no", "匹配批次号", "同一次匹配请求的批次号。", searchable=1),
_col("teacher_name", "教师姓名", "被匹配教师。", searchable=1),
_col("department", "所属学院", "教师所属学院。", searchable=1),
_col("season", "指南年度", "匹配所用指南年度。", searchable=1),
_col("top_json", "Top3推荐", "Top 3 赛道(JSON,含得分与理由)。", visible=0, display_type="textarea"),
_col("tiers_json", "分层清单", "强烈推荐/可尝试/不建议分层(JSON)。", visible=0, display_type="textarea"),
_col("eligible_count", "可匹配赛道数", "通过硬门槛过滤后的赛道数。", searchable=1, display_type="number"),
_col("excluded_json", "剔除清单", "被硬门槛剔除的赛道及原因(JSON)。", visible=0, display_type="textarea"),
_col("report_file", "建议书文件", "《申报策略建议书》HTML 文件路径。", visible=0),
_col("created_at", "创建时间", "报告创建时间(Unix 秒级,自动生成)。"),
)
def _init_teacher_profiles_metadata(cursor) -> None:
cursor.execute(_TEACHER_PROFILES_DDL)
init_updated_at_trigger(cursor, "teacher_profiles")
upsert_table_metadata(
cursor,
table_name="teacher_profiles",
display_name="教师画像",
description="教师学术画像:研究标签、成果与职称,用于赛道匹配。",
grid_help="研究标签是匹配主依据(逗号分隔);顶栏按钮可录入,行内可编辑。",
visible=1,
readonly=0,
default_sort_json='[{"column":"id","direction":"desc"}]',
default_visible_columns_json='["teacher_name","department","title","research_tags","sci_q1_count"]',
)
seed_column_metadata_batch(
cursor,
table_name="teacher_profiles",
columns=_TEACHER_PROFILES_COLUMNS,
valid_column_names=[str(c["column_name"]) for c in _TEACHER_PROFILES_COLUMNS],
)
def _init_guide_tracks_metadata(cursor) -> None:
cursor.execute(_GUIDE_TRACKS_DDL)
init_updated_at_trigger(cursor, "guide_tracks")
upsert_table_metadata(
cursor,
table_name="guide_tracks",
display_name="指南赛道",
description="当年申报指南的结构化赛道库:需求标签、优先方向与硬门槛。",
grid_help="按年度管理;需求标签为匹配主依据,硬门槛用于过滤不满足条件的教师。",
visible=1,
readonly=0,
default_sort_json='[{"column":"year","direction":"desc"},{"column":"id","direction":"asc"}]',
default_visible_columns_json='["year","program_name","track_name","track_tags","min_title","min_sci_q1"]',
)
seed_column_metadata_batch(
cursor,
table_name="guide_tracks",
columns=_GUIDE_TRACKS_COLUMNS,
valid_column_names=[str(c["column_name"]) for c in _GUIDE_TRACKS_COLUMNS],
)
def _init_match_reports_metadata(cursor) -> None:
cursor.execute(_MATCH_REPORTS_DDL)
cursor.execute(_MATCH_REPORTS_INDEX)
upsert_table_metadata(
cursor,
table_name="match_reports",
display_name="推荐报告",
description="每次匹配的 Top 3 赛道、分层清单与剔除原因;可重建建议书。",
grid_help="推荐仅供参考;顶栏按钮可重建《申报策略建议书》。",
visible=1,
readonly=1,
default_sort_json='[{"column":"id","direction":"desc"}]',
default_visible_columns_json='["id","season","teacher_name","department","eligible_count"]',
)
seed_column_metadata_batch(
cursor,
table_name="match_reports",
columns=_MATCH_REPORTS_COLUMNS,
valid_column_names=[str(c["column_name"]) for c in _MATCH_REPORTS_COLUMNS],
)
def init_db() -> dict[str, int]:
"""幂等创建/迁移本地 SQLite 与数据管理元数据。"""
conn = get_conn()
try:
cur = conn.cursor()
init_display_metadata(cur)
_init_teacher_profiles_metadata(cur)
_init_guide_tracks_metadata(cur)
_init_match_reports_metadata(cur)
conn.commit()
return {}
finally:
conn.close()

View File

@@ -0,0 +1,324 @@
"""匠厂数据管理展示元数据(_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]

View File

@@ -0,0 +1,592 @@
"""数据管理展示元数据校验(模板开发自检,不依赖宿主)。"""
from __future__ import annotations
import json
import os
import re
import sqlite3
from dataclasses import dataclass, field
from db.display_metadata import (
DATETIME_UNIX_SECONDS,
METADATA_TABLES,
STANDARD_TIMESTAMP_COLUMNS,
iter_user_tables,
list_editable_business_columns,
list_physical_column_names,
table_allows_manual_create,
)
from db.timestamp_columns import has_updated_at_trigger
from util.constants import SKILL_SLUG
from util.slug_naming import LEGACY_EXEMPT_SLUGS, validate_slug_semantics
_TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"})
_ENGLISH_PLACEHOLDER_NAME_PATTERNS = (
re.compile(r"^my\s+skill$", re.I),
re.compile(r"^your[-_\s]?skill", re.I),
re.compile(r"^skill[-_\s]?template$", re.I),
re.compile(r"^[a-z0-9_-]+$"),
)
_KEBAB_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
_CJK_RE = re.compile(r"[\u4e00-\u9fff]")
_FK_DDL_PATTERN = re.compile(r"\bFOREIGN\s+KEY\b|\bREFERENCES\b", re.I)
@dataclass
class ValidationReport:
errors: list[str] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
@property
def ok(self) -> bool:
return not self.errors
def _has_cjk(text: str) -> bool:
return bool(_CJK_RE.search(text))
def _looks_like_raw_physical_name(display_name: str, physical_name: str) -> bool:
if not display_name.strip():
return True
if display_name.strip().lower() == physical_name.strip().lower():
return True
compact = display_name.replace(" ", "").replace("_", "").lower()
if compact == physical_name.replace("_", "").lower():
return True
return False
def _parse_frontmatter_field(text: str, field_name: str) -> str:
parts = text.split("---", 2)
if len(parts) < 2:
return ""
for line in parts[1].splitlines():
match = re.match(rf"^{re.escape(field_name)}:\s*(.+)\s*$", line.strip())
if match:
return match.group(1).strip().strip("\"'")
return ""
def _parse_openclaw_slug(text: str) -> str:
parts = text.split("---", 2)
if len(parts) < 2:
return ""
lines = parts[1].splitlines()
for index, line in enumerate(lines):
if re.match(r"^\s+openclaw:\s*$", line):
for inner in lines[index + 1 :]:
if inner.strip() and inner[0] not in (" ", "\t"):
break
match = re.match(r"^(\s+)slug:\s*(.+)\s*$", inner)
if match:
return match.group(2).strip().strip("\"'")
return ""
def parse_skill_frontmatter(skill_root: str) -> dict[str, str]:
path = os.path.join(skill_root, "SKILL.md")
with open(path, encoding="utf-8") as f:
text = f.read()
return {
"name": _parse_frontmatter_field(text, "name"),
"slug": _parse_openclaw_slug(text),
}
def validate_skill_frontmatter(skill_root: str, *, allow_template_placeholders: bool = True) -> ValidationReport:
report = ValidationReport()
data = parse_skill_frontmatter(skill_root)
name = str(data.get("name") or "").strip()
slug = str(data.get("slug") or "").strip()
if not name:
report.errors.append("SKILL.md frontmatter.name 缺失")
elif not _has_cjk(name):
if allow_template_placeholders and slug in _TEMPLATE_PLACEHOLDER_SLUGS:
report.warnings.append("SKILL.md frontmatter.name 建议使用清晰的中文业务名称(模板占位可暂保留)")
else:
report.errors.append("SKILL.md frontmatter.name 应使用用户可见的中文业务名称")
else:
for pattern in _ENGLISH_PLACEHOLDER_NAME_PATTERNS:
if pattern.fullmatch(name):
if allow_template_placeholders and slug in _TEMPLATE_PLACEHOLDER_SLUGS:
report.warnings.append(f"SKILL.md frontmatter.name 仍为占位风格:{name!r}")
else:
report.errors.append(f"SKILL.md frontmatter.name 不应为英文占位名:{name!r}")
break
if not slug:
report.errors.append("SKILL.md metadata.openclaw.slug 缺失")
elif not _KEBAB_SLUG.fullmatch(slug):
report.errors.append(f"metadata.openclaw.slug 必须为 kebab-case:{slug!r}")
elif slug in _TEMPLATE_PLACEHOLDER_SLUGS and not allow_template_placeholders:
report.errors.append(f"metadata.openclaw.slug 仍为模板占位:{slug!r}")
constants_slug = SKILL_SLUG.strip()
if slug and constants_slug and slug != constants_slug:
report.warnings.append(
f"SKILL.md slug ({slug!r}) 与 util.constants.SKILL_SLUG ({constants_slug!r}) 不一致"
)
if slug and slug not in LEGACY_EXEMPT_SLUGS and _KEBAB_SLUG.fullmatch(slug):
naming_errors, naming_warnings = validate_slug_semantics(
slug,
strict=not allow_template_placeholders,
)
report.errors.extend(naming_errors)
report.warnings.extend(naming_warnings)
return report
def _quote_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def _table_column_info(cursor, table_name: str) -> dict[str, dict[str, object]]:
cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
info: dict[str, dict[str, object]] = {}
for row in cursor.fetchall():
info[str(row[1])] = {
"cid": int(row[0]),
"name": str(row[1]),
"type": str(row[2] or ""),
"notnull": int(row[3] or 0),
"dflt_value": row[4],
"pk": int(row[5] or 0),
}
return info
def _table_column_xinfo(cursor, table_name: str) -> dict[str, dict[str, object]]:
"""PRAGMA table_xinfo 可识别 generated/hidden 字段(SQLite 3.31+)。"""
cursor.execute(f"PRAGMA table_xinfo({_quote_identifier(table_name)})")
info: dict[str, dict[str, object]] = {}
for row in cursor.fetchall():
hidden = int(row[6] or 0) if len(row) > 6 else 0
info[str(row[1])] = {
"cid": int(row[0]),
"name": str(row[1]),
"type": str(row[2] or ""),
"notnull": int(row[3] or 0),
"dflt_value": row[4],
"pk": int(row[5] or 0),
"hidden": hidden,
}
return info
def _is_generated_column(xinfo: dict[str, object]) -> bool:
hidden = int(xinfo.get("hidden") or 0)
return hidden in (2, 3)
def _is_blob_column(column_info: dict[str, object]) -> bool:
return "BLOB" in str(column_info.get("type") or "").upper()
def _default_uses_unixepoch(default_value: object) -> bool:
if default_value is None:
return False
text = str(default_value).strip().lower()
return "unixepoch" in text
def validate_options_json(options_json: str | None) -> tuple[list[str], list[str]]:
"""校验 options_json:静态枚举数组,或 lookup 对象。"""
errors: list[str] = []
warnings: list[str] = []
if options_json is None or str(options_json).strip() == "":
return errors, warnings
try:
parsed = json.loads(str(options_json))
except json.JSONDecodeError as exc:
errors.append(f"options_json 不是合法 JSON:{exc}")
return errors, warnings
if isinstance(parsed, dict):
for key in ("ref_table", "value_column", "label_column"):
if not str(parsed.get(key) or "").strip():
errors.append(f"options_json lookup 必须包含非空 {key}")
ident = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
for key in ("ref_table", "value_column", "label_column", "filter_column"):
raw = parsed.get(key)
if raw is None or raw == "":
continue
if not ident.match(str(raw)):
errors.append(f"options_json.{key} 必须是安全标识符,当前为 {raw!r}")
if parsed.get("filter_column") not in (None, "") and "filter_value" not in parsed:
warnings.append("options_json lookup 声明了 filter_column 但未提供 filter_value")
if "multiple" in parsed and not isinstance(parsed.get("multiple"), bool):
errors.append("options_json lookup.multiple 必须是布尔值")
return errors, warnings
if not isinstance(parsed, list):
errors.append("options_json 必须是 JSON 数组(静态枚举)或对象(lookup)")
return errors, warnings
for index, item in enumerate(parsed):
if not isinstance(item, dict):
errors.append(f"options_json[{index}] 必须是对象")
continue
if "value" not in item or "label" not in item:
errors.append(f"options_json[{index}] 必须包含 value 与 label")
return errors, warnings
def validate_filter_search_metadata(
cursor,
table_name: str,
) -> tuple[list[str], list[str]]:
"""对可见但未开启 searchable 的枚举 / boolean 列给出软建议。"""
warnings: list[str] = []
cursor.execute(
"""
SELECT column_name, visible, searchable, display_type, options_json
FROM _jiangchang_columns
WHERE table_name = ?
""",
(table_name,),
)
for column_name, visible, searchable, display_type, options_json in cursor.fetchall():
if visible is not None and int(visible) == 0:
continue
if searchable is not None and int(searchable) != 0:
continue
display_type_norm = str(display_type or "").strip()
is_boolean = display_type_norm == "boolean"
has_enum = False
if options_json is not None and str(options_json).strip():
opt_errors, _ = validate_options_json(options_json)
if not opt_errors:
try:
parsed = json.loads(str(options_json))
except json.JSONDecodeError:
parsed = None
has_enum = isinstance(parsed, list) and len(parsed) > 0
if not has_enum and not is_boolean:
continue
warnings.append(
f"{table_name}.{column_name}: 枚举/状态类列建议 searchable=1,"
"以便宿主列筛选与按中文 label 全局搜索(见 references/SCHEMA.md)"
)
return [], warnings
def validate_column_permissions(
cursor,
table_name: str,
*,
physical_columns: list[str] | None = None,
) -> tuple[list[str], list[str]]:
"""校验常见系统字段 editable=0,并识别技术上可写但不适合手工新增的表。"""
errors: list[str] = []
warnings: list[str] = []
ordered = physical_columns or list_physical_column_names(cursor, table_name)
column_info = _table_column_info(cursor, table_name)
column_xinfo = _table_column_xinfo(cursor, table_name)
cursor.execute(
"""
SELECT column_name, editable, options_json
FROM _jiangchang_columns
WHERE table_name = ?
""",
(table_name,),
)
meta_by_column = {
str(row[0]): {"editable": row[1], "options_json": row[2]}
for row in cursor.fetchall()
}
for column_name, meta in meta_by_column.items():
physical = column_info.get(column_name) or {}
xinfo = column_xinfo.get(column_name) or physical
if not physical and column_name not in column_xinfo:
continue
pk = int(physical.get("pk") or xinfo.get("pk") or 0)
editable = meta["editable"]
if pk and editable is not None and int(editable) != 0:
errors.append(f"{table_name}.{column_name} 是主键,editable 应为 0")
if column_name == "id" and editable is not None and int(editable) != 0:
errors.append(f"{table_name}.id 的 editable 应为 0(自增主键)")
if _is_generated_column(xinfo) and editable is not None and int(editable) != 0:
errors.append(f"{table_name}.{column_name} 是 generated 字段,editable 应为 0")
if _is_blob_column(physical or xinfo) and editable is not None and int(editable) != 0:
errors.append(f"{table_name}.{column_name} 是 BLOB 字段,editable 应为 0")
opt_errors, _ = validate_options_json(meta.get("options_json"))
for item in opt_errors:
errors.append(f"{table_name}.{column_name}: {item}")
cursor.execute(
"SELECT readonly FROM _jiangchang_tables WHERE table_name = ?",
(table_name,),
)
readonly_row = cursor.fetchone()
readonly = int(readonly_row[0]) if readonly_row else None
editable_columns = list_editable_business_columns(cursor, table_name)
if readonly == 0 and not editable_columns:
warnings.append(
f"{table_name}: 表级 readonly=0 但所有字段 editable=0,"
"宿主不应打开空白新增表单;数据应由技能业务流程产生"
)
if readonly == 1 and editable_columns:
warnings.append(
f"{table_name}: 表级 readonly=1 但存在 editable=1 字段,"
"宿主应以表级只读为准"
)
return errors, warnings
def _expected_timestamp_display_name(column_name: str) -> str:
if column_name == "created_at":
return "创建时间"
if column_name == "updated_at":
return "更新时间"
return ""
def validate_standard_timestamp_columns(
cursor,
table_name: str,
*,
physical_columns: list[str] | None = None,
) -> tuple[list[str], list[str]]:
errors: list[str] = []
warnings: list[str] = []
ordered = physical_columns or list_physical_column_names(cursor, table_name)
column_info = _table_column_info(cursor, table_name)
for column_name in STANDARD_TIMESTAMP_COLUMNS:
if column_name not in ordered:
continue
physical = column_info.get(column_name)
if not physical:
errors.append(f"{table_name}.{column_name} 缺少物理字段定义")
continue
if "INT" not in str(physical["type"]).upper():
errors.append(
f"{table_name}.{column_name} 字段类型应为 INTEGER(Unix 秒级时间戳),当前为 {physical['type']!r}"
)
if column_name == "created_at" and not _default_uses_unixepoch(physical["dflt_value"]):
errors.append(
f"{table_name}.created_at 应设置数据库默认值 DEFAULT (unixepoch()),当前默认值为 {physical['dflt_value']!r}"
)
if column_name == "updated_at" and not _default_uses_unixepoch(physical["dflt_value"]):
errors.append(
f"{table_name}.updated_at 应设置数据库初始默认值 DEFAULT (unixepoch()),当前默认值为 {physical['dflt_value']!r}"
)
cursor.execute(
"""
SELECT display_name, display_type, editable, searchable, display_order
FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(table_name, column_name),
)
meta_row = cursor.fetchone()
if not meta_row:
errors.append(f"{table_name}.{column_name} 缺少 _jiangchang_columns 元数据")
continue
display_name, display_type, editable, searchable, display_order = meta_row
expected_name = _expected_timestamp_display_name(column_name)
if str(display_name or "").strip() != expected_name:
errors.append(
f"{table_name}.{column_name} 的 display_name 应为 {expected_name!r},当前为 {display_name!r}"
)
if str(display_type or "").strip() != DATETIME_UNIX_SECONDS:
errors.append(
f"{table_name}.{column_name} 的 display_type 应为 {DATETIME_UNIX_SECONDS!r},当前为 {display_type!r}"
)
if editable is None or int(editable) != 0:
errors.append(f"{table_name}.{column_name} 的 editable 应为 0(系统时间字段不可编辑)")
if display_order is not None:
errors.append(f"{table_name}.{column_name} 的 display_order 应为 NULL")
if searchable is not None and int(searchable) != 0:
warnings.append(f"{table_name}.{column_name} 的时间字段通常不必开启 searchable")
if "created_at" in ordered and "updated_at" in ordered:
if ordered.index("updated_at") < ordered.index("created_at"):
warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前")
if "updated_at" in ordered and not has_updated_at_trigger(cursor, table_name):
errors.append(
f"{table_name}.updated_at 缺少自动维护 trigger(期望 {table_name}_set_updated_at)"
)
return errors, warnings
def column_order_warnings(table_name: str, ordered_names: list[str]) -> list[str]:
warnings: list[str] = []
if "id" in ordered_names and ordered_names.index("id") != 0:
warnings.append(f"{table_name}: 字段 id 通常应定义在 CREATE TABLE 的第一列")
if "created_at" in ordered_names:
idx = ordered_names.index("created_at")
trailing = ordered_names[idx + 1 :]
if trailing and trailing != ["updated_at"]:
warnings.append(
f"{table_name}: created_at 之后通常只应跟 updated_at,不应再放置其他业务字段"
)
if "updated_at" in ordered_names and "created_at" in ordered_names:
if ordered_names.index("updated_at") < ordered_names.index("created_at"):
warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前")
return warnings
def _table_exists(cursor, table_name: str) -> bool:
cursor.execute(
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1",
(table_name,),
)
return cursor.fetchone() is not None
def validate_no_sqlite_foreign_keys(cursor) -> tuple[list[str], list[str]]:
"""业务表 DDL 不得含 FOREIGN KEY / REFERENCES(关联走 repository/service)。"""
errors: list[str] = []
cursor.execute(
"""
SELECT name, sql FROM sqlite_master
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
"""
)
for table_name, sql in cursor.fetchall():
if table_name in METADATA_TABLES:
continue
ddl = str(sql or "")
if ddl and _FK_DDL_PATTERN.search(ddl):
errors.append(
f"业务表 {table_name!r} 的 DDL 含 SQLite FOREIGN KEY/REFERENCES;"
"请改为业务 id 列 + repository/service 逻辑关联(见 references/SCHEMA.md)"
)
return errors, []
def validate_display_metadata(conn: sqlite3.Connection, *, check_column_order: bool = True) -> ValidationReport:
report = ValidationReport()
cur = conn.cursor()
has_tables_meta = _table_exists(cur, "_jiangchang_tables")
has_columns_meta = _table_exists(cur, "_jiangchang_columns")
if not has_tables_meta:
report.errors.append("缺少数据管理元数据表 _jiangchang_tables")
if not has_columns_meta:
report.errors.append("缺少数据管理元数据表 _jiangchang_columns")
if not has_tables_meta or not has_columns_meta:
return report
fk_errors, _ = validate_no_sqlite_foreign_keys(cur)
report.errors.extend(fk_errors)
for table_name in iter_user_tables(cur):
cur.execute(
"SELECT display_name FROM _jiangchang_tables WHERE table_name = ?",
(table_name,),
)
row = cur.fetchone()
if not row or not str(row[0] or "").strip():
report.errors.append(f"业务表 {table_name!r} 缺少 _jiangchang_tables.display_name")
elif not _has_cjk(str(row[0])):
report.errors.append(f"业务表 {table_name!r} 的 display_name 应使用中文业务名称")
physical_columns = list_physical_column_names(cur, table_name)
if check_column_order:
report.warnings.extend(column_order_warnings(table_name, physical_columns))
for column_name in physical_columns:
cur.execute(
"""
SELECT display_name, visible
FROM _jiangchang_columns
WHERE table_name = ? AND column_name = ?
""",
(table_name, column_name),
)
col_row = cur.fetchone()
if not col_row:
report.errors.append(
f"字段 {table_name}.{column_name} 缺少 _jiangchang_columns 元数据"
)
continue
display_name = str(col_row[0] or "").strip()
visible = col_row[1] is None or int(col_row[1]) != 0
if visible and not display_name:
report.errors.append(f"字段 {table_name}.{column_name} 的 display_name 为空")
elif visible and _looks_like_raw_physical_name(display_name, column_name):
report.errors.append(
f"字段 {table_name}.{column_name} 的 display_name 不应直接等于英文物理名:{display_name!r}"
)
elif visible and not _has_cjk(display_name):
report.errors.append(
f"字段 {table_name}.{column_name} 的 display_name 应使用中文:{display_name!r}"
)
ts_errors, ts_warnings = validate_standard_timestamp_columns(
cur,
table_name,
physical_columns=physical_columns,
)
report.errors.extend(ts_errors)
report.warnings.extend(ts_warnings)
perm_errors, perm_warnings = validate_column_permissions(
cur,
table_name,
physical_columns=physical_columns,
)
report.errors.extend(perm_errors)
report.warnings.extend(perm_warnings)
_, filter_warnings = validate_filter_search_metadata(cur, table_name)
report.warnings.extend(filter_warnings)
cur.execute("SELECT table_name, column_name FROM _jiangchang_columns")
for table_name, column_name in cur.fetchall():
if table_name in METADATA_TABLES:
continue
cur.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?",
(table_name,),
)
if not cur.fetchone():
report.errors.append(f"元数据引用了不存在的表:{table_name!r}")
continue
physical = list_physical_column_names(cur, str(table_name))
if str(column_name) not in physical:
report.errors.append(f"元数据引用了不存在的字段:{table_name}.{column_name}")
cur.execute("SELECT table_name FROM _jiangchang_tables")
for (table_name,) in cur.fetchall():
if table_name in METADATA_TABLES:
continue
cur.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?",
(table_name,),
)
if not cur.fetchone():
report.errors.append(f"元数据引用了不存在的表:{table_name!r}")
return report
def validate_template_database(conn: sqlite3.Connection) -> ValidationReport:
"""模板自检入口:校验展示元数据完整性,并扫描业务表 DDL 是否含 SQLite 外键。"""
return validate_display_metadata(conn)

View File

@@ -0,0 +1,50 @@
"""标准时间字段:Unix 秒级时间戳、默认值与 updated_at 自动维护。
业务表若包含 created_at / updated_at,在 CREATE TABLE 后调用
``init_updated_at_trigger(cursor, table_name)``。模板默认不建业务表。
"""
from __future__ import annotations
STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})
DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
def updated_at_trigger_name(table_name: str) -> str:
return f"{table_name}_set_updated_at"
def _quote_identifier(name: str) -> str:
return '"' + name.replace('"', '""') + '"'
def init_updated_at_trigger(cursor, table_name: str) -> None:
"""幂等创建 {table}_set_updated_at:业务字段 UPDATE 后刷新 updated_at。"""
trigger = updated_at_trigger_name(table_name)
quoted = _quote_identifier(table_name)
quoted_trigger = _quote_identifier(trigger)
cursor.execute(
f"""
CREATE TRIGGER IF NOT EXISTS {quoted_trigger}
AFTER UPDATE ON {quoted}
FOR EACH ROW
BEGIN
UPDATE {quoted}
SET updated_at = unixepoch()
WHERE id = NEW.id AND updated_at = OLD.updated_at;
END
"""
)
def has_updated_at_trigger(cursor, table_name: str) -> bool:
trigger_name = updated_at_trigger_name(table_name)
cursor.execute(
"""
SELECT 1 FROM sqlite_master
WHERE type = 'trigger' AND name = ? AND tbl_name = ?
LIMIT 1
""",
(trigger_name, table_name),
)
return cursor.fetchone() is not None