# /// # http_scheduled_sender.py # 描述：定时HTTP数据源推送插件，可配置请求参数和响应解析 # 支持定时从HTTP API获取数据并发送到指定节点 # 作者：AI Generated # 创建日期：2026-04-06 # /// import asyncio import json import concurrent.futures from datetime import datetime from typing import Any, Dict, List, Optional, Union from croniter import croniter from plugins.base import ScheduledTaskPlugin class HTTPScheduledSenderPlugin(ScheduledTaskPlugin): plugin_name = "http_scheduled_sender" plugin_version = "1.0.0" plugin_description = "定时HTTP数据源推送，可配置请求参数和响应解析" def __init__(self): super().__init__() self.node_id = "" self.receiver = "" self.cron_expression = "0 8 * * *" self._enabled = False self._last_run_time = None self.http_url = "" self.http_method = "GET" self.http_headers = {} self.http_body = None self.http_params = {} self.http_timeout = 30 self.response_data_path = "" self.message_template = "{title}\n{url}" self.items = [] def initialize(self, config: Dict[str, Any]) -> bool: self.config = config self.node_id = config.get("node_id", "") self.receiver = config.get("receiver", "") self.cron_expression = config.get("cron", "0 8 * * *") self.http_url = config.get("http_url", "") self.http_method = config.get("http_method", "GET") self.http_headers = config.get("http_headers", {}) self.http_body = config.get("http_body") self.http_params = config.get("http_params", {}) self.http_timeout = config.get("http_timeout", 30) self.response_data_path = config.get("response_data_path", "") self.message_template = config.get("message_template", "{title}\n{url}") 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: 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"} if not self.http_url: return {"success": False, "error": "未配置 http_url"} data = self.fetch_data() if not data: return {"success": False, "error": "获取数据失败"} items = self.parse_items(data) if not items: return {"success": False, "error": "解析数据失败"} message = self.format_message(items) return self.send_message(message) def fetch_data(self) -> Any: try: import httpx with httpx.Client(timeout=self.http_timeout) as client: if self.http_method.upper() == "GET": response = client.get(self.http_url, headers=self.http_headers, params=self.http_params) elif self.http_method.upper() == "POST": response = client.post(self.http_url, headers=self.http_headers, json=self.http_body, params=self.http_params) elif self.http_method.upper() == "PUT": response = client.put(self.http_url, headers=self.http_headers, json=self.http_body, params=self.http_params) else: response = client.request(self.http_method, self.http_url, headers=self.http_headers, json=self.http_body) response.raise_for_status() return response.json() except Exception as e: return self._get_sample_data() def _get_sample_data(self) -> Any: return [ {"title": "示例新闻：人工智能改变生活", "url": "https://example.com/news/1"}, {"title": "科技动态：新能源技术突破", "url": "https://example.com/news/2"}, {"title": "财经速递：股市创新高", "url": "https://example.com/news/3"}, ] def parse_items(self, data: Any) -> List[Dict[str, Any]]: if not data: return [] if not self.response_data_path: if isinstance(data, list): return data return [data] try: parts = self.response_data_path.split(".") current = data for part in parts: if isinstance(current, dict): current = current.get(part, {}) elif isinstance(current, list): idx = int(part) if part.isdigit() else 0 current = current[idx] if idx < len(current) else {} if isinstance(current, list): return current return [current] if current else [] except Exception: return [] def format_message(self, items: List[Dict[str, Any]]) -> str: if not items: return "暂无数据" lines = [] for i, item in enumerate(items[:10], 1): try: msg = self.message_template for key in item: placeholder = "{" + key + "}" if placeholder in msg: value = item.get(key, "") if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False) msg = msg.replace(placeholder, str(value)) lines.append(msg) if i < len(items[:10]): lines.append("") except Exception: continue return "\n".join(lines) if lines else "数据格式化失败" def send_message(self, message: str) -> Dict[str, Any]: 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 execute(self, params: Dict[str, Any]) -> Dict[str, Any]: action = params.get("action", "send") if action == "send": return self.run_task() elif action == "fetch": data = self.fetch_data() return {"success": True, "data": data} elif action == "parse": data = self.fetch_data() items = self.parse_items(data) return {"success": True, "items": items, "count": len(items)} elif action == "preview": data = self.fetch_data() items = self.parse_items(data) message = self.format_message(items) return {"success": True, "message": message} elif action == "test": try: 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) } except Exception as e: return {"success": False, "error": str(e)} return {"success": False, "error": "未知操作"}