# /// # news_plugin.py # 描述:新闻定时推送插件,定时获取新闻并发送到指定节点,支持 cron 表达式 # 作者:AI Generated # 创建日期:2026-04-06 # /// import asyncio import time import concurrent.futures from datetime import datetime from typing import Any, Dict, List from croniter import croniter from plugins.base import ScheduledTaskPlugin class NewsScheduledPlugin(ScheduledTaskPlugin): plugin_name = "news_scheduled" plugin_version = "1.0.0" plugin_description = "新闻定时推送插件,支持 cron 表达式定时发送新闻到微信" def __init__(self): super().__init__() self.node_id = "" self.receiver = "" self.news_count = 5 self.cron_expression = "0 8 * * *" self.news_api_url = "https://news.runoob.com/api/news" self._enabled = False self._last_run_time = None def initialize(self, config: Dict[str, Any]) -> bool: self.config = config self.node_id = config.get("node_id", "") self.receiver = config.get("receiver", "") self.news_count = config.get("news_count", 5) self.cron_expression = config.get("cron", "0 8 * * *") self.news_api_url = config.get("news_api_url", "https://news.runoob.com/api/news") return True def enable(self): self._enabled = True self.enabled = True def disable(self): self._enabled = False self.enabled = False def get_cron_expression(self) -> str: return self.cron_expression def should_run_and_get_next(self) -> tuple: try: now = datetime.now() cron = croniter(self.cron_expression, now) prev_run = cron.get_prev(datetime) next_run = cron.get_next(datetime) if self._last_run_time is None or self._last_run_time != prev_run: self._last_run_time = prev_run return True, next_run return False, next_run except Exception as e: return False, None def run_task(self) -> Dict[str, Any]: if not self.node_id or not self.receiver: return {"success": False, "error": "未配置 node_id 或 receiver"} news = self.fetch_news() if not news: return {"success": False, "error": "获取新闻失败"} message = self.format_news_message(news) try: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(self._send_message_sync, message) return future.result(timeout=30) except Exception as e: return {"success": False, "error": str(e)} def _send_message_sync(self, message: str) -> Dict[str, Any]: from services.node_manager import node_manager from database_service import get_db_service db_service = get_db_service() if db_service: node = db_service.get_node(self.node_id) if node: existing = node_manager.nodes.get(self.node_id) if existing: existing.api_url = node.api_url existing.api_key = node.api_key else: from services.node_manager import Node new_node = Node( node_id=node.node_id, name=node.name, api_url=node.api_url, api_key=node.api_key, description=node.description or "", group=node.group or "default" ) node_manager.nodes[node.node_id] = new_node try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( node_manager.send_message( node_id=self.node_id, who=self.receiver, msg=message, msg_type="text" ) ) finally: loop.close() return result except Exception as e: return {"success": False, "error": str(e)} def fetch_news(self) -> List[Dict[str, Any]]: try: import httpx response = httpx.get(self.news_api_url, timeout=10) if response.status_code == 200: data = response.json() if isinstance(data, list): return data[:self.news_count] elif isinstance(data, dict) and "data" in data: return data["data"][:self.news_count] return self._get_sample_news() except Exception as e: return self._get_sample_news() def _get_sample_news(self) -> List[Dict[str, Any]]: return [ {"title": "科技新闻:人工智能新突破", "url": "https://example.com/tech-ai"}, {"title": "财经:股市今日上涨", "url": "https://example.com/stock"}, {"title": "体育:足球比赛精彩回顾", "url": "https://example.com/football"}, {"title": "娱乐:电影票房排行", "url": "https://example.com/movie"}, {"title": "社会:民生政策解读", "url": "https://example.com/policy"}, ][:self.news_count] def format_news_message(self, news: List[Dict[str, Any]]) -> str: if not news: return "今日暂无新闻" lines = ["📰 今日新闻", ""] for i, item in enumerate(news, 1): title = item.get("title", item.get("name", "无标题")) url = item.get("url", "") lines.append(f"{i}. {title}") if url: lines.append(f" {url}") lines.append("") return "\n".join(lines) def execute(self, params: Dict[str, Any]) -> Dict[str, Any]: action = params.get("action", "send") if action == "send": return self.run_task() elif action == "fetch": news = self.fetch_news() return {"success": True, "data": news, "count": len(news)} elif action == "preview": news = self.fetch_news() message = self.format_news_message(news) return {"success": True, "message": message} elif action == "test": cron = croniter(self.cron_expression, datetime.now()) prev_run = cron.get_prev(datetime) next_run = cron.get_next(datetime) return { "success": True, "cron": self.cron_expression, "prev_run": str(prev_run), "next_run": str(next_run) } return {"success": False, "error": "未知操作"}