osiman/reader
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add run-state persistence for FreshRSS pipeline

Browse Source
This commit is contained in:
root
2026-04-07 10:48:00 +08:00
parent 3a85d47f00
commit 72a6853c03
5 changed files with 802 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/summary_mcp/workflows/freshrss_pipeline.py
+421 -249
View File
@@ -1,14 +1,8 @@
from __future__ import annotations
# FreshRSS 全链路管道：拉取未读条目 -> 内容提取 -> LLM 摘要 -> 规则过滤 ->
# 构建 OpenClaw delivery payload -> 写盘 -> 词元统计 -> 标记已读。
# 生产入口：run_freshrss_pipeline()，由 MCP 工具 run_freshrss_openclaw_pipeline 调用。
import json
import os
from datetime import date, datetime, timezone
UTC = timezone.utc
from pathlib import Path
from typing import Any
@@ -32,8 +26,10 @@ from summary_mcp.models.openclaw_delivery import (
build_openclaw_digest_brief,
)
from summary_mcp.models.summary_io import ExtractionInput
from summary_mcp.runtime import RunStore
UTC = timezone.utc
REPO_ROOT = Path(__file__).resolve().parents[3]
OUTPUT_ROOT = REPO_ROOT / "outputs"
FRESHRSS_OUTPUT_ROOT = OUTPUT_ROOT / "freshrss"
@@ -44,6 +40,14 @@ DEFAULT_TERM_ALIASES_PATH = REPO_ROOT / "configs" / "term_aliases.json"
DEFAULT_TERM_STOPWORDS_PATH = REPO_ROOT / "configs" / "term_stopwords.json"
DEFAULT_TERM_DAILY_DIR = DATA_ROOT / "daily"
DEFAULT_TERM_STATS_PATH = DATA_ROOT / "term_stats.json"
WORKFLOW_NAME = "freshrss_daily_digest"
RUN_TYPE = "daily_digest"
FETCH_STAGE = "fetch_feed"
EXTRACT_STAGE = "extract_articles"
SUMMARY_STAGE = "generate_summaries"
FILTER_STAGE = "apply_filters"
DELIVERY_STAGE = "build_delivery_payload"
REPORT_STAGE = "write_run_report"
def _save_json(path: Path, payload: dict[str, Any] | list[Any]) -> None:
@@ -56,7 +60,6 @@ def _load_json(path: Path) -> dict[str, Any]:
def _load_required_env(name: str, value: str | None) -> str:
# 优先使用传入的 value，否则读取同名环境变量；两者均缺失时抛出 RuntimeError
if value:
return value
env_value = os.environ.get(name)
@@ -75,25 +78,7 @@ def _maybe_path(enabled: bool, path: Path) -> Path | None:
return path if enabled else None
def _process_item(
*,
index: int,
item: Any,
resolved_output_dir: Path,
resolved_prompt_path: Path,
resolved_run_id: str,
debug_artifacts: bool,
loaded_rules: list,
filter_context: Any,
max_retries: int,
timeout_seconds: float,
resolved_llm_api_key: str,
resolved_llm_model: str,
resolved_llm_api_url: str,
) -> dict[str, Any]:
# 处理单条 item：提取 -> LLM 摘要 -> 规则过滤 -> 候选构建
# 返回 item_report dict；delivered 时额外携带 _candidate/_external_id 供调用方解包
# 步骤 1：确定各中间文件路径（debug_artifacts=False 时大部分路径为 None，不写盘）
def _build_item_context(*, index: int, item: Any, resolved_output_dir: Path, debug_artifacts: bool) -> dict[str, Any]:
item_key = f"item-{index:02d}"
item_path = _maybe_path(debug_artifacts, resolved_output_dir / "items" / f"{item_key}.item.json")
extracted_path = resolved_output_dir / "extracted" / f"{item_key}.extracted.json"
@@ -102,10 +87,6 @@ def _process_item(
record_path = _maybe_path(debug_artifacts, resolved_output_dir / "candidates" / f"{item_key}.article-candidate-record.json")
openclaw_path = _maybe_path(debug_artifacts, resolved_output_dir / "candidates" / f"{item_key}.openclaw-candidate-input.json")
if item_path is not None:
_save_json(item_path, item.model_dump(mode="json"))
# 步骤 2：初始化 item_report，记录基础元信息；debug 模式下附加各中间文件路径
item_report: dict[str, Any] = {
"item_key": item_key,
"item_id": item.item_id,
@@ -117,115 +98,27 @@ def _process_item(
if debug_artifacts:
item_report["paths"] = {
"item": str(item_path) if item_path else None,
"extracted": str(extracted_path) if extracted_path else None,
"extracted": str(extracted_path),
"summary": str(summary_output) if summary_output else None,
"filter": str(filter_path) if filter_path else None,
"article_candidate": str(record_path) if record_path else None,
"openclaw_candidate": str(openclaw_path) if openclaw_path else None,
}
# 步骤 3：内容提取（RSS 内联内容 or 回源抓取）；extracted.json 始终写盘
extraction = extract_content(ExtractionInput(item=item))
extracted_payload = extraction.model_dump(mode="json")
if extracted_path is not None:
_save_json(extracted_path, extracted_payload)
if not extraction.success or extraction.article is None:
item_report["status"] = "extract_failed"
item_report["error"] = extraction.error.model_dump(mode="json") if extraction.error else None
return item_report
# 步骤 4：LLM 摘要循环，失败时最多重试 max_retries 次
item_report["status"] = "extracted"
summary_exit_code, summary_payload, summary_report = run_loop_payload(
extracted_payload=extracted_payload,
prompt_path=resolved_prompt_path,
output_path=summary_output,
max_retries=max_retries,
timeout_seconds=timeout_seconds,
api_key=resolved_llm_api_key,
model=resolved_llm_model,
api_url=resolved_llm_api_url,
)
if summary_exit_code != 0 or summary_payload is None:
item_report["status"] = "summary_failed"
if summary_report is not None:
item_report["summary_errors"] = summary_report.errors
return item_report
# 步骤 5：规则引擎过滤，产出 keep/review/drop 决策及 digest_rank
summary = LlmSummaryResult.model_validate(summary_payload)
decision = evaluate_filter_rules(
FilterInput(item=item, article=extraction.article, summary=summary, context=filter_context),
loaded_rules,
)
if filter_path is not None:
_save_json(filter_path, decision.model_dump(mode="json"))
# 步骤 6：构建 ArticleCandidateRecord 和 OpenClawCandidateInput
record = build_article_candidate_record(
summary=summary,
article=extraction.article,
filter_result=decision,
item=item,
source_refs=CandidateSourceRefs(
item_path=str(item_path) if item_path else None,
extracted_path=str(extracted_path) if extracted_path else None,
summary_path=str(summary_output) if summary_output else None,
filter_path=str(filter_path) if filter_path else None,
),
metadata=CandidateMetadata(
generated_at=datetime.now(tz=UTC),
producer="run_freshrss_pipeline",
run_id=resolved_run_id,
),
)
openclaw_input = build_openclaw_candidate_input(record)
if record_path is not None:
_save_json(record_path, record.model_dump(mode="json"))
if openclaw_path is not None:
_save_json(openclaw_path, openclaw_input.model_dump(mode="json"))
# 步骤 7：标记 delivered，附加临时键 _candidate/_external_id 供主函数解包后加入投递列表
item_report["status"] = "delivered"
item_report["selection_decision"] = decision.decision
item_report["candidate_id"] = openclaw_input.candidate_id
item_report["_candidate"] = openclaw_input
item_report["_external_id"] = item.external_id
return item_report
def _build_and_persist_delivery(
*,
delivered_candidates: list[OpenClawCandidateInput],
resolved_run_id: str,
resolved_delivery_date: Any,
delivery_output: Path,
) -> tuple[OpenClawDeliveryPayload, Path, dict[str, Any]]:
# 按 digest_rank 排序，构建 delivery payload；同时派生一个更轻量的 digest-brief.json 供日报生成使用。
delivered_candidates.sort(key=lambda candidate: candidate.digest_rank, reverse=True)
delivery_payload = build_openclaw_delivery_payload(
delivered_candidates,
run_id=resolved_run_id,
for_date=resolved_delivery_date,
)
_save_json(delivery_output, delivery_payload.model_dump(mode="json"))
digest_brief = build_openclaw_digest_brief(delivery_payload)
digest_brief_output = delivery_output.with_name("digest-brief.json")
_save_json(digest_brief_output, digest_brief.model_dump(mode="json"))
keyword_index_result = persist_keyword_indexes(
delivery_payload.candidates,
for_date=delivery_payload.date,
digest_id=delivery_payload.run_id,
source="openclaw_delivery_payload",
daily_dir=DEFAULT_TERM_DAILY_DIR,
stats_path=DEFAULT_TERM_STATS_PATH,
aliases_path=DEFAULT_TERM_ALIASES_PATH,
stopwords_path=DEFAULT_TERM_STOPWORDS_PATH,
)
return delivery_payload, digest_brief_output, keyword_index_result
return {
"item": item,
"item_key": item_key,
"item_path": item_path,
"extracted_path": extracted_path,
"summary_output": summary_output,
"filter_path": filter_path,
"record_path": record_path,
"openclaw_path": openclaw_path,
"item_report": item_report,
"extraction": None,
"extracted_payload": None,
"summary_payload": None,
}
def _build_run_report(
@@ -233,8 +126,8 @@ def _build_run_report(
resolved_run_id: str,
started_at: datetime,
limit: int,
items: list,
delivered_candidates: list,
items: list[Any],
delivered_candidates: list[OpenClawCandidateInput],
marked_count: int,
mark_read: bool,
debug_artifacts: bool,
@@ -244,7 +137,6 @@ def _build_run_report(
keyword_index_result: dict[str, Any],
item_reports: list[dict[str, Any]],
) -> dict[str, Any]:
# 统计各状态计数，组装 run report dict
status_counts: dict[str, int] = {}
for item_report in item_reports:
status = str(item_report["status"])
@@ -269,6 +161,12 @@ def _build_run_report(
}
def _final_run_status(item_reports: list[dict[str, Any]]) -> str:
if any(item_report.get("status") in {"extract_failed", "summary_failed"} for item_report in item_reports):
return "partial"
return "success"
def run_freshrss_pipeline(
*,
api_base_url: str | None = None,
@@ -293,139 +191,413 @@ def run_freshrss_pipeline(
delivery_date: date | None = None,
output_dir: Path | None = None,
) -> dict[str, Any]:
# --- 阶段 1：初始化 run_id、输出路径、凭证 ---
started_at = datetime.now(tz=UTC)
resolved_output_dir = output_dir or default_output_dir()
run_stamp = started_at.strftime("%Y%m%d-%H%M%S")
resolved_run_id = run_id or f"freshrss-pipeline-{run_stamp}"
resolved_delivery_date = delivery_date or datetime.now(tz=UTC).date()
resolved_api_base_url = _load_required_env("FRESHRSS_API_BASE_URL", api_base_url)
resolved_username = _load_required_env("FRESHRSS_USERNAME", username)
resolved_api_password = _load_required_env("FRESHRSS_API_PASSWORD", api_password)
resolved_llm_api_key, resolved_llm_model, resolved_llm_api_url = resolve_llm_settings(
api_key=llm_api_key,
model=llm_model,
api_url=llm_api_url,
)
resolved_prompt_path = prompt or DEFAULT_PROMPT_PATH
resolved_rules_path = rules or DEFAULT_RULES_PATH
raw_output = resolved_output_dir / "raw" / "freshrss.raw.json"
delivery_output = resolved_output_dir / "candidates" / "openclaw-delivery-payload.json"
digest_brief_output = delivery_output.with_name("digest-brief.json")
report_output = resolved_output_dir / "run-report.json"
run_state_output = resolved_output_dir / "run-state.json"
items_list_output = _maybe_path(debug_artifacts, resolved_output_dir / "items" / "freshrss.items.json")
# --- 阶段 2：登录 FreshRSS，拉取未读条目，写原始 payload ---
client = FreshRSSClient(
api_base_url=resolved_api_base_url,
username=resolved_username,
api_password=resolved_api_password,
timeout_seconds=timeout_seconds,
run_store = RunStore.create(
path=run_state_output,
run_id=resolved_run_id,
workflow=WORKFLOW_NAME,
run_type=RUN_TYPE,
started_at=started_at,
input_payload={
"limit": limit,
"mark_read": mark_read,
"include_read": include_read,
"debug_artifacts": debug_artifacts,
"continuation": continuation,
"stream_id": stream_id,
},
repo_root=REPO_ROOT,
)
auth_token = client.client_login()
payload = client.fetch_stream_contents(
auth_token=auth_token,
stream_id=stream_id,
limit=limit,
continuation=continuation,
exclude_targets=[] if include_read else [READ_TAG],
)
entries = payload.get("items")
if not isinstance(entries, list):
raise RuntimeError("FreshRSS stream response does not contain an items array.")
run_store.save()
_save_json(raw_output, payload)
items = [map_entry_to_item(entry) for entry in entries]
if items_list_output is not None:
_save_json(items_list_output, [item.model_dump(mode="json") for item in items])
loaded_rules = load_filter_rules(resolved_rules_path)
# context 优先使用直接传入的 dict，其次读取 context_path 文件，两者均缺失则使用空 context
if context is not None:
filter_context = FilterContext.model_validate(context)
elif context_path is not None:
filter_context = FilterContext.model_validate(_load_json(context_path))
else:
filter_context = FilterContext()
# --- 阶段 3：逐条处理（提取 -> LLM 摘要 -> 规则过滤 -> 候选构建） ---
client: FreshRSSClient | None = None
auth_token: str | None = None
items: list[Any] = []
item_contexts: list[dict[str, Any]] = []
item_reports: list[dict[str, Any]] = []
delivered_candidates: list[OpenClawCandidateInput] = []
delivered_item_ids: list[str] = []
item_reports: list[dict[str, Any]] = []
for index, item in enumerate(items, start=1):
result = _process_item(
index=index,
item=item,
resolved_output_dir=resolved_output_dir,
resolved_prompt_path=resolved_prompt_path,
resolved_run_id=resolved_run_id,
debug_artifacts=debug_artifacts,
loaded_rules=loaded_rules,
filter_context=filter_context,
max_retries=max_retries,
timeout_seconds=timeout_seconds,
resolved_llm_api_key=resolved_llm_api_key,
resolved_llm_model=resolved_llm_model,
resolved_llm_api_url=resolved_llm_api_url,
)
if result.get("status") == "delivered":
delivered_candidates.append(result.pop("_candidate"))
external_id = result.pop("_external_id", None)
if external_id:
delivered_item_ids.append(external_id)
item_reports.append(result)
# --- 阶段 4+5：构建 delivery payload 并持久化词元索引 ---
delivery_payload, digest_brief_output, keyword_index_result = _build_and_persist_delivery(
delivered_candidates=delivered_candidates,
resolved_run_id=resolved_run_id,
resolved_delivery_date=resolved_delivery_date,
delivery_output=delivery_output,
)
# --- 阶段 6：标记已读，汇总报告，返回结果 ---
keyword_index_result: dict[str, Any] = {}
marked_count = 0
if mark_read and delivered_item_ids:
# 仅标记成功投递（delivered）的 item；drop/review 的 item 保持未读状态
client.mark_items_as_read(auth_token=auth_token, item_ids=delivered_item_ids)
marked_count = len({item_id for item_id in delivered_item_ids if item_id})
report = _build_run_report(
resolved_run_id=resolved_run_id,
started_at=started_at,
limit=limit,
items=items,
delivered_candidates=delivered_candidates,
marked_count=marked_count,
mark_read=mark_read,
debug_artifacts=debug_artifacts,
raw_output=raw_output,
delivery_output=delivery_output,
digest_brief_output=digest_brief_output,
keyword_index_result=keyword_index_result,
item_reports=item_reports,
)
_save_json(report_output, report)
try:
run_store.start_stage(FETCH_STAGE, outputs={"output_dir": str(resolved_output_dir)})
resolved_api_base_url = _load_required_env("FRESHRSS_API_BASE_URL", api_base_url)
resolved_username = _load_required_env("FRESHRSS_USERNAME", username)
resolved_api_password = _load_required_env("FRESHRSS_API_PASSWORD", api_password)
resolved_llm_api_key, resolved_llm_model, resolved_llm_api_url = resolve_llm_settings(
api_key=llm_api_key,
model=llm_model,
api_url=llm_api_url,
)
return {
"run_id": resolved_run_id,
"output_dir": str(resolved_output_dir),
"raw_output": str(raw_output),
"delivery_output": str(delivery_output),
"digest_brief_output": str(digest_brief_output),
"report_output": str(report_output),
"keyword_index": keyword_index_result,
"pulled_count": len(items),
"delivered_count": len(delivered_candidates),
"marked_read_count": marked_count,
"status_counts": report["status_counts"],
"debug_artifacts": debug_artifacts,
"delivery_payload": delivery_payload.model_dump(mode="json"),
"items": item_reports,
}
client = FreshRSSClient(
api_base_url=resolved_api_base_url,
username=resolved_username,
api_password=resolved_api_password,
timeout_seconds=timeout_seconds,
)
auth_token = client.client_login()
payload = client.fetch_stream_contents(
auth_token=auth_token,
stream_id=stream_id,
limit=limit,
continuation=continuation,
exclude_targets=[] if include_read else [READ_TAG],
)
entries = payload.get("items")
if not isinstance(entries, list):
raise RuntimeError("FreshRSS stream response does not contain an items array.")
_save_json(raw_output, payload)
run_store.register_artifact(name="raw_output", path=raw_output, kind="json", stage=FETCH_STAGE)
items = [map_entry_to_item(entry) for entry in entries]
if items_list_output is not None:
_save_json(items_list_output, [item.model_dump(mode="json") for item in items])
run_store.register_artifact(name="items_output", path=items_list_output, kind="json", stage=FETCH_STAGE)
loaded_rules = load_filter_rules(resolved_rules_path)
if context is not None:
filter_context = FilterContext.model_validate(context)
elif context_path is not None:
filter_context = FilterContext.model_validate(_load_json(context_path))
else:
filter_context = FilterContext()
run_store.finish_stage(
FETCH_STAGE,
outputs={
"pulled_count": len(items),
"raw_output": str(raw_output),
"items_output": str(items_list_output) if items_list_output else None,
},
)
run_store.start_stage(
EXTRACT_STAGE,
outputs={
"expected_items": len(items),
"completed_items": 0,
"success_count": 0,
"failed_count": 0,
},
)
extracted_success_count = 0
extracted_failed_count = 0
for index, item in enumerate(items, start=1):
item_context = _build_item_context(
index=index,
item=item,
resolved_output_dir=resolved_output_dir,
debug_artifacts=debug_artifacts,
)
item_contexts.append(item_context)
item_report = item_context["item_report"]
item_reports.append(item_report)
item_path = item_context["item_path"]
if item_path is not None:
_save_json(item_path, item.model_dump(mode="json"))
extraction = extract_content(ExtractionInput(item=item))
extracted_payload = extraction.model_dump(mode="json")
item_context["extraction"] = extraction
item_context["extracted_payload"] = extracted_payload
_save_json(item_context["extracted_path"], extracted_payload)
run_store.register_artifact(name="extracted_dir", path=resolved_output_dir / "extracted", kind="directory", stage=EXTRACT_STAGE)
if not extraction.success or extraction.article is None:
item_report["status"] = "extract_failed"
item_report["error"] = extraction.error.model_dump(mode="json") if extraction.error else None
extracted_failed_count += 1
else:
item_report["status"] = "extracted"
extracted_success_count += 1
run_store.update_stage(
EXTRACT_STAGE,
outputs={
"expected_items": len(items),
"completed_items": extracted_success_count + extracted_failed_count,
"success_count": extracted_success_count,
"failed_count": extracted_failed_count,
},
)
run_store.finish_stage(
EXTRACT_STAGE,
outputs={
"expected_items": len(items),
"completed_items": extracted_success_count + extracted_failed_count,
"success_count": extracted_success_count,
"failed_count": extracted_failed_count,
"extracted_dir": str(resolved_output_dir / "extracted"),
},
)
run_store.start_stage(
SUMMARY_STAGE,
outputs={
"expected_items": extracted_success_count,
"completed_items": 0,
"success_count": 0,
"failed_count": 0,
},
)
summary_success_count = 0
summary_failed_count = 0
summary_candidates = [ctx for ctx in item_contexts if ctx["extraction"] is not None and ctx["extraction"].success]
for item_context in summary_candidates:
item_report = item_context["item_report"]
summary_exit_code, summary_payload, summary_report = run_loop_payload(
extracted_payload=item_context["extracted_payload"],
prompt_path=resolved_prompt_path,
output_path=item_context["summary_output"],
max_retries=max_retries,
timeout_seconds=timeout_seconds,
api_key=resolved_llm_api_key,
model=resolved_llm_model,
api_url=resolved_llm_api_url,
)
if summary_exit_code != 0 or summary_payload is None:
item_report["status"] = "summary_failed"
if summary_report is not None:
item_report["summary_errors"] = summary_report.errors
summary_failed_count += 1
else:
item_context["summary_payload"] = summary_payload
item_report["status"] = "summarized"
summary_success_count += 1
run_store.update_stage(
SUMMARY_STAGE,
outputs={
"expected_items": extracted_success_count,
"completed_items": summary_success_count + summary_failed_count,
"success_count": summary_success_count,
"failed_count": summary_failed_count,
},
)
if debug_artifacts and (resolved_output_dir / "summary").exists():
run_store.register_artifact(name="summary_dir", path=resolved_output_dir / "summary", kind="directory", stage=SUMMARY_STAGE)
run_store.finish_stage(
SUMMARY_STAGE,
outputs={
"expected_items": extracted_success_count,
"completed_items": summary_success_count + summary_failed_count,
"success_count": summary_success_count,
"failed_count": summary_failed_count,
},
)
run_store.start_stage(
FILTER_STAGE,
outputs={
"expected_items": summary_success_count,
"completed_items": 0,
"candidate_count": 0,
"keep_count": 0,
"review_count": 0,
"drop_count": 0,
},
)
filter_completed_count = 0
keep_count = 0
review_count = 0
drop_count = 0
for item_context in [ctx for ctx in item_contexts if ctx["summary_payload"] is not None]:
item = item_context["item"]
item_report = item_context["item_report"]
extraction = item_context["extraction"]
summary = LlmSummaryResult.model_validate(item_context["summary_payload"])
decision = evaluate_filter_rules(
FilterInput(item=item, article=extraction.article, summary=summary, context=filter_context),
loaded_rules,
)
filter_path = item_context["filter_path"]
if filter_path is not None:
_save_json(filter_path, decision.model_dump(mode="json"))
record = build_article_candidate_record(
summary=summary,
article=extraction.article,
filter_result=decision,
item=item,
source_refs=CandidateSourceRefs(
item_path=str(item_context["item_path"]) if item_context["item_path"] else None,
extracted_path=str(item_context["extracted_path"]),
summary_path=str(item_context["summary_output"]) if item_context["summary_output"] else None,
filter_path=str(filter_path) if filter_path else None,
),
metadata=CandidateMetadata(
generated_at=datetime.now(tz=UTC),
producer="run_freshrss_pipeline",
run_id=resolved_run_id,
),
)
openclaw_input = build_openclaw_candidate_input(record)
item_context["candidate"] = openclaw_input
if item_context["record_path"] is not None:
_save_json(item_context["record_path"], record.model_dump(mode="json"))
if item_context["openclaw_path"] is not None:
_save_json(item_context["openclaw_path"], openclaw_input.model_dump(mode="json"))
item_report["status"] = "delivered"
item_report["selection_decision"] = decision.decision
item_report["candidate_id"] = openclaw_input.candidate_id
delivered_candidates.append(openclaw_input)
if item.external_id:
delivered_item_ids.append(item.external_id)
if decision.decision == "keep":
keep_count += 1
elif decision.decision == "review":
review_count += 1
elif decision.decision == "drop":
drop_count += 1
filter_completed_count += 1
run_store.update_stage(
FILTER_STAGE,
outputs={
"expected_items": summary_success_count,
"completed_items": filter_completed_count,
"candidate_count": len(delivered_candidates),
"keep_count": keep_count,
"review_count": review_count,
"drop_count": drop_count,
},
)
if debug_artifacts and (resolved_output_dir / "candidates").exists():
run_store.register_artifact(name="candidate_dir", path=resolved_output_dir / "candidates", kind="directory", stage=FILTER_STAGE)
run_store.finish_stage(
FILTER_STAGE,
outputs={
"expected_items": summary_success_count,
"completed_items": filter_completed_count,
"candidate_count": len(delivered_candidates),
"keep_count": keep_count,
"review_count": review_count,
"drop_count": drop_count,
},
)
run_store.start_stage(DELIVERY_STAGE, outputs={"candidate_count": len(delivered_candidates)})
delivered_candidates.sort(key=lambda candidate: candidate.digest_rank, reverse=True)
delivery_payload = build_openclaw_delivery_payload(
delivered_candidates,
run_id=resolved_run_id,
for_date=resolved_delivery_date,
)
_save_json(delivery_output, delivery_payload.model_dump(mode="json"))
run_store.register_artifact(name="delivery_payload", path=delivery_output, kind="json", stage=DELIVERY_STAGE)
digest_brief = build_openclaw_digest_brief(delivery_payload)
_save_json(digest_brief_output, digest_brief.model_dump(mode="json"))
run_store.register_artifact(name="digest_brief", path=digest_brief_output, kind="json", stage=DELIVERY_STAGE)
keyword_index_result = persist_keyword_indexes(
delivery_payload.candidates,
for_date=delivery_payload.date,
digest_id=delivery_payload.run_id,
source="openclaw_delivery_payload",
daily_dir=DEFAULT_TERM_DAILY_DIR,
stats_path=DEFAULT_TERM_STATS_PATH,
aliases_path=DEFAULT_TERM_ALIASES_PATH,
stopwords_path=DEFAULT_TERM_STOPWORDS_PATH,
)
run_store.register_artifact(
name="keyword_daily_index",
path=Path(str(keyword_index_result["daily_output"])),
kind="json",
stage=DELIVERY_STAGE,
)
run_store.register_artifact(
name="keyword_stats_index",
path=Path(str(keyword_index_result["stats_output"])),
kind="json",
stage=DELIVERY_STAGE,
)
run_store.finish_stage(
DELIVERY_STAGE,
outputs={
"candidate_count": len(delivered_candidates),
"delivery_output": str(delivery_output),
"digest_brief_output": str(digest_brief_output),
"keyword_daily_output": str(keyword_index_result["daily_output"]),
"keyword_stats_output": str(keyword_index_result["stats_output"]),
},
)
run_store.start_stage(REPORT_STAGE, outputs={"mark_read_requested": mark_read})
if mark_read and delivered_item_ids:
client.mark_items_as_read(auth_token=auth_token, item_ids=delivered_item_ids)
marked_count = len({item_id for item_id in delivered_item_ids if item_id})
report = _build_run_report(
resolved_run_id=resolved_run_id,
started_at=started_at,
limit=limit,
items=items,
delivered_candidates=delivered_candidates,
marked_count=marked_count,
mark_read=mark_read,
debug_artifacts=debug_artifacts,
raw_output=raw_output,
delivery_output=delivery_output,
digest_brief_output=digest_brief_output,
keyword_index_result=keyword_index_result,
item_reports=item_reports,
)
_save_json(report_output, report)
run_store.register_artifact(name="run_report", path=report_output, kind="json", stage=REPORT_STAGE)
run_store.finish_stage(
REPORT_STAGE,
outputs={
"marked_read_count": marked_count,
"report_output": str(report_output),
},
)
run_store.finish_run(status=_final_run_status(item_reports))
return {
"run_id": resolved_run_id,
"output_dir": str(resolved_output_dir),
"raw_output": str(raw_output),
"delivery_output": str(delivery_output),
"digest_brief_output": str(digest_brief_output),
"report_output": str(report_output),
"keyword_index": keyword_index_result,
"pulled_count": len(items),
"delivered_count": len(delivered_candidates),
"marked_read_count": marked_count,
"status_counts": report["status_counts"],
"debug_artifacts": debug_artifacts,
"delivery_payload": delivery_payload.model_dump(mode="json"),
"items": item_reports,
}
except Exception as error:
failed_stage = run_store.state.current_stage or FETCH_STAGE
run_store.fail_stage(failed_stage, error=error)
raise
def read_delivery_payload(path: Path) -> OpenClawDeliveryPayload: