62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SRC_ROOT = REPO_ROOT / "src"
|
|
|
|
if str(SRC_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(SRC_ROOT))
|
|
|
|
from summary_mcp.models.document import ExtractedArticle
|
|
from summary_mcp.models.filtering import FilterDecisionResult
|
|
from summary_mcp.models.item import Item
|
|
from summary_mcp.models.llm_result import LlmSummaryResult
|
|
from summary_mcp.models.sink import SinkInput, SinkMetadata
|
|
from summary_mcp.sinks.markdown import MarkdownSink
|
|
|
|
|
|
def _load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Write a filtered result into the Markdown sink.")
|
|
parser.add_argument("--summary", type=Path, required=True, help="Structured LLM summary JSON file")
|
|
parser.add_argument("--extracted", type=Path, required=True, help="Extracted article JSON file")
|
|
parser.add_argument("--filter", type=Path, required=True, help="Filter decision JSON file")
|
|
parser.add_argument("--item", type=Path, default=None, help="Optional normalized item JSON file")
|
|
parser.add_argument(
|
|
"--base-dir",
|
|
type=Path,
|
|
default=REPO_ROOT / "knowledge-base",
|
|
help="Markdown sink base directory",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
summary = LlmSummaryResult.model_validate(_load_json(args.summary))
|
|
extracted_payload = _load_json(args.extracted)
|
|
article = ExtractedArticle.model_validate(extracted_payload.get("article", extracted_payload))
|
|
decision = FilterDecisionResult.model_validate(_load_json(args.filter))
|
|
item = Item.model_validate(_load_json(args.item)) if args.item else None
|
|
|
|
sink_input = SinkInput(
|
|
item=item,
|
|
article=article,
|
|
summary=summary,
|
|
filter_decision=decision,
|
|
metadata=SinkMetadata(generated_at=datetime.now(tz=UTC), source="run_markdown_sink.py"),
|
|
)
|
|
result = MarkdownSink(args.base_dir).write(sink_input)
|
|
print(json.dumps(result.model_dump(mode="json"), ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|