diff --git a/src/talking/README.md b/src/talking/README.md index 85dddd1..5c15f6a 100644 --- a/src/talking/README.md +++ b/src/talking/README.md @@ -1,3 +1,13 @@ -前往控制台 >> 插件管理 >> 插件配置，管理自定义回复 +# 自定义回复插件 + +前往控制台 >> 插件管理 >> 插件配置，管理自定义回复。 > 不校验前缀词。 + +## 功能特性 + +- 支持多种关键词匹配方式：包含关键词、精确匹配、正则表达式 +- 支持群聊/私聊分别配置 +- 支持变量替换：`{nickname}`、`{url:链接}` + +详细使用说明请查看 [README_USE.md](README_USE.md) diff --git a/src/talking/README_USE.md b/src/talking/README_USE.md index 54bfb49..c1c4986 100644 --- a/src/talking/README_USE.md +++ b/src/talking/README_USE.md @@ -1 +1,38 @@ -无使用说明 +## 自定义回复插件使用说明 + +### 基本使用 + +在控制台 >> 插件管理 >> 插件配置 中添加回复规则。 + +### 可用变量 + +| 变量 | 说明 | 示例 | +|------|------|------| +| `{nickname}` | 用户昵称 | `你好{nickname}！` | +| `{url:链接}` | 获取链接内容（支持图片和文本） | `今日壁纸：{url:https://example.com/img.jpg}` | + +### URL 变量说明 + +`{url:链接}` 会智能检测内容类型： + +- **图片链接**：自动发送图片（支持 PNG/JPG/GIF/WebP） +- **文本内容**：自动插入到回复中 +- **混合使用**：文本和图片可同时发送 + +### 使用示例 + +**基础问答：** +- 关键词：`你好` +- 回复：`你好{nickname}，很高兴见到你！` + +**发送图片：** +- 关键词：`壁纸` +- 回复：`今日壁纸：{url:https://example.com/wallpaper.jpg}` + +**获取网页内容：** +- 关键词：`天气` +- 回复：`今天的天气是：{url:https://wttr.in/Beijing?format=3}` + +**组合使用：** +- 关键词：`帮助` +- 回复：`{nickname}，这是帮助信息：{url:https://example.com/help}` diff --git a/src/talking/main.py b/src/talking/main.py index 0afd108..4be3bbf 100644 --- a/src/talking/main.py +++ b/src/talking/main.py @@ -1,5 +1,7 @@ import os import re +import time +import aiohttp from amiyabot import Message, Chain from core import AmiyaBotPluginInstance @@ -7,12 +9,142 @@ curr_dir = os.path.dirname(__file__) +# Image URL cache: url -> (text, image_url_or_path, timestamp) +_url_cache: dict[str, tuple[str, str | None, float]] = {} +_cache_timeout = 300 # 5 minutes cache + + +def _get_cached(url: str) -> tuple[str, str | None] | None: + """Get cached content if not expired.""" + if url in _url_cache: + text, img, timestamp = _url_cache[url] + if time.time() - timestamp < _cache_timeout: + return text, img + else: + # Remove expired cache + del _url_cache[url] + # Clean up temp file + if img and os.path.exists(img) and img.startswith('/tmp/'): + try: + os.remove(img) + except Exception: + pass + return None + + +def _is_image_by_content(data: bytes) -> tuple[bool, str]: + """Detect if content is an image by magic bytes. + + Returns: + tuple: (is_image, extension) + """ + if len(data) < 4: + return False, '' + # PNG, JPEG, GIF, WebP magic bytes + signatures = [ + (b'\x89PNG\r\n\x1a\n', '.png'), + (b'\xff\xd8\xff', '.jpg'), + (b'GIF87a', '.gif'), + (b'GIF89a', '.gif'), + (b'RIFF', '.webp'), # WebP starts with RIFF....WEBP + (b'BM', '.bmp'), # BMP + ] + for sig, ext in signatures: + if data.startswith(sig): + return True, ext + return False, '' + + +def _is_image_by_url(url: str) -> tuple[bool, str]: + """Detect if URL likely points to an image by extension.""" + image_exts = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.ico'} + lower = url.lower() + for ext in image_exts: + if lower.endswith(ext) or ('?' in lower and lower.split('?')[0].endswith(ext)): + return True, ext + return False, '' + + +async def fetch_url_content(url: str) -> tuple[str, str | None]: + """Fetch content from a URL with intelligent detection. + + Returns: + tuple: (text content, image path if it's an image, else None) + """ + # Check cache with expiration + cached = _get_cached(url) + if cached: + return cached + + cache_path = f'/tmp/url_content_{abs(hash(url))}' + + try: + # Always download and save content first + async with aiohttp.ClientSession() as session: + async with session.get( + url, + timeout=aiohttp.ClientTimeout(total=10), + ssl=False # Skip SSL verification for self-signed certificates + ) as response: + if response.status == 200: + data = await response.read() + + # Save raw content + with open(cache_path, 'wb') as f: + f.write(data) + + # Then check if it's an image by content + is_image, ext = _is_image_by_content(data) + + if is_image: + # Rename to proper extension + img_path = f'{cache_path}{ext}' + os.rename(cache_path, img_path) + _url_cache[url] = ('', img_path, time.time()) + return '', img_path + + # Return text content + text = data.decode('utf-8', errors='ignore') + _url_cache[url] = (text, None, time.time()) + return text, None + + except Exception as e: + print(f'[TalkingPlugin] Fetch URL error: {url}, error: {e}') + + _url_cache[url] = ('', None, time.time()) + return '', None + + +async def parse_reply_content(reply: str, data: Message) -> tuple[str, str | None]: + """Parse reply content, handling URL expressions and nickname placeholder. + + Returns: + tuple: (text content, image path if there's an image, else None) + """ + image_path = None + + # Handle URL expression: {url:https://example.com} + url_pattern = r'\{url:([^}]+)\}' + + for match in re.finditer(url_pattern, reply): + url = match.group(1) + content, img_path = await fetch_url_content(url) + if img_path: + image_path = img_path + reply = reply.replace(match.group(0), '') + elif content: + reply = reply.replace(match.group(0), content) + + # Replace nickname placeholder + return reply.replace('{nickname}', data.nickname), image_path + + class TalkPluginInstance(AmiyaBotPluginInstance): ... bot = TalkPluginInstance( name='自定义回复', - version='1.6', + version='1.7', plugin_id='amiyabot-talking', plugin_type='official', description='可以自定义一问一答的简单对话', @@ -56,4 +188,12 @@ async def _(data: Message): if os.path.exists(reply): return Chain(data, at=is_at).image(reply) - return Chain(data, at=is_at).text(reply.replace('{nickname}', data.nickname)) + reply, image_path = await parse_reply_content(reply, data) + + chain = Chain(data, at=is_at) + if image_path: + chain = chain.image(image_path) + if reply.strip(): + chain = chain.text(reply) + + return chain