diff --git a/plugins/github_opengraph/README.md b/plugins/github_opengraph/README.md new file mode 100644 index 0000000..018e75b --- /dev/null +++ b/plugins/github_opengraph/README.md @@ -0,0 +1,31 @@ +# GitHub OpenGraph 插件 + +## 功能说明 +自动识别群聊/私聊中的 `github.com` 链接,将其转换为 GitHub OpenGraph 预览图并发送图片消息。 + +## 触发规则 +1. `fuzzy` 模式:消息中只要包含 GitHub 链接就会触发。 +2. `exact` 模式:消息内容必须是完整 GitHub 链接才会触发。 + +## 权限控制 +插件注册了群功能开关: +- `FEATURE_KEY`: `GITHUB_OPENGRAPH` +- `FEATURE_DESCRIPTION`: `🧩 GitHub链接卡片 [自动转OpenGraph图片]` + +如果某群关闭该功能,该群消息不会触发处理。 + +## 配置项 +见 [config.toml](/D:/learn/abot/plugins/github_opengraph/config.toml): +- `enable`:总开关 +- `match_mode`:匹配模式(`fuzzy` / `exact`) +- `max_links_per_message`:单条消息最大处理链接数 +- `hash_salt`:OpenGraph 哈希盐值 +- `request_timeout_seconds`:拉图超时秒数 + +## 典型示例 +原链接: +`https://github.com/python/cpython/issues/12345` + +转换后: +`https://opengraph.githubassets.com//python/cpython/issues/12345` + diff --git a/plugins/github_opengraph/__init__.py b/plugins/github_opengraph/__init__.py new file mode 100644 index 0000000..4165b5c --- /dev/null +++ b/plugins/github_opengraph/__init__.py @@ -0,0 +1,8 @@ +# -*- coding: utf-8 -*- +from .main import GithubOpenGraphPlugin + + +def get_plugin(): + """返回插件实例。""" + return GithubOpenGraphPlugin() + diff --git a/plugins/github_opengraph/config.toml b/plugins/github_opengraph/config.toml new file mode 100644 index 0000000..071fd29 --- /dev/null +++ b/plugins/github_opengraph/config.toml @@ -0,0 +1,17 @@ +[GithubOpenGraph] +enable = true + +# 匹配模式: +# - fuzzy:在任意文本中提取 GitHub 链接(推荐) +# - exact:整条消息必须是 GitHub 链接才触发 +match_mode = "fuzzy" + +# 每条消息最多处理多少个 GitHub 链接,避免刷屏 +max_links_per_message = 3 + +# OpenGraph 哈希前缀盐值(可选,不填会基于链接本身计算) +hash_salt = "" + +# 拉取 OpenGraph 图片的超时时间(秒) +request_timeout_seconds = 15 + diff --git a/plugins/github_opengraph/main.py b/plugins/github_opengraph/main.py new file mode 100644 index 0000000..d290bd8 --- /dev/null +++ b/plugins/github_opengraph/main.py @@ -0,0 +1,267 @@ +# -*- coding: utf-8 -*- +import hashlib +import re +from typing import Any, Dict, List, Optional, Set, Tuple +from urllib.parse import urlparse + +import aiohttp +from loguru import logger + +from base.plugin_common.message_plugin_interface import MessagePluginInterface +from base.plugin_common.plugin_interface import PluginStatus +from utils.decorator.plugin_decorators import plugin_stats_decorator +from utils.robot_cmd.robot_command import GroupBotManager, PermissionStatus +from wechat_ipad.models.message import MessageType + + +class GithubOpenGraphPlugin(MessagePluginInterface): + """GitHub 链接 OpenGraph 图片插件。 + + 设计目标: + 1. 自动识别消息中的 GitHub 链接; + 2. 转换为 OpenGraph 图片地址并发送,便于群聊快速预览; + 3. 兼容群权限开关,满足不同群的启停需求。 + """ + + # 功能权限常量:用于接入现有“群插件权限开关”体系。 + FEATURE_KEY = "GITHUB_OPENGRAPH" + FEATURE_DESCRIPTION = "🧩 GitHub链接卡片 [自动转OpenGraph图片]" + + # 链接提取正则:提取消息中可能的 GitHub URL。 + GITHUB_URL_PATTERN = re.compile(r"https?://(?:www\.)?github\.com/[^\s<>\u3000]+", re.IGNORECASE) + + # 去除 URL 末尾常见标点,避免“链接后跟句号/括号”导致请求失败。 + TRAILING_PUNCTUATION = ".,!?;:,。!?;:'\"`)]}>" + + @property + def name(self) -> str: + return "GitHub OpenGraph" + + @property + def version(self) -> str: + return "1.0.0" + + @property + def description(self) -> str: + return "自动将GitHub链接转换为OpenGraph预览图片。" + + @property + def author(self) -> str: + return "ABOT Team" + + @property + def command_prefix(self) -> Optional[str]: + # 本插件采用“自动识别链接”模式,不使用命令前缀。 + return "" + + @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.LOG = logger + self.enable = True + self.match_mode = "fuzzy" + self.max_links_per_message = 3 + self.hash_salt = "" + self.request_timeout_seconds = 15 + # 注册插件功能权限,后续在群权限设置中可独立开关。 + self.feature = self.register_feature() + + def initialize(self, context: Dict[str, Any]) -> bool: + """初始化插件配置。""" + cfg = self._config.get("GithubOpenGraph", {}) + self.enable = bool(cfg.get("enable", True)) + self.match_mode = str(cfg.get("match_mode", "fuzzy") or "fuzzy").strip().lower() + if self.match_mode not in {"fuzzy", "exact"}: + self.LOG.warning(f"[{self.name}] match_mode={self.match_mode} 非法,已回退为 fuzzy") + self.match_mode = "fuzzy" + self.max_links_per_message = max(1, int(cfg.get("max_links_per_message", 3) or 3)) + self.hash_salt = str(cfg.get("hash_salt", "") or "") + self.request_timeout_seconds = max(3, int(cfg.get("request_timeout_seconds", 15) or 15)) + self.LOG.info( + f"[{self.name}] 初始化完成: enable={self.enable}, mode={self.match_mode}, " + f"max_links={self.max_links_per_message}, timeout={self.request_timeout_seconds}s" + ) + 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: + """快速判断是否需要进入处理流程。""" + if not self.enable: + return False + if message.get("type") != MessageType.TEXT: + return False + content = str(message.get("content", "") or "").strip() + if not content: + return False + + # fuzzy 模式:消息内出现 github.com 即可认为可能命中。 + if self.match_mode == "fuzzy": + return "github.com/" in content.lower() + + # exact 模式:整条消息必须是可解析的 GitHub 链接。 + normalized = self._normalize_github_url(content) + return bool(normalized) + + @plugin_stats_decorator(plugin_name="GitHub OpenGraph") + async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: + """处理文本消息并发送 GitHub OpenGraph 预览图。""" + room_id = str(message.get("roomid", "") or "").strip() + sender = str(message.get("sender", "") or "").strip() + target = room_id if room_id else sender + gbm: GroupBotManager = message.get("gbm") + bot = message.get("bot") + content = str(message.get("content", "") or "").strip() + + # 群聊场景权限检查:关闭时直接不处理。 + if room_id and gbm and self.feature and gbm.get_group_permission(room_id, self.feature) == PermissionStatus.DISABLED: + self.LOG.debug(f"[{self.name}] 群权限关闭,跳过: room_id={room_id}") + return False, "没有权限" + + github_links = self._extract_github_links(content) + if not github_links: + return False, "未匹配到GitHub链接" + + # 单条消息做去重并限制处理数量,避免多次发送同一张图。 + unique_links = self._deduplicate_keep_order(github_links)[: self.max_links_per_message] + sent_count = 0 + failed_links: List[str] = [] + + for link in unique_links: + og_url = self._build_opengraph_url(link) + image_bytes = await self._download_image_bytes(og_url) + if not image_bytes: + failed_links.append(link) + continue + + try: + # send_image_message 支持 bytes,内部会自动转 base64 上传。 + await bot.send_image_message(target, image_bytes) + sent_count += 1 + self.LOG.info(f"[{self.name}] 发送OpenGraph成功: target={target}, link={link}") + except Exception as e: + failed_links.append(link) + self.LOG.error(f"[{self.name}] 发送OpenGraph失败: target={target}, link={link}, error={e}") + + if sent_count <= 0: + return False, f"GitHub链接解析失败,失败数量={len(failed_links)}" + + summary = f"已发送{sent_count}张GitHub预览图" + if failed_links: + summary += f",失败{len(failed_links)}条" + return True, summary + + def _extract_github_links(self, content: str) -> List[str]: + """根据配置的匹配模式提取 GitHub 链接。""" + content = str(content or "").strip() + if not content: + return [] + + if self.match_mode == "exact": + exact_url = self._normalize_github_url(content) + return [exact_url] if exact_url else [] + + # fuzzy 模式:从文本中抽取所有候选链接。 + matches = self.GITHUB_URL_PATTERN.findall(content) + normalized: List[str] = [] + for raw in matches: + url = self._normalize_github_url(raw) + if url: + normalized.append(url) + return normalized + + def _normalize_github_url(self, raw_url: str) -> str: + """标准化 GitHub 链接。 + + 标准化规则: + 1. 去掉首尾空白和末尾标点; + 2. 仅接受 github.com / www.github.com; + 3. 必须包含 path(至少 /owner/repo 级别); + 4. 丢弃 query 与 fragment,降低重复预览概率。 + """ + if not raw_url: + return "" + + cleaned = str(raw_url).strip().rstrip(self.TRAILING_PUNCTUATION) + if not cleaned.lower().startswith(("http://", "https://")): + return "" + + try: + parsed = urlparse(cleaned) + except Exception: + return "" + + host = str(parsed.netloc or "").lower() + if host not in {"github.com", "www.github.com"}: + return "" + + # path 至少应包含 owner/repo,避免把 GitHub 首页当成预览链接。 + path = str(parsed.path or "").strip() + if not path or path == "/": + return "" + path_parts = [p for p in path.split("/") if p] + if len(path_parts) < 2: + return "" + + # 统一域名到 github.com,保留路径原貌。 + normalized_path = "/" + "/".join(path_parts) + return f"https://github.com{normalized_path}" + + def _build_opengraph_url(self, github_url: str) -> str: + """把 GitHub 链接转换成 OpenGraph 图片链接。""" + parsed = urlparse(github_url) + path = str(parsed.path or "/") + hash_text = f"{self.hash_salt}|{github_url}" if self.hash_salt else github_url + digest = hashlib.sha256(hash_text.encode("utf-8")).hexdigest()[:20] + return f"https://opengraph.githubassets.com/{digest}{path}" + + async def _download_image_bytes(self, image_url: str) -> bytes: + """下载 OpenGraph 图片,失败时返回空字节。""" + timeout = aiohttp.ClientTimeout(total=self.request_timeout_seconds) + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.get(image_url) as resp: + if resp.status != 200: + self.LOG.warning(f"[{self.name}] 拉取OpenGraph失败: status={resp.status}, url={image_url}") + return b"" + content_type = str(resp.headers.get("Content-Type", "") or "").lower() + if "image" not in content_type: + self.LOG.warning( + f"[{self.name}] 返回内容不是图片: content_type={content_type}, url={image_url}" + ) + return b"" + return await resp.read() + except Exception as e: + self.LOG.warning(f"[{self.name}] 拉取OpenGraph异常: url={image_url}, error={e}") + return b"" + + @staticmethod + def _deduplicate_keep_order(items: List[str]) -> List[str]: + """按顺序去重,保持原始出现顺序。""" + seen: Set[str] = set() + result: List[str] = [] + for item in items: + if item in seen: + continue + seen.add(item) + result.append(item) + return result +