839 lines
32 KiB
Python
839 lines
32 KiB
Python
# ///
|
|
# advanced_sender_plugin.py
|
|
# 描述：高级多节点数据推送插件，支持表单化配置和变量赋值
|
|
# 功能：多数据源获取 → 去重入库 → 按记录时间路由发送到微信
|
|
# 作者：User
|
|
# 创建日期：2026-04-06
|
|
# requires: requests, pymysql
|
|
# ///
|
|
|
|
import os
|
|
import sys
|
|
import re
|
|
import time
|
|
import json
|
|
import logging
|
|
import asyncio
|
|
import threading
|
|
from datetime import datetime, time as dtime
|
|
from typing import Any, Dict, List, Optional
|
|
from contextlib import contextmanager
|
|
from croniter import croniter
|
|
from plugins.base import ScheduledTaskPlugin
|
|
from plugins.plugin_config_schema import (
|
|
PluginConfigSchema, ConfigSection, ConfigField, FieldType,
|
|
create_variable_assignment_field, cron_validator
|
|
)
|
|
|
|
PLUGIN_LOG_DIR = "/app/data/logs"
|
|
os.makedirs(PLUGIN_LOG_DIR, exist_ok=True)
|
|
|
|
plugin_logger = logging.getLogger("advanced_sender")
|
|
plugin_logger.setLevel(logging.INFO)
|
|
|
|
file_handler = logging.FileHandler(os.path.join(PLUGIN_LOG_DIR, "advanced_sender.log"), encoding='utf-8')
|
|
file_handler.setLevel(logging.INFO)
|
|
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
|
|
|
|
if not plugin_logger.handlers:
|
|
plugin_logger.addHandler(file_handler)
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.INFO)
|
|
plugin_logger.addHandler(console_handler)
|
|
|
|
logger = plugin_logger
|
|
|
|
|
|
def get_config_schema() -> PluginConfigSchema:
|
|
schema = PluginConfigSchema("advanced_sender")
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="basic",
|
|
label="基本设置",
|
|
fields=[
|
|
ConfigField(
|
|
name="cron",
|
|
label="执行周期",
|
|
field_type=FieldType.CRON,
|
|
required=True,
|
|
default="*/2 * * * *",
|
|
description="Cron表达式"
|
|
),
|
|
ConfigField(
|
|
name="enable_cron",
|
|
label="启用定时任务",
|
|
field_type=FieldType.BOOLEAN,
|
|
default=False
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="node",
|
|
label="默认设置（可被数据源覆盖）",
|
|
fields=[
|
|
ConfigField(
|
|
name="node_id",
|
|
label="默认节点ID",
|
|
field_type=FieldType.STRING,
|
|
default="wx1",
|
|
placeholder="如 wx1"
|
|
),
|
|
ConfigField(
|
|
name="receiver",
|
|
label="默认接收人",
|
|
field_type=FieldType.STRING,
|
|
default="asq",
|
|
placeholder="联系人名称"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="database",
|
|
label="外部数据库",
|
|
fields=[
|
|
ConfigField(
|
|
name="db_host",
|
|
label="数据库地址",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
default="192.168.2.27",
|
|
placeholder="192.168.1.100"
|
|
),
|
|
ConfigField(
|
|
name="db_port",
|
|
label="端口",
|
|
field_type=FieldType.NUMBER,
|
|
default=3306,
|
|
min_value=1,
|
|
max_value=65535
|
|
),
|
|
ConfigField(
|
|
name="db_user",
|
|
label="用户名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
default="root2"
|
|
),
|
|
ConfigField(
|
|
name="db_password",
|
|
label="密码",
|
|
field_type=FieldType.PASSWORD,
|
|
default="root@root"
|
|
),
|
|
ConfigField(
|
|
name="db_name",
|
|
label="数据库名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
default="addb"
|
|
),
|
|
ConfigField(
|
|
name="db_table",
|
|
label="表名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
default="user_data"
|
|
),
|
|
ConfigField(
|
|
name="sendto_field",
|
|
label="发送状态字段",
|
|
field_type=FieldType.STRING,
|
|
default="sendto"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="cookie_source1",
|
|
label="数据源1",
|
|
fields=[
|
|
ConfigField(
|
|
name="cookie_1",
|
|
label="Cookie",
|
|
field_type=FieldType.TEXTAREA,
|
|
placeholder="PHPSESSID=xxx; think_var=zh-cn; keeplogin=...",
|
|
description="数据源1的完整Cookie"
|
|
),
|
|
ConfigField(
|
|
name="node_id_1",
|
|
label="节点ID",
|
|
field_type=FieldType.STRING,
|
|
default="",
|
|
placeholder="留空使用默认节点"
|
|
),
|
|
ConfigField(
|
|
name="receiver_1",
|
|
label="接收人",
|
|
field_type=FieldType.STRING,
|
|
default="",
|
|
placeholder="留空使用默认接收人"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="cookie_source2",
|
|
label="数据源2",
|
|
fields=[
|
|
ConfigField(
|
|
name="cookie_2",
|
|
label="Cookie",
|
|
field_type=FieldType.TEXTAREA,
|
|
placeholder="PHPSESSID=xxx; think_var=zh-cn; keeplogin=..."
|
|
),
|
|
ConfigField(
|
|
name="node_id_2",
|
|
label="节点ID",
|
|
field_type=FieldType.STRING,
|
|
default="",
|
|
placeholder="留空使用默认节点"
|
|
),
|
|
ConfigField(
|
|
name="receiver_2",
|
|
label="接收人",
|
|
field_type=FieldType.STRING,
|
|
default="",
|
|
placeholder="留空使用默认接收人"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="fetch",
|
|
label="抓取设置",
|
|
fields=[
|
|
ConfigField(
|
|
name="fetch_timeout",
|
|
label="请求超时(秒)",
|
|
field_type=FieldType.NUMBER,
|
|
default=15,
|
|
min_value=5,
|
|
max_value=60
|
|
),
|
|
ConfigField(
|
|
name="send_interval",
|
|
label="数据源间隔(秒)",
|
|
field_type=FieldType.NUMBER,
|
|
default=20,
|
|
min_value=0,
|
|
max_value=60,
|
|
description="两个数据源之间的等待时间"
|
|
)
|
|
]
|
|
))
|
|
|
|
return schema
|
|
|
|
|
|
class CircuitBreaker:
|
|
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
|
|
self.failure_threshold = failure_threshold
|
|
self.timeout = timeout
|
|
self.failures = 0
|
|
self.last_failure_time: Optional[float] = None
|
|
self.state = "closed"
|
|
self._lock = threading.Lock()
|
|
|
|
def call(self, func, *args, **kwargs):
|
|
with self._lock:
|
|
if self.state == "open":
|
|
if time.time() - self.last_failure_time >= self.timeout:
|
|
self.state = "half-open"
|
|
logger.info("Circuit breaker: OPEN -> HALF-OPEN")
|
|
else:
|
|
raise CircuitBreakerOpen("Circuit breaker is OPEN")
|
|
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
with self._lock:
|
|
if self.state == "half-open":
|
|
self.state = "closed"
|
|
self.failures = 0
|
|
logger.info("Circuit breaker: HALF-OPEN -> CLOSED")
|
|
return result
|
|
except Exception as e:
|
|
with self._lock:
|
|
self.failures += 1
|
|
self.last_failure_time = time.time()
|
|
if self.failures >= self.failure_threshold:
|
|
self.state = "open"
|
|
logger.warning(f"Circuit breaker: CLOSED -> OPEN (failures={self.failures})")
|
|
raise
|
|
|
|
|
|
class CircuitBreakerOpen(Exception):
|
|
pass
|
|
|
|
|
|
class AdvancedSenderPlugin(ScheduledTaskPlugin):
|
|
plugin_name = "advanced_sender"
|
|
plugin_version = "6.0.0"
|
|
plugin_description = "高级多节点数据推送插件，支持表单化配置和Cookie变量"
|
|
plugin_author = "User"
|
|
config_schema = staticmethod(get_config_schema)
|
|
|
|
_instance_lock = threading.Lock()
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._enabled = False
|
|
self._last_run_time: Optional[datetime] = None
|
|
self._running = False
|
|
self._semaphore: Optional[asyncio.Semaphore] = None
|
|
|
|
self._cookie_error_reported_1 = False
|
|
self._cookie_error_reported_2 = False
|
|
|
|
self.cron_expression = "*/2 * * * *"
|
|
|
|
self.db_host = "192.168.2.27"
|
|
self.db_port = 3306
|
|
self.db_user = "root2"
|
|
self.db_password = "root@root"
|
|
self.db_name = "addb"
|
|
self.db_table = "user_data"
|
|
self.sendto_field = "sendto"
|
|
|
|
self.cookie_1 = ""
|
|
self.cookie_2 = ""
|
|
|
|
self.node_id = "wx1"
|
|
self.receiver = "asq"
|
|
|
|
self.node_id_1 = ""
|
|
self.receiver_1 = ""
|
|
self.node_id_2 = ""
|
|
self.receiver_2 = ""
|
|
self.fetch_timeout = 15
|
|
self.send_interval = 20
|
|
|
|
self._api_circuit = CircuitBreaker(failure_threshold=3, timeout=30)
|
|
self._db_circuit = CircuitBreaker(failure_threshold=3, timeout=30)
|
|
|
|
def _get_external_db_config(self) -> Dict[str, Any]:
|
|
return {
|
|
"host": self.db_host,
|
|
"user": self.db_user,
|
|
"password": self.db_password,
|
|
"database": self.db_name,
|
|
"port": self.db_port,
|
|
"charset": "utf8mb4"
|
|
}
|
|
|
|
def _get_datasources(self) -> List[Dict[str, Any]]:
|
|
return [
|
|
{
|
|
"name": "source1",
|
|
"url": "https://xiansuo2.lfgzyx.com/admin/customer/sales/index?addtabs=1&sort=id&order=desc&offset=0&limit=50",
|
|
"headers": {
|
|
"Cookie": self.cookie_1,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/javascript, */*; q=0.01",
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36 Edg/140.0.0.0",
|
|
"X-Requested-With": "XMLHttpRequest"
|
|
},
|
|
"verify_ssl": True,
|
|
"zhid": "1",
|
|
"node_id": self.node_id_1 or self.node_id,
|
|
"receiver": self.receiver_1 or self.receiver
|
|
},
|
|
{
|
|
"name": "source2",
|
|
"url": "https://xiansuo2.lfgzyx.com/admin/customer/sales/index?addtabs=1&sort=id&order=desc&offset=0&limit=50",
|
|
"headers": {
|
|
"Cookie": self.cookie_2,
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/javascript, */*; q=0.01",
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
|
|
"X-Requested-With": "XMLHttpRequest"
|
|
},
|
|
"verify_ssl": True,
|
|
"zhid": "2",
|
|
"node_id": self.node_id_2 or self.node_id,
|
|
"receiver": self.receiver_2 or self.receiver
|
|
}
|
|
]
|
|
|
|
def initialize(self, config: Dict[str, Any]) -> bool:
|
|
self.config = config
|
|
self.cron_expression = config.get("cron", "*/2 * * * *")
|
|
|
|
self.db_host = config.get("db_host", self.db_host)
|
|
self.db_port = config.get("db_port", 3306)
|
|
self.db_user = config.get("db_user", self.db_user)
|
|
self.db_password = config.get("db_password", self.db_password)
|
|
self.db_name = config.get("db_name", self.db_name)
|
|
self.db_table = config.get("db_table", "user_data")
|
|
self.sendto_field = config.get("sendto_field", "sendto")
|
|
|
|
self.cookie_1 = config.get("cookie_1", "")
|
|
self.cookie_2 = config.get("cookie_2", "")
|
|
|
|
self.node_id = config.get("node_id", "wx1")
|
|
self.receiver = config.get("receiver", "asq")
|
|
|
|
self.node_id_1 = config.get("node_id_1", "")
|
|
self.receiver_1 = config.get("receiver_1", "")
|
|
self.node_id_2 = config.get("node_id_2", "")
|
|
self.receiver_2 = config.get("receiver_2", "")
|
|
|
|
self.fetch_timeout = max(5, min(60, config.get("fetch_timeout", 15)))
|
|
self.send_interval = config.get("send_interval", 20)
|
|
|
|
logger.info(f"AdvancedSenderPlugin initialized: cron={self.cron_expression}")
|
|
logger.info(f"Cookie1: {self.cookie_1[:30] if self.cookie_1 else 'empty'}..., Cookie2: {self.cookie_2[:30] if self.cookie_2 else 'empty'}...")
|
|
return True
|
|
|
|
def enable(self):
|
|
self._enabled = True
|
|
self.enabled = True
|
|
logger.info("AdvancedSenderPlugin enabled")
|
|
|
|
def disable(self):
|
|
self._enabled = False
|
|
self.enabled = False
|
|
logger.info("AdvancedSenderPlugin disabled")
|
|
|
|
def get_cron_expression(self) -> str:
|
|
return self.cron_expression
|
|
|
|
def should_run_and_get_next(self) -> tuple:
|
|
try:
|
|
now = datetime.now()
|
|
current_hour = now.hour
|
|
|
|
if current_hour < 8 or current_hour >= 22:
|
|
interval_minutes = 15
|
|
else:
|
|
interval_minutes = 2
|
|
|
|
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:
|
|
logger.error(f"should_run_and_get_next error: {e}")
|
|
return False, None
|
|
|
|
@contextmanager
|
|
def _db_connection(self):
|
|
conn = None
|
|
try:
|
|
import pymysql
|
|
conn = pymysql.connect(
|
|
host=self.db_host,
|
|
user=self.db_user,
|
|
password=self.db_password,
|
|
database=self.db_name,
|
|
port=self.db_port,
|
|
charset="utf8mb4",
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
connect_timeout=5,
|
|
read_timeout=10,
|
|
write_timeout=10
|
|
)
|
|
yield conn
|
|
finally:
|
|
if conn:
|
|
try:
|
|
conn.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def _fetch_from_datasource(self, datasource: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
source_name = datasource.get('name', 'unknown')
|
|
source_num = datasource.get('zhid', '0')
|
|
try:
|
|
import requests
|
|
url = datasource["url"]
|
|
if "_=" in url:
|
|
url = url + str(int(time.time() * 1000))
|
|
|
|
logger.info(f"Fetching from {source_name}: {url}")
|
|
logger.info(f"Cookie: {datasource['headers'].get('Cookie', '')[:50]}...")
|
|
|
|
response = requests.get(
|
|
url,
|
|
headers=datasource.get("headers", {}),
|
|
verify=datasource.get("verify_ssl", True),
|
|
timeout=self.fetch_timeout
|
|
)
|
|
|
|
if response.status_code == 403 or response.status_code == 401:
|
|
error_msg = f"数据源{source_name}Cookie失效 (HTTP {response.status_code})"
|
|
logger.error(error_msg)
|
|
if source_num == '1' and not self._cookie_error_reported_1:
|
|
self._cookie_error_reported_1 = True
|
|
self.notify_error("Cookie失效告警", error_msg)
|
|
elif source_num == '2' and not self._cookie_error_reported_2:
|
|
self._cookie_error_reported_2 = True
|
|
self.notify_error("Cookie失效告警", error_msg)
|
|
return []
|
|
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
total = data.get('total', 0)
|
|
rows = data.get("rows", []) if isinstance(data, dict) else []
|
|
|
|
if total == 0 and len(rows) == 0:
|
|
error_msg = f"数据源{source_name}返回空数据，Cookie可能已失效"
|
|
logger.warning(error_msg)
|
|
if source_num == '1' and not self._cookie_error_reported_1:
|
|
self._cookie_error_reported_1 = True
|
|
self.notify_warning("数据源为空告警", error_msg)
|
|
elif source_num == '2' and not self._cookie_error_reported_2:
|
|
self._cookie_error_reported_2 = True
|
|
self.notify_warning("数据源为空告警", error_msg)
|
|
return []
|
|
|
|
if source_num == '1' and self._cookie_error_reported_1:
|
|
self._cookie_error_reported_1 = False
|
|
self.notify_success("数据源恢复", f"数据源{source_name}已恢复正常")
|
|
elif source_num == '2' and self._cookie_error_reported_2:
|
|
self._cookie_error_reported_2 = False
|
|
self.notify_success("数据源恢复", f"数据源{source_name}已恢复正常")
|
|
|
|
logger.info(f"Fetched data: total={total}")
|
|
return rows
|
|
|
|
except Exception as e:
|
|
error_msg = f"数据源{source_name}请求失败: {str(e)}"
|
|
logger.error(error_msg)
|
|
if source_num == '1' and not self._cookie_error_reported_1:
|
|
self._cookie_error_reported_1 = True
|
|
self.notify_error("数据源请求失败", error_msg)
|
|
elif source_num == '2' and not self._cookie_error_reported_2:
|
|
self._cookie_error_reported_2 = True
|
|
self.notify_error("数据源请求失败", error_msg)
|
|
return []
|
|
|
|
def _check_id_exists(self, conn, id_value: Any) -> bool:
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = f"SELECT id FROM {self.db_table} WHERE id = %s"
|
|
cursor.execute(sql, (id_value,))
|
|
result = cursor.fetchone()
|
|
return result is not None
|
|
except Exception as e:
|
|
logger.error(f"Check ID exists failed: {e}")
|
|
return False
|
|
|
|
def _insert_data(self, conn, data: Dict[str, Any], zhid: str) -> bool:
|
|
try:
|
|
def get_val(field):
|
|
val = data.get(field)
|
|
return val if val is not None else None
|
|
|
|
other_text = json.dumps(data.get("other_text")) if data.get("other_text") else None
|
|
|
|
sql = f"""
|
|
INSERT INTO {self.db_table} (
|
|
id, source_id, name, phone, location, province_id, city_id, address,
|
|
app_name, keshi, create_time, update_time, admin_id, distribute_admin_id,
|
|
distribute_time, follow_up_status, is_recover, recover_time, `desc`,
|
|
other, wx, ggmc, gid, recover_time_text, weight_id_text,
|
|
follow_up_status_text, follow_text, origin_id_text, admin_id_text,
|
|
sendto, zhid
|
|
) VALUES (
|
|
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s, %s, %s, %s, %s, 0, %s
|
|
)
|
|
"""
|
|
|
|
values = (
|
|
get_val("id"), get_val("source_id"), get_val("name"), get_val("phone"),
|
|
get_val("location"), get_val("province_id"), get_val("city_id"), get_val("address"),
|
|
get_val("app_name"), get_val("keshi"), get_val("create_time"), get_val("update_time"),
|
|
get_val("admin_id"), get_val("distribute_admin_id"), get_val("distribute_time"),
|
|
get_val("follow_up_status"), get_val("is_recover"), get_val("recover_time"),
|
|
get_val("desc"), other_text, get_val("wx"), get_val("ggmc"),
|
|
get_val("gid"), get_val("recover_time_text"), get_val("weight_id_text"),
|
|
get_val("follow_up_status_text"), get_val("follow_text"), get_val("origin_id_text"),
|
|
get_val("admin_id_text"), zhid
|
|
)
|
|
|
|
with conn.cursor() as cursor:
|
|
cursor.execute(sql, values)
|
|
conn.commit()
|
|
logger.info(f"Successfully inserted data, ID: {data.get('id')}")
|
|
return True
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error(f"Insert data failed (ID: {data.get('id')}): {e}")
|
|
return False
|
|
|
|
def _get_pending_records(self, conn) -> List[Dict[str, Any]]:
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = f"SELECT * FROM {self.db_table} WHERE {self.sendto_field} = 0 ORDER BY id DESC"
|
|
cursor.execute(sql)
|
|
return cursor.fetchall()
|
|
except Exception as e:
|
|
logger.error(f"Get pending records failed: {e}")
|
|
return []
|
|
|
|
def _get_pending_records_by_zhid(self, conn, zhid: str) -> List[Dict[str, Any]]:
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
sql = f"SELECT * FROM {self.db_table} WHERE {self.sendto_field} = 0 AND zhid = %s ORDER BY id DESC"
|
|
cursor.execute(sql, (zhid,))
|
|
return cursor.fetchall()
|
|
except Exception as e:
|
|
logger.error(f"Get pending records by zhid failed: {e}")
|
|
return []
|
|
|
|
def _update_sendto(self, conn, ids: List[int]) -> bool:
|
|
if not ids:
|
|
return True
|
|
try:
|
|
with conn.cursor() as cursor:
|
|
placeholders = ", ".join(["%s"] * len(ids))
|
|
sql = f"UPDATE {self.db_table} SET {self.sendto_field} = 1 WHERE id IN ({placeholders})"
|
|
cursor.execute(sql, ids)
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
conn.rollback()
|
|
logger.error(f"Update sendto failed: {e}")
|
|
return False
|
|
|
|
def _sync_node_to_manager(self):
|
|
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
|
|
logger.info(f"Node {self.node_id} synced to node_manager")
|
|
|
|
def _send_via_center(self, node_id: str, who: str, msg: str) -> Dict[str, Any]:
|
|
from services.node_manager import node_manager
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
self._sync_node_to_manager()
|
|
effective_node_id = node_id or self.node_id
|
|
|
|
def _send_in_thread():
|
|
async def _send_async():
|
|
return await node_manager.send_message(
|
|
node_id=effective_node_id,
|
|
who=who,
|
|
msg=msg,
|
|
msg_type="text"
|
|
)
|
|
return asyncio.run(_send_async())
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
future = executor.submit(_send_in_thread)
|
|
result = future.result(timeout=15)
|
|
logger.info(f"Message sent via center to {who} via {effective_node_id}: {result}")
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Send via center failed: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def _process_pending_records(self, conn, records: List[Dict[str, Any]], datasource: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not records:
|
|
logger.info("No records to process")
|
|
return {"success": True, "sent": 0, "skipped": 0}
|
|
|
|
target_node_id = datasource.get("node_id", self.node_id)
|
|
target_receiver = datasource.get("receiver", self.receiver)
|
|
source_name = datasource.get("name", "unknown")
|
|
|
|
logger.info(f"Processing {len(records)} pending records for {source_name} -> {target_node_id}/{target_receiver}")
|
|
|
|
sent_count = 0
|
|
skipped_count = 0
|
|
|
|
for record in records:
|
|
record_id = record.get("id")
|
|
phone = record.get("phone", "")
|
|
create_time = record.get("create_time")
|
|
myid = record.get("myid", 0)
|
|
|
|
logger.info(f"Record ID: {record_id}")
|
|
|
|
if create_time:
|
|
create_time_dt = create_time if hasattr(create_time, 'hour') else datetime.now()
|
|
hour = create_time_dt.hour
|
|
|
|
if 0 <= hour <= 7:
|
|
logger.info(f"Hour {hour} in 0-7: skip")
|
|
skipped_count += 1
|
|
continue
|
|
elif 8 <= hour <= 16:
|
|
if myid % 2 == 0:
|
|
target = target_receiver
|
|
else:
|
|
target = target_receiver
|
|
logger.info(f"Hour {hour}, myid={myid}: sending to {target}")
|
|
elif 17 <= hour <= 21:
|
|
target = target_receiver
|
|
logger.info(f"Hour {hour}: sending to {target}")
|
|
elif 22 <= hour <= 23:
|
|
logger.info(f"Hour {hour} in 22-23: skip")
|
|
skipped_count += 1
|
|
continue
|
|
else:
|
|
target = target_receiver
|
|
|
|
msg = f"{phone}\r\n时间：{create_time}"
|
|
result = self._send_via_center(target_node_id, target, msg)
|
|
if result.get("success"):
|
|
sent_count += 1
|
|
|
|
return {"success": True, "sent": sent_count, "skipped": skipped_count, "total": len(records)}
|
|
|
|
def run_task(self) -> Dict[str, Any]:
|
|
if self._running:
|
|
logger.warning("Previous task still running, skipping")
|
|
return {"success": False, "error": "上一次执行尚未完成"}
|
|
|
|
self._running = True
|
|
try:
|
|
logger.info(f"Starting task at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
try:
|
|
with self._db_connection() as conn:
|
|
datasources = self._get_datasources()
|
|
all_pending_records = []
|
|
|
|
for idx, datasource in enumerate(datasources):
|
|
rows = self._fetch_from_datasource(datasource)
|
|
logger.info(f"Source {datasource.get('name')}: fetched {len(rows)} rows")
|
|
|
|
for row in rows:
|
|
record_id = row.get("id")
|
|
if not record_id:
|
|
logger.warning("Record missing ID, skip")
|
|
continue
|
|
|
|
if self._check_id_exists(conn, record_id):
|
|
logger.info(f"ID {record_id} exists, skip")
|
|
continue
|
|
|
|
self._insert_data(conn, row, datasource.get("zhid", "0"))
|
|
|
|
if idx < len(datasources) - 1 and self.send_interval > 0:
|
|
logger.info(f"Waiting {self.send_interval}s before next source...")
|
|
time.sleep(self.send_interval)
|
|
|
|
for datasource in datasources:
|
|
records = self._get_pending_records_by_zhid(conn, datasource.get("zhid", "0"))
|
|
if records:
|
|
send_result = self._process_pending_records(conn, records, datasource)
|
|
logger.info(f"Source {datasource.get('name')} send result: {send_result}")
|
|
all_pending_records.extend(records)
|
|
|
|
if all_pending_records:
|
|
ids = [r["id"] for r in all_pending_records]
|
|
self._update_sendto(conn, ids)
|
|
logger.info(f"Updated {len(ids)} records sendto=1")
|
|
|
|
return {"success": True, "message": "Task completed", "total": len(all_pending_records)}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Database operation error: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
|
|
except Exception as e:
|
|
logger.error(f"run_task error: {e}")
|
|
return {"success": False, "error": str(e)}
|
|
finally:
|
|
self._running = False
|
|
|
|
def aggregate_results(self, results: List[Dict[str, Any]], shift_name: str = "") -> Dict[str, Any]:
|
|
success_count = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
|
|
total_sent = sum(r.get("sent", 0) for r in results if isinstance(r, dict))
|
|
total_skipped = sum(r.get("skipped", 0) for r in results if isinstance(r, dict))
|
|
return {
|
|
"success": success_count == len(results),
|
|
"shift": shift_name,
|
|
"total": len(results),
|
|
"success_count": success_count,
|
|
"total_sent": total_sent,
|
|
"total_skipped": total_skipped,
|
|
"results": results
|
|
}
|
|
|
|
def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
|
action = params.get("action", "send")
|
|
|
|
if action == "send":
|
|
return self.run_task()
|
|
elif action == "test":
|
|
return self.test_config()
|
|
elif action == "test_send":
|
|
msg = params.get("msg", "测试消息")
|
|
node_id = params.get("node_id", self.node_id)
|
|
receiver = params.get("receiver", self.receiver)
|
|
return self._send_via_center(node_id, receiver, msg)
|
|
elif action == "query_pending":
|
|
try:
|
|
with self._db_connection() as conn:
|
|
records = self._get_pending_records(conn)
|
|
return {"success": True, "count": len(records), "records": records[:10]}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
return {"success": False, "error": "未知操作"}
|
|
|
|
def test_config(self) -> Dict[str, Any]:
|
|
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),
|
|
"node_id": self.node_id,
|
|
"receiver": self.receiver,
|
|
"datasource1": {
|
|
"node_id": self.node_id_1 or "(default)",
|
|
"receiver": self.receiver_1 or "(default)"
|
|
},
|
|
"datasource2": {
|
|
"node_id": self.node_id_2 or "(default)",
|
|
"receiver": self.receiver_2 or "(default)"
|
|
},
|
|
"db_host": self.db_host,
|
|
"db_name": self.db_name,
|
|
"db_table": self.db_table,
|
|
"cookie_1": self.cookie_1[:50] + "..." if len(self.cookie_1) > 50 else self.cookie_1,
|
|
"cookie_2": self.cookie_2[:50] + "..." if len(self.cookie_2) > 50 else self.cookie_2,
|
|
"config": {
|
|
"fetch_timeout": self.fetch_timeout,
|
|
"send_interval": self.send_interval
|
|
}
|
|
}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)} |