252 lines
9.3 KiB
Python
252 lines
9.3 KiB
Python
# ///
|
|
# 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": "未知操作"}
|