cmbuyer/scripts/vikunja_export.py

#!/usr/bin/env python3
"""把 Vikunja 上的任务内容单向导出到 docs/tasks/T-XXX.md 的标记区块。

设计约束（改动前先读 docs/tasks/T-008.md 的「方案」与「边界」）：

- **单向**。本脚本只对 Vikunja 发 GET。任何写回路径都会重新引入两个权威，
  因此 http_get() 把方法写死为 GET，并拒绝其余方法；--selftest 会验证这一点。
- **只写标记区块内**。区块外是 git 原生内容（write_paths、context_ref、
  ## 边界），承载安全边界的审计链，脚本绝不触碰。
- **只用标准库**。T-002 完成前仓库没有 Python 虚拟环境，本脚本必须能用系统
  python3 直接跑通，和 validate_agent_context.py 一致。
- **未支持的 HTML 标签一律报错退出**，不静默丢弃。静默丢弃会让安全相关文字
  无声消失，与本项目 fail closed 的要求冲突。
"""

from __future__ import annotations

import argparse
import hashlib
import html
import json
import re
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from html.parser import HTMLParser
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent
MANIFEST = REPO_ROOT / "docs" / "agent-context.json"
ENV_CANDIDATES = (
    REPO_ROOT / "vikunja.env",
    Path.home() / ".config" / "vikunja" / "env",
    Path.home() / ".claude" / "vikunja.env",
)
TIMEOUT_SECONDS = 20

EXPORT_BEGIN_TEMPLATE = (
    "<!-- BEGIN VIKUNJA EXPORT id={task_id} synced={synced} sha256={sha256} -->"
)
EXPORT_END = "<!-- END VIKUNJA EXPORT -->"
EXPORT_BEGIN_RE = re.compile(
    r"<!-- BEGIN VIKUNJA EXPORT id=(?P<id>\d+) synced=(?P<synced>\S+) "
    r"sha256=(?P<sha>[0-9a-f]{64}) -->"
)


class ExportError(RuntimeError):
    """导出失败。一律以非零码退出，不产生半截文件。"""


# --------------------------------------------------------------------------
# 凭据与 HTTP（只读）
# --------------------------------------------------------------------------


def read_credentials() -> tuple[str, str]:
    for candidate in ENV_CANDIDATES:
        if candidate.is_file():
            values: dict[str, str] = {}
            for line in candidate.read_text(encoding="utf-8").splitlines():
                line = line.strip().lstrip("")
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, value = line.split("=", 1)
                values[key.strip().lower()] = value.strip()
            apiurl, apikey = values.get("apiurl"), values.get("apikey")
            if not apiurl or not apikey:
                raise ExportError(f"{candidate} 缺少 apiurl 或 apikey。")
            return apiurl.rstrip("/"), apikey
    raise ExportError(
        "找不到凭据文件，尝试过："
        + "、".join(str(p) for p in ENV_CANDIDATES)
        + "（参照 vikunja.env.example 创建）"
    )


def http_get(apiurl: str, apikey: str, path: str, method: str = "GET") -> Any:
    """只读取。method 参数存在是为了让「拒绝写方法」这条约束可被测试断言。"""
    if method != "GET":
        raise ExportError(
            f"本脚本只允许 GET，收到 {method}。"
            "写回 Vikunja 会重新引入两个权威，见 docs/tasks/T-008.md 的边界。"
        )
    request = urllib.request.Request(
        f"{apiurl}{path}",
        headers={"Authorization": f"Bearer {apikey}", "Accept": "application/json"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        raise ExportError(f"GET {path} 返回 HTTP {error.code}。") from error
    except (urllib.error.URLError, TimeoutError, OSError) as error:
        raise ExportError(
            f"GET {path} 连接失败：{error}。"
            "Vikunja 不可达时不导出，本地投影保持上一次的内容。"
        ) from error


# --------------------------------------------------------------------------
# HTML -> Markdown
# --------------------------------------------------------------------------

INLINE_MARKS = {
    "strong": "**",
    "b": "**",
    "em": "*",
    "i": "*",
    "del": "~~",
    "s": "~~",
}
BLOCK_TAGS = {
    "p", "h1", "h2", "h3", "h4", "h5", "h6", "ul", "ol", "li",
    "pre", "blockquote", "table", "thead", "tbody", "tr", "th", "td", "hr", "div",
}
SUPPORTED = BLOCK_TAGS | set(INLINE_MARKS) | {"a", "code", "br", "span"}


class _Node:
    __slots__ = ("tag", "attrs", "children")

    def __init__(self, tag: str, attrs: dict[str, str] | None = None) -> None:
        self.tag = tag
        self.attrs = attrs or {}
        self.children: list[Any] = []


class _TreeBuilder(HTMLParser):
    """把 Vikunja(TipTap) 产出的 HTML 建成树；遇到未知标签立即失败。"""

    VOID = {"br", "hr"}

    def __init__(self) -> None:
        super().__init__(convert_charrefs=True)
        self.root = _Node("#root")
        self.stack = [self.root]

    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        if tag not in SUPPORTED:
            raise ExportError(
                f"description 含未支持的 HTML 标签 <{tag}>。"
                "不静默丢弃：请在 scripts/vikunja_export.py 显式支持它，"
                "或改用已支持的写法。"
            )
        node = _Node(tag, {k: (v or "") for k, v in attrs})
        self.stack[-1].children.append(node)
        if tag not in self.VOID:
            self.stack.append(node)

    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
        self.handle_starttag(tag, attrs)

    def handle_endtag(self, tag: str) -> None:
        if tag in self.VOID:
            return
        for index in range(len(self.stack) - 1, 0, -1):
            if self.stack[index].tag == tag:
                del self.stack[index:]
                return

    def handle_data(self, data: str) -> None:
        self.stack[-1].children.append(data)


def _render_inline(node: _Node) -> str:
    out: list[str] = []
    for child in node.children:
        if isinstance(child, str):
            out.append(child.replace("\xa0", " "))
        elif child.tag == "br":
            out.append("\n")
        elif child.tag in INLINE_MARKS:
            inner = _render_inline(child).strip()
            out.append(f"{INLINE_MARKS[child.tag]}{inner}{INLINE_MARKS[child.tag]}" if inner else "")
        elif child.tag == "code":
            out.append(f"`{_render_inline(child)}`")
        elif child.tag == "a":
            text = _render_inline(child).strip()
            href = child.attrs.get("href", "")
            out.append(f"[{text}]({href})" if href else text)
        elif child.tag == "span":
            out.append(_render_inline(child))
        else:
            out.append(_render_block(child, 0))
    return "".join(out)


def _render_rows(node: _Node) -> list[list[str]]:
    rows: list[list[str]] = []
    for child in node.children:
        if isinstance(child, str):
            continue
        if child.tag in {"thead", "tbody"}:
            rows.extend(_render_rows(child))
        elif child.tag == "tr":
            cells = [
                _render_inline(cell).strip().replace("\n", " ").replace("|", "\\|")
                for cell in child.children
                if isinstance(cell, _Node) and cell.tag in {"th", "td"}
            ]
            rows.append(cells)
    return rows


def _render_block(node: _Node, depth: int) -> str:
    tag = node.tag
    if tag in {"p", "div"}:
        return _render_inline(node).strip()
    if tag in {"h1", "h2", "h3", "h4", "h5", "h6"}:
        return "#" * int(tag[1]) + " " + _render_inline(node).strip()
    if tag == "hr":
        return "---"
    if tag == "blockquote":
        inner = _render_children(node, depth)
        return "\n".join(f"> {line}" if line else ">" for line in inner.split("\n"))
    if tag == "pre":
        text = "".join(_collect_text(child) for child in node.children)
        return "```\n" + text.rstrip("\n") + "\n```"
    if tag in {"ul", "ol"}:
        items: list[str] = []
        counter = 0
        for child in node.children:
            if not isinstance(child, _Node) or child.tag != "li":
                continue
            counter += 1
            marker = "- " if tag == "ul" else f"{counter}. "
            body = _render_children(child, depth + 1).strip()
            pad = " " * len(marker)
            lines = body.split("\n")
            items.append(
                marker + lines[0]
                + "".join("\n" + (pad + line if line else "") for line in lines[1:])
            )
        return "\n".join(items)
    if tag == "table":
        rows = _render_rows(node)
        if not rows:
            return ""
        width = max(len(row) for row in rows)
        rows = [row + [""] * (width - len(row)) for row in rows]
        head, *body = rows
        out = ["| " + " | ".join(head) + " |",
               "| " + " | ".join(["---"] * width) + " |"]
        out.extend("| " + " | ".join(row) + " |" for row in body)
        return "\n".join(out)
    return _render_inline(node).strip()


def _collect_text(node: Any) -> str:
    if isinstance(node, str):
        return node
    return "".join(_collect_text(child) for child in node.children)


def _render_children(node: _Node, depth: int) -> str:
    blocks: list[str] = []
    inline_buffer: list[Any] = []

    def flush() -> None:
        if inline_buffer:
            holder = _Node("p")
            holder.children = list(inline_buffer)
            text = _render_inline(holder).strip()
            if text:
                blocks.append(text)
            inline_buffer.clear()

    for child in node.children:
        if isinstance(child, str):
            if child.strip():
                inline_buffer.append(child)
        elif child.tag in BLOCK_TAGS:
            flush()
            rendered = _render_block(child, depth)
            if rendered:
                blocks.append(rendered)
        else:
            inline_buffer.append(child)
    flush()
    return "\n\n".join(blocks)


def html_to_markdown(source: str) -> str:
    if not source or not source.strip():
        return ""
    builder = _TreeBuilder()
    builder.feed(source)
    builder.close()
    text = _render_children(builder.root, 0)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


# --------------------------------------------------------------------------
# 组装区块
# --------------------------------------------------------------------------


def build_block_body(task: dict[str, Any], comments: list[dict[str, Any]]) -> str:
    parts = [html_to_markdown(task.get("description") or "")]

    records: list[str] = []
    for comment in comments:
        author = (comment.get("author") or {}).get("username", "unknown")
        created = comment.get("created", "")
        body = html_to_markdown(comment.get("comment") or "")
        records.append(f"### {created} · {author}\n\n{body}".rstrip())

    parts.append("## 执行记录\n\n" + ("\n\n".join(records) if records else "（暂无）"))
    body = "\n\n".join(part for part in parts if part.strip())
    return body.strip("\n") + "\n"


def splice(original: str, task_id: int, body: str, synced: str) -> str:
    """替换标记区块，区块外内容原样保留。"""
    digest = hashlib.sha256(body.encode("utf-8")).hexdigest()
    begin = EXPORT_BEGIN_TEMPLATE.format(task_id=task_id, synced=synced, sha256=digest)
    block = f"{begin}\n{body}{EXPORT_END}\n"

    lines = original.splitlines(keepends=True)
    in_fence = False
    begin_idx = end_idx = None
    for index, line in enumerate(lines):
        if line.lstrip().startswith("```"):
            in_fence = not in_fence
            continue
        if in_fence:
            continue
        stripped = line.rstrip("\n")
        if begin_idx is None and EXPORT_BEGIN_RE.fullmatch(stripped):
            begin_idx = index
        elif stripped == EXPORT_END and end_idx is None:
            end_idx = index

    if begin_idx is None or end_idx is None or end_idx < begin_idx:
        raise ExportError(
            "目标文件缺少成对的导出标记。首次迁移时请先手工插入空区块："
            f"\n{EXPORT_BEGIN_TEMPLATE.format(task_id=task_id, synced=synced, sha256='0' * 64)}"
            f"\n{EXPORT_END}"
        )

    return "".join(lines[:begin_idx]) + block + "".join(lines[end_idx + 1 :])


def frontmatter_task_id(text: str) -> int | None:
    parts = text.split("---\n", 2)
    if len(parts) < 3:
        return None
    match = re.search(r"^vikunja_task_id:[ \t]*(\d+)", parts[1], re.MULTILINE)
    return int(match.group(1)) if match else None


def frontmatter_status(text: str) -> str | None:
    parts = text.split("---\n", 2)
    if len(parts) < 3:
        return None
    match = re.search(r"^status:[ \t]*(\S+)", parts[1], re.MULTILINE)
    return match.group(1) if match else None


def report_status_drift(path: Path, status: str | None, task: dict[str, Any]) -> None:
    """只读比对 frontmatter 的 status 与 Vikunja 的 done 标志。

    git 是 status 的权威，看板只是人类视图，所以这里只提示、不修正，
    更不会反向写回——反向同步会重新引入两个权威。
    """
    if status is None:
        return
    expected_done = status == "DONE"
    actual_done = bool(task.get("done"))
    if expected_done != actual_done:
        name = path.resolve().relative_to(REPO_ROOT)
        print(
            f"提示 {name}：frontmatter status={status}，但 Vikunja done={actual_done}。"
            "以 git 为准；看板卡片需要人工归位。",
            file=sys.stderr,
        )


# --------------------------------------------------------------------------
# 自检
# --------------------------------------------------------------------------


def selftest() -> int:
    failures: list[str] = []

    def check(name: str, actual: Any, expected: Any) -> None:
        if actual != expected:
            failures.append(f"{name}\n  实际: {actual!r}\n  期望: {expected!r}")

    check("段落与粗体", html_to_markdown("<p>纸面<b>价格</b></p>"), "纸面**价格**")
    check("行内代码", html_to_markdown("<p>调用 <code>SavePerson</code></p>"),
          "调用 `SavePerson`")
    check("链接", html_to_markdown('<p><a href="https://x/y">文档</a></p>'),
          "[文档](https://x/y)")
    check("无序列表", html_to_markdown("<ul><li><p>甲</p></li><li><p>乙</p></li></ul>"),
          "- 甲\n- 乙")
    check("有序列表", html_to_markdown("<ol><li><p>先</p></li><li><p>后</p></li></ol>"),
          "1. 先\n2. 后")
    check("标题", html_to_markdown("<h2>方案</h2>"), "## 方案")
    check("代码块", html_to_markdown("<pre><code>go test ./...</code></pre>"),
          "```\ngo test ./...\n```")
    check("表格",
          html_to_markdown("<table><tr><th>项</th><th>值</th></tr>"
                           "<tr><td>闸门</td><td>三道</td></tr></table>"),
          "| 项 | 值 |\n| --- | --- |\n| 闸门 | 三道 |")
    check("引用", html_to_markdown("<blockquote><p>不碰钱</p></blockquote>"), "> 不碰钱")
    check("换行", html_to_markdown("<p>上<br>下</p>"), "上\n下")
    check("空输入", html_to_markdown(""), "")

    # 未支持标签必须失败，不得静默丢弃。
    try:
        html_to_markdown("<p><marquee>价格</marquee></p>")
        failures.append("未支持标签 <marquee> 应当报错，但通过了。")
    except ExportError:
        pass

    # 只读约束：任何非 GET 方法必须被拒绝。
    for method in ("POST", "PUT", "PATCH", "DELETE"):
        try:
            http_get("http://unused.invalid", "tk_x", "/tasks/1", method=method)
            failures.append(f"{method} 应当被拒绝，但通过了。")
        except ExportError as error:
            if "只允许 GET" not in str(error):
                failures.append(f"{method} 被拒绝的原因不对：{error}")

    # status 漂移检测：只提示不修正，且不得反向写回。
    import io
    from contextlib import redirect_stderr

    for status, done, should_warn in (
        ("DONE", True, False), ("DOING", False, False),
        ("DONE", False, True), ("DOING", True, True),
    ):
        buffer = io.StringIO()
        with redirect_stderr(buffer):
            report_status_drift(REPO_ROOT / "docs/tasks/T-000.md", status, {"done": done})
        warned = "提示" in buffer.getvalue()
        if warned != should_warn:
            failures.append(
                f"漂移检测 status={status} done={done}：预期 {'提示' if should_warn else '静默'}，"
                f"实际 {'提示' if warned else '静默'}"
            )

    # 幂等：同样输入两次拼接结果必须一致。
    template = ("---\nid: T-999\nvikunja_task_id: 7\n---\n\n"
                f"{EXPORT_BEGIN_TEMPLATE.format(task_id=7, synced='x', sha256='0' * 64)}\n"
                f"旧内容\n{EXPORT_END}\n\n## 边界\n\n不可放宽。\n")
    first = splice(template, 7, "新内容\n", "2026-08-03T00:00:00Z")
    second = splice(first, 7, "新内容\n", "2026-08-03T00:00:00Z")
    check("幂等", first, second)
    if "## 边界" not in first or "不可放宽。" not in first:
        failures.append("区块外的 git 原生内容被破坏。")

    for failure in failures:
        print(f"FAIL {failure}", file=sys.stderr)
    print(f"自检：{'通过' if not failures else str(len(failures)) + ' 项失败'}")
    return 1 if failures else 0


# --------------------------------------------------------------------------


def export_file(path: Path, apiurl: str, apikey: str, check_only: bool) -> bool:
    text = path.read_text(encoding="utf-8")
    task_id = frontmatter_task_id(text)
    if task_id is None:
        return False

    # 有连接键但没有区块 = 刻意未迁移（例如内容已冻结的 DONE 任务）。
    # 与门禁的检查 4 保持一致：豁免而非报错。先判再联网，避免无谓请求。
    if not EXPORT_BEGIN_RE.search(text) and EXPORT_END not in text:
        print(f"跳过 {path.resolve().relative_to(REPO_ROOT)}"
              f"（vikunja#{task_id}，无导出区块，未迁移）")
        return False

    task = http_get(apiurl, apikey, f"/tasks/{task_id}")
    comments = http_get(apiurl, apikey, f"/tasks/{task_id}/comments") or []
    report_status_drift(path, frontmatter_status(text), task)
    body = build_block_body(task, comments)

    existing = EXPORT_BEGIN_RE.search(text)
    if existing and existing.group("sha") == hashlib.sha256(body.encode("utf-8")).hexdigest():
        return False  # 内容未变，不重写，避免噪音 diff。

    synced = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    updated = splice(text, task_id, body, synced)
    if check_only:
        print(f"DIFF {path.resolve().relative_to(REPO_ROOT)}")
        return True
    path.write_text(updated, encoding="utf-8", newline="\n")
    print(f"更新 {path.resolve().relative_to(REPO_ROOT)}（vikunja#{task_id}）")
    return True


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--file", type=Path, help="只导出指定任务文件")
    parser.add_argument("--check", action="store_true", help="只报告差异，不写文件")
    parser.add_argument("--selftest", action="store_true", help="离线自检，不访问网络")
    args = parser.parse_args()

    if args.selftest:
        return selftest()

    try:
        apiurl, apikey = read_credentials()
        targets = [args.file] if args.file else sorted(
            (REPO_ROOT / "docs" / "tasks").glob("T-*.md")
        )
        changed = sum(export_file(p, apiurl, apikey, args.check) for p in targets)
    except ExportError as error:
        print(f"ERROR: {error}", file=sys.stderr)
        return 1

    if args.check and changed:
        print(f"{changed} 个文件与 Vikunja 不同步。", file=sys.stderr)
        return 1
    print(f"完成：{changed} 个文件有变更。")
    return 0


if __name__ == "__main__":
    sys.exit(main())