diff --git a/src/plugins/built_in/maizone/__init__.py b/src/plugins/built_in/maizone/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/src/plugins/built_in/maizone/_manifest.json b/src/plugins/built_in/maizone/_manifest.json deleted file mode 100644 index d9999bf5a..000000000 --- a/src/plugins/built_in/maizone/_manifest.json +++ /dev/null @@ -1,50 +0,0 @@ -{ - "manifest_version": 1, - "name": "MaiZone（麦麦空间）", - "version": "2.0.0", - "description": "让你的麦麦发QQ空间说说、评论、点赞，支持AI配图、定时发送和自动监控功能", - "author": { - "name": "MaiBot-Plus", - "url": "https://github.com/MaiBot-Plus" - }, - "license": "AGPL-v3.0", - - "host_application": { - "min_version": "0.8.0", - "max_version": "0.10.0" - }, - "homepage_url": "https://github.com/MaiBot-Plus/MaiMbot-Pro-Max", - "repository_url": "https://github.com/MaiBot-Plus/MaiMbot-Pro-Max", - "keywords": ["QQ空间", "说说", "动态", "评论", "点赞", "自动化", "AI配图"], - "categories": ["社交", "自动化", "QQ空间"], - - "plugin_info": { - "is_built_in": false, - "plugin_type": "social", - "components": [ - { - "type": "action", - "name": "send_feed", - "description": "根据指定主题发送一条QQ空间说说" - }, - { - "type": "action", - "name": "read_feed", - "description": "读取指定好友最近的说说，并评论点赞" - }, - { - "type": "command", - "name": "send_feed", - "description": "通过命令发送QQ空间说说" - } - ], - "features": [ - "智能生成说说内容", - "AI自动配图（硅基流动）", - "自动点赞评论好友说说", - "定时发送说说", - "权限管理系统", - "历史记录避重" - ] - } -} \ No newline at end of file diff --git a/src/plugins/built_in/maizone/config_loader.py b/src/plugins/built_in/maizone/config_loader.py deleted file mode 100644 index 0a9652a80..000000000 --- a/src/plugins/built_in/maizone/config_loader.py +++ /dev/null @@ -1,283 +0,0 @@ -""" -MaiZone插件配置加载器 - -简化的配置文件加载系统，专注于基本的配置文件读取和写入功能。 -支持TOML格式的配置文件，具有基本的类型转换和默认值处理。 -""" - -import toml -from typing import Dict, Any, Optional -from pathlib import Path - -from src.common.logger import get_logger - -logger = get_logger("MaiZone.ConfigLoader") - - -class MaiZoneConfigLoader: - """MaiZone插件配置加载器 - 简化版""" - - def __init__(self, plugin_dir: str, config_filename: str = "config.toml"): - """ - 初始化配置加载器 - - Args: - plugin_dir: 插件目录路径 - config_filename: 配置文件名 - """ - self.plugin_dir = Path(plugin_dir) - self.config_filename = config_filename - self.config_file_path = self.plugin_dir / config_filename - self.config_data: Dict[str, Any] = {} - - # 确保插件目录存在 - self.plugin_dir.mkdir(parents=True, exist_ok=True) - - def load_config(self) -> bool: - """ - 加载配置文件 - - Returns: - bool: 是否成功加载 - """ - try: - # 如果配置文件不存在，创建默认配置 - if not self.config_file_path.exists(): - logger.info(f"配置文件不存在，创建默认配置: {self.config_file_path}") - self._create_default_config() - - # 加载配置文件 - with open(self.config_file_path, 'r', encoding='utf-8') as f: - self.config_data = toml.load(f) - - logger.info(f"成功加载配置文件: {self.config_file_path}") - return True - - except Exception as e: - logger.error(f"加载配置文件失败: {e}") - # 如果加载失败，使用默认配置 - self.config_data = self._get_default_config() - return False - - def _create_default_config(self): - """创建默认配置文件""" - default_config = self._get_default_config() - self._save_config_to_file(default_config) - self.config_data = default_config - - def _get_default_config(self) -> Dict[str, Any]: - """获取默认配置""" - return { - "plugin": { - "enabled": True, - "name": "MaiZone", - "version": "2.1.0" - }, - "qzone": { - "qq": "", - "auto_login": True, - "check_interval": 300, - "max_retries": 3 - }, - "ai": { - "enabled": False, - "model": "gpt-3.5-turbo", - "max_tokens": 150, - "temperature": 0.7 - }, - "monitor": { - "enabled": False, - "keywords": [], - "check_friends": True, - "check_groups": False - }, - "scheduler": { - "enabled": False, - "schedules": [] - } - } - - def _save_config_to_file(self, config_data: Dict[str, Any]): - """保存配置到文件""" - try: - with open(self.config_file_path, 'w', encoding='utf-8') as f: - toml.dump(config_data, f) - logger.debug(f"配置已保存到: {self.config_file_path}") - except Exception as e: - logger.error(f"保存配置文件失败: {e}") - raise - - def get_config(self, key: str, default: Any = None) -> Any: - """ - 获取配置值，支持嵌套键访问 - - Args: - key: 配置键名，支持嵌套访问如 "section.field" - default: 默认值 - - Returns: - Any: 配置值或默认值 - """ - if not self.config_data: - logger.warning("配置数据为空，返回默认值") - return default - - keys = key.split('.') - current = self.config_data - - try: - for k in keys: - if isinstance(current, dict) and k in current: - current = current[k] - else: - return default - return current - except Exception as e: - logger.warning(f"获取配置失败 {key}: {e}") - return default - - def set_config(self, key: str, value: Any) -> bool: - """ - 设置配置值 - - Args: - key: 配置键名，格式为 "section.field" - value: 配置值 - - Returns: - bool: 是否设置成功 - """ - try: - keys = key.split('.') - if len(keys) < 2: - logger.error(f"配置键格式错误: {key}，应为 'section.field' 格式") - return False - - # 获取或创建嵌套字典结构 - current = self.config_data - for k in keys[:-1]: - if k not in current: - current[k] = {} - elif not isinstance(current[k], dict): - logger.error(f"配置路径冲突: {k} 不是字典类型") - return False - current = current[k] - - # 设置最终值 - current[keys[-1]] = value - logger.debug(f"设置配置: {key} = {value}") - return True - - except Exception as e: - logger.error(f"设置配置失败 {key}: {e}") - return False - - def save_config(self) -> bool: - """ - 保存当前配置到文件 - - Returns: - bool: 是否保存成功 - """ - try: - self._save_config_to_file(self.config_data) - logger.info(f"配置已保存到: {self.config_file_path}") - return True - except Exception as e: - logger.error(f"保存配置失败: {e}") - return False - - def reload_config(self) -> bool: - """ - 重新加载配置文件 - - Returns: - bool: 是否重新加载成功 - """ - return self.load_config() - - def get_section(self, section_name: str) -> Optional[Dict[str, Any]]: - """ - 获取整个配置节 - - Args: - section_name: 配置节名称 - - Returns: - Optional[Dict[str, Any]]: 配置节数据或None - """ - return self.config_data.get(section_name) - - def set_section(self, section_name: str, section_data: Dict[str, Any]) -> bool: - """ - 设置整个配置节 - - Args: - section_name: 配置节名称 - section_data: 配置节数据 - - Returns: - bool: 是否设置成功 - """ - try: - if not isinstance(section_data, dict): - logger.error(f"配置节数据必须为字典类型: {section_name}") - return False - - self.config_data[section_name] = section_data - logger.debug(f"设置配置节: {section_name}") - return True - except Exception as e: - logger.error(f"设置配置节失败 {section_name}: {e}") - return False - - def has_config(self, key: str) -> bool: - """ - 检查配置项是否存在 - - Args: - key: 配置键名 - - Returns: - bool: 配置项是否存在 - """ - keys = key.split('.') - current = self.config_data - - try: - for k in keys: - if isinstance(current, dict) and k in current: - current = current[k] - else: - return False - return True - except Exception: - return False - - def get_config_info(self) -> Dict[str, Any]: - """ - 获取配置信息 - - Returns: - Dict[str, Any]: 配置信息 - """ - return { - "config_file": str(self.config_file_path), - "config_exists": self.config_file_path.exists(), - "sections": list(self.config_data.keys()) if self.config_data else [], - "loaded": bool(self.config_data) - } - - def reset_to_default(self) -> bool: - """ - 重置为默认配置 - - Returns: - bool: 是否重置成功 - """ - try: - self.config_data = self._get_default_config() - return self.save_config() - except Exception as e: - logger.error(f"重置配置失败: {e}") - return False diff --git a/src/plugins/built_in/maizone/monitor.py b/src/plugins/built_in/maizone/monitor.py deleted file mode 100644 index df1c170a5..000000000 --- a/src/plugins/built_in/maizone/monitor.py +++ /dev/null @@ -1,240 +0,0 @@ -import asyncio -import random -import time -import traceback -from typing import Dict, Any - -from src.common.logger import get_logger -from src.plugin_system.apis import llm_api, config_api - -# 导入工具模块 -import sys -import os -sys.path.append(os.path.dirname(__file__)) - -from qzone_utils import QZoneManager - -# 获取日志记录器 -logger = get_logger('MaiZone-Monitor') - - -class MonitorManager: - """监控管理器 - 负责自动监控好友说说并点赞评论""" - - def __init__(self, plugin): - """初始化监控管理器""" - self.plugin = plugin - self.is_running = False - self.task = None - self.last_check_time = 0 - - logger.info("监控管理器初始化完成") - - async def start(self): - """启动监控任务""" - if self.is_running: - logger.warning("监控任务已在运行中") - return - - self.is_running = True - self.task = asyncio.create_task(self._monitor_loop()) - logger.info("说说监控任务已启动") - - async def stop(self): - """停止监控任务""" - if not self.is_running: - return - - self.is_running = False - - if self.task: - self.task.cancel() - try: - await self.task - except asyncio.CancelledError: - logger.info("监控任务已被取消") - - logger.info("说说监控任务已停止") - - async def _monitor_loop(self): - """监控任务主循环""" - while self.is_running: - try: - # 获取监控间隔配置 - interval_minutes = int(self.plugin.get_config("monitor.interval_minutes", 10) or 10) - - # 等待指定时间间隔 - await asyncio.sleep(interval_minutes * 60) - - # 执行监控检查 - await self._check_and_process_feeds() - - except asyncio.CancelledError: - logger.info("监控循环被取消") - break - except Exception as e: - logger.error(f"监控任务出错: {str(e)}") - logger.error(traceback.format_exc()) - # 出错后等待5分钟再重试 - await asyncio.sleep(300) - - async def _check_and_process_feeds(self): - """检查并处理好友说说""" - try: - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - read_num = 10 # 监控时读取较少的说说数量 - - logger.info("监控任务: 开始检查好友说说") - - # 创建QZone管理器 (监控模式不需要stream_id) - qzone_manager = QZoneManager() - - # 获取监控说说列表 - feeds_list = await qzone_manager.monitor_read_feed(qq_account, read_num) - - if not feeds_list: - logger.info("监控任务: 未发现新说说") - return - - logger.info(f"监控任务: 发现 {len(feeds_list)} 条新说说") - - # 处理每条说说 - for feed in feeds_list: - try: - await self._process_monitor_feed(feed, qzone_manager) - # 每条说说之间随机延迟 - await asyncio.sleep(3 + random.random() * 2) - except Exception as e: - logger.error(f"处理监控说说失败: {str(e)}") - - except Exception as e: - logger.error(f"监控检查失败: {str(e)}") - - async def _process_monitor_feed(self, feed: Dict[str, Any], qzone_manager: QZoneManager): - """处理单条监控说说""" - try: - # 提取说说信息 - target_qq = feed.get("target_qq", "") - tid = feed.get("tid", "") - content = feed.get("content", "") - images = feed.get("images", []) - rt_con = feed.get("rt_con", "") - - # 构建完整内容用于显示 - full_content = content - if images: - full_content += f" [图片: {len(images)}张]" - if rt_con: - full_content += f" [转发: {rt_con[:20]}...]" - - logger.info(f"监控处理说说: {target_qq} - {full_content[:30]}...") - - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - like_possibility = float(self.plugin.get_config("read.like_possibility", 1.0) or 1.0) - comment_possibility = float(self.plugin.get_config("read.comment_possibility", 0.3) or 0.3) - - # 随机决定是否评论 - if random.random() <= comment_possibility: - comment = await self._generate_monitor_comment(content, rt_con, target_qq) - if comment: - success = await qzone_manager.comment_feed(qq_account, target_qq, tid, comment) - if success: - logger.info(f"监控评论成功: '{comment}'") - else: - logger.error(f"监控评论失败: {content[:20]}...") - - # 随机决定是否点赞 - if random.random() <= like_possibility: - success = await qzone_manager.like_feed(qq_account, target_qq, tid) - if success: - logger.info(f"监控点赞成功: {content[:20]}...") - else: - logger.error(f"监控点赞失败: {content[:20]}...") - - except Exception as e: - logger.error(f"处理监控说说异常: {str(e)}") - - async def _generate_monitor_comment(self, content: str, rt_con: str, target_qq: str) -> str: - """生成监控评论内容""" - try: - # 获取模型配置 - models = llm_api.get_available_models() - text_model = str(self.plugin.get_config("models.text_model", "replyer_1")) - model_config = models.get(text_model) - - if not model_config: - logger.error("未配置LLM模型") - return "" - - # 获取机器人信息 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_expression = config_api.get_global_config("expression.expression_style", "内容积极向上") - - # 构建提示词 - if not rt_con: - prompt = f""" - 你是'{bot_personality}'，你正在浏览你好友'{target_qq}'的QQ空间， - 你看到了你的好友'{target_qq}'qq空间上内容是'{content}'的说说，你想要发表你的一条评论， - {bot_expression}，回复的平淡一些，简短一些，说中文， - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出回复内容 - """ - else: - prompt = f""" - 你是'{bot_personality}'，你正在浏览你好友'{target_qq}'的QQ空间， - 你看到了你的好友'{target_qq}'在qq空间上转发了一条内容为'{rt_con}'的说说，你的好友的评论为'{content}' - 你想要发表你的一条评论，{bot_expression}，回复的平淡一些，简短一些，说中文， - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出回复内容 - """ - - logger.info(f"正在为 {target_qq} 的说说生成评论...") - - # 生成评论 - success, comment, reasoning, model_name = await llm_api.generate_with_model( - prompt=prompt, - model_config=model_config, - request_type="story.generate", - temperature=0.3, - max_tokens=1000 - ) - - if success: - logger.info(f"成功生成监控评论: '{comment}'") - return comment - else: - logger.error("生成监控评论失败") - return "" - - except Exception as e: - logger.error(f"生成监控评论异常: {str(e)}") - return "" - - def get_status(self) -> Dict[str, Any]: - """获取监控状态""" - return { - "is_running": self.is_running, - "interval_minutes": self.plugin.get_config("monitor.interval_minutes", 10), - "last_check_time": self.last_check_time, - "enabled": self.plugin.get_config("monitor.enable_auto_monitor", False) - } - - async def manual_check(self) -> Dict[str, Any]: - """手动执行一次监控检查""" - try: - logger.info("执行手动监控检查") - await self._check_and_process_feeds() - - return { - "success": True, - "message": "手动监控检查完成", - "timestamp": time.time() - } - - except Exception as e: - logger.error(f"手动监控检查失败: {str(e)}") - return { - "success": False, - "message": f"手动监控检查失败: {str(e)}", - "timestamp": time.time() - } diff --git a/src/plugins/built_in/maizone/plugin.py b/src/plugins/built_in/maizone/plugin.py deleted file mode 100644 index edf966bd6..000000000 --- a/src/plugins/built_in/maizone/plugin.py +++ /dev/null @@ -1,819 +0,0 @@ -import asyncio -import random -import time -from typing import List, Tuple, Type - -from src.common.logger import get_logger -from src.plugin_system import ( - BasePlugin, register_plugin, BaseAction, BaseCommand, - ComponentInfo, ActionActivationType, ChatMode -) -from src.plugin_system.apis import llm_api, config_api, person_api, generator_api -from src.plugin_system.base.config_types import ConfigField - -# 导入插件工具模块 -import sys -import os -sys.path.append(os.path.dirname(__file__)) - -from qzone_utils import ( - QZoneManager, generate_image_by_sf, get_send_history -) -from scheduler import ScheduleManager -from config_loader import MaiZoneConfigLoader - -# 获取日志记录器 -logger = get_logger('MaiZone') - - -# ===== 发送说说命令组件 ===== -class SendFeedCommand(BaseCommand): - """发送说说命令 - 响应 /send_feed 命令""" - - command_name = "send_feed" - command_description = "发送一条QQ空间说说" - command_pattern = r"^/send_feed(?:\s+(?P\w+))?$" - command_help = "发一条主题为或随机的说说" - command_examples = ["/send_feed", "/send_feed 日常"] - intercept_message = True - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # 获取配置加载器引用 - self.config_loader = None - self._init_config_loader() - - def _init_config_loader(self): - """初始化配置加载器""" - try: - plugin_dir = os.path.dirname(__file__) - self.config_loader = MaiZoneConfigLoader(plugin_dir) - self.config_loader.load_config() - except Exception as e: - logger.error(f"初始化配置加载器失败: {e}") - - def get_config(self, key: str, default=None): - """获取配置值""" - if self.config_loader: - return self.config_loader.get_config(key, default) - return default - - def check_permission(self, qq_account: str) -> bool: - """检查用户权限""" - - permission_list = self.get_config("send.permission", []) - permission_type = self.get_config("send.permission_type", "whitelist") - - logger.info(f'权限检查: {permission_type}:{permission_list}') - - if not isinstance(permission_list, list): - logger.error("权限列表配置错误") - return False - - if permission_type == 'whitelist': - return qq_account in permission_list - elif permission_type == 'blacklist': - return qq_account not in permission_list - else: - logger.error('权限类型配置错误，应为 whitelist 或 blacklist') - return False - - async def execute(self) -> Tuple[bool, str, bool]: - """执行发送说说命令""" - try: - # 获取用户信息 - user_id = self.message.message_info.user_info.user_id if self.message and self.message.message_info and self.message.message_info.user_info else None - - # 权限检查 - if not user_id or not self.check_permission(user_id): - logger.info(f"用户 {user_id} 权限不足") - await self.send_text("权限不足，无法使用此命令") - return False, "权限不足", True - - # 获取主题 - topic = self.matched_groups.get("topic", "") - - # 生成说说内容 - story = await self._generate_story_content(topic) - if not story: - return False, "生成说说内容失败", True - - # 处理图片 - await self._handle_images(story) - - # 发送说说 - success = await self._send_feed(story) - if success: - if self.get_config("send.enable_reply", True): - await self.send_text(f"已发送说说：\n{story}") - return True, "发送成功", True - else: - return False, "发送说说失败", True - - except Exception as e: - logger.error(f"发送说说命令执行失败: {str(e)}") - return False, "命令执行失败", True - - async def _generate_story_content(self, topic: str) -> str: - """生成说说内容""" - try: - # 获取模型配置 - models = llm_api.get_available_models() - text_model = str(self.get_config("models.text_model", "replyer_1")) - model_config = models.get(text_model) - - if not model_config: - logger.error("未配置LLM模型") - return "" - - # 获取机器人信息 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_expression = config_api.get_global_config("personality.reply_style", "内容积极向上") - qq_account = config_api.get_global_config("bot.qq_account", "") - - # 构建提示词 - if topic: - prompt = f""" - 你是'{bot_personality}'，你想写一条主题是'{topic}'的说说发表在qq空间上， - {bot_expression} - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，可以适当使用颜文字， - 只输出一条说说正文的内容，不要有其他的任何正文以外的冗余输出 - """ - else: - prompt = f""" - 你是'{bot_personality}'，你想写一条说说发表在qq空间上，主题不限 - {bot_expression} - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，可以适当使用颜文字， - 只输出一条说说正文的内容，不要有其他的任何正文以外的冗余输出 - """ - - # 添加历史记录 - prompt += "\n以下是你以前发过的说说，写新说说时注意不要在相隔不长的时间发送相同主题的说说" - history_block = await get_send_history(qq_account) - if history_block: - prompt += history_block - - # 生成内容 - success, story, reasoning, model_name = await llm_api.generate_with_model( - prompt=prompt, - model_config=model_config, - request_type="story.generate", - temperature=0.3, - max_tokens=1000 - ) - - if success: - logger.info(f"成功生成说说内容：'{story}'") - return story - else: - logger.error("生成说说内容失败") - return "" - - except Exception as e: - logger.error(f"生成说说内容异常: {str(e)}") - return "" - - async def _handle_images(self, story: str): - """处理说说配图""" - try: - enable_ai_image = bool(self.get_config("send.enable_ai_image", False)) - apikey = str(self.get_config("models.siliconflow_apikey", "")) - image_dir = str(self.get_config("send.image_directory", "./plugins/Maizone/images")) - image_num_raw = self.get_config("send.ai_image_number", 1) - image_num = int(image_num_raw if image_num_raw is not None else 1) - - if enable_ai_image and apikey: - await generate_image_by_sf( - api_key=apikey, - story=story, - image_dir=image_dir, - batch_size=image_num - ) - elif enable_ai_image and not apikey: - logger.error('启用了AI配图但未填写API密钥') - - except Exception as e: - logger.error(f"处理配图失败: {str(e)}") - - async def _send_feed(self, story: str) -> bool: - """发送说说到QQ空间""" - try: - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - enable_image = bool(self.get_config("send.enable_image", False)) - image_dir = str(self.get_config("send.image_directory", "./plugins/Maizone/images")) - - # 获取聊天流ID - stream_id = self.message.chat_stream.stream_id if self.message and self.message.chat_stream else None - - # 创建QZone管理器并发送 - qzone_manager = QZoneManager(stream_id) - success = await qzone_manager.send_feed(story, image_dir, qq_account, enable_image) - - return success - - except Exception as e: - logger.error(f"发送说说失败: {str(e)}") - return False - - -# ===== 发送说说动作组件 ===== -class SendFeedAction(BaseAction): - """发送说说动作 - 当用户要求发说说时激活""" - - action_name = "send_feed" - action_description = "发一条相应主题的说说" - activation_type = ActionActivationType.KEYWORD - mode_enable = ChatMode.ALL - - activation_keywords = ["说说", "空间", "动态"] - keyword_case_sensitive = False - - action_parameters = { - "topic": "要发送的说说主题", - "user_name": "要求你发说说的好友的qq名称", - } - action_require = [ - "用户要求发说说时使用", - "当有人希望你更新qq空间时使用", - "当你认为适合发说说时使用", - ] - associated_types = ["text"] - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # 获取配置加载器引用 - self.config_loader = None - self._init_config_loader() - - def _init_config_loader(self): - """初始化配置加载器""" - try: - plugin_dir = os.path.dirname(__file__) - self.config_loader = MaiZoneConfigLoader(plugin_dir) - self.config_loader.load_config() - except Exception as e: - logger.error(f"初始化配置加载器失败: {e}") - - def get_config(self, key: str, default=None): - """获取配置值""" - if self.config_loader: - return self.config_loader.get_config(key, default) - return default - - def check_permission(self, qq_account: str) -> bool: - """检查用户权限""" - permission_list = self.get_config("send.permission", []) - permission_type = self.get_config("send.permission_type", "whitelist") - - logger.info(f'权限检查: {permission_type}:{permission_list}') - - if isinstance(permission_list, list): - if permission_type == 'whitelist': - return qq_account in permission_list - elif permission_type == 'blacklist': - return qq_account not in permission_list - - logger.error('权限类型配置错误') - return False - - async def execute(self) -> Tuple[bool, str]: - """执行发送说说动作""" - try: - # 获取用户信息 - user_name = self.action_data.get("user_name", "") - person_id = person_api.get_person_id_by_name(user_name) - user_id = await person_api.get_person_value(person_id, "user_id") - - # 权限检查 - if not self.check_permission(user_id): - logger.info(f"用户 {user_id} 权限不足") - success, reply_set, _ = await generator_api.generate_reply( - chat_stream=self.chat_stream, - action_data={"extra_info_block": f'{user_name}无权命令你发送说说，请用符合你人格特点的方式拒绝请求'} - ) - if success and reply_set: - for reply_type, reply_content in reply_set: - if reply_type == "text": - await self.send_text(reply_content) - return False, "权限不足" - - # 获取主题并生成内容 - topic = self.action_data.get("topic", "") - story = await self._generate_story_content(topic) - if not story: - return False, "生成说说内容失败" - - # 处理图片 - await self._handle_images(story) - - # 发送说说 - success = await self._send_feed(story) - if success: - logger.info(f"成功发送说说: {story}") - - # 生成回复 - success, reply_set, _ = await generator_api.generate_reply( - chat_stream=self.chat_stream, - action_data={"extra_info_block": f'你刚刚发了一条说说，内容为{story}'} - ) - - if success and reply_set: - for reply_type, reply_content in reply_set: - if reply_type == "text": - await self.send_text(reply_content) - return True, '发送成功' - else: - await self.send_text('我发了一条说说啦~') - return True, '发送成功但回复生成失败' - else: - return False, "发送说说失败" - - except Exception as e: - logger.error(f"发送说说动作执行失败: {str(e)}") - return False, "动作执行失败" - - async def _generate_story_content(self, topic: str) -> str: - """生成说说内容""" - try: - # 获取模型配置 - models = llm_api.get_available_models() - text_model = str(self.get_config("models.text_model", "replyer_1")) - model_config = models.get(text_model) - - if not model_config: - return "" - - # 获取机器人信息 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_expression = config_api.get_global_config("expression.expression_style", "内容积极向上") - qq_account = config_api.get_global_config("bot.qq_account", "") - - # 构建提示词 - prompt = f""" - 你是{bot_personality}，你想写一条主题是{topic}的说说发表在qq空间上， - {bot_expression} - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，可以适当使用颜文字， - 只输出一条说说正文的内容，不要有其他的任何正文以外的冗余输出 - """ - - # 添加历史记录 - prompt += "\n以下是你以前发过的说说，写新说说时注意不要在相隔不长的时间发送相同主题的说说" - history_block = await get_send_history(qq_account) - if history_block: - prompt += history_block - - # 生成内容 - success, story, reasoning, model_name = await llm_api.generate_with_model( - prompt=prompt, - model_config=model_config, - request_type="story.generate", - temperature=0.3, - max_tokens=1000 - ) - - if success: - return story - else: - return "" - - except Exception as e: - logger.error(f"生成说说内容异常: {str(e)}") - return "" - - async def _handle_images(self, story: str): - """处理说说配图""" - try: - enable_ai_image = bool(self.get_config("send.enable_ai_image", False)) - apikey = str(self.get_config("models.siliconflow_apikey", "")) - image_dir = str(self.get_config("send.image_directory", "./plugins/Maizone/images")) - image_num_raw = self.get_config("send.ai_image_number", 1) - image_num = int(image_num_raw if image_num_raw is not None else 1) - - if enable_ai_image and apikey: - await generate_image_by_sf( - api_key=apikey, - story=story, - image_dir=image_dir, - batch_size=image_num - ) - elif enable_ai_image and not apikey: - logger.error('启用了AI配图但未填写API密钥') - - except Exception as e: - logger.error(f"处理配图失败: {str(e)}") - - async def _send_feed(self, story: str) -> bool: - """发送说说到QQ空间""" - try: - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - enable_image = bool(self.get_config("send.enable_image", False)) - image_dir = str(self.get_config("send.image_directory", "./plugins/Maizone/images")) - - # 获取聊天流ID - stream_id = self.chat_stream.stream_id if self.chat_stream else None - - # 创建QZone管理器并发送 - qzone_manager = QZoneManager(stream_id) - success = await qzone_manager.send_feed(story, image_dir, qq_account, enable_image) - - return success - - except Exception as e: - logger.error(f"发送说说失败: {str(e)}") - return False - - -# ===== 阅读说说动作组件 ===== -class ReadFeedAction(BaseAction): - """阅读说说动作 - 当用户要求读说说时激活""" - - action_name = "read_feed" - action_description = "读取好友最近的动态/说说/qq空间并评论点赞" - activation_type = ActionActivationType.KEYWORD - mode_enable = ChatMode.ALL - - activation_keywords = ["说说", "空间", "动态"] - keyword_case_sensitive = False - - action_parameters = { - "target_name": "需要阅读动态的好友的qq名称", - "user_name": "要求你阅读动态的好友的qq名称" - } - - action_require = [ - "需要阅读某人动态、说说、QQ空间时使用", - "当有人希望你评价某人的动态、说说、QQ空间", - "当你认为适合阅读说说、动态、QQ空间时使用", - ] - associated_types = ["text"] - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # 获取配置加载器引用 - self.config_loader = None - self._init_config_loader() - - def _init_config_loader(self): - """初始化配置加载器""" - try: - plugin_dir = os.path.dirname(__file__) - self.config_loader = MaiZoneConfigLoader(plugin_dir) - self.config_loader.load_config() - except Exception as e: - logger.error(f"初始化配置加载器失败: {e}") - - def get_config(self, key: str, default=None): - """获取配置值""" - if self.config_loader: - return self.config_loader.get_config(key, default) - return default - - def check_permission(self, qq_account: str) -> bool: - """检查用户权限""" - permission_list = self.get_config("read.permission", []) - permission_type = self.get_config("read.permission_type", "blacklist") - - if not isinstance(permission_list, list): - return False - - logger.info(f'权限检查: {permission_type}:{permission_list}') - - if permission_type == 'whitelist': - return qq_account in permission_list - elif permission_type == 'blacklist': - return qq_account not in permission_list - else: - logger.error('权限类型配置错误') - return False - - async def execute(self) -> Tuple[bool, str]: - """执行阅读说说动作""" - try: - # 获取用户信息 - user_name = self.action_data.get("user_name", "") - person_id = person_api.get_person_id_by_name(user_name) - user_id = await person_api.get_person_value(person_id, "user_id") - - # 权限检查 - if not self.check_permission(user_id): - logger.info(f"用户 {user_id} 权限不足") - success, reply_set, _ = await generator_api.generate_reply( - chat_stream=self.chat_stream, - action_data={"extra_info_block": f'{user_name}无权命令你阅读说说，请用符合人格的方式进行拒绝的回复'} - ) - if success and reply_set: - for reply_type, reply_content in reply_set: - if reply_type == "text": - await self.send_text(reply_content) - return False, "权限不足" - - # 获取目标用户 - target_name = self.action_data.get("target_name", "") - target_person_id = person_api.get_person_id_by_name(target_name) - target_qq = await person_api.get_person_value(target_person_id, "user_id") - - # 读取并处理说说 - success = await self._read_and_process_feeds(target_qq, target_name) - - if success: - # 生成回复 - success, reply_set, _ = await generator_api.generate_reply( - chat_stream=self.chat_stream, - action_data={"extra_info_block": f'你刚刚成功读了{target_name}的说说，请告知你已经读了说说'} - ) - - if success and reply_set: - for reply_type, reply_content in reply_set: - if reply_type == "text": - await self.send_text(reply_content) - return True, '阅读成功' - return True, '阅读成功但回复生成失败' - else: - return False, "阅读说说失败" - - except Exception as e: - logger.error(f"阅读说说动作执行失败: {str(e)}") - return False, "动作执行失败" - - async def _read_and_process_feeds(self, target_qq: str, target_name: str) -> bool: - """读取并处理说说""" - try: - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - num_raw = self.get_config("read.read_number", 5) - num = int(num_raw if num_raw is not None else 5) - like_raw = self.get_config("read.like_possibility", 1.0) - like_possibility = float(like_raw if like_raw is not None else 1.0) - comment_raw = self.get_config("read.comment_possibility", 1.0) - comment_possibility = float(comment_raw if comment_raw is not None else 1.0) - - # 获取聊天流ID - stream_id = self.chat_stream.stream_id if self.chat_stream else None - - # 创建QZone管理器并读取说说 - qzone_manager = QZoneManager(stream_id) - feeds_list = await qzone_manager.read_feed(qq_account, target_qq, num) - - # 处理错误情况 - if isinstance(feeds_list, list) and len(feeds_list) > 0 and isinstance(feeds_list[0], dict) and 'error' in feeds_list[0]: - success, reply_set, _ = await generator_api.generate_reply( - chat_stream=self.chat_stream, - action_data={"extra_info_block": f'你在读取说说的时候出现了错误，错误原因：{feeds_list[0].get("error")}'} - ) - - if success and reply_set: - for reply_type, reply_content in reply_set: - if reply_type == "text": - await self.send_text(reply_content) - return True - - # 处理说说列表 - if isinstance(feeds_list, list): - logger.info(f"成功读取到{len(feeds_list)}条说说") - - for feed in feeds_list: - # 随机延迟 - time.sleep(3 + random.random()) - - # 处理说说内容 - await self._process_single_feed( - feed, target_qq, target_name, - like_possibility, comment_possibility, qzone_manager - ) - - return True - else: - return False - - except Exception as e: - logger.error(f"读取并处理说说失败: {str(e)}") - return False - - async def _process_single_feed(self, feed: dict, target_qq: str, target_name: str, - like_possibility: float, comment_possibility: float, - qzone_manager): - """处理单条说说""" - try: - content = feed.get("content", "") - images = feed.get("images", []) - if images: - for image in images: - content = content + str(image) - fid = feed.get("tid", "") - rt_con = feed.get("rt_con", "") - - # 随机评论 - if random.random() <= comment_possibility: - comment = await self._generate_comment(content, rt_con, target_name) - if comment: - success = await qzone_manager.comment_feed( - config_api.get_global_config("bot.qq_account", ""), - target_qq, fid, comment - ) - if success: - logger.info(f"发送评论'{comment}'成功") - else: - logger.error(f"评论说说'{content[:20]}...'失败") - - # 随机点赞 - if random.random() <= like_possibility: - success = await qzone_manager.like_feed( - config_api.get_global_config("bot.qq_account", ""), - target_qq, fid - ) - if success: - logger.info(f"点赞说说'{content[:10]}..'成功") - else: - logger.error(f"点赞说说'{content[:20]}...'失败") - - except Exception as e: - logger.error(f"处理单条说说失败: {str(e)}") - - async def _generate_comment(self, content: str, rt_con: str, target_name: str) -> str: - """生成评论内容""" - try: - # 获取模型配置 - models = llm_api.get_available_models() - text_model = str(self.get_config("models.text_model", "replyer_1")) - model_config = models.get(text_model) - - if not model_config: - return "" - - # 获取机器人信息 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_expression = config_api.get_global_config("expression.expression_style", "内容积极向上") - - # 构建提示词 - if not rt_con: - prompt = f""" - 你是'{bot_personality}'，你正在浏览你好友'{target_name}'的QQ空间， - 你看到了你的好友'{target_name}'qq空间上内容是'{content}'的说说，你想要发表你的一条评论， - {bot_expression}，回复的平淡一些，简短一些，说中文， - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出回复内容 - """ - else: - prompt = f""" - 你是'{bot_personality}'，你正在浏览你好友'{target_name}'的QQ空间， - 你看到了你的好友'{target_name}'在qq空间上转发了一条内容为'{rt_con}'的说说，你的好友的评论为'{content}' - 你想要发表你的一条评论，{bot_expression}，回复的平淡一些，简短一些，说中文， - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，不要输出多余内容(包括前后缀，冒号和引号，括号()，表情包，at或 @等 )。只输出回复内容 - """ - - logger.info(f"正在评论'{target_name}'的说说：{content[:20]}...") - - # 生成评论 - success, comment, reasoning, model_name = await llm_api.generate_with_model( - prompt=prompt, - model_config=model_config, - request_type="story.generate", - temperature=0.3, - max_tokens=1000 - ) - - if success: - logger.info(f"成功生成评论内容：'{comment}'") - return comment - else: - logger.error("生成评论内容失败") - return "" - - except Exception as e: - logger.error(f"生成评论内容异常: {str(e)}") - return "" - - -# ===== 插件主类 ===== -@register_plugin -class MaiZonePlugin(BasePlugin): - """MaiZone插件 - 让麦麦发QQ空间""" - - # 插件基本信息 - plugin_name: str = "MaiZonePlugin" - enable_plugin: bool = True - dependencies: List[str] = [] - python_dependencies: List[str] = [] - config_file_name: str = "config.toml" - - # 配置节描述 - config_section_descriptions = { - "plugin": "插件基础配置", - "models": "模型相关配置", - "send": "发送说说配置", - "read": "阅读说说配置", - "monitor": "自动监控配置", - "schedule": "定时发送配置", - } - - # 配置模式定义 - config_schema: dict = { - "plugin": { - "enable": ConfigField(type=bool, default=True, description="是否启用插件"), - "config_version": ConfigField(type=str, default="2.1.0", description="配置文件版本"), - }, - "models": { - "text_model": ConfigField(type=str, default="replyer_1", description="生成文本的模型名称"), - "siliconflow_apikey": ConfigField(type=str, default="", description="硅基流动AI生图API密钥"), - }, - "send": { - "permission": ConfigField(type=list, default=['1145141919810'], description="发送权限QQ号列表"), - "permission_type": ConfigField(type=str, default='whitelist', description="权限类型：whitelist(白名单) 或 blacklist(黑名单)"), - "enable_image": ConfigField(type=bool, default=False, description="是否启用说说配图"), - "enable_ai_image": ConfigField(type=bool, default=False, description="是否启用AI生成配图"), - "enable_reply": ConfigField(type=bool, default=True, description="生成完成时是否发出回复"), - "ai_image_number": ConfigField(type=int, default=1, description="AI生成图片数量(1-4张)"), - "image_directory": ConfigField(type=str, default="./plugins/built_in/Maizone/images", description="图片存储目录") - }, - "read": { - "permission": ConfigField(type=list, default=[], description="阅读权限QQ号列表"), - "permission_type": ConfigField(type=str, default='blacklist', description="权限类型：whitelist(白名单) 或 blacklist(黑名单)"), - "read_number": ConfigField(type=int, default=5, description="一次读取的说说数量"), - "like_possibility": ConfigField(type=float, default=1.0, description="点赞概率(0.0-1.0)"), - "comment_possibility": ConfigField(type=float, default=0.3, description="评论概率(0.0-1.0)"), - }, - "monitor": { - "enable_auto_monitor": ConfigField(type=bool, default=False, description="是否启用自动监控好友说说"), - "interval_minutes": ConfigField(type=int, default=10, description="监控间隔时间(分钟)"), - }, - "schedule": { - "enable_schedule": ConfigField(type=bool, default=False, description="是否启用基于日程表的定时发送说说"), - }, - } - - def __init__(self, *args, **kwargs): - """初始化插件""" - super().__init__(*args, **kwargs) - - # 设置插件信息 - self.plugin_name = "MaiZone" - self.plugin_description = "让麦麦实现QQ空间点赞、评论、发说说功能" - self.plugin_version = "2.0.0" - self.plugin_author = "重构版" - self.config_file_name = "config.toml" - - # 初始化独立配置加载器 - plugin_dir = self.plugin_dir - if plugin_dir is None: - plugin_dir = os.path.dirname(__file__) - self.config_loader = MaiZoneConfigLoader(plugin_dir, self.config_file_name) - - # 加载配置 - if not self.config_loader.load_config(): - logger.error("配置加载失败，使用默认设置") - - # 获取启用状态 - self.enable_plugin = self.config_loader.get_config("plugin.enable", True) - - # 初始化管理器 - self.monitor_manager = None - self.schedule_manager = None - - # 根据配置启动功能 - if self.enable_plugin: - self._init_managers() - - def _init_managers(self): - """初始化管理器""" - try: - # 初始化监控管理器 - if self.config_loader.get_config("monitor.enable_auto_monitor", False): - from .monitor import MonitorManager - self.monitor_manager = MonitorManager(self) - asyncio.create_task(self._start_monitor_delayed()) - - # 初始化定时管理器 - if self.config_loader.get_config("schedule.enable_schedule", False): - logger.info("定时任务启用状态: true") - self.schedule_manager = ScheduleManager(self) - asyncio.create_task(self._start_scheduler_delayed()) - - except Exception as e: - logger.error(f"初始化管理器失败: {str(e)}") - - async def _start_monitor_delayed(self): - """延迟启动监控管理器""" - try: - await asyncio.sleep(10) # 等待插件完全初始化 - if self.monitor_manager: - await self.monitor_manager.start() - except Exception as e: - logger.error(f"启动监控管理器失败: {str(e)}") - - async def _start_scheduler_delayed(self): - """延迟启动定时管理器""" - try: - await asyncio.sleep(10) # 等待插件完全初始化 - if self.schedule_manager: - await self.schedule_manager.start() - except Exception as e: - logger.error(f"启动定时管理器失败: {str(e)}") - - def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]: - """获取插件组件列表""" - return [ - (SendFeedAction.get_action_info(), SendFeedAction), - (ReadFeedAction.get_action_info(), ReadFeedAction), - (SendFeedCommand.get_command_info(), SendFeedCommand) - ] diff --git a/src/plugins/built_in/maizone/qzone_utils.py b/src/plugins/built_in/maizone/qzone_utils.py deleted file mode 100644 index 73844c202..000000000 --- a/src/plugins/built_in/maizone/qzone_utils.py +++ /dev/null @@ -1,1060 +0,0 @@ -import base64 -import json -import os -import random -import time -import datetime -from typing import List, Dict, Any, Optional -from pathlib import Path - -import requests -import bs4 -import json5 - -from src.chat.utils.utils_image import get_image_manager -from src.common.logger import get_logger -from src.plugin_system.apis import llm_api, config_api, emoji_api, send_api -from src.chat.message_receive.chat_stream import get_chat_manager - -# 获取日志记录器 -logger = get_logger('MaiZone-Utils') - - -class CookieManager: - """Cookie管理类 - 负责处理QQ空间的认证Cookie""" - - @staticmethod - def get_cookie_file_path(uin: str) -> str: - """获取Cookie文件路径""" - # 使用当前文件所在目录作为基础路径，更稳定可靠 - current_dir = Path(__file__).resolve().parent - - # 尝试多种可能的根目录查找方式 - # 方法1：直接在当前插件目录下存储（最稳定） - cookie_dir = current_dir / "cookies" - cookie_dir.mkdir(exist_ok=True) # 确保目录存在 - - return str(cookie_dir / f"cookies-{uin}.json") - - @staticmethod - def parse_cookie_string(cookie_str: str) -> Dict[str, str]: - """解析Cookie字符串为字典""" - cookies: Dict[str, str] = {} - if not cookie_str: - return cookies - - for pair in cookie_str.split("; "): - if not pair or "=" not in pair: - continue - key, value = pair.split("=", 1) - cookies[key.strip()] = value.strip() - return cookies - - @staticmethod - def extract_uin_from_cookie(cookie_str: str) -> str: - """从Cookie中提取用户UIN""" - for item in cookie_str.split("; "): - if item.startswith("uin=") or item.startswith("o_uin="): - _, value = item.split("=", 1) - return value.lstrip("o") - raise ValueError("无法从Cookie字符串中提取UIN") - - @staticmethod - async def fetch_cookies(domain: str, stream_id: Optional[str] = None) -> Dict[str, Any]: - """通过适配器API从NapCat获取Cookie""" - logger.info(f"正在通过适配器API获取Cookie，域名: {domain}") - - try: - if stream_id is None: - response = await send_api.adapter_command_to_stream( - action="get_cookies", - params={"domain": domain}, - platform="qq", - timeout=40.0, - storage_message=False - ) - # 使用适配器命令API获取cookie - else: - response = await send_api.adapter_command_to_stream( - action="get_cookies", - params={"domain": domain}, - platform="qq", - stream_id=stream_id, - timeout=40.0, - storage_message=False - ) - - logger.info(f"适配器响应: {response}") - - if response.get("status") == "ok": - data = response.get("data", {}) - if "cookies" in data: - logger.info("成功通过适配器API获取Cookie") - return data - else: - raise RuntimeError(f"适配器返回的数据中缺少cookies字段: {data}") - else: - error_msg = response.get("message", "未知错误") - raise RuntimeError(f"适配器API获取Cookie失败: {error_msg}") - - except Exception as e: - logger.error(f"通过适配器API获取Cookie失败: {str(e)}") - raise - - @staticmethod - async def renew_cookies(stream_id: Optional[str] = None) -> bool: - """更新Cookie文件""" - try: - domain = "user.qzone.qq.com" - cookie_data = await CookieManager.fetch_cookies(domain, stream_id) - cookie_str = cookie_data["cookies"] - parsed_cookies = CookieManager.parse_cookie_string(cookie_str) - uin = CookieManager.extract_uin_from_cookie(cookie_str) - - file_path = CookieManager.get_cookie_file_path(uin) - - with open(file_path, "w", encoding="utf-8") as f: - json.dump(parsed_cookies, f, indent=4, ensure_ascii=False) - - logger.info(f"Cookie已更新并保存至: {file_path}") - return True - - except Exception as e: - logger.error(f"更新Cookie失败: {str(e)}") - return False - - @staticmethod - def load_cookies(qq_account: str) -> Optional[Dict[str, str]]: - """加载Cookie文件""" - cookie_file = CookieManager.get_cookie_file_path(qq_account) - - if os.path.exists(cookie_file): - try: - with open(cookie_file, 'r', encoding='utf-8') as f: - return json.load(f) - except Exception as e: - logger.error(f"加载Cookie文件失败: {str(e)}") - return None - else: - logger.warning(f"Cookie文件不存在: {cookie_file}") - return None - - -class QZoneAPI: - """QQ空间API类 - 封装QQ空间的核心操作""" - - # QQ空间API地址常量 - UPLOAD_IMAGE_URL = "https://up.qzone.qq.com/cgi-bin/upload/cgi_upload_image" - EMOTION_PUBLISH_URL = "https://user.qzone.qq.com/proxy/domain/taotao.qzone.qq.com/cgi-bin/emotion_cgi_publish_v6" - DOLIKE_URL = "https://user.qzone.qq.com/proxy/domain/w.qzone.qq.com/cgi-bin/likes/internal_dolike_app" - COMMENT_URL = "https://user.qzone.qq.com/proxy/domain/taotao.qzone.qq.com/cgi-bin/emotion_cgi_re_feeds" - LIST_URL = "https://user.qzone.qq.com/proxy/domain/taotao.qq.com/cgi-bin/emotion_cgi_msglist_v6" - ZONE_LIST_URL = "https://user.qzone.qq.com/proxy/domain/ic2.qzone.qq.com/cgi-bin/feeds/feeds3_html_more" - - def __init__(self, cookies_dict: Optional[Dict[str, str]] = None): - """初始化QZone API""" - self.cookies = cookies_dict or {} - self.gtk2 = '' - self.uin = 0 - self.qzonetoken = '' - - # 生成gtk2 - p_skey = self.cookies.get('p_skey') or self.cookies.get('p_skey'.upper()) - if p_skey: - self.gtk2 = self._generate_gtk(p_skey) - - # 提取UIN - uin_raw = self.cookies.get('uin') or self.cookies.get('o_uin') or self.cookies.get('p_uin') - if isinstance(uin_raw, str) and uin_raw: - uin_str = uin_raw.lstrip('o') - try: - self.uin = int(uin_str) - except Exception: - logger.error(f"UIN格式错误: {uin_raw}") - - def _generate_gtk(self, skey: str) -> str: - """生成GTK令牌""" - hash_val = 5381 - for i in range(len(skey)): - hash_val += (hash_val << 5) + ord(skey[i]) - return str(hash_val & 2147483647) - - async def _do_request( - self, - method: str, - url: str, - params: Optional[Dict] = None, - data: Optional[Dict] = None, - headers: Optional[Dict] = None, - timeout: int = 10 - ) -> requests.Response: - """执行HTTP请求""" - try: - return requests.request( - method=method, - url=url, - params=params or {}, - data=data or {}, - headers=headers or {}, - cookies=self.cookies, - timeout=timeout - ) - except Exception as e: - logger.error(f"HTTP请求失败: {str(e)}") - raise - - async def validate_token(self, retry: int = 3) -> bool: - """验证Token有效性""" - # 简单验证 - 检查必要的Cookie是否存在 - required_cookies = ['p_skey', 'uin'] - for cookie in required_cookies: - if cookie not in self.cookies and cookie.upper() not in self.cookies: - logger.error(f"缺少必要的Cookie: {cookie}") - return False - return True - - def _image_to_base64(self, image: bytes) -> str: - """将图片转换为Base64""" - pic_base64 = base64.b64encode(image) - return str(pic_base64)[2:-1] - - async def _get_image_base64_by_url(self, url: str) -> str: - """通过URL获取图片的Base64编码""" - try: - res = await self._do_request("GET", url, timeout=60) - image_data = res.content - base64_str = base64.b64encode(image_data).decode('utf-8') - return base64_str - except Exception as e: - logger.error(f"获取图片Base64失败: {str(e)}") - raise - - async def upload_image(self, image: bytes) -> Dict[str, Any]: - """上传图片到QQ空间""" - try: - res = await self._do_request( - method="POST", - url=self.UPLOAD_IMAGE_URL, - data={ - "filename": "filename", - "zzpanelkey": "", - "uploadtype": "1", - "albumtype": "7", - "exttype": "0", - "skey": self.cookies["skey"], - "zzpaneluin": self.uin, - "p_uin": self.uin, - "uin": self.uin, - "p_skey": self.cookies['p_skey'], - "output_type": "json", - "qzonetoken": "", - "refer": "shuoshuo", - "charset": "utf-8", - "output_charset": "utf-8", - "upload_hd": "1", - "hd_width": "2048", - "hd_height": "10000", - "hd_quality": "96", - "backUrls": "http://upbak.photo.qzone.qq.com/cgi-bin/upload/cgi_upload_image,http://119.147.64.75/cgi-bin/upload/cgi_upload_image", - "url": "https://up.qzone.qq.com/cgi-bin/upload/cgi_upload_image?g_tk=" + self.gtk2, - "base64": "1", - "picfile": self._image_to_base64(image), - }, - headers={ - 'referer': 'https://user.qzone.qq.com/' + str(self.uin), - 'origin': 'https://user.qzone.qq.com' - }, - timeout=60 - ) - - if res.status_code == 200: - # 解析返回的JSON数据 - response_text = res.text - json_start = response_text.find('{') - json_end = response_text.rfind('}') + 1 - json_str = response_text[json_start:json_end] - return eval(json_str) # 使用eval解析，因为可能不是标准JSON - else: - raise Exception(f"上传图片失败，状态码: {res.status_code}") - - except Exception as e: - logger.error(f"上传图片异常: {str(e)}") - raise - - def _get_picbo_and_richval(self, upload_result: Dict[str, Any]) -> tuple[str, str]: - """从上传结果中提取picbo和richval""" - try: - if upload_result.get('ret') != 0: - raise Exception("上传图片失败") - - picbo_spt = upload_result['data']['url'].split('&bo=') - if len(picbo_spt) < 2: - raise Exception("解析图片URL失败") - picbo = picbo_spt[1] - - data = upload_result['data'] - richval = f",{data['albumid']},{data['lloc']},{data['sloc']},{data['type']},{data['height']},{data['width']},,{data['height']},{data['width']}" - - return picbo, richval - - except Exception as e: - logger.error(f"提取图片信息失败: {str(e)}") - raise - - async def publish_emotion(self, content: str, images: Optional[List[bytes]] = None) -> str: - """发布说说""" - if images is None: - images = [] - - try: - post_data = { - "syn_tweet_verson": "1", - "paramstr": "1", - "who": "1", - "con": content, - "feedversion": "1", - "ver": "1", - "ugc_right": "1", - "to_sign": "0", - "hostuin": self.uin, - "code_version": "1", - "format": "json", - "qzreferrer": "https://user.qzone.qq.com/" + str(self.uin) - } - - # 处理图片 - if len(images) > 0: - pic_bos = [] - richvals = [] - - for img in images: - upload_result = await self.upload_image(img) - picbo, richval = self._get_picbo_and_richval(upload_result) - pic_bos.append(picbo) - richvals.append(richval) - - post_data['pic_bo'] = ','.join(pic_bos) - post_data['richtype'] = '1' - post_data['richval'] = '\t'.join(richvals) - - res = await self._do_request( - method="POST", - url=self.EMOTION_PUBLISH_URL, - params={'g_tk': self.gtk2, 'uin': self.uin}, - data=post_data, - headers={ - 'referer': 'https://user.qzone.qq.com/' + str(self.uin), - 'origin': 'https://user.qzone.qq.com' - } - ) - - if res.status_code == 200: - result = res.json() - return result.get('tid', '') - else: - raise Exception(f"发表说说失败，状态码: {res.status_code}") - - except Exception as e: - logger.error(f"发表说说异常: {str(e)}") - raise - - async def like_feed(self, fid: str, target_qq: str) -> bool: - """点赞说说""" - try: - post_data = { - 'qzreferrer': f'https://user.qzone.qq.com/{self.uin}', - 'opuin': self.uin, - 'unikey': f'http://user.qzone.qq.com/{target_qq}/mood/{fid}', - 'curkey': f'http://user.qzone.qq.com/{target_qq}/mood/{fid}', - 'appid': 311, - 'from': 1, - 'typeid': 0, - 'abstime': int(time.time()), - 'fid': fid, - 'active': 0, - 'format': 'json', - 'fupdate': 1, - } - - res = await self._do_request( - method="POST", - url=self.DOLIKE_URL, - params={'g_tk': self.gtk2}, - data=post_data, - headers={ - 'referer': 'https://user.qzone.qq.com/' + str(self.uin), - 'origin': 'https://user.qzone.qq.com' - } - ) - - return res.status_code == 200 - - except Exception as e: - logger.error(f"点赞说说异常: {str(e)}") - return False - - async def comment_feed(self, fid: str, target_qq: str, content: str) -> bool: - """评论说说""" - try: - post_data = { - "topicId": f'{target_qq}_{fid}__1', - "uin": self.uin, - "hostUin": target_qq, - "feedsType": 100, - "inCharset": "utf-8", - "outCharset": "utf-8", - "plat": "qzone", - "source": "ic", - "platformid": 52, - "format": "fs", - "ref": "feeds", - "content": content, - } - - res = await self._do_request( - method="POST", - url=self.COMMENT_URL, - params={"g_tk": self.gtk2}, - data=post_data, - headers={ - 'referer': 'https://user.qzone.qq.com/' + str(self.uin), - 'origin': 'https://user.qzone.qq.com' - } - ) - - return res.status_code == 200 - - except Exception as e: - logger.error(f"评论说说异常: {str(e)}") - return False - - async def get_feed_list(self, target_qq: str, num: int) -> List[Dict[str, Any]]: - """获取指定用户的说说列表""" - try: - logger.info(f'获取用户 {target_qq} 的说说列表') - - res = await self._do_request( - method="GET", - url=self.LIST_URL, - params={ - 'g_tk': self.gtk2, - "uin": target_qq, - "ftype": 0, - "sort": 0, - "pos": 0, - "num": num, - "replynum": 100, - "callback": "_preloadCallback", - "code_version": 1, - "format": "jsonp", - "need_comment": 1, - "need_private_comment": 1 - }, - headers={ - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", - "Referer": f"https://user.qzone.qq.com/{target_qq}", - "Host": "user.qzone.qq.com", - "Connection": "keep-alive" - } - ) - - if res.status_code != 200: - raise Exception(f"访问失败，状态码: {res.status_code}") - - # 解析JSONP响应 - data = res.text - if data.startswith('_preloadCallback(') and data.endswith(');'): - json_str = data[len('_preloadCallback('):-2] - else: - json_str = data - - json_data = json.loads(json_str) - - if json_data.get('code') != 0: - return [{"error": json_data.get('message', '未知错误')}] - - # 解析说说列表 - return await self._parse_feed_list(json_data, target_qq) - - except Exception as e: - logger.error(f"获取说说列表失败: {str(e)}") - return [{"error": f'获取说说列表失败: {str(e)}'}] - - async def _parse_feed_list(self, json_data: Dict[str, Any], target_qq: str) -> List[Dict[str, Any]]: - """解析说说列表数据""" - try: - feeds_list = [] - login_info = json_data.get('logininfo', {}) - uin_nickname = login_info.get('name', '') - - for msg in json_data.get("msglist", []): - # 检查是否已经评论过 - is_commented = False - commentlist = msg.get("commentlist", []) - - if isinstance(commentlist, list): - for comment in commentlist: - if comment.get("name") == uin_nickname: - logger.info('已评论过此说说，跳过') - is_commented = True - break - - if not is_commented: - # 解析说说信息 - feed_info = await self._parse_single_feed(msg) - if feed_info: - feeds_list.append(feed_info) - - if len(feeds_list) == 0: - return [{"error": '你已经看过所有说说了，没有必要再看一遍'}] - - return feeds_list - - except Exception as e: - logger.error(f"解析说说列表失败: {str(e)}") - return [{"error": f'解析说说列表失败: {str(e)}'}] - - async def _parse_single_feed(self, msg: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """解析单条说说信息""" - try: - # 基本信息 - timestamp = msg.get("created_time", "") - created_time = "unknown" - if timestamp: - time_tuple = time.localtime(timestamp) - created_time = time.strftime('%Y-%m-%d %H:%M:%S', time_tuple) - - tid = msg.get("tid", "") - content = msg.get("content", "") - - logger.debug(f"正在解析说说: {content[:20]}...") - - # 解析图片 - images = [] - if 'pic' in msg: - for pic in msg['pic']: - url = pic.get('url1') or pic.get('pic_id') or pic.get('smallurl') - if url: - try: - image_base64 = await self._get_image_base64_by_url(url) - image_manager = get_image_manager() - image_description = await image_manager.get_image_description(image_base64) - images.append(image_description) - except Exception as e: - logger.warning(f"处理图片失败: {str(e)}") - - # 解析视频 - videos = [] - if 'video' in msg: - for video in msg['video']: - # 视频缩略图 - video_image_url = video.get('url1') or video.get('pic_url') - if video_image_url: - try: - image_base64 = await self._get_image_base64_by_url(video_image_url) - image_manager = get_image_manager() - image_description = await image_manager.get_image_description(image_base64) - images.append(f"视频缩略图: {image_description}") - except Exception as e: - logger.warning(f"处理视频缩略图失败: {str(e)}") - - # 视频URL - url = video.get('url3') - if url: - videos.append(url) - - # 解析转发内容 - rt_con = "" - if "rt_con" in msg: - rt_con_data = msg.get("rt_con") - if isinstance(rt_con_data, dict): - rt_con = rt_con_data.get("content", "") - - return { - "tid": tid, - "created_time": created_time, - "content": content, - "images": images, - "videos": videos, - "rt_con": rt_con - } - - except Exception as e: - logger.error(f"解析单条说说失败: {str(e)}") - return None - - async def get_monitor_feed_list(self, num: int) -> List[Dict[str, Any]]: - """获取监控用的说说列表（所有好友的最新动态）""" - try: - res = await self._do_request( - method="GET", - url=self.ZONE_LIST_URL, - params={ - "uin": self.uin, - "scope": 0, - "view": 1, - "filter": "all", - "flag": 1, - "applist": "all", - "pagenum": 1, - "count": num, - "aisortEndTime": 0, - "aisortOffset": 0, - "aisortBeginTime": 0, - "begintime": 0, - "format": "json", - "g_tk": self.gtk2, - "useutf8": 1, - "outputhtmlfeed": 1 - }, - headers={ - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", - "Referer": f"https://user.qzone.qq.com/{self.uin}", - "Host": "user.qzone.qq.com", - "Connection": "keep-alive" - } - ) - - if res.status_code != 200: - raise Exception(f"访问失败，状态码: {res.status_code}") - - # 解析响应数据 - data = res.text - if data.startswith('_Callback(') and data.endswith(');'): - data = data[len('_Callback('):-2] - - data = data.replace('undefined', 'null') - - try: - json_data = json5.loads(data) - if json_data and isinstance(json_data, dict): - feeds_data = json_data.get('data', {}).get('data', []) - else: - feeds_data = [] - except Exception as e: - logger.error(f"解析JSON数据失败: {str(e)}") - return [] - - # 解析说说列表 - return await self._parse_monitor_feeds(feeds_data) - - except Exception as e: - logger.error(f"获取监控说说列表失败: {str(e)}") - return [] - - async def _parse_monitor_feeds(self, feeds_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: - """解析监控说说数据""" - try: - feeds_list = [] - current_uin = str(self.uin) - - for feed in feeds_data: - if not feed: - continue - - # 过滤广告和非说说内容 - appid = str(feed.get('appid', '')) - if appid != '311': - continue - - target_qq = feed.get('uin', '') - tid = feed.get('key', '') - - if not target_qq or not tid: - continue - - # 过滤自己的说说 - if target_qq == current_uin: - continue - - # 解析HTML内容 - html_content = feed.get('html', '') - if not html_content: - continue - - feed_info = await self._parse_monitor_html(html_content, target_qq, tid) - if feed_info: - feeds_list.append(feed_info) - - logger.info(f"成功解析 {len(feeds_list)} 条未读说说") - return feeds_list - - except Exception as e: - logger.error(f"解析监控说说数据失败: {str(e)}") - return [] - - async def _parse_monitor_html(self, html_content: str, target_qq: str, tid: str) -> Optional[Dict[str, Any]]: - """解析监控说说的HTML内容""" - try: - soup = bs4.BeautifulSoup(html_content, 'html.parser') - - # 检查是否已经点赞（判断是否已读） - like_btn = soup.find('a', class_='qz_like_btn_v3') - if not like_btn: - like_btn = soup.find('a', attrs={'data-islike': True}) - - if isinstance(like_btn, bs4.element.Tag): - data_islike = like_btn.get('data-islike') - if data_islike == '1': # 已点赞，跳过 - return None - - # 提取文字内容 - text_div = soup.find('div', class_='f-info') - text = text_div.get_text(strip=True) if text_div else "" - - # 提取转发内容 - rt_con = "" - txt_box = soup.select_one('div.txt-box') - if txt_box: - rt_con = txt_box.get_text(strip=True) - if '：' in rt_con: - rt_con = rt_con.split('：', 1)[1].strip() - - # 提取图片 - images = [] - img_box = soup.find('div', class_='img-box') - if isinstance(img_box, bs4.element.Tag): - for img in img_box.find_all('img'): - src = img.get('src') if isinstance(img, bs4.element.Tag) else None - if src and isinstance(src, str) and not src.startswith('http://qzonestyle.gtimg.cn'): - try: - image_base64 = await self._get_image_base64_by_url(src) - image_manager = get_image_manager() - description = await image_manager.get_image_description(image_base64) - images.append(description) - except Exception as e: - logger.warning(f"处理图片失败: {str(e)}") - - # 视频缩略图 - img_tag = soup.select_one('div.video-img img') - if isinstance(img_tag, bs4.element.Tag): - src = img_tag.get('src') - if src and isinstance(src, str): - try: - image_base64 = await self._get_image_base64_by_url(src) - image_manager = get_image_manager() - description = await image_manager.get_image_description(image_base64) - images.append(f"视频缩略图: {description}") - except Exception as e: - logger.warning(f"处理视频缩略图失败: {str(e)}") - - # 视频URL - videos = [] - video_div = soup.select_one('div.img-box.f-video-wrap.play') - if video_div and 'url3' in video_div.attrs: - videos.append(video_div['url3']) - - return { - 'target_qq': target_qq, - 'tid': tid, - 'content': text, - 'images': images, - 'videos': videos, - 'rt_con': rt_con, - } - - except Exception as e: - logger.error(f"解析监控HTML失败: {str(e)}") - return None - - -class QZoneManager: - """QQ空间管理器 - 高级封装类""" - - def __init__(self, stream_id: Optional[str] = None): - """初始化QZone管理器""" - self.stream_id = stream_id - self.cookie_manager = CookieManager() - - async def _get_qzone_api(self, qq_account: str) -> Optional[QZoneAPI]: - """获取QZone API实例""" - try: - # 更新Cookie - await self.cookie_manager.renew_cookies(self.stream_id) - - # 加载Cookie - cookies = self.cookie_manager.load_cookies(qq_account) - if not cookies: - logger.error("无法加载Cookie") - return None - - # 创建API实例 - qzone_api = QZoneAPI(cookies) - - # 验证Token - if not await qzone_api.validate_token(): - logger.error("Token验证失败") - return None - - return qzone_api - - except Exception as e: - logger.error(f"获取QZone API失败: {str(e)}") - return None - - async def send_feed(self, message: str, image_directory: str, qq_account: str, enable_image: bool) -> bool: - """发送说说""" - try: - # 获取API实例 - qzone_api = await self._get_qzone_api(qq_account) - if not qzone_api: - return False - - # 处理图片 - images = [] - if enable_image: - images = await self._load_images(image_directory, message) - - # 发送说说 - tid = await qzone_api.publish_emotion(message, images) - if tid: - logger.info(f"成功发送说说，TID: {tid}") - return True - else: - logger.error("发送说说失败") - return False - - except Exception as e: - logger.error(f"发送说说异常: {str(e)}") - return False - - async def _load_images(self, image_directory: str, message: str) -> List[bytes]: - """加载图片文件""" - images = [] - - try: - if os.path.exists(image_directory): - # 获取所有未处理的图片文件 - all_files = [f for f in os.listdir(image_directory) - if os.path.isfile(os.path.join(image_directory, f))] - unprocessed_files = [f for f in all_files if not f.startswith("done_")] - unprocessed_files_sorted = sorted(unprocessed_files) - - for image_file in unprocessed_files_sorted: - full_path = os.path.join(image_directory, image_file) - try: - with open(full_path, "rb") as img: - images.append(img.read()) - - # 重命名已处理的文件 - timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") - new_filename = f"done_{timestamp}_{image_file}" - new_path = os.path.join(image_directory, new_filename) - os.rename(full_path, new_path) - - except Exception as e: - logger.warning(f"处理图片文件 {image_file} 失败: {str(e)}") - - # 如果没有图片文件，尝试获取表情包 - if not images: - image = await emoji_api.get_by_description(message) - if image: - image_base64, description, scene = image - image_data = base64.b64decode(image_base64) - images.append(image_data) - - except Exception as e: - logger.error(f"加载图片失败: {str(e)}") - - return images - - async def read_feed(self, qq_account: str, target_qq: str, num: int) -> List[Dict[str, Any]]: - """读取指定用户的说说""" - try: - # 获取API实例 - qzone_api = await self._get_qzone_api(qq_account) - if not qzone_api: - return [{"error": "无法获取QZone API"}] - - # 获取说说列表 - feeds_list = await qzone_api.get_feed_list(target_qq, num) - return feeds_list - - except Exception as e: - logger.error(f"读取说说失败: {str(e)}") - return [{"error": f"读取说说失败: {str(e)}"}] - - async def monitor_read_feed(self, qq_account: str, num: int) -> List[Dict[str, Any]]: - """监控读取所有好友的说说""" - try: - # 获取API实例 - qzone_api = await self._get_qzone_api(qq_account) - if not qzone_api: - return [] - - # 获取监控说说列表 - feeds_list = await qzone_api.get_monitor_feed_list(num) - return feeds_list - - except Exception as e: - logger.error(f"监控读取说说失败: {str(e)}") - return [] - - async def like_feed(self, qq_account: str, target_qq: str, fid: str) -> bool: - """点赞说说""" - try: - # 获取API实例 - qzone_api = await self._get_qzone_api(qq_account) - if not qzone_api: - return False - - # 点赞说说 - success = await qzone_api.like_feed(fid, target_qq) - return success - - except Exception as e: - logger.error(f"点赞说说失败: {str(e)}") - return False - - async def comment_feed(self, qq_account: str, target_qq: str, fid: str, content: str) -> bool: - """评论说说""" - try: - # 获取API实例 - qzone_api = await self._get_qzone_api(qq_account) - if not qzone_api: - return False - - # 评论说说 - success = await qzone_api.comment_feed(fid, target_qq, content) - return success - - except Exception as e: - logger.error(f"评论说说失败: {str(e)}") - return False - - -# ===== 辅助功能函数 ===== - -async def generate_image_by_sf(api_key: str, story: str, image_dir: str, batch_size: int = 1) -> bool: - """使用硅基流动API生成图片""" - try: - logger.info(f"正在生成图片，保存路径: {image_dir}") - - # 获取模型配置 - models = llm_api.get_available_models() - prompt_model = "replyer_1" - model_config = models.get(prompt_model) - - if not model_config: - logger.error('配置模型失败') - return False - - # 生成图片提示词 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_details = config_api.get_global_config("identity.identity_detail", "未知") - - success, prompt, reasoning, model_name = await llm_api.generate_with_model( - prompt=f""" - 请根据以下QQ空间说说内容配图，并构建生成配图的风格和prompt。 - 说说主人信息：'{bot_personality},{str(bot_details)}'。 - 说说内容:'{story}'。 - 请注意：仅回复用于生成图片的prompt，不要有其他的任何正文以外的冗余输出""", - model_config=model_config, - request_type="story.generate", - temperature=0.3, - max_tokens=1000 - ) - - if not success: - logger.error('生成说说配图prompt失败') - return False - - logger.info(f'即将生成说说配图：{prompt}') - - # 调用硅基流动API - sf_url = "https://api.siliconflow.cn/v1/images/generations" - sf_headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - sf_data = { - "model": "Kwai-Kolors/Kolors", - "prompt": prompt, - "negative_prompt": "lowres, bad anatomy, bad hands, text, error, cropped, worst quality, low quality, normal quality, jpeg artifacts, signature, watermark, username, blurry", - "image_size": "1024x1024", - "batch_size": batch_size, - "seed": random.randint(1, 9999999999), - "num_inference_steps": 20, - "guidance_scale": 7.5, - } - - res = requests.post(sf_url, headers=sf_headers, json=sf_data) - - if res.status_code != 200: - logger.error(f'生成图片出错，错误码: {res.status_code}') - return False - - json_data = res.json() - image_urls = [img["url"] for img in json_data["images"]] - - # 确保目录存在 - Path(image_dir).mkdir(parents=True, exist_ok=True) - - # 下载并保存图片 - for i, img_url in enumerate(image_urls): - try: - img_response = requests.get(img_url) - filename = f"sf_{i}_{int(time.time())}.png" - save_path = Path(image_dir) / filename - - with open(save_path, "wb") as f: - f.write(img_response.content) - - logger.info(f"图片已保存至: {save_path}") - - except Exception as e: - logger.error(f"下载图片失败: {str(e)}") - return False - - return True - - except Exception as e: - logger.error(f"生成图片失败: {str(e)}") - return False - - -async def get_send_history(qq_account: str) -> str: - """获取发送历史记录""" - try: - cookie_manager = CookieManager() - cookies = cookie_manager.load_cookies(qq_account) - - if not cookies: - return "" - - qzone_api = QZoneAPI(cookies) - - if not await qzone_api.validate_token(): - logger.error("Token验证失败") - return "" - - feeds_list = await qzone_api.get_feed_list(target_qq=qq_account, num=5) - - if not isinstance(feeds_list, list) or len(feeds_list) == 0: - return "" - - history_lines = ["==================="] - - for feed in feeds_list: - if not isinstance(feed, dict): - continue - - created_time = feed.get("created_time", "") - content = feed.get("content", "") - images = feed.get("images", []) - rt_con = feed.get("rt_con", "") - - if not rt_con: - history_lines.append( - f"\n时间：'{created_time}'\n说说内容：'{content}'\n图片：'{images}'\n===================" - ) - else: - history_lines.append( - f"\n时间: '{created_time}'\n转发了一条说说，内容为: '{rt_con}'\n图片: '{images}'\n对该说说的评论为: '{content}'\n===================" - ) - - return "".join(history_lines) - - except Exception as e: - logger.error(f"获取发送历史失败: {str(e)}") - return "" \ No newline at end of file diff --git a/src/plugins/built_in/maizone/scheduler.py b/src/plugins/built_in/maizone/scheduler.py deleted file mode 100644 index 5761258a7..000000000 --- a/src/plugins/built_in/maizone/scheduler.py +++ /dev/null @@ -1,303 +0,0 @@ -import asyncio -import datetime -import time -import traceback -import os -from typing import Dict, Any - -from src.common.logger import get_logger -from src.plugin_system.apis import llm_api, config_api -from src.manager.schedule_manager import schedule_manager -from src.common.database.sqlalchemy_database_api import get_db_session -from src.common.database.sqlalchemy_models import MaiZoneScheduleStatus -from sqlalchemy import select - -# 导入工具模块 -import sys -sys.path.append(os.path.dirname(__file__)) - -from qzone_utils import QZoneManager, get_send_history - -# 获取日志记录器 -logger = get_logger('MaiZone-Scheduler') - - -class ScheduleManager: - """定时任务管理器 - 根据日程表定时发送说说""" - - def __init__(self, plugin): - """初始化定时任务管理器""" - self.plugin = plugin - self.is_running = False - self.task = None - self.last_activity_hash = None # 记录上次处理的活动哈希，避免重复发送 - - logger.info("定时任务管理器初始化完成 - 将根据日程表发送说说") - - async def start(self): - """启动定时任务""" - if self.is_running: - logger.warning("定时任务已在运行中") - return - - self.is_running = True - self.task = asyncio.create_task(self._schedule_loop()) - logger.info("定时发送说说任务已启动 - 基于日程表") - - async def stop(self): - """停止定时任务""" - if not self.is_running: - return - - self.is_running = False - - if self.task: - self.task.cancel() - try: - await self.task - except asyncio.CancelledError: - logger.info("定时任务已被取消") - - logger.info("定时发送说说任务已停止") - - async def _schedule_loop(self): - """定时任务主循环 - 根据日程表检查活动""" - while self.is_running: - try: - # 检查定时任务是否启用 - if not self.plugin.get_config("schedule.enable_schedule", False): - logger.info("定时任务已禁用，等待下次检查") - await asyncio.sleep(60) - continue - - # 获取当前活动 - current_activity = schedule_manager.get_current_activity() - - if current_activity: - # 获取当前小时的时间戳格式 YYYY-MM-DD HH - current_datetime_hour = datetime.datetime.now().strftime("%Y-%m-%d %H") - - # 检查数据库中是否已经处理过这个小时的日程 - is_already_processed = await self._check_if_already_processed(current_datetime_hour, current_activity) - - if not is_already_processed: - logger.info(f"检测到新的日程活动: {current_activity} (时间: {current_datetime_hour})") - success, story_content = await self._execute_schedule_based_send(current_activity) - - # 更新处理状态到数据库 - await self._update_processing_status(current_datetime_hour, current_activity, success, story_content) - else: - logger.debug(f"当前小时的日程活动已处理过: {current_activity} (时间: {current_datetime_hour})") - else: - logger.debug("当前时间没有日程活动") - - # 每5分钟检查一次，避免频繁检查 - await asyncio.sleep(300) - - except asyncio.CancelledError: - logger.info("定时任务循环被取消") - break - except Exception as e: - logger.error(f"定时任务循环出错: {str(e)}") - logger.error(traceback.format_exc()) - # 出错后等待5分钟再重试 - await asyncio.sleep(300) - - async def _check_if_already_processed(self, datetime_hour: str, activity: str) -> bool: - """检查数据库中是否已经处理过这个小时的日程""" - try: - with get_db_session() as session: - # 查询是否存在已处理的记录 - query = session.query(MaiZoneScheduleStatus).filter( - MaiZoneScheduleStatus.datetime_hour == datetime_hour, - MaiZoneScheduleStatus.activity == activity, - MaiZoneScheduleStatus.is_processed == True - ).first() - - return query is not None - - except Exception as e: - logger.error(f"检查日程处理状态时出错: {str(e)}") - # 如果查询出错，为了安全起见返回False，允许重新处理 - return False - - async def _update_processing_status(self, datetime_hour: str, activity: str, success: bool, story_content: str = ""): - """更新日程处理状态到数据库""" - try: - with get_db_session() as session: - # 先查询是否已存在记录 - existing_record = session.query(MaiZoneScheduleStatus).filter( - MaiZoneScheduleStatus.datetime_hour == datetime_hour, - MaiZoneScheduleStatus.activity == activity - ).first() - - if existing_record: - # 更新现有记录 - existing_record.is_processed = True - existing_record.processed_at = datetime.datetime.now() - existing_record.send_success = success - if story_content: - existing_record.story_content = story_content - existing_record.updated_at = datetime.datetime.now() - else: - # 创建新记录 - new_record = MaiZoneScheduleStatus( - datetime_hour=datetime_hour, - activity=activity, - is_processed=True, - processed_at=datetime.datetime.now(), - story_content=story_content or "", - send_success=success - ) - session.add(new_record) - - session.commit() - logger.info(f"已更新日程处理状态: {datetime_hour} - {activity} - 成功: {success}") - - except Exception as e: - logger.error(f"更新日程处理状态时出错: {str(e)}") - - async def _execute_schedule_based_send(self, activity: str) -> tuple[bool, str]: - """根据日程活动执行发送任务，返回(成功状态, 故事内容)""" - try: - logger.info(f"根据日程活动生成说说: {activity}") - - # 生成基于活动的说说内容 - story = await self._generate_activity_story(activity) - if not story: - logger.error("生成活动相关说说内容失败") - return False, "" - - logger.info(f"基于日程活动生成说说内容: '{story}'") - - # 处理配图 - await self._handle_images(story) - - # 发送说说 - success = await self._send_scheduled_feed(story) - - if success: - logger.info(f"基于日程活动的说说发送成功: {story}") - else: - logger.error(f"基于日程活动的说说发送失败: {activity}") - - return success, story - - except Exception as e: - logger.error(f"执行基于日程的发送任务失败: {str(e)}") - return False, "" - - async def _generate_activity_story(self, activity: str) -> str: - """根据日程活动生成说说内容""" - try: - # 获取模型配置 - models = llm_api.get_available_models() - text_model = str(self.plugin.get_config("models.text_model", "replyer_1")) - model_config = models.get(text_model) - - if not model_config: - logger.error("未配置LLM模型") - return "" - - # 获取机器人信息 - bot_personality = config_api.get_global_config("personality.personality_core", "一个机器人") - bot_expression = config_api.get_global_config("expression.expression_style", "内容积极向上") - qq_account = config_api.get_global_config("bot.qq_account", "") - - # 构建基于活动的提示词 - prompt = f""" - 你是'{bot_personality}'，根据你当前的日程安排，你正在'{activity}'。 - 请基于这个活动写一条说说发表在qq空间上， - {bot_expression} - 说说内容应该自然地反映你正在做的事情或你的想法， - 不要刻意突出自身学科背景，不要浮夸，不要夸张修辞，可以适当使用颜文字， - 只输出一条说说正文的内容，不要有其他的任何正文以外的冗余输出 - - 注意： - - 如果活动是学习相关的，可以分享学习心得或感受 - - 如果活动是休息相关的，可以分享放松的感受 - - 如果活动是日常生活相关的，可以分享生活感悟 - - 让说说内容贴近你当前正在做的事情，显得自然真实 - """ - - # 添加历史记录避免重复 - prompt += "\n\n以下是你最近发过的说说，写新说说时注意不要在相隔不长的时间发送相似内容的说说\n" - history_block = await get_send_history(qq_account) - if history_block: - prompt += history_block - - # 生成内容 - success, story, reasoning, model_name = await llm_api.generate_with_model( - prompt=prompt, - model_config=model_config, - request_type="story.generate", - temperature=0.7, # 稍微提高创造性 - max_tokens=1000 - ) - - if success: - return story - else: - logger.error("生成基于活动的说说内容失败") - return "" - - except Exception as e: - logger.error(f"生成基于活动的说说内容异常: {str(e)}") - return "" - - async def _handle_images(self, story: str): - """处理定时说说配图""" - try: - enable_ai_image = bool(self.plugin.get_config("send.enable_ai_image", False)) - apikey = str(self.plugin.get_config("models.siliconflow_apikey", "")) - image_dir = str(self.plugin.get_config("send.image_directory", "./plugins/Maizone/images")) - image_num = int(self.plugin.get_config("send.ai_image_number", 1) or 1) - - if enable_ai_image and apikey: - from qzone_utils import generate_image_by_sf - await generate_image_by_sf( - api_key=apikey, - story=story, - image_dir=image_dir, - batch_size=image_num - ) - logger.info("基于日程活动的AI配图生成完成") - elif enable_ai_image and not apikey: - logger.warning('启用了AI配图但未填写API密钥') - - except Exception as e: - logger.error(f"处理基于日程的说说配图失败: {str(e)}") - - async def _send_scheduled_feed(self, story: str) -> bool: - """发送基于日程的说说""" - try: - # 获取配置 - qq_account = config_api.get_global_config("bot.qq_account", "") - enable_image = self.plugin.get_config("send.enable_image", False) - image_dir = str(self.plugin.get_config("send.image_directory", "./plugins/Maizone/images")) - - # 创建QZone管理器并发送 (定时任务不需要stream_id) - qzone_manager = QZoneManager() - success = await qzone_manager.send_feed(story, image_dir, qq_account, enable_image) - - if success: - logger.info(f"基于日程的说说发送成功: {story}") - else: - logger.error("基于日程的说说发送失败") - - return success - - except Exception as e: - logger.error(f"发送基于日程的说说失败: {str(e)}") - return False - - def get_status(self) -> Dict[str, Any]: - """获取定时任务状态""" - current_activity = schedule_manager.get_current_activity() - return { - "is_running": self.is_running, - "enabled": self.plugin.get_config("schedule.enable_schedule", False), - "schedule_mode": "based_on_daily_schedule", - "current_activity": current_activity, - "last_activity_hash": self.last_activity_hash - } \ No newline at end of file