feat(docs-center): CMAKnowledge reads md single source via new /api/cma/docs
ci/woodpecker/push/main Pipeline failed

New router app/api/doc_center.py: GET /api/cma/docs (list, ?with_content) and GET /api/cma/docs/{slug}; runtime disk read, role-gated (ceo/finance/business/it), no existing API semantics changed.

Content dir /srv/hermes/shared/cma-content (docs/*.md + categories.json). Placed outside the repo on purpose: the deploy tree equals the committed tree (deploy.sh checkout -f + clean -fdx), so in-repo content would be reset on every deploy and content edits would fail the deploy-tree dirty check.

CMAKnowledge.vue: 20 hardcoded items migrated to md, page now fetches at runtime and renders md with the existing marked dep (no new deps); table borders restored via :deep() since scoped styles never reach v-html content.

Tests: backend/tests/test_doc_center_20260924.py (20 cases). Full suite 1448 passed (baseline 1428 + 20).
This commit is contained in:
Hermes CI Fix
2026-09-24 01:07:29 +08:00
parent 72d4d97c83
commit ede0524c22
4 changed files with 512 additions and 286 deletions
+233
View File
@@ -0,0 +1,233 @@
"""CMA 文档中心 API —— md 单一来源(任务 yanxue-20260924-cma-docs-center-poc)
口径(任总 2026-09-24 00:54 拍板):**指标进 UI、结论出 md、机器走 API/MCP。**
设计要点
--------
1. 内容单一来源 = 运行期目录(默认 `/srv/hermes/shared/cma-content`,可用 `CMA_DOCS_DIR` 覆盖),
一篇一个 `docs/<slug>.md`,front-matter 带 title/category/category_name/icon/summary/order。
2. **故意不放在仓库内**:CMA 部署链是「部署树 = origin/main 已提交树」(deploy.sh `[0]/[1]`
`git checkout -f` + `git clean -fdx`):仓内内容目录会被每次部署重置,且内容改动会让
部署树出现「未提交修改」而被 `[1.1]` 判失败、**阻断整个部署**。放仓外 ⇒ 改内容不需要
git/部署/重启,dev 树与部署树天然共读同一份。
3. **本模块每次请求实时读盘,无缓存、不做 build-time 打包** ⇒ 新增/修改一个 md 后,
页面刷新即变,前端无需改码、无需 rebuild。
4. 只**新增**端点,不改任何既有接口语义;内容为系统级共享知识(与 `knowledge_articles`
`entity_id IS NULL` 的全局共享同族),不做账套分区,故不接 `entity_id`。
5. 写口 = 文件系统(Bot/Agent 直接写 md),**不提供写 API**,避免出现"第二来源"。
"""
from __future__ import annotations
import logging
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from app.auth_middleware import require_role
logger = logging.getLogger("cma.doc_center_api")
DEFAULT_DOCS_DIR = "/srv/hermes/shared/cma-content"
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
router = APIRouter(
prefix="/api/cma/docs",
tags=["文档中心"],
dependencies=[Depends(require_role("ceo", "finance", "business", "it"))],
)
# ────────────────────────── 内容目录(运行期读盘)──────────────────────────
def content_root() -> Path:
"""内容根目录(每次调用实时解析,便于测试/迁移换目录)"""
return Path(os.environ.get("CMA_DOCS_DIR", DEFAULT_DOCS_DIR))
def docs_dir() -> Path:
return content_root() / "docs"
def _parse_scalar(raw: str):
raw = raw.strip()
if len(raw) >= 2 and raw[0] == raw[-1] and raw[0] in "\"'":
body = raw[1:-1]
return body.replace('\\"', '"').replace("\\\\", "\\")
if re.fullmatch(r"-?\d+", raw or ""):
return int(raw)
return raw
def parse_front_matter(text: str) -> Optional[dict]:
"""极简 YAML front-matter 解析(只支持 `key: value` 标量,避免引入新依赖)"""
if not text.startswith("---"):
return None
end = text.find("\n---", 3)
if end < 0:
return None
meta = {}
for line in text[3:end].splitlines():
if not line.strip() or line.lstrip().startswith("#"):
continue
key, sep, val = line.partition(":")
if not sep:
continue
meta[key.strip()] = _parse_scalar(val)
body = text[end + 4:]
if body.startswith("\n"):
body = body[1:]
return {"meta": meta, "body": body}
def _category_registry() -> list:
"""分类注册表(categories.json):决定分类顺序与展示名/图标"""
path = content_root() / "categories.json"
cats = []
if path.is_file():
import json
try:
data = json.loads(path.read_text(encoding="utf-8"))
cats = list(data.get("categories") or [])
except Exception as exc: # 注册表损坏不阻断读内容,只在响应里暴露
logger.warning("categories.json 解析失败: %s", exc)
return []
return cats
def _read_doc(path: Path):
try:
text = path.read_text(encoding="utf-8")
except Exception as exc:
return None, f"读取失败: {exc}"
parsed = parse_front_matter(text)
if parsed is None:
return None, "缺少 front-matter(--- 包裹的 title/category/... )"
meta = parsed["meta"]
for field in ("title", "category", "summary"):
if not str(meta.get(field, "")).strip():
return None, f"front-matter 缺必填字段: {field}"
stat = path.stat()
return {
"slug": path.stem,
"title": str(meta["title"]),
"category": str(meta["category"]),
"summary": str(meta["summary"]),
"order": int(meta.get("order") or 0),
"icon": str(meta.get("icon") or ""),
"category_name": str(meta.get("category_name") or ""),
"bytes": stat.st_size,
"updated_at": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
"content": parsed["body"],
}, None
def load_docs() -> tuple:
"""实时读盘:返回 (docs, skipped, available)。docs 已按 (category order, doc order, slug) 排序"""
d = docs_dir()
if not d.is_dir():
return [], [], False
cats = _category_registry()
cat_order = {c.get("key"): int(c.get("order") or 0) for c in cats}
docs, skipped = [], []
for path in sorted(d.glob("*.md")):
if not SLUG_RE.match(path.stem):
skipped.append({"file": path.name, "reason": "文件名不符合 slug 规则 ^[a-z0-9][a-z0-9-]{0,63}$"})
continue
doc, err = _read_doc(path)
if err:
skipped.append({"file": path.name, "reason": err})
continue
docs.append(doc)
docs.sort(key=lambda x: (cat_order.get(x["category"], 10_000), x["category"], x["order"], x["slug"]))
return docs, skipped, True
def _categories_view(docs: list) -> list:
"""分类视图:按注册表 `order` 升序(未登记分类补在末尾,不丢内容)"""
cats = _category_registry()
seen, out, seq = set(), [], 0
for c in cats:
key = c.get("key")
if not key or key in seen:
continue
seen.add(key)
seq += 1
out.append({
"key": key,
"name": c.get("name") or key,
"icon": c.get("icon") or "",
"order": int(c.get("order") or 0),
"_seq": seq,
"count": len([d for d in docs if d["category"] == key]),
})
for d in docs:
if d["category"] in seen:
continue
seen.add(d["category"])
seq += 1
out.append({
"key": d["category"],
"name": d["category_name"] or d["category"],
"icon": d["icon"] or "",
"order": 10_000,
"_seq": seq,
"count": len([x for x in docs if x["category"] == d["category"]]),
})
out.sort(key=lambda c: (c["order"], c["_seq"]))
for c in out:
c.pop("_seq")
return out
def _meta_only(doc: dict) -> dict:
return {k: v for k, v in doc.items() if k != "content"}
# ────────────────────────── 端点(只新增)──────────────────────────
@router.get("")
def list_docs(
with_content: bool = Query(False, description="是否带上 md 正文(默认只返回元数据)"),
category: Optional[str] = Query(None, description="按分类 key 过滤"),
):
"""文档列表(实时读盘)。
- `data`:按「分类顺序 → 篇内 order → slug」排序的文档元数据(`with_content=true` 时含正文)
- `categories`:分类视图(name/icon/count),前端按它分组渲染
- `source`:内容目录绝对路径(机器侧取内容用同一来源,满足「机器走 API」)
- `skipped`:被跳过的 md 及原因(缺 front-matter/文件名非法),便于 Bot 自诊断
"""
docs, skipped, available = load_docs()
all_docs = docs
if category:
docs = [d for d in docs if d["category"] == category]
return {
"count": len(docs),
"available": available,
"source": str(docs_dir()),
"categories": _categories_view(all_docs), # 分类视图按全量内容统计(不受 category 过滤影响)
"data": docs if with_content else [_meta_only(d) for d in docs],
"skipped": skipped,
"format": "markdown",
}
@router.get("/{slug}")
def get_doc(slug: str):
"""单篇文档(含 md 正文,实时读盘)"""
if not SLUG_RE.match(slug or ""):
raise HTTPException(400, "无效的文档标识(只允许 [a-z0-9-],1-64 位,字母数字开头)")
path = docs_dir() / f"{slug}.md"
if not path.is_file():
raise HTTPException(404, "文档不存在")
doc, err = _read_doc(path)
if err:
raise HTTPException(422, f"文档格式不合法: {err}")
return doc
+2 -1
View File
@@ -7,7 +7,7 @@ from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from dotenv import load_dotenv
from app.database import init_db, get_db
from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury
from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure, cash_vouchers_json, ext_input_ledgers, treasury, doc_center
from app.utils.cache import clear_all as clear_cache, delete as delete_cache
from scripts.erp_sync import run_sync as run_erp_sync
from app.auth_middleware import require_auth, require_auth_401
@@ -100,6 +100,7 @@ app.include_router(tax_compliance.router)
app.include_router(verify.router)
app.include_router(ext_input_ledgers.router) # 外部输入台账(M2 准入3 派生,2026-09-15)
app.include_router(treasury.router) # 司库(资金)阶段一:账户池/头寸/内控闸门(2026-09-15)
app.include_router(doc_center.router) # 文档中心:md 单一来源(运行期读盘,2026-09-24)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
+215
View File
@@ -0,0 +1,215 @@
"""文档中心(md 单一来源)端点测试 —— 任务 yanxue-20260924-cma-docs-center-poc
覆盖:
1. 列表:元数据/分类视图/排序/默认不带正文、`with_content=true` 带正文
2. 单篇:200 / 未知 slug 404 / 非法 slug 400(含路径穿越)
3. 鉴权:无 token → 401(不得成为 B 类缺口);持 token → 200
4. 健壮性:缺 front-matter / 缺必填字段 → skipped 且不影响其它文档;目录缺失 → available=false 不 500
5. **核心判据**:改 md 后同一进程再次请求即返回新内容(运行期读盘、无缓存、无 build-time 打包)
6. 生产内容目录金丝雀(目录不存在时跳过)
"""
import json
import os
import pytest
from fastapi.testclient import TestClient
from tests.conftest import create_test_user, get_token_for_user, auth_header
DOCS_ENV = "CMA_DOCS_DIR"
PROD_CONTENT = "/srv/hermes/shared/cma-content"
def write_doc(root, slug, body="#### 标题\n\n正文段落。\n", title="标题", category="term",
summary="摘要", category_name="术语解释", icon="📖", order=10, extra=""):
path = root / "docs" / f"{slug}.md"
path.write_text(
"---\n"
f'title: "{title}"\n'
f"category: {category}\n"
f"category_name: {category_name}\n"
f'icon: "{icon}"\n'
f'summary: "{summary}"\n'
f"order: {order}\n"
f"{extra}"
"---\n" + body,
encoding="utf-8",
)
return path
@pytest.fixture
def content_dir(tmp_path, monkeypatch):
"""隔离的临时内容目录(不动生产目录)"""
root = tmp_path / "cma-content"
(root / "docs").mkdir(parents=True)
(root / "categories.json").write_text(json.dumps({"categories": [
{"key": "faq", "name": "常见问题", "icon": "❓", "order": 40},
{"key": "term", "name": "术语解释", "icon": "📖", "order": 10},
]}, ensure_ascii=False), encoding="utf-8")
monkeypatch.setenv(DOCS_ENV, str(root))
return root
@pytest.fixture
def token(client: TestClient, db) -> str:
create_test_user(db)
return get_token_for_user(client)
class TestListDocs:
def test_meta_only_and_category_order(self, client, content_dir, token):
write_doc(content_dir, "term-b", title="B术语", order=20)
write_doc(content_dir, "term-a", title="A术语", order=10)
write_doc(content_dir, "faq-x", title="常见问题X", category="faq",
category_name="常见问题", icon="❓", order=10)
r = client.get("/api/cma/docs", headers=auth_header(token))
assert r.status_code == 200, r.text
body = r.json()
assert body["count"] == 3 and body["available"] is True and body["format"] == "markdown"
assert body["source"].endswith("/docs")
# 分类顺序 = categories.json 的 order(term 10 在 faq 40 之前,与文件写入顺序无关)
assert [c["key"] for c in body["categories"]] == ["term", "faq"]
assert [c["count"] for c in body["categories"]] == [2, 1]
assert body["categories"][0]["name"] == "术语解释" and body["categories"][0]["icon"] == "📖"
# 文档顺序 = (分类 order, 篇内 order, slug)
assert [d["slug"] for d in body["data"]] == ["term-a", "term-b", "faq-x"]
assert "content" not in body["data"][0], "默认不得带正文"
assert set(body["data"][0]) >= {"slug", "title", "summary", "category",
"category_name", "icon", "order", "bytes", "updated_at"}
assert body["skipped"] == []
def test_with_content(self, client, content_dir, token):
write_doc(content_dir, "term-a", body="#### H\n\n**加粗** 正文\n")
r = client.get("/api/cma/docs?with_content=true", headers=auth_header(token))
assert r.status_code == 200
assert r.json()["data"][0]["content"] == "#### H\n\n**加粗** 正文\n"
def test_filter_by_category(self, client, content_dir, token):
write_doc(content_dir, "term-a")
write_doc(content_dir, "faq-x", category="faq", category_name="常见问题")
r = client.get("/api/cma/docs?category=faq", headers=auth_header(token))
body = r.json()
assert [d["slug"] for d in body["data"]] == ["faq-x"]
# 分类视图按全量统计,不受过滤影响
assert [c["count"] for c in body["categories"]] == [1, 1]
def test_unregistered_category_still_listed(self, client, content_dir, token):
write_doc(content_dir, "other-a", category="其他类", category_name="其他类")
r = client.get("/api/cma/docs", headers=auth_header(token))
body = r.json()
assert body["count"] == 1
assert [c["key"] for c in body["categories"]] == ["term", "faq", "其他类"]
class TestGetDoc:
def test_detail(self, client, content_dir, token):
write_doc(content_dir, "term-a", body="#### H\n\n正文x\n")
r = client.get("/api/cma/docs/term-a", headers=auth_header(token))
assert r.status_code == 200
d = r.json()
assert d["slug"] == "term-a" and d["title"] == "标题" and d["content"] == "#### H\n\n正文x\n"
def test_unknown_slug_404(self, client, content_dir, token):
r = client.get("/api/cma/docs/no-such-doc", headers=auth_header(token))
assert r.status_code == 404
@pytest.mark.parametrize("slug", ["Upper", "with space", "a@b", "-lead", "x" * 65])
def test_invalid_slug_400(self, client, content_dir, token, slug):
r = client.get(f"/api/cma/docs/{slug}", headers=auth_header(token))
assert r.status_code in (400, 404), f"非法 slug 不得进入文件系统查找: {slug} → {r.status_code}"
if r.status_code == 400:
assert "无效的文档标识" in r.text
def test_path_traversal_blocked(self, content_dir):
"""越权取文件:`../`、绝对路径、URL 编码一律被 slug 白名单拦下(无 HTTP 层绕行)"""
from app.api import doc_center
from fastapi import HTTPException
for bad in ["../categories", "..%2F..%2Fetc%2Fpasswd", "/etc/passwd", "%2e%2e%2fsecret",
"....//x", "a/b"]:
with pytest.raises(HTTPException) as ei:
doc_center.get_doc(bad)
assert ei.value.status_code == 400
# 正对照:合法 slug 走查文件路径(不存在 → 404,不是 400)
with pytest.raises(HTTPException) as ei:
doc_center.get_doc("term-a")
assert ei.value.status_code == 404
class TestAuth:
def test_no_token_401(self, client, content_dir):
assert client.get("/api/cma/docs").status_code == 401
assert client.get("/api/cma/docs/term-a").status_code == 401
def test_with_token_200(self, client, content_dir, token):
write_doc(content_dir, "term-a")
assert client.get("/api/cma/docs", headers=auth_header(token)).status_code == 200
class TestRobustness:
def test_bad_md_skipped_not_fatal(self, client, content_dir, token):
(content_dir / "docs" / "no-front.md").write_text("没有 front-matter\n", encoding="utf-8")
write_doc(content_dir, "missing-title", title="", summary="s")
(content_dir / "docs" / "BadSlug.md").write_text(
'---\ntitle: "T"\ncategory: term\nsummary: "S"\n---\nX\n', encoding="utf-8")
write_doc(content_dir, "term-good")
r = client.get("/api/cma/docs", headers=auth_header(token))
assert r.status_code == 200
body = r.json()
assert [d["slug"] for d in body["data"]] == ["term-good"]
files = {s["file"]: s["reason"] for s in body["skipped"]}
assert "no-front.md" in files and "front-matter" in files["no-front.md"]
assert "missing-title.md" in files and "title" in files["missing-title.md"]
assert "BadSlug.md" in files and "slug" in files["BadSlug.md"]
def test_missing_dir_not_500(self, client, tmp_path, monkeypatch, token):
monkeypatch.setenv(DOCS_ENV, str(tmp_path / "not-exist"))
r = client.get("/api/cma/docs", headers=auth_header(token))
assert r.status_code == 200
body = r.json()
assert body["count"] == 0 and body["available"] is False and body["data"] == []
def test_front_matter_multi_colon_and_quotes(self, content_dir):
from app.api import doc_center
p = write_doc(content_dir, "term-q", title='带:冒号"引号"的标题', summary="摘要: 含冒号")
doc, err = doc_center._read_doc(p)
assert err is None
assert doc["title"] == '带:冒号"引号"的标题' and doc["summary"] == "摘要: 含冒号"
class TestRuntimeSource:
"""核心判据:内容运行期读盘 —— 改 md 即生效,无需改前端 / rebuild / 重启"""
def test_edit_md_visible_on_next_request(self, client, content_dir, token):
p = write_doc(content_dir, "term-a", body="#### 旧内容\n")
r1 = client.get("/api/cma/docs/term-a", headers=auth_header(token)).json()
assert r1["content"] == "#### 旧内容\n"
# 只改 md 文件(不改码、不重启、不构建)
p.write_text(p.read_text(encoding="utf-8").replace("旧内容", "新内容"), encoding="utf-8")
r2 = client.get("/api/cma/docs/term-a", headers=auth_header(token)).json()
assert r2["content"] == "#### 新内容\n"
def test_add_md_visible_without_restart(self, client, content_dir, token):
assert client.get("/api/cma/docs", headers=auth_header(token)).json()["count"] == 0
write_doc(content_dir, "term-new", title="新增篇目")
body = client.get("/api/cma/docs?with_content=true", headers=auth_header(token)).json()
assert body["count"] == 1 and body["data"][0]["title"] == "新增篇目"
@pytest.mark.skipif(not os.path.isdir(PROD_CONTENT), reason="生产内容目录不存在(非本机环境)")
class TestProdContentCanary:
"""生产内容目录金丝雀:真实目录必须可读、无 skipped、分类齐备(防内容被写坏)"""
def test_prod_content_readable(self, client, token, monkeypatch):
monkeypatch.setenv(DOCS_ENV, PROD_CONTENT)
r = client.get("/api/cma/docs", headers=auth_header(token))
assert r.status_code == 200
body = r.json()
assert body["available"] is True and body["skipped"] == [], body["skipped"]
assert body["count"] >= 20, f"迁移应有 20 篇,实际 {body['count']}"
assert [c["key"] for c in body["categories"]] == ["term", "formula", "practice", "faq"]
for slug in ("term-kpi", "formula-achievement-rate", "practice-threshold-tips", "faq-data-sync"):
d = client.get(f"/api/cma/docs/{slug}", headers=auth_header(token)).json()
assert d["title"] and d["content"].strip()