""" 随机图片插件 支持命令触发和LLM工具调用 """ import asyncio import aiohttp import uuid from pathlib import Path from datetime import datetime from typing import List, Optional from loguru import logger from utils.plugin_base import PluginBase from utils.decorators import on_text_message from WechatHook import WechatHookClient class RandomImage(PluginBase): """随机图片插件""" description = "随机图片插件 - 支持黑丝、白丝、原神等图片" author = "Assistant" version = "1.0.0" def __init__(self): super().__init__() self.images_dir = None async def async_init(self): """异步初始化""" self.images_dir = Path(__file__).parent / "images" self.images_dir.mkdir(exist_ok=True) logger.success("随机图片插件初始化完成") async def _download_image(self, url: str) -> Optional[tuple]: """下载图片到本地，返回(本地路径, 原始URL)""" try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session: async with session.get(url) as response: if response.status == 200: content = await response.read() ts = datetime.now().strftime("%Y%m%d_%H%M%S") uid = uuid.uuid4().hex[:8] file_path = self.images_dir / f"random_{ts}_{uid}.jpg" with open(file_path, "wb") as f: f.write(content) logger.info(f"图片下载成功: {file_path}") return (str(file_path), url) except Exception as e: logger.error(f"下载图片失败: {e}") return None async def _fetch_random_image(self) -> Optional[str]: """从三个接口随机选一个，获取一张图片，返回本地路径""" import random # 三个随机图片接口 api_urls = [ "https://v2.xxapi.cn/api/heisi", "https://v2.xxapi.cn/api/meinvpic", "https://v2.xxapi.cn/api/yscos" ] # 随机选择一个接口 url = random.choice(api_urls) logger.info(f"随机选择接口: {url}") try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=10)) as session: async with session.get(url) as response: if response.status == 200: data = await response.json() if data.get("code") == 200 and "data" in data: image_url = data["data"] result = await self._download_image(image_url) if result: return result[0] # 返回本地路径 except Exception as e: logger.error(f"获取随机图片失败: {e}") return None async def _send_as_chat_record(self, bot: WechatHookClient, from_wxid: str, title: str, images: List[tuple]): """将图片以聊天记录格式发送""" try: import time import hashlib import xml.etree.ElementTree as ET is_group = from_wxid.endswith("@chatroom") recordinfo = ET.Element("recordinfo") ET.SubElement(recordinfo, "info").text = title ET.SubElement(recordinfo, "isChatRoom").text = "1" if is_group else "0" datalist = ET.SubElement(recordinfo, "datalist") datalist.set("count", str(len(images))) ET.SubElement(recordinfo, "desc").text = title ET.SubElement(recordinfo, "fromscene").text = "3" for i, (local_path, img_url) in enumerate(images): di = ET.SubElement(datalist, "dataitem") di.set("datatype", "5") di.set("dataid", uuid.uuid4().hex) src_local_id = str((int(time.time() * 1000) % 90000) + 10000) new_msg_id = str(int(time.time() * 1000) + i) create_time = str(int(time.time()) - len(images) + i) ET.SubElement(di, "srcMsgLocalid").text = src_local_id ET.SubElement(di, "sourcetime").text = time.strftime("%Y-%m-%d %H:%M", time.localtime(int(create_time))) ET.SubElement(di, "fromnewmsgid").text = new_msg_id ET.SubElement(di, "srcMsgCreateTime").text = create_time ET.SubElement(di, "sourcename").text = "图片助手" ET.SubElement(di, "sourceheadurl").text = "" ET.SubElement(di, "datatitle").text = f"图片 {i+1}" ET.SubElement(di, "datadesc").text = "点击查看" ET.SubElement(di, "datafmt").text = "url" ET.SubElement(di, "link").text = img_url ET.SubElement(di, "ischatroom").text = "1" if is_group else "0" weburlitem = ET.SubElement(di, "weburlitem") ET.SubElement(weburlitem, "thumburl").text = img_url ET.SubElement(weburlitem, "link").text = img_url ET.SubElement(weburlitem, "title").text = f"图片 {i+1}" ET.SubElement(weburlitem, "desc").text = "点击查看" dataitemsource = ET.SubElement(di, "dataitemsource") ET.SubElement(dataitemsource, "hashusername").text = hashlib.sha256(from_wxid.encode("utf-8")).hexdigest() record_xml = ET.tostring(recordinfo, encoding="unicode") appmsg_xml = "".join([ "", f"{title}", f"{title}", "19", "https://support.weixin.qq.com/cgi-bin/mmsupport-bin/readtemplate?t=page/favorite_record__w_unsupport", "", f"", "0", "" ]) await bot._send_data_async(11214, {"to_wxid": from_wxid, "content": appmsg_xml}) logger.success(f"已发送聊天记录: {title}") except Exception as e: logger.error(f"发送聊天记录失败: {e}") @on_text_message(priority=70) async def handle_message(self, bot: WechatHookClient, message: dict): """处理文本消息""" content = message.get("Content", "").strip() from_wxid = message.get("FromWxid", "") # 精确匹配命令 if content not in ["随机图片", "图来", "黑丝", "白丝"]: return True logger.info(f"收到随机图片请求: {content}") try: # 获取一张随机图片 image_path = await self._fetch_random_image() if image_path: # 直接发送图片 await bot.send_image(from_wxid, image_path) logger.success(f"已发送随机图片: {image_path}") else: await bot.send_text(from_wxid, "❌ 获取图片失败，请稍后重试") except Exception as e: logger.error(f"处理失败: {e}") await bot.send_text(from_wxid, f"❌ 处理失败: {str(e)}") return False def get_llm_tools(self) -> List[dict]: """返回LLM工具定义""" return [{ "type": "function", "function": { "name": "get_random_image", "description": "获取随机图片，从三个接口中随机选择一个返回一张图片", "parameters": { "type": "object", "properties": {}, "required": [] } } }] async def execute_llm_tool(self, tool_name: str, arguments: dict, bot: WechatHookClient, from_wxid: str) -> dict: """执行LLM工具调用""" if tool_name != "get_random_image": return None try: logger.info(f"LLM工具调用获取随机图片") # 获取一张随机图片 image_path = await self._fetch_random_image() if image_path: # 直接发送图片 await bot.send_image(from_wxid, image_path) return { "success": True, "message": "已发送随机图片", "no_reply": True # 不需要 AI 再回复 } else: return {"success": False, "message": "获取图片失败"} except Exception as e: logger.error(f"LLM工具执行失败: {e}") return {"success": False, "message": f"执行失败: {str(e)}"}