"""Wiki 页面 frontmatter 的 Pydantic 校验模型。
用法：
python schema.py .wiki/my-research/
python schema.py .wiki/my-research/entities/alpha-corp.md
Agent 在 ingest 完成后应自动运行校验。lint 操作也会调用。
"""
from __future__ import annotations
import sys
from datetime import date
from enum import Enum
from pathlib import Path
from typing import Any, Optional, Union
import yaml
from pydantic import BaseModel, Field, field_validator, model_validator
# ── 枚举 ──────────────────────────────────────────────────
class PageType(str, Enum):
source = "source"
entity = "entity"
concept = "concept"
analysis = "analysis"
mental_model = "mental-model"
class Confidence(str, Enum):
high = "high"
medium = "medium"
low = "low"
contested = "contested"
class SourceType(str, Enum):
primary = "一手"
authoritative_secondary = "二手·权威"
secondary = "二手"
hearsay = "转述"
inference = "推断"
oral = "口述"
# ── 关系 ──────────────────────────────────────────────────
class Relation(BaseModel):
"""页面间的类型化关系。"""
target: str # 目标页面 slug
type: str # 关系类型
@field_validator("type")
@classmethod
def known_type(cls, v: str) -> str:
"""建议使用标准关系类型，但不强制。"""
standard = {
"part_of", "manages", "regulated_by", "competes_with",
"implements", "derived_from", "contradicts", "influenced_by",
"applies_to", "supplies", "subsidiary_of", "contrasted_with",
}
if v not in standard:
import warnings
warnings.warn(
f"非标准关系类型 '{v}'（允许使用，但建议对齐标准类型：{', '.join(sorted(standard))}）",
stacklevel=2,
)
return v
# ── 三重验证（cognitive 类型专用）──────────────────────────
class MentalModelVerification(BaseModel):
"""心智模型的三重验证结果。"""
cross_domain: bool = False # 跨域复现
generative: bool = False # 有生成力
exclusive: bool = False # 有排他性
domains: list[str] = Field(default_factory=list) # 出现过的领域
# ── 页面基类 ──────────────────────────────────────────────
class BasePage(BaseModel):
"""所有页面的公共字段。"""
title: str
type: PageType
created: Union[str, date] # YYYY-MM-DD（YAML 可能解析为 date 对象）
updated: Union[str, date] # YYYY-MM-DD
sources: list[str] = Field(default_factory=list)
confidence: Confidence = Confidence.high
@field_validator("created", "updated", mode="before")
@classmethod
def coerce_date(cls, v: Any) -> str:
"""YAML 会把 2026-04-07 自动解析为 datetime.date，统一转为 str。"""
if isinstance(v, date):
return v.isoformat()
if isinstance(v, str):
try:
date.fromisoformat(v)
except ValueError:
raise ValueError(f"日期必须为 YYYY-MM-DD 格式: {v}")
return v
raise ValueError(f"日期类型错误: {type(v)}")
class SourcePage(BasePage):
"""source 类型页面。"""
type: PageType = PageType.source
source_type: SourceType
source_origin: str # 来源出处
source_date: Union[str, date] # 原始材料日期
source_url: str = "" # 来源 URL
@field_validator("source_date", mode="before")
@classmethod
def coerce_source_date(cls, v: Any) -> str:
if isinstance(v, date):
return v.isoformat()
return v
@model_validator(mode="after")
def sources_should_be_empty(self) -> "SourcePage":
if self.sources:
raise ValueError("source 类型页面的 sources 应为空列表")
return self
class DataPage(BasePage):
"""entity / concept / analysis 页面。
结构化数据（data/history）存在 data.db 中，不在 frontmatter。
relations 保留在 frontmatter（Obsidian wikilink 渲染），同时写入 data.db。
"""
relations: list[Relation] = Field(default_factory=list)
@model_validator(mode="after")
def sources_not_empty(self) -> "DataPage":
if self.type != PageType.source and not self.sources:
raise ValueError(f"{self.type.value} 类型页面的 sources 不能为空")
return self
class MentalModelPage(BasePage):
"""mental-model 类型页面（cognitive wiki 专用）。"""
type: PageType = PageType.mental_model
verification: Optional[MentalModelVerification] = None
relations: list[Relation] = Field(default_factory=list)
# ── 解析与校验 ────────────────────────────────────────────
def parse_frontmatter(path: Path) -> dict[str, Any]:
"""从 markdown 文件中提取 YAML frontmatter。"""
text = path.read_text(encoding="utf-8")
if not text.startswith("---"):
raise ValueError(f"页面缺少 YAML frontmatter: {path}")
parts = text.split("---", 2)
if len(parts) < 3:
raise ValueError(f"YAML frontmatter 格式错误: {path}")
return yaml.safe_load(parts[1]) or {}
def validate_page(path: Path) -> tuple[bool, str]:
"""校验单个页面。返回 (通过?, 消息)。"""
try:
fm = parse_frontmatter(path)
except Exception as e:
return False, f"PARSE ERROR: {e}"
page_type = fm.get("type", "")
warnings_list: list[str] = []
# data/history 应在 data.db 中，不在 frontmatter
if "data" in fm and fm["data"]:
warnings_list.append("MIGRATE: frontmatter 中有 data 字段，应迁移到 data.db（python store.py init）")
if "history" in fm and fm["history"]:
warnings_list.append("MIGRATE: frontmatter 中有 history 字段，应迁移到 data.db")
# 从 fm 中移除 data/history 避免 Pydantic 报错（模型已不包含这些字段）
fm_clean = {k: v for k, v in fm.items() if k not in ("data", "history")}
try:
if page_type == "source":
SourcePage(**fm_clean)
elif page_type == "mental-model":
MentalModelPage(**fm_clean)
elif page_type in ("entity", "concept", "analysis"):
DataPage(**fm_clean)
else:
return False, f"UNKNOWN TYPE: {page_type}"
if warnings_list:
return True, "OK (⚠️ " + "; ".join(warnings_list) + ")"
return True, "OK"
except Exception as e:
return False, f"VALIDATION ERROR: {e}"
def validate_wiki(wiki_dir: Path) -> list[tuple[str, bool, str]]:
"""校验整个 wiki 目录。返回 [(文件名, 通过?, 消息)]。"""
results = []
for subdir in ["sources", "entities", "concepts", "analyses", "mental-models"]:
d = wiki_dir / subdir
if not d.exists():
continue
for f in sorted(d.glob("*.md")):
if f.name.startswith("_"):
continue
ok, msg = validate_page(f)
rel = f.relative_to(wiki_dir)
results.append((str(rel), ok, msg))
return results
# ── Report 数据提取 ──────────────────────────────────────
import json
import re
from collections import Counter
def _page_body(path: Path) -> str:
"""读取 markdown 正文（frontmatter 之后的部分）。"""
text = path.read_text(encoding="utf-8")
parts = text.split("---", 2)
return parts[2] if len(parts) >= 3 else ""
def _extract_wikilinks(body: str) -> list[str]:
"""提取正文中的 [[slug]] 或 [[slug|display]] 链接。"""
return re.findall(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]", body)
def collect_report_data(wiki_dir: Path) -> dict[str, Any]:
"""扫描 wiki 目录，提取完整的报告数据。"""
meta_path = wiki_dir / "meta.yaml"
meta = {}
if meta_path.exists():
meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
nodes: list[dict] = []
edges: list[dict] = []
data_rows: list[dict] = []
type_counts: Counter = Counter()
confidence_counts: Counter = Counter()
contested_pages: list[str] = []
inlink_counts: Counter = Counter()
page_slugs: set[str] = set()
freshness: list[dict] = []
subdirs = ["sources", "entities", "concepts", "analyses", "mental-models"]
for subdir in subdirs:
d = wiki_dir / subdir
if not d.exists():
continue
for f in sorted(d.glob("*.md")):
if f.name.startswith("_"):
continue
slug = f.stem
page_slugs.add(slug)
rel_path = str(f.relative_to(wiki_dir))
try:
fm = parse_frontmatter(f)
except Exception:
continue
page_type = fm.get("type", "unknown")
confidence = fm.get("confidence", "unknown")
title = fm.get("title", slug)
updated = str(fm.get("updated", ""))
type_counts[page_type] += 1
confidence_counts[confidence] += 1
if confidence == "contested":
contested_pages.append(rel_path)
# 节点颜色按类型
color_map = {
"entity": "#4A90D9",
"concept": "#7B68EE",
"source": "#50C878",
"analysis": "#FF8C00",
"mental-model": "#E91E63",
}
nodes.append({
"id": slug,
"label": title,
"type": page_type,
"color": color_map.get(page_type, "#999"),
"confidence": confidence,
"updated": updated,
"path": rel_path,
})
freshness.append({"slug": slug, "updated": updated, "type": page_type})
# 关系 → 边
for rel in fm.get("relations", []):
target = rel.get("target", "")
rel_type = rel.get("type", "")
if target:
edges.append({
"from": slug,
"to": target,
"label": rel_type,
})
# 正文 wikilinks → 入链统计
body = _page_body(f)
for linked_slug in _extract_wikilinks(body):
inlink_counts[linked_slug] += 1
# 从 data.db 读取结构化数据（唯一数据源）
db_path = wiki_dir / "data.db"
if db_path.exists():
import sqlite3 as _sqlite3
_conn = _sqlite3.connect(str(db_path))
_conn.row_factory = _sqlite3.Row
for r in _conn.execute("SELECT * FROM data_points ORDER BY page_slug, field").fetchall():
page_title = r["page_slug"]
pg = _conn.execute("SELECT title FROM pages WHERE slug=?", (r["page_slug"],)).fetchone()
if pg:
page_title = pg["title"]
data_rows.append({
"page": page_title, "slug": r["page_slug"], "field": r["field"],
"value": r["value"], "unit": r["unit"], "period": r["period"],
"source": r["source_slug"], "verified": r["verified"],
"confidence": r["confidence"] or "high",
})
# DB relations 补充 frontmatter edges
edge_set = {(e["from"], e["to"], e["label"]) for e in edges}
for r in _conn.execute("SELECT * FROM relations").fetchall():
key = (r["from_slug"], r["to_slug"], r["type"])
if key not in edge_set:
edges.append({"from": r["from_slug"], "to": r["to_slug"], "label": r["type"]})
# DB contested 补充
contested_set = set(contested_pages)
for r in _conn.execute("SELECT DISTINCT page_slug FROM data_points WHERE confidence='contested'").fetchall():
for n in nodes:
if n["id"] == r["page_slug"] and n["path"] not in contested_set:
contested_pages.append(n["path"])
contested_set.add(n["path"])
_conn.close()
# frontmatter relations 也算入链
for edge in edges:
inlink_counts[edge["to"]] += 1
# 覆盖度分析
coverage_gaps: list[dict] = []
for slug in page_slugs:
inlinks = inlink_counts.get(slug, 0)
node = next((n for n in nodes if n["id"] == slug), None)
if not node:
continue
if inlinks == 0 and node["type"] != "source":
coverage_gaps.append({
"slug": slug,
"issue": "orphan",
"detail": f"零入链（无其他页面引用）",
})
# 被大量引用但内容可能薄的页面（通过 wikilink 被引用但不在 page_slugs 中）
for linked, count in inlink_counts.most_common():
if linked not in page_slugs and count >= 2:
coverage_gaps.append({
"slug": linked,
"issue": "missing",
"detail": f"被引用 {count} 次但页面不存在",
})
return {
"name": meta.get("name", wiki_dir.name),
"ontology_type": meta.get("ontology_type", "unknown"),
"description": meta.get("description", ""),
"seed": meta.get("seed", ""),
"total_pages": len(nodes),
"type_counts": dict(type_counts),
"confidence_counts": dict(confidence_counts),
"contested_pages": contested_pages,
"nodes": nodes,
"edges": edges,
"data_rows": data_rows,
"freshness": freshness,
"coverage_gaps": coverage_gaps,
}
REPORT_HTML_TEMPLATE = r"""
Wiki Report: {{WIKI_NAME}}
"""
def generate_report(wiki_dir: Path) -> Path:
"""生成 wiki 可视化报告 HTML。返回输出文件路径。"""
data = collect_report_data(wiki_dir)
html = REPORT_HTML_TEMPLATE
html = html.replace("{{WIKI_NAME}}", data["name"])
html = html.replace("{{WIKI_DESC}}", data.get("description", ""))
html = html.replace("{{ONTOLOGY_TYPE}}", data.get("ontology_type", ""))
html = html.replace("{{TOTAL_PAGES}}", str(data["total_pages"]))
html = html.replace("{{JSON_DATA}}", json.dumps(data, ensure_ascii=False, default=str))
out_path = wiki_dir / "_report.html"
out_path.write_text(html, encoding="utf-8")
return out_path
# ── CLI ───────────────────────────────────────────────────
def main():
if len(sys.argv) < 2:
print("Usage:")
print(" python schema.py — validate all pages")
print(" python schema.py — validate one page")
print(" python schema.py --report — generate visual report")
sys.exit(1)
# --report 模式
if sys.argv[1] == "--report":
if len(sys.argv) < 3:
print("Usage: python schema.py --report ")
sys.exit(1)
target = Path(sys.argv[2])
if not target.is_dir():
print(f"Not a directory: {target}")
sys.exit(1)
out = generate_report(target)
print(f"Report generated: {out}")
sys.exit(0)
target = Path(sys.argv[1])
if target.is_file():
ok, msg = validate_page(target)
status = "✅" if ok else "❌"
print(f"{status} {target.name}: {msg}")
sys.exit(0 if ok else 1)
if target.is_dir():
results = validate_wiki(target)
if not results:
print("No pages found.")
sys.exit(0)
passed = sum(1 for _, ok, _ in results if ok)
failed = sum(1 for _, ok, _ in results if not ok)
print(f"\n{'='*60}")
print(f"Wiki Validation: {target.name}")
print(f"{'='*60}\n")
for name, ok, msg in results:
status = "✅" if ok else "❌"
print(f" {status} {name}")
if not ok:
# 缩进显示错误详情
for line in str(msg).split("\n"):
print(f" {line}")
print(f"\n{'─'*60}")
print(f" Total: {len(results)} | Passed: {passed} | Failed: {failed}")
if failed == 0:
print(" Result: ALL PASSED ✅")
else:
print(f" Result: {failed} FAILED ❌")
print(f"{'─'*60}\n")
sys.exit(0 if failed == 0 else 1)
print(f"Not a file or directory: {target}")
sys.exit(1)
if __name__ == "__main__":
main()