feat(knowledge): 工单验收自动生成草稿，核对后发布，来源可追

- knowledge_doc 与 item 1:1，除 source_kind=manual 外必须有来源（库级约束） - 状态机 draft/in_review/published/archived，发布要 owner/admin；非法动作回 allowed_from 与 allowed_actions - 工单验收通过自动落 postmortem 草稿（generated_by=rule），同一来源只生成一次 - 接口：列表/详情/手记/改内容/状态动作/从事项沉淀；已发布禁改，改内容带乐观锁 - Web 知识页 + 详情抽屉；任务与工单详情里可「沉淀成知识」 - 8 个后端用例 + e2e 7 项断言（沉淀 → 待核对 → 发布）
2026-10-28 14:02:35 +08:00
parent 3471b07ac1
commit e9addeed25
17 changed files with 1957 additions and 1 deletions
@@ -0,0 +1,187 @@
+"""知识域状态机与"从事项生成草稿"的纯函数（M3）。
+
+为什么单独放一个模块：这两件事都是纯逻辑，不该缠在 HTTP 层或 ORM 里。
+状态机能被 5 行测试覆盖；草稿拼装能被快照测试比对。放进 API 模块就只能靠 e2e 兜。
+
+两个刻意的设计：
+
+1. **草稿默认落 `draft`，不自动发布**（用户拍板）。自动生成的东西一定会有"看着像那么回事、
+   其实漏了关键前提"的情况，人扫一眼的成本远低于让团队按错的结论办事。
+2. **草稿正文如实转述来源**，不写"建议""应该"这类推断。LLM 阶段（M4）接进来时，
+   `generated_by` 会从 `rule` 变成 `llm`，但"只转述、不改写结论"这条不变。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from ..errors import AppError, ErrorCode
+
+KNOWLEDGE_TRANSITIONS: dict[str, tuple[str, ...]] = {
+    "draft": ("in_review", "published", "archived"),
+    "in_review": ("draft", "published", "archived"),
+    "published": ("archived",),
+    "archived": ("draft",),
+}
+
+KNOWLEDGE_ACTIONS: dict[str, tuple[tuple[str, ...], str]] = {
+    # action: (允许的前置状态, 目标状态)
+    "submit_review": (("draft",), "in_review"),
+    "withdraw": (("in_review",), "draft"),
+    "publish": (("draft", "in_review"), "published"),
+    "archive": (("draft", "in_review", "published"), "archived"),
+    "restore": (("archived",), "draft"),
+}
+
+# 生成草稿时给读者的一句提示，防止有人把草稿当已确认的结论转发
+DRAFT_BANNER = "> 这份草稿由系统根据来源自动整理，**请先核对再发布**；发布后才会进入团队检索。"
+
+ACTION_LABEL: dict[str, str] = {
+    "created": "创建",
+    "status_changed": "状态变更",
+    "promoted_to_ticket": "升级成工单",
+    "ticket_created": "工单创建",
+    "ticket_accept": "受理",
+    "ticket_start": "开始办理",
+    "ticket_submit": "提交验收",
+    "ticket_approve": "验收通过",
+    "ticket_reject": "验收驳回",
+    "ticket_reassign": "改派",
+    "ticket_cancel": "取消",
+    "comment_added": "评论",
+    "knowledge_created": "沉淀为知识",
+}
+
+
+def target_status(action: str, current: str) -> str:
+    """动作 → 目标状态。非法组合抛 7003，并把"现在还能做什么"带回去。"""
+    rule = KNOWLEDGE_ACTIONS.get(action)
+    if rule is None:
+        raise AppError(
+            ErrorCode.KNOWLEDGE_STATUS_INVALID,
+            message=f"未知动作 {action}",
+            details={"action": action, "allowed_actions": sorted(KNOWLEDGE_ACTIONS)},
+        )
+    allowed_from, to_status = rule
+    if current not in allowed_from:
+        allowed_actions = [a for a, (src, _) in KNOWLEDGE_ACTIONS.items() if current in src]
+        raise AppError(
+            ErrorCode.KNOWLEDGE_STATUS_INVALID,
+            message=f"当前状态（{current}）不能{_label(action)}",
+            details={
+                "action": action,
+                "from": current,
+                "allowed_from": list(allowed_from),
+                "allowed_actions": allowed_actions,
+            },
+        )
+    return to_status
+
+
+def ensure_transition(current: str, target: str) -> None:
+    """直接改状态（不经过动作）时的校验。"""
+    if target not in KNOWLEDGE_TRANSITIONS.get(current, ()):
+        raise AppError(
+            ErrorCode.KNOWLEDGE_STATUS_INVALID,
+            message=f"不能从 {current} 直接变成 {target}",
+            details={
+                "from": current,
+                "to": target,
+                "allowed": list(KNOWLEDGE_TRANSITIONS.get(current, ())),
+            },
+        )
+
+
+def sidebar_fields(target_status: str, now: datetime) -> dict[str, Any]:
+    """状态跃迁时要顺手维护的时间戳。"""
+    if target_status == "published":
+        return {"published_at": now, "archived_at": None}
+    if target_status == "archived":
+        return {"archived_at": now}
+    return {}
+
+
+def can_publish_role(role: str) -> bool:
+    """发布是"团队背书"，guest/member 只能提交审议。"""
+    return role in ("owner", "admin")
+
+
+def _label(action: str) -> str:
+    return {
+        "submit_review": "提交审议",
+        "withdraw": "撤回审议",
+        "publish": "发布",
+        "archive": "归档",
+        "restore": "恢复",
+    }.get(action, action)
+
+
+def render_draft(
+    *,
+    source_kind: str,
+    title: str,
+    body_md: str = "",
+    resolution_md: str = "",
+    created_at: datetime | None = None,
+    closed_at: datetime | None = None,
+    people: dict[str, str] | None = None,
+    timeline: list[dict[str, Any]] | None = None,
+    comments: list[dict[str, Any]] | None = None,
+) -> str:
+    """把一件事（任务或工单）整理成知识草稿的正文。
+
+    只做"结构化的转述"：来源里有什么就写什么，没有的段落直接省略，
+    不补写"经验教训"这类需要人判断的话。
+    """
+    people = people or {}
+    out: list[str] = [DRAFT_BANNER, ""]
+
+    when = _fmt(closed_at or created_at)
+    origin = "工单" if source_kind == "ticket" else "任务"
+    out.append(f"自动沉淀自{origin}「{title}」{('，' + when + ' 结束') if when else ''}。")
+    out.append("")
+
+    if people:
+        line = "、".join(f"{k}：{v}" for k, v in people.items() if v)
+        if line:
+            out.append(f"- {line}")
+            out.append("")
+
+    out.append("## 背景")
+    out.append("")
+    out.append(
+        body_md.strip() if body_md.strip() else "（来源里没有正文，补充一句这件事要解决什么。）"
+    )
+    out.append("")
+
+    rows = [r for r in (timeline or []) if r.get("action") in ACTION_LABEL]
+    if rows:
+        out.append("## 过程")
+        out.append("")
+        for r in rows:
+            out.append(f"- {_fmt(r.get('at'))} {ACTION_LABEL[r['action']]}")
+        out.append("")
+
+    if resolution_md.strip():
+        out.append("## 结论")
+        out.append("")
+        out.append(resolution_md.strip())
+        out.append("")
+
+    said = [c for c in (comments or []) if (c.get("body") or "").strip()]
+    if said:
+        out.append("## 讨论")
+        out.append("")
+        for c in said:
+            who = c.get("author") or "某人"
+            out.append(f"- **{who}**：{c['body'].strip()}")
+        out.append("")
+
+    return "\n".join(out).rstrip() + "\n"
+
+
+def _fmt(value: Any) -> str:
+    if isinstance(value, datetime):
+        return value.strftime("%Y-%m-%d %H:%M")
+    return ""