128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
import asyncio
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from base.plugin_common.message_plugin_interface import MessagePluginInterface
|
||
from base.plugin_common.plugin_interface import PluginStatus
|
||
from plugins.sehuatang_push.shehuatang import pdf_file_path
|
||
from plugins.sehuatang_push.shehuatang_undetected import pdf_file_path_undetected
|
||
|
||
|
||
class SehuatangPushPlugin(MessagePluginInterface):
|
||
"""涩图 PDF 推送插件。"""
|
||
|
||
FEATURE_KEY = "PDF_CAPABILITY"
|
||
FEATURE_DESCRIPTION = "📄 sehuatang PDF能力 [无]"
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "涩图推送"
|
||
|
||
@property
|
||
def version(self) -> str:
|
||
return "1.0.0"
|
||
|
||
@property
|
||
def description(self) -> str:
|
||
return "将 sehuatang PDF 推送从系统任务迁移到插件任务。"
|
||
|
||
@property
|
||
def author(self) -> str:
|
||
return "ABOT Team"
|
||
|
||
@property
|
||
def commands(self) -> List[str]:
|
||
return []
|
||
|
||
@property
|
||
def feature_key(self) -> Optional[str]:
|
||
return self.FEATURE_KEY
|
||
|
||
@property
|
||
def feature_description(self) -> Optional[str]:
|
||
return self.FEATURE_DESCRIPTION
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.feature = self.register_feature()
|
||
|
||
def initialize(self, context: Dict[str, Any]) -> bool:
|
||
return True
|
||
|
||
def start(self) -> bool:
|
||
self.status = PluginStatus.RUNNING
|
||
return True
|
||
|
||
def stop(self) -> bool:
|
||
self.status = PluginStatus.STOPPED
|
||
return True
|
||
|
||
def can_process(self, message: Dict[str, Any]) -> bool:
|
||
return False
|
||
|
||
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
|
||
return False, None
|
||
|
||
def get_schedule_actions(self) -> List[Dict[str, Any]]:
|
||
return [
|
||
{
|
||
"action_key": "daily_pdf_push",
|
||
"name": "涩图PDF推送",
|
||
"description": "每天生成并推送涩图 PDF 提醒",
|
||
"trigger_type": "at_times",
|
||
"trigger_config": {"time_list": ["15:30"]},
|
||
"target_scope": "all_enabled_groups",
|
||
"target_config": {},
|
||
"payload": {"at_user": "Jyunere"},
|
||
"default_enabled": True,
|
||
}
|
||
]
|
||
|
||
async def run_scheduled_action(self, action_key: str, context: Dict[str, Any]) -> Dict[str, Any]:
|
||
if action_key != "daily_pdf_push":
|
||
return {
|
||
"success": False,
|
||
"summary": f"不支持的动作: {action_key}",
|
||
"detail": {"action_key": action_key},
|
||
}
|
||
if not self.bot:
|
||
return {"success": False, "summary": "bot 未注入", "detail": {}}
|
||
|
||
target_groups = [str(g).strip() for g in (context.get("target_groups") or []) if str(g).strip()]
|
||
if not target_groups:
|
||
return {"success": False, "summary": "没有可推送目标群", "detail": {"target_count": 0}}
|
||
|
||
payload = context.get("payload") or {}
|
||
at_user = str(payload.get("at_user", "Jyunere") or "Jyunere").strip()
|
||
|
||
# 兼容历史逻辑：优先使用 undetected 方案，失败后回退普通抓取。
|
||
try:
|
||
ok, path = await asyncio.to_thread(pdf_file_path_undetected)
|
||
if not ok:
|
||
ok, path = await asyncio.to_thread(pdf_file_path)
|
||
if not ok:
|
||
return {"success": False, "summary": "PDF 生成失败", "detail": {}}
|
||
except Exception as e:
|
||
return {"success": False, "summary": f"PDF 生成异常: {e}", "detail": {"error": str(e)}}
|
||
|
||
success_groups = []
|
||
failed_groups = {}
|
||
for gid in target_groups:
|
||
try:
|
||
# 历史系统任务仅发送提醒文本，这里保持一致，避免引入文件发送兼容风险。
|
||
await self.bot.send_at_message(gid, f"98堂 PDF已就绪，请手动发送！", [at_user])
|
||
success_groups.append(gid)
|
||
except Exception as e:
|
||
failed_groups[gid] = str(e)
|
||
|
||
return {
|
||
"success": len(failed_groups) == 0,
|
||
"summary": f"涩图推送完成: 成功{len(success_groups)}群, 失败{len(failed_groups)}群",
|
||
"detail": {
|
||
"target_count": len(target_groups),
|
||
"success_groups": success_groups,
|
||
"failed_groups": failed_groups,
|
||
"pdf_path": path,
|
||
},
|
||
}
|