from __future__ import annotations import importlib import io import sys from contextlib import redirect_stdout from datetime import datetime from pathlib import Path from time import perf_counter from typing import Any from backend.agents.base import BaseAgent from backend.config import settings from backend.events import event_logger from backend.maintenance import cleanup_rss_artifacts from backend.memory.base import memory_store from backend.models import AgentRequest, AgentResponse class RSSDigestAdapter(BaseAgent): """Expose rss_digest as one information/news platform agent.""" def run(self, request: AgentRequest) -> AgentResponse: event_logger.emit("agent_started", agent_id=self.agent_id, task_id=request.task_id) try: output, artifacts = self._run_with_artifacts(request) except Exception as exc: output = f"资讯员运行失败：{type(exc).__name__}: {exc}" artifacts = {"error": str(exc), "error_type": type(exc).__name__} print(f"[rss_digest][error] {output}") memory_store.add(self.agent_id, f"input={request.input} output={output}") event = event_logger.emit( "agent_completed", agent_id=self.agent_id, task_id=request.task_id, payload={ "output_preview": output[:200], "artifact_keys": sorted(artifacts.keys()), }, ) return AgentResponse( agent_id=self.agent_id, output=output, artifacts=artifacts, events=[event], ) def _run(self, request: AgentRequest) -> str: output, _ = self._run_with_artifacts(request) return output def _run_with_artifacts(self, request: AgentRequest) -> tuple[str, dict[str, Any]]: root_dir = Path(settings.rss_digest_root).resolve() data_root = Path(settings.rss_digest_data_root).resolve() cleanup_stats = cleanup_rss_artifacts() print(f"[rss_digest] start {datetime.now().isoformat(timespec='seconds')} input={request.input[:80]}") if not root_dir.exists(): message = f"rss_digest 项目路径不存在，无法运行资讯员：{root_dir}" print(f"[rss_digest][error] {message}") return message, { "ready": False, "rss_digest_root": str(root_dir), "rss_digest_data_root": str(data_root), "cleanup": cleanup_stats, } if request.context.get("mode") == "group_chat": digest_path = self._latest_digest_path(data_root) print("[rss_digest] skipped: group_chat guard") if digest_path: return ( f"资讯员已就绪。最新 RSS 简报：{digest_path}", { "skipped": True, "reason": "batch_guard", "digest_path": str(digest_path), "rss_digest_data_root": str(data_root), "cleanup": cleanup_stats, }, ) return ( "资讯员是较长耗时流程。请单独使用 @rss_digest 生成或更新 RSS 中文简报。", {"skipped": True, "reason": "batch_guard", "cleanup": cleanup_stats}, ) if request.context.get("dry_run"): print("[rss_digest] dry_run ok") return ( "资讯员已接入 rss_digest，真实运行会拉取 RSS、生成中文摘要并输出 HTML 简报。", { "ready": True, "rss_digest_root": str(root_dir), "rss_digest_data_root": str(data_root), "cleanup": cleanup_stats, }, ) modules = self._load_rss_modules(root_dir) force_refresh = bool(request.context.get("force_refresh")) or self._is_force_refresh(request.input) today_digest_path = self._today_digest_path(data_root) if today_digest_path and not force_refresh: print("[rss_digest] skipped: today digest exists") recent_articles = self._recent_articles(root_dir, data_root, modules, limit=8) digest_url = self._digest_url(today_digest_path) run_stats = { "skipped": True, "reason": "today_digest_exists", "digest_article_count": len(recent_articles), "llm_enabled": True, } return self._format_output(today_digest_path, digest_url, recent_articles, run_stats), { "skipped": True, "reason": "today_digest_exists", "rss_digest_root": str(root_dir), "rss_digest_data_root": str(data_root), "digest_path": str(today_digest_path), "digest_url": digest_url, "recent_articles": recent_articles, "run_stats": run_stats, "cleanup": cleanup_stats, } stdout_buffer = io.StringIO() print("[rss_digest] running pipeline") started = perf_counter() with redirect_stdout(stdout_buffer): run_stats = modules["pipeline"].run_pipeline(root_dir, data_root) run_stats["adapter_total_seconds"] = round(perf_counter() - started, 3) print( "[rss_digest] complete " f"discovered={run_stats.get('discovered', 0)} " f"extracted={run_stats.get('extracted', 0)} " f"summarized={run_stats.get('summarized', 0)} " f"digest_articles={run_stats.get('digest_article_count', 0)} " f"seconds={run_stats.get('adapter_total_seconds')}" ) digest_path = self._latest_digest_path(data_root) digest_url = self._digest_url(digest_path) recent_articles = self._recent_articles(root_dir, data_root, modules, limit=8) output = self._format_output(digest_path, digest_url, recent_articles, run_stats) artifacts = { "rss_digest_root": str(root_dir), "rss_digest_data_root": str(data_root), "digest_path": str(digest_path) if digest_path else None, "digest_url": digest_url, "recent_articles": recent_articles, "run_stats": run_stats, "stdout": stdout_buffer.getvalue().strip(), "cleanup": cleanup_stats, } return output, artifacts @staticmethod def _load_rss_modules(root_dir: Path) -> dict[str, Any]: src_dir = root_dir / "src" src_text = str(src_dir) if src_text not in sys.path: sys.path.insert(0, src_text) return { "pipeline": importlib.import_module("rss_digest.pipeline"), "config": importlib.import_module("rss_digest.config"), "db": importlib.import_module("rss_digest.db"), } @staticmethod def _latest_digest_path(data_root: Path) -> Path | None: digest_dir = data_root / "runs" / "digests" files = sorted(digest_dir.glob("digest_*.html"), key=lambda path: path.stat().st_mtime, reverse=True) return files[0] if files else None @staticmethod def _today_digest_path(data_root: Path) -> Path | None: digest_path = data_root / "runs" / "digests" / f"digest_{datetime.now().strftime('%Y-%m-%d')}.html" return digest_path if digest_path.exists() else None @staticmethod def _is_force_refresh(text: str) -> bool: normalized = text.lower() return any(token in normalized for token in ("强制", "重新生成", "刷新", "force", "refresh")) @staticmethod def _digest_url(digest_path: Path | None) -> str | None: if not digest_path: return None return f"/rss-digests/{digest_path.name}" @staticmethod def _recent_articles(root_dir: Path, data_root: Path, modules: dict[str, Any], limit: int) -> list[dict[str, Any]]: cfg = modules["config"].build_config(root_dir, data_root) conn = modules["db"].connect(cfg.db_path) modules["db"].init_db(conn) rows = modules["db"].get_recent_articles(conn, limit=limit) return [ { "title": row.get("title", ""), "source_name": row.get("source_name", ""), "category": row.get("category", ""), "published_at": row.get("published_at", ""), "link": row.get("link", ""), "article_score": row.get("article_score"), "one_line": row.get("one_line"), "worth_reading": row.get("worth_reading"), } for row in rows ] @staticmethod def _format_output( digest_path: Path | None, digest_url: str | None, articles: list[dict[str, Any]], run_stats: dict[str, Any] | None, ) -> str: lines = ["资讯员已完成 RSS 更新和中文摘要生成。"] if run_stats: lines.append( "本轮统计：" f"RSS新增 {run_stats.get('discovered', 0)}，" f"正文抽取 {run_stats.get('extracted', 0)}，" f"LLM摘要 {run_stats.get('summarized', 0)}，" f"本次简报文章 {run_stats.get('digest_article_count', 0)}，" f"LLM启用 {run_stats.get('llm_enabled', False)}。" ) if run_stats.get("no_new_articles"): lines.append("提示：本次没有新的未读文章进入简报，已避免重复展示今天看过的内容。") if run_stats.get("llm_enabled") and run_stats.get("summarized", 0) == 0: lines.append("提示：LLM 已配置，但本轮没有成功摘要新文章，可查看任务 artifacts 中的 stdout 和 run_stats。") if not run_stats.get("llm_enabled"): lines.append("提示：LLM 未启用，请检查 .env 中的 LLM_MODEL_ID、LLM_API_KEY、LLM_BASE_URL。") if digest_path: lines.append(f"最新 HTML 简报：{digest_path}") if digest_url: lines.append(f"点击打开：{digest_url}") if articles: lines.append("") lines.append("最新文章：") for index, article in enumerate(articles[:5], start=1): title = article.get("title") or "未命名文章" source = article.get("source_name") or "未知来源" score = article.get("article_score") score_text = f"，评分 {score}" if score is not None else "" lines.append(f"{index}. {title}，{source}{score_text}") one_line = article.get("one_line") if one_line: lines.append(f" {one_line}") return "\n".join(lines)