Add markdown sink for filtered knowledge notes

src/summary_mcp/sinks/__init__.py

@@ -0,0 +1,3 @@
+from .markdown import MarkdownSink
+
+__all__ = ["MarkdownSink"]

src/summary_mcp/sinks/markdown.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from pathlib import Path
+import re
+
+from summary_mcp.models.sink import SinkInput, SinkResult
+
+
+_DECISION_DIR = {
+    "keep": "inbox",
+    "review": "review",
+    "drop": "archive",
+}
+
+
+def _slugify(text: str) -> str:
+    slug = text.lower()
+    slug = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", "-", slug)
+    slug = re.sub(r"-+", "-", slug).strip("-")
+    return slug or "untitled"
+
+
+def _yaml_scalar(value: str) -> str:
+    escaped = value.replace('"', '\\"')
+    return f'"{escaped}"'
+
+
+def _yaml_list(values: list[str]) -> str:
+    if not values:
+        return "[]"
+    return "\n".join(f"  - {_yaml_scalar(value)}" for value in values)
+
+
+def _frontmatter(payload: SinkInput) -> str:
+    article = payload.article
+    summary = payload.summary
+    decision = payload.filter_decision
+    generated_at = payload.metadata.generated_at or datetime.now(tz=UTC)
+    published_at = article.published_at.isoformat() if article.published_at else ""
+    source_id = article.source_id or (payload.item.source_id if payload.item else "")
+    item_id = article.item_id or (payload.item.item_id if payload.item else "")
+
+    return "\n".join(
+        [
+            "---",
+            f"title: {_yaml_scalar(summary.title)}",
+            f"url: {_yaml_scalar(str(summary.url))}",
+            f"source_id: {_yaml_scalar(source_id)}",
+            f"item_id: {_yaml_scalar(item_id)}",
+            f"extract_id: {_yaml_scalar(article.extract_id)}",
+            f"category: {_yaml_scalar(summary.category)}",
+            f"worth_keeping: {'true' if summary.worth_keeping else 'false'}",
+            f"decision: {_yaml_scalar(decision.decision)}",
+            f"priority: {decision.priority}",
+            "topics:",
+            _yaml_list(summary.topics),
+            "keywords:",
+            _yaml_list(summary.keywords),
+            "labels:",
+            _yaml_list(decision.labels),
+            f"published_at: {_yaml_scalar(published_at)}",
+            f"saved_at: {_yaml_scalar(generated_at.isoformat())}",
+            "quality_flags:",
+            f"  is_paywalled: {'true' if article.quality_flags.is_paywalled else 'false'}",
+            f"  is_truncated: {'true' if article.quality_flags.is_truncated else 'false'}",
+            f"  is_low_content: {'true' if article.quality_flags.is_low_content else 'false'}",
+            "---",
+        ]
+    )
+
+
+def _render_markdown(payload: SinkInput) -> str:
+    summary = payload.summary
+    article = payload.article
+    decision = payload.filter_decision
+    highlights = "\n".join(f"- {item}" for item in summary.highlights)
+    reasons = "\n".join(f"- {item}" for item in decision.reasons)
+
+    return (
+        f"{_frontmatter(payload)}\n\n"
+        f"# 摘要\n\n{summary.summary}\n\n"
+        f"# 要点\n\n{highlights}\n\n"
+        f"# 过滤决策\n\n"
+        f"- decision: `{decision.decision}`\n"
+        f"- priority: `{decision.priority}`\n"
+        f"- matched rules: {', '.join(decision.matched_rules) if decision.matched_rules else 'none'}\n\n"
+        f"## reasons\n\n{reasons}\n\n"
+        f"# 原文信息\n\n"
+        f"- 标题: {summary.title}\n"
+        f"- 链接: {summary.url}\n"
+        f"- 作者: {article.author or 'unknown'}\n"
+        f"- 发布时间: {article.published_at.isoformat() if article.published_at else 'unknown'}\n"
+        f"- 内容类型: {article.content_kind}\n\n"
+        f"# 正文摘录\n\n{article.plain_text.strip()}\n"
+    )
+
+
+class MarkdownSink:
+    def __init__(self, base_dir: Path) -> None:
+        self.base_dir = base_dir
+
+    def _target_dir(self, decision: str, published_at: datetime | None) -> Path:
+        bucket = _DECISION_DIR.get(decision, "archive")
+        if published_at is None:
+            return self.base_dir / bucket / "unknown-date"
+        return self.base_dir / bucket / published_at.strftime("%Y") / published_at.strftime("%Y-%m")
+
+    def _target_name(self, payload: SinkInput) -> str:
+        article = payload.article
+        published_at = article.published_at or datetime.now(tz=UTC)
+        title = payload.summary.title or article.title or "untitled"
+        return f"{published_at.strftime('%Y-%m-%d')}-{_slugify(title)}.md"
+
+    def write(self, payload: SinkInput) -> SinkResult:
+        target_dir = self._target_dir(payload.filter_decision.decision, payload.article.published_at)
+        target_dir.mkdir(parents=True, exist_ok=True)
+        output_path = target_dir / self._target_name(payload)
+        output_path.write_text(_render_markdown(payload), encoding="utf-8")
+        return SinkResult(
+            target="markdown",
+            success=True,
+            output_path=str(output_path),
+            message="Markdown note written successfully.",
+        )
+
+