From 78b74d1bb1da806dd69e35435e2d2aaa1930b978 Mon Sep 17 00:00:00 2001 From: shuiyu1123 Date: Thu, 30 Apr 2026 03:01:11 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=207=20=E4=B8=AA?= =?UTF-8?q?=E5=AE=9E=E7=94=A8=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - json_toolkit: JSON 工具箱(格式化/压缩/校验/路径查询/差异比较/Schema推断) - snippet_manager: 代码片段管理器(保存/搜索/分类/标签/导入导出) - regex_tester: 正则表达式测试器(匹配/提取/替换/解释/常用模式库/性能测试) - id_generator: UUID/ID 生成器(UUID v1/v4/v5/NanoID/短ID/雪花ID/ULID/自定义模板) - color_picker: 颜色工具(HEX/RGB/HSL/HSV/CMYK转换/WCAG对比度/调色板/色盲模拟) - bookmark_manager: 书签管理器(分类/标签/搜索/导入导出/可访问性检查) - markdown_tool: Markdown 工具(TOC/统计/表格生成/待办提取/语法检查/字数统计) 所有插件均支持 Plugin/Liugin 双协议和 MCP 定义。 --- plugins/bookmark_manager.py | 489 ++++++++++++++++++++++++++++++++ plugins/color_picker.py | 459 ++++++++++++++++++++++++++++++ plugins/id_generator.py | 383 +++++++++++++++++++++++++ plugins/json_toolkit.py | 386 ++++++++++++++++++++++++++ plugins/markdown_tool.py | 540 ++++++++++++++++++++++++++++++++++++ plugins/regex_tester.py | 442 +++++++++++++++++++++++++++++ plugins/snippet_manager.py | 428 ++++++++++++++++++++++++++++ 7 files changed, 3127 insertions(+) create mode 100644 plugins/bookmark_manager.py create mode 100644 plugins/color_picker.py create mode 100644 plugins/id_generator.py create mode 100644 plugins/json_toolkit.py create mode 100644 plugins/markdown_tool.py create mode 100644 plugins/regex_tester.py create mode 100644 plugins/snippet_manager.py diff --git a/plugins/bookmark_manager.py b/plugins/bookmark_manager.py new file mode 100644 index 0000000..e50b118 --- /dev/null +++ b/plugins/bookmark_manager.py @@ -0,0 +1,489 @@ +""" +书签管理器插件 - 管理开发相关链接和资源 +支持分类、标签、搜索、导入导出,JSON 持久化 +""" + +import json +import os +import hashlib +import time +from datetime import datetime +from typing import List, Optional + + +class Plugin: + """书签管理器 - 开发资源和链接管理""" + + def __init__(self): + self.usage = """书签管理器 + +操作: + add -n <名称> [-c 分类] [-t 标签1,标签2] [-d 描述] + - 添加书签 + list [-c 分类] [-t 标签] - 列出书签 + open <名称或ID> - 获取书签 URL + search <关键词> - 搜索书签 + delete <名称或ID> - 删除书签 + edit <名称或ID> [-n 名称] [-d 描述] [-c 分类] [-t 标签] + - 编辑书签 + categories - 列出所有分类 + tags - 列出所有标签 + export [文件] - 导出书签 + import <文件> - 导入书签 + stats - 统计信息 + recent [-n 数量] - 最近添加的书签 + check - 检查书签是否可访问 + favicon - 获取网站 favicon URL + +示例: + bookmark_manager add https://github.com -n GitHub -c 开发 -t 代码,托管 -d 代码托管平台 + bookmark_manager list -c 开发 + bookmark_manager search github + bookmark_manager open GitHub + bookmark_manager categories +""" + self.cli = None + self.bookmarks_file = "bookmarks.json" + self.bookmarks = [] + self._load_bookmarks() + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "bookmark_manager", + "description": "书签管理器 - 管理开发相关链接和资源,支持分类、标签、搜索", + "keywords": ["书签", "链接", "收藏", "bookmark", "网址", "资源", "网站", "URL"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "bookmark_manager", + "description": "书签管理器,支持添加、搜索、获取、删除书签", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["add", "list", "open", "search", "delete", + "edit", "categories", "tags", "export", + "import", "stats", "recent", "check"], + "description": "操作类型" + }, + "url": {"type": "string", "description": "URL"}, + "name": {"type": "string", "description": "书签名称"}, + "category": {"type": "string", "description": "分类"}, + "tags": {"type": "string", "description": "标签(逗号分隔)"}, + "description": {"type": "string", "description": "描述"}, + "keyword": {"type": "string", "description": "搜索关键词"}, + "file": {"type": "string", "description": "文件路径"}, + "count": {"type": "integer", "description": "数量"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + name = arguments.get("name", "") + return f"{op} {name}".strip() + + def _load_bookmarks(self): + try: + if os.path.exists(self.bookmarks_file): + with open(self.bookmarks_file, 'r', encoding='utf-8') as f: + self.bookmarks = json.load(f) + except Exception: + self.bookmarks = [] + + def _save_bookmarks(self): + try: + with open(self.bookmarks_file, 'w', encoding='utf-8') as f: + json.dump(self.bookmarks, f, ensure_ascii=False, indent=2) + return None + except Exception as e: + return f"保存失败: {e}" + + def _gen_id(self, url: str) -> str: + return hashlib.md5(f"{url}{time.time()}".encode()).hexdigest()[:8] + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return self._op_list([]) + + operation = parts[0].lower() + handlers = { + "add": self._op_add, + "list": self._op_list, + "open": self._op_open, + "search": self._op_search, + "delete": self._op_delete, + "edit": self._op_edit, + "categories": self._op_categories, + "tags": self._op_tags, + "export": self._op_export, + "import": self._op_import, + "stats": self._op_stats, + "recent": self._op_recent, + "check": self._op_check, + "favicon": self._op_favicon, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"书签管理错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _op_add(self, args: List[str]) -> str: + if not args: + return "错误:请提供 URL" + url = args[0] + name, category, tags, desc = "", "", [], "" + + i = 1 + while i < len(args): + if args[i] == "-n" and i + 1 < len(args): + name = args[i + 1]; i += 1 + elif args[i] == "-c" and i + 1 < len(args): + category = args[i + 1]; i += 1 + elif args[i] == "-t" and i + 1 < len(args): + tags = [t.strip() for t in args[i + 1].split(",") if t.strip()]; i += 1 + elif args[i] == "-d" and i + 1 < len(args): + desc = args[i + 1]; i += 1 + i += 1 + + if not name: + # 从 URL 推断名称 + from urllib.parse import urlparse + parsed = urlparse(url if '://' in url else 'https://' + url) + name = parsed.netloc.replace('www.', '').split('.')[0].capitalize() + url = parsed.geturl() + + # 检查重复 + for b in self.bookmarks: + if b["url"] == url: + return f"错误:URL 已存在 (书签: {b['name']})" + + bookmark = { + "id": self._gen_id(url), + "name": name, + "url": url, + "category": category, + "tags": tags, + "description": desc, + "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "access_count": 0 + } + self.bookmarks.append(bookmark) + err = self._save_bookmarks() + if err: + return err + return f"✅ 已添加书签 '{name}'\n URL: {url}\n 分类: {category or '未分类'}" + + def _op_list(self, args: List[str]) -> str: + cat_filter, tag_filter = None, None + i = 0 + while i < len(args): + if args[i] == "-c" and i + 1 < len(args): + cat_filter = args[i + 1].lower(); i += 1 + elif args[i] == "-t" and i + 1 < len(args): + tag_filter = args[i + 1].lower(); i += 1 + i += 1 + + filtered = self.bookmarks + if cat_filter: + filtered = [b for b in filtered if b.get("category", "").lower() == cat_filter] + if tag_filter: + filtered = [b for b in filtered if any(tag_filter in t.lower() for t in b.get("tags", []))] + + if not filtered: + return "📭 没有找到书签" + + # 按分类分组 + grouped = {} + for b in filtered: + cat = b.get("category", "未分类") + grouped.setdefault(cat, []).append(b) + + result = f"\n 书签列表 (共 {len(filtered)} 个)\n" + result += "═" * 55 + "\n" + + for cat, items in sorted(grouped.items()): + result += f"\n {cat} ({len(items)} 个)\n" + result += "─" * 45 + "\n" + for b in items: + tags = ", ".join(b.get("tags", [])) if b.get("tags") else "" + result += f" [{b['id']}] {b['name']}\n" + result += f" {b['url']}\n" + if tags: + result += f" ️ {tags}\n" + if b.get("description"): + result += f" {b['description'][:50]}\n" + result += "\n" + + return result.rstrip() + + def _op_open(self, args: List[str]) -> str: + if not args: + return "错误:请提供书签名称或 ID" + key = args[0] + bookmark = self._find_bookmark(key) + if not bookmark: + return f"❌ 未找到书签: {key}" + + bookmark["access_count"] = bookmark.get("access_count", 0) + 1 + self._save_bookmarks() + + return (f" {bookmark['name']}\n" + f" URL: {bookmark['url']}\n" + f" 访问次数: {bookmark.get('access_count', 0)}") + + def _op_search(self, args: List[str]) -> str: + if not args: + return "错误:请提供搜索关键词" + keyword = " ".join(args).lower() + + results = [] + for b in self.bookmarks: + searchable = f"{b['name']} {b['url']} {b.get('description', '')} {' '.join(b.get('tags', []))} {b.get('category', '')}".lower() + if keyword in searchable: + results.append(b) + + if not results: + return f" 未找到匹配 '{keyword}' 的书签" + + output = f" 搜索结果 ({len(results)} 个):\n" + output += "─" * 50 + "\n" + for b in results: + output += f" [{b['id']}] {b['name']} - {b['url']}\n" + if b.get("description"): + output += f" {b['description'][:40]}\n" + return output.rstrip() + + def _op_delete(self, args: List[str]) -> str: + if not args: + return "错误:请提供书签名称或 ID" + key = args[0] + for i, b in enumerate(self.bookmarks): + if b["name"].lower() == key.lower() or b["id"] == key: + removed = self.bookmarks.pop(i) + self._save_bookmarks() + return f" ️ 已删除书签: {removed['name']} ({removed['url']})" + return f"❌ 未找到书签: {key}" + + def _op_edit(self, args: List[str]) -> str: + if not args: + return "错误:请提供书签名称或 ID" + key = args[0] + bookmark = self._find_bookmark(key) + if not bookmark: + return f"❌ 未找到书签: {key}" + + i = 1 + while i < len(args): + if args[i] == "-n" and i + 1 < len(args): + bookmark["name"] = args[i + 1]; i += 1 + elif args[i] == "-d" and i + 1 < len(args): + bookmark["description"] = args[i + 1]; i += 1 + elif args[i] == "-c" and i + 1 < len(args): + bookmark["category"] = args[i + 1]; i += 1 + elif args[i] == "-t" and i + 1 < len(args): + bookmark["tags"] = [t.strip() for t in args[i + 1].split(",") if t.strip()]; i += 1 + i += 1 + + bookmark["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + self._save_bookmarks() + return f"✅ 已更新书签: {bookmark['name']}" + + def _op_categories(self, args: List[str]) -> str: + cat_count = {} + for b in self.bookmarks: + cat = b.get("category", "未分类") + cat_count[cat] = cat_count.get(cat, 0) + 1 + + if not cat_count: + return " 暂无分类" + + result = " 分类列表:\n" + for cat, count in sorted(cat_count.items(), key=lambda x: -x[1]): + bar = "█" * min(count, 20) + result += f" {cat:15s} {bar} ({count})\n" + return result.rstrip() + + def _op_tags(self, args: List[str]) -> str: + tag_count = {} + for b in self.bookmarks: + for t in b.get("tags", []): + tag_count[t] = tag_count.get(t, 0) + 1 + + if not tag_count: + return " ️ 暂无标签" + + result = " ️ 标签列表:\n" + for tag, count in sorted(tag_count.items(), key=lambda x: -x[1]): + result += f" {tag} ({count})\n" + return result.rstrip() + + def _op_export(self, args: List[str]) -> str: + output_file = args[0] if args else "bookmarks_export.json" + try: + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(self.bookmarks, f, ensure_ascii=False, indent=2) + return f"✅ 已导出 {len(self.bookmarks)} 个书签到: {output_file}" + except Exception as e: + return f"❌ 导出失败: {e}" + + def _op_import(self, args: List[str]) -> str: + if not args: + return "错误:请提供导入文件路径" + file_path = args[0] + if not os.path.isfile(file_path): + return f"❌ 文件不存在: {file_path}" + try: + with open(file_path, 'r', encoding='utf-8') as f: + imported = json.load(f) + if not isinstance(imported, list): + return "❌ 文件格式错误:需要 JSON 数组" + + existing_urls = {b["url"] for b in self.bookmarks} + added = 0 + for b in imported: + if isinstance(b, dict) and b.get("url") and b["url"] not in existing_urls: + b["id"] = self._gen_id(b["url"]) + self.bookmarks.append(b) + existing_urls.add(b["url"]) + added += 1 + + self._save_bookmarks() + return f"✅ 已导入 {added} 个书签 (跳过 {len(imported) - added} 个重复)" + except Exception as e: + return f"❌ 导入失败: {e}" + + def _op_stats(self, args: List[str]) -> str: + total = len(self.bookmarks) + if not total: + return "📊 暂无书签" + + cat_count = {} + total_access = 0 + for b in self.bookmarks: + cat = b.get("category", "未分类") + cat_count[cat] = cat_count.get(cat, 0) + 1 + total_access += b.get("access_count", 0) + + most_accessed = max(self.bookmarks, key=lambda b: b.get("access_count", 0)) + + result = f"📊 书签统计\n" + result += "─" * 35 + "\n" + result += f" 总数: {total}\n" + result += f" 总访问: {total_access} 次\n" + result += f" 分类数: {len(cat_count)}\n" + result += f" 最常访问: {most_accessed['name']} ({most_accessed.get('access_count', 0)} 次)\n" + result += f"\n 分类分布:\n" + for cat, count in sorted(cat_count.items(), key=lambda x: -x[1]): + result += f" {cat}: {count}\n" + return result.rstrip() + + def _op_recent(self, args: List[str]) -> str: + count = 5 + for i, a in enumerate(args): + if a == "-n" and i + 1 < len(args): + count = int(args[i + 1]) + + recent = sorted(self.bookmarks, key=lambda b: b.get("created_at", ""), reverse=True)[:count] + if not recent: + return "📭 暂无书签" + + result = f"⏰ 最近 {len(recent)} 个书签:\n" + for b in recent: + result += f" [{b['id']}] {b['name']} - {b.get('created_at', '')}\n" + return result.rstrip() + + def _op_check(self, args: List[str]) -> str: + import socket + from urllib.parse import urlparse + + bookmarks = self.bookmarks + if args: + # 检查指定书签 + bookmark = self._find_bookmark(args[0]) + if not bookmark: + return f"❌ 未找到书签: {args[0]}" + bookmarks = [bookmark] + + result = f" ️ 书签可访问性检查 ({len(bookmarks)} 个):\n" + result += "─" * 50 + "\n" + + for b in bookmarks[:20]: + url = b["url"] + try: + parsed = urlparse(url) + host = parsed.netloc or parsed.path.split('/')[0] + socket.create_connection((host, 80), timeout=3) + result += f" {b['name']} ({host})\n" + except (socket.timeout, socket.error, OSError): + result += f" {b['name']} ({host}) - 不可达\n" + + if len(bookmarks) > 20: + result += f" ... 仅检查前 20 个\n" + return result.rstrip() + + def _op_favicon(self, args: List[str]) -> str: + if not args: + return "错误:请提供 URL" + url = args[0] + if '://' not in url: + url = 'https://' + url + + from urllib.parse import urlparse + parsed = urlparse(url) + base = f"{parsed.scheme}://{parsed.netloc}" + favicon_url = f"{base}/favicon.ico" + + return (f" Favicon:\n" + f" 网站: {base}\n" + f" Favicon: {favicon_url}\n" + f" Google: https://www.google.com/s2/favicons?domain={parsed.netloc}&sz=64") + + def _find_bookmark(self, key: str) -> Optional[dict]: + for b in self.bookmarks: + if b["name"].lower() == key.lower() or b["id"] == key: + return b + return None + + +def test_plugin(): + plugin = Plugin() + print("书签管理器测试:") + print(plugin.handle('add https://github.com -n GitHub -c 开发 -t 代码,托管 -d 代码托管平台')) + print(plugin.handle('add https://stackoverflow.com -n StackOverflow -c 学习 -t 问答,编程')) + print(plugin.handle('list')) + print(plugin.handle('search github')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/color_picker.py b/plugins/color_picker.py new file mode 100644 index 0000000..45da008 --- /dev/null +++ b/plugins/color_picker.py @@ -0,0 +1,459 @@ +""" +颜色工具插件 - 颜色格式转换、调色板、对比度检查 +支持 HEX/RGB/HSL/HSV/CMYK 格式互转,WCAG 对比度检查 +""" + +import re +import colorsys +from typing import List, Tuple + + +class Plugin: + """颜色工具 - 格式转换、调色板、对比度检查""" + + def __init__(self): + self.usage = """颜色工具 + +操作: + convert <颜色> - 自动检测并转换为所有格式 + hex - 转换为 HEX + rgb - 转换为 RGB + hsl - 转换为 HSL + hsv - 转换为 HSV + cmyk - 转换为 CMYK + contrast <颜色1> <颜色2> - WCAG 对比度检查 + palette <颜色> [-n 数量] - 生成调色板(明暗变体) + complementary <颜色> - 互补色 + analogous <颜色> - 类似色 + triadic <颜色> - 三色组 + random [-n 数量] - 随机颜色 + mix <颜色1> <颜色2> [-w 权重] - 混合两种颜色 + blind <颜色> <类型> - 色盲模拟 + named - 列出 CSS 命名颜色 + +颜色格式: + HEX: #FF6600, #F60, FF6600 + RGB: rgb(255,102,0) 或 255,102,0 + HSL: hsl(24,100%,50%) + HSV: hsv(24,100%,100%) + CMYK: cmyk(0,60,100,0) + +示例: + color_picker convert "#3498db" + color_picker rgb "hsl(207,62%,54%)" + color_picker contrast "#FFFFFF" "#000000" + color_picker palette "#e74c3c" -n 9 + color_picker complementary "#3498db" +""" + self.cli = None + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "color_picker", + "description": "颜色工具 - HEX/RGB/HSL/HSV/CMYK 转换、对比度检查、调色板生成", + "keywords": ["颜色", "color", "HEX", "RGB", "HSL", "调色板", "对比度", "色盲", "转换"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "color_picker", + "description": "颜色工具,支持格式转换、对比度检查、调色板生成", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["convert", "hex", "rgb", "hsl", "hsv", "cmyk", + "contrast", "palette", "complementary", "analogous", + "triadic", "random", "mix", "blind"], + "description": "操作类型" + }, + "color": {"type": "string", "description": "颜色值"}, + "color2": {"type": "string", "description": "第二个颜色值"}, + "count": {"type": "integer", "description": "数量"}, + "weight": {"type": "number", "description": "混合权重 (0-1)"}, + "blind_type": {"type": "string", "description": "色盲类型"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + color = arguments.get("color", "") + return f"{op} {color}".strip() + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return self._op_random(["1"]) + + operation = parts[0].lower() + handlers = { + "convert": self._op_convert, + "hex": self._op_hex, + "rgb": self._op_rgb, + "hsl": self._op_hsl, + "hsv": self._op_hsv, + "cmyk": self._op_cmyk, + "contrast": self._op_contrast, + "palette": self._op_palette, + "complementary": self._op_complementary, + "analogous": self._op_analogous, + "triadic": self._op_triadic, + "random": self._op_random, + "mix": self._op_mix, + "blind": self._op_blind, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"颜色工具错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _parse_opts(self, args: List[str]) -> dict: + opts = {} + i = 0 + while i < len(args): + if args[i].startswith("-") and i + 1 < len(args): + key = args[i].lstrip("-") + opts[key] = args[i + 1] + i += 1 + i += 1 + return opts + + def _parse_color(self, color_str: str) -> Tuple[int, int, int]: + """解析颜色字符串为 RGB (0-255)""" + s = color_str.strip().lower() + + # HEX + hex_match = re.match(r'^#?([0-9a-f]{3,8})$', s) + if hex_match: + h = hex_match.group(1) + if len(h) == 3: + h = h[0]*2 + h[1]*2 + h[2]*2 + elif len(h) == 8: + h = h[:6] # ignore alpha + if len(h) == 6: + return (int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16)) + + # RGB + rgb_match = re.match(r'rgb\s*\(\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)\s*\)', s) + if rgb_match: + return (int(rgb_match.group(1)), int(rgb_match.group(2)), int(rgb_match.group(3))) + + # plain r,g,b + plain_match = re.match(r'^(\d+)\s*,\s*(\d+)\s*,\s*(\d+)$', s) + if plain_match: + return (int(plain_match.group(1)), int(plain_match.group(2)), int(plain_match.group(3))) + + # HSL + hsl_match = re.match(r'hsl\s*\(\s*(\d+)\s*,\s*(\d+)%,\s*(\d+)%\s*\)', s) + if hsl_match: + h = int(hsl_match.group(1)) / 360 + sl = int(hsl_match.group(2)) / 100 + l = int(hsl_match.group(3)) / 100 + r, g, b = colorsys.hls_to_rgb(h, l, sl) + return (int(r * 255), int(g * 255), int(b * 255)) + + # HSV/HSB + hsv_match = re.match(r'hsv\s*\(\s*(\d+)\s*,\s*(\d+)%,\s*(\d+)%\s*\)', s) + if hsv_match: + h = int(hsv_match.group(1)) / 360 + sv = int(hsv_match.group(2)) / 100 + v = int(hsv_match.group(3)) / 100 + r, g, b = colorsys.hsv_to_rgb(h, sv, v) + return (int(r * 255), int(g * 255), int(b * 255)) + + # CMYK + cmyk_match = re.match(r'cmyk\s*\(\s*(\d+)%?\s*,\s*(\d+)%?\s*,\s*(\d+)%?\s*,\s*(\d+)%?\s*\)', s) + if cmyk_match: + c = int(cmyk_match.group(1)) / 100 + m = int(cmyk_match.group(2)) / 100 + y = int(cmyk_match.group(3)) / 100 + k = int(cmyk_match.group(4)) / 100 + r = int(255 * (1 - c) * (1 - k)) + g = int(255 * (1 - m) * (1 - k)) + b = int(255 * (1 - y) * (1 - k)) + return (r, g, b) + + raise ValueError(f"无法解析颜色: {color_str}") + + def _rgb_to_hex(self, r: int, g: int, b: int) -> str: + return f"#{r:02X}{g:02X}{b:02X}" + + def _rgb_to_hsl(self, r: int, g: int, b: int) -> Tuple[int, int, int]: + h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) + return (int(h * 360), int(s * 100), int(l * 100)) + + def _rgb_to_hsv(self, r: int, g: int, b: int) -> Tuple[int, int, int]: + h, s, v = colorsys.rgb_to_hsv(r / 255, g / 255, b / 255) + return (int(h * 360), int(s * 100), int(v * 100)) + + def _rgb_to_cmyk(self, r: int, g: int, b: int) -> Tuple[int, int, int, int]: + if r == g == b == 0: + return (0, 0, 0, 100) + c = 1 - r / 255 + m = 1 - g / 255 + y = 1 - b / 255 + k = min(c, m, y) + c = int((c - k) / (1 - k) * 100) + m = int((m - k) / (1 - k) * 100) + y = int((y - k) / (1 - k) * 100) + k = int(k * 100) + return (c, m, y, k) + + def _luminance(self, r: int, g: int, b: int) -> float: + """计算相对亮度 (WCAG 2.0)""" + def linearize(c): + c = c / 255 + return c / 12.92 if c <= 0.03928 else ((c + 0.055) / 1.055) ** 2.4 + return 0.2126 * linearize(r) + 0.7152 * linearize(g) + 0.0722 * linearize(b) + + def _format_color_block(self, r: int, g: int, b: int) -> str: + """生成颜色预览块""" + return f" ■ RGB({r}, {g}, {b})" + + def _op_convert(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + h, s, l = self._rgb_to_hsl(r, g, b) + hh, sv, vv = self._rgb_to_hsv(r, g, b) + c, m, y, k = self._rgb_to_cmyk(r, g, b) + + result = f" 颜色转换: {args[0]}\n" + result += self._format_color_block(r, g, b) + "\n" + result += "─" * 35 + "\n" + result += f" HEX: {self._rgb_to_hex(r, g, b)}\n" + result += f" RGB: rgb({r}, {g}, {b})\n" + result += f" HSL: hsl({h}, {s}%, {l}%)\n" + result += f" HSV: hsv({hh}, {sv}%, {vv}%)\n" + result += f" CMYK: cmyk({c}%, {m}%, {y}%, {k}%)\n" + return result.rstrip() + + def _op_hex(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + return f" HEX: {self._rgb_to_hex(r, g, b)}" + + def _op_rgb(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + return f" RGB: rgb({r}, {g}, {b})" + + def _op_hsl(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + h, s, l = self._rgb_to_hsl(r, g, b) + return f" HSL: hsl({h}, {s}%, {l}%)" + + def _op_hsv(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + h, s, v = self._rgb_to_hsv(r, g, b) + return f" HSV: hsv({h}, {s}%, {v}%)" + + def _op_cmyk(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色值" + r, g, b = self._parse_color(args[0]) + c, m, y, k = self._rgb_to_cmyk(r, g, b) + return f" CMYK: cmyk({c}%, {m}%, {y}%, {k}%)" + + def _op_contrast(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:请提供两种颜色" + r1, g1, b1 = self._parse_color(args[0]) + r2, g2, b2 = self._parse_color(args[1]) + + l1 = self._luminance(r1, g1, b1) + l2 = self._luminance(r2, g2, b2) + lighter = max(l1, l2) + darker = min(l1, l2) + ratio = (lighter + 0.05) / (darker + 0.05) + + aa_large = ratio >= 3 + aa_normal = ratio >= 4.5 + aaa_large = ratio >= 4.5 + aaa_normal = ratio >= 7 + + result = f" WCAG 对比度检查\n" + result += "─" * 40 + "\n" + result += f" 颜色1: {self._rgb_to_hex(r1, g1, b1)} {self._format_color_block(r1, g1, b1)}\n" + result += f" 颜色2: {self._rgb_to_hex(r2, g2, b2)} {self._format_color_block(r2, g2, b1)}\n" + result += f" 对比度: {ratio:.2f}:1\n" + result += "─" * 40 + "\n" + result += f" {' ' if aa_normal else '❌'} AA 正文 (≥4.5:1)\n" + result += f" {' ' if aa_large else '❌'} AA 大字 (≥3:1)\n" + result += f" {' ' if aaa_normal else '❌'} AAA 正文 (≥7:1)\n" + result += f" {' ' if aaa_large else '❌'} AAA 大字 (≥4.5:1)\n" + return result.rstrip() + + def _op_palette(self, args: List[str]) -> str: + if not args: + return "错误:请提供基础颜色" + r, g, b = self._parse_color(args[0]) + opts = self._parse_opts(args[1:]) + count = int(opts.get("n", 9)) + count = max(3, min(count, 15)) + + h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) + + result = f" 调色板 (基于 {args[0]}):\n" + result += "─" * 50 + "\n" + + steps = count + for i in range(steps): + # 从暗到亮 + new_l = 0.1 + (0.8 * i / (steps - 1)) + nr, ng, nb = colorsys.hls_to_rgb(h, new_l, s) + nr, ng, nb = int(nr * 255), int(ng * 255), int(nb * 255) + hex_val = self._rgb_to_hex(nr, ng, nb) + bar = "█" * int(new_l * 20) + result += f" {hex_val} {bar} L:{int(new_l*100)}%\n" + + return result.rstrip() + + def _op_complementary(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色" + r, g, b = self._parse_color(args[0]) + h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) + # 互补色: 色相旋转180° + comp_h = (h + 0.5) % 1.0 + cr, cg, cb = colorsys.hls_to_rgb(comp_h, l, s) + cr, cg, cb = int(cr * 255), int(cg * 255), int(cb * 255) + + result = f" 互补色:\n" + result += f" 原色: {self._rgb_to_hex(r, g, b)}\n" + result += f" 互补: {self._rgb_to_hex(cr, cg, cb)}\n" + return result + + def _op_analogous(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色" + r, g, b = self._parse_color(args[0]) + h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) + + result = f" 类似色:\n" + for offset in [-30, -15, 0, 15, 30]: + ah = (h + offset / 360) % 1.0 + ar, ag, ab = colorsys.hls_to_rgb(ah, l, s) + ar, ag, ab = int(ar * 255), int(ag * 255), int(ab * 255) + marker = " ◀" if offset == 0 else "" + result += f" {self._rgb_to_hex(ar, ag, ab)} ({offset:+d}°){marker}\n" + return result.rstrip() + + def _op_triadic(self, args: List[str]) -> str: + if not args: + return "错误:请提供颜色" + r, g, b = self._parse_color(args[0]) + h, l, s = colorsys.rgb_to_hls(r / 255, g / 255, b / 255) + + result = f" 三色组:\n" + for i, offset in enumerate([0, 120, 240]): + th = (h + offset / 360) % 1.0 + tr, tg, tb = colorsys.hls_to_rgb(th, l, s) + tr, tg, tb = int(tr * 255), int(tg * 255), int(tb * 255) + result += f" {i+1}. {self._rgb_to_hex(tr, tg, tb)} ({offset}°)\n" + return result.rstrip() + + def _op_random(self, args: List[str]) -> str: + opts = self._parse_opts(args) + count = int(opts.get("n", 5)) + count = min(count, 20) + + result = f" 随机颜色 ({count} 个):\n" + for _ in range(count): + r, g, b = __import__('random').randint(0, 255), __import__('random').randint(0, 255), __import__('random').randint(0, 255) + result += f" {self._rgb_to_hex(r, g, b)} rgb({r}, {g}, {b})\n" + return result.rstrip() + + def _op_mix(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:请提供两种颜色" + r1, g1, b1 = self._parse_color(args[0]) + r2, g2, b2 = self._parse_color(args[1]) + opts = self._parse_opts(args[2:]) + w = float(opts.get("w", 0.5)) + w = max(0, min(1, w)) + + mr = int(r1 * (1 - w) + r2 * w) + mg = int(g1 * (1 - w) + g2 * w) + mb = int(b1 * (1 - w) + b2 * w) + + result = f" 颜色混合 (权重 {w}):\n" + result += f" 颜色1: {self._rgb_to_hex(r1, g1, b1)}\n" + result += f" 颜色2: {self._rgb_to_hex(r2, g2, b2)}\n" + result += f" 混合: {self._rgb_to_hex(mr, mg, mb)}\n" + return result + + def _op_blind(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:请提供颜色和色盲类型\n类型: protanopia, deuteranopia, tritanopia" + r, g, b = self._parse_color(args[0]) + blind_type = args[1].lower() + + # 简化的色盲模拟矩阵 + matrices = { + "protanopia": [[0.567, 0.433, 0.0], [0.558, 0.442, 0.0], [0.0, 0.242, 0.758]], + "deuteranopia": [[0.625, 0.375, 0.0], [0.7, 0.3, 0.0], [0.0, 0.3, 0.7]], + "tritanopia": [[0.95, 0.05, 0.0], [0.0, 0.433, 0.567], [0.0, 0.475, 0.525]], + } + + matrix = matrices.get(blind_type) + if not matrix: + return f"错误:未知类型 '{blind_type}'\n支持: protanopia, deuteranopia, tritanopia" + + nr = int(matrix[0][0] * r + matrix[0][1] * g + matrix[0][2] * b) + ng = int(matrix[1][0] * r + matrix[1][1] * g + matrix[1][2] * b) + nb = int(matrix[2][0] * r + matrix[2][1] * g + matrix[2][2] * b) + nr, ng, nb = max(0, min(255, nr)), max(0, min(255, ng)), max(0, min(255, nb)) + + result = f" 色盲模拟 ({blind_type}):\n" + result += f" 原色: {self._rgb_to_hex(r, g, b)}\n" + result += f" 模拟: {self._rgb_to_hex(nr, ng, nb)}\n" + return result + + +def test_plugin(): + plugin = Plugin() + print("颜色工具测试:") + print(plugin.handle('convert "#3498db"')) + print(plugin.handle('contrast "#FFFFFF" "#000000"')) + print(plugin.handle('palette "#e74c3c" -n 7')) + print(plugin.handle('complementary "#3498db"')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/id_generator.py b/plugins/id_generator.py new file mode 100644 index 0000000..b3c768a --- /dev/null +++ b/plugins/id_generator.py @@ -0,0 +1,383 @@ +""" +UUID / ID 生成器插件 - 生成各种格式的唯一标识符 +支持 UUID v1/v4/v5、NanoID、短ID、雪花ID、自定义格式 +""" + +import uuid +import hashlib +import time +import random +import string +import json +import os +import struct +from datetime import datetime +from typing import List + + +class Plugin: + """UUID / ID 生成器 - 多格式唯一标识符生成""" + + def __init__(self): + self.usage = """UUID / ID 生成器 + +操作: + uuid [-v 版本] [-n 数量] - 生成 UUID(默认 v4) + v1=时间戳 v4=随机 v5=命名 + nanoid [-s 长度] [-n 数量] - 生成 NanoID(默认21位) + short [-s 长度] [-n 数量] - 生成短ID(默认8位) + snowflake [-n 数量] - 生成雪花ID(时间排序) + custom <模板> [-n 数量] - 自定义格式生成 + 模板: {U}=大写字母 {L}=小写 {D}=数字 {X}=十六进制 + ulid [-n 数量] - 生成 ULID(时间排序+随机) + timestamp - 当前时间戳(秒/毫秒/微秒) + from_name <命名空间> <名称> - 基于名称生成确定性 UUID (v5) + batch <类型> [-n 数量] [-o 文件] - 批量生成并保存到文件 + +示例: + id_generator uuid + id_generator uuid -v 1 -n 5 + id_generator nanoid -s 16 + id_generator short -s 6 -n 10 + id_generator custom "USER-{D}{D}{D}{D}{U}{U}" -n 5 + id_generator snowflake -n 3 + id_generator batch uuid -n 100 -o ids.txt +""" + self.cli = None + self._machine_id = random.randint(0, 1023) + self._sequence = 0 + self._last_ts = 0 + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "id_generator", + "description": "UUID/ID 生成器 - 支持 UUID、NanoID、短ID、雪花ID、ULID、自定义格式", + "keywords": ["uuid", "id", "生成", "唯一", "标识", "nanoid", "snowflake", "ulid", "短id"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "id_generator", + "description": "UUID/ID 生成器,支持多种格式", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["uuid", "nanoid", "short", "snowflake", "custom", + "ulid", "timestamp", "from_name", "batch"], + "description": "操作类型" + }, + "version": {"type": "integer", "description": "UUID 版本 (1/4/5)"}, + "length": {"type": "integer", "description": "ID 长度"}, + "count": {"type": "integer", "description": "生成数量"}, + "template": {"type": "string", "description": "自定义模板"}, + "namespace": {"type": "string", "description": "命名空间"}, + "name": {"type": "string", "description": "名称"}, + "output": {"type": "string", "description": "输出文件"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + count = arguments.get("count", "") + parts = [op] + if count: + parts.extend(["-n", str(count)]) + return " ".join(parts) + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return self._op_uuid([]) + + operation = parts[0].lower() + handlers = { + "uuid": self._op_uuid, + "nanoid": self._op_nanoid, + "short": self._op_short, + "snowflake": self._op_snowflake, + "custom": self._op_custom, + "ulid": self._op_ulid, + "timestamp": self._op_timestamp, + "from_name": self._op_from_name, + "batch": self._op_batch, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"ID 生成错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _parse_opts(self, args: List[str]) -> dict: + opts = {} + i = 0 + while i < len(args): + if args[i].startswith("-") and i + 1 < len(args): + key = args[i].lstrip("-") + opts[key] = args[i + 1] + i += 1 + i += 1 + return opts + + def _op_uuid(self, args: List[str]) -> str: + opts = self._parse_opts(args) + version = int(opts.get("v", 4)) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + ids = [] + for _ in range(count): + if version == 1: + ids.append(str(uuid.uuid1())) + elif version == 4: + ids.append(str(uuid.uuid4())) + elif version == 5: + ids.append(str(uuid.uuid5(uuid.NAMESPACE_DNS, "example.com"))) + else: + return f"错误:不支持的 UUID 版本: {version}" + + header = f" UUID v{version} ({count} 个):\n" + if count <= 20: + return header + "\n".join(f" {u}" for u in ids) + else: + return header + "\n".join(f" {u}" for u in ids[:10]) + f"\n ... (共 {count} 个)" + + def _op_nanoid(self, args: List[str]) -> str: + opts = self._parse_opts(args) + size = int(opts.get("s", 21)) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + alphabet = string.ascii_letters + string.digits + "_-" + ids = [] + for _ in range(count): + ids.append(''.join(random.choices(alphabet, k=size))) + + header = f" NanoID (长度 {size}, {count} 个):\n" + if count <= 20: + return header + "\n".join(f" {i}" for i in ids) + else: + return header + "\n".join(f" {i}" for i in ids[:10]) + f"\n ... (共 {count} 个)" + + def _op_short(self, args: List[str]) -> str: + opts = self._parse_opts(args) + size = int(opts.get("s", 8)) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + chars = string.ascii_letters + string.digits + ids = [] + for _ in range(count): + ids.append(''.join(random.choices(chars, k=size))) + + header = f" 短ID (长度 {size}, {count} 个):\n" + if count <= 20: + return header + "\n".join(f" {i}" for i in ids) + else: + return header + "\n".join(f" {i}" for i in ids[:10]) + f"\n ... (共 {count} 个)" + + def _op_snowflake(self, args: List[str]) -> str: + opts = self._parse_opts(args) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + # Twitter Snowflake: 41位时间戳 + 10位机器ID + 12位序列号 + epoch = 1288834974657 # Twitter epoch (2010-11-04) + + ids = [] + for _ in range(count): + ts = int(time.time() * 1000) - epoch + if ts == self._last_ts: + self._sequence = (self._sequence + 1) & 0xFFF + if self._sequence == 0: + while ts <= self._last_ts: + ts = int(time.time() * 1000) - epoch + else: + self._sequence = 0 + self._last_ts = ts + + snowflake_id = (ts << 22) | (self._machine_id << 12) | self._sequence + ids.append(str(snowflake_id)) + + header = f"❄️ 雪花ID ({count} 个):\n" + if count <= 20: + return header + "\n".join(f" {i}" for i in ids) + else: + return header + "\n".join(f" {i}" for i in ids[:10]) + f"\n ... (共 {count} 个)" + + def _op_custom(self, args: List[str]) -> str: + if not args: + return "错误:请提供模板\n示例: custom \"ORDER-{D}{D}{D}{D}-{U}{U}{U}\"" + template = args[0] + opts = self._parse_opts(args[1:]) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + ids = [] + for _ in range(count): + result = [] + for ch in template: + if ch == '{': + continue # handled below + result.append(ch) + # Actually use proper template parsing + ids.append(self._expand_template(template)) + + header = f" ️ 自定义ID (模板: {template}, {count} 个):\n" + if count <= 20: + return header + "\n".join(f" {i}" for i in ids) + else: + return header + "\n".join(f" {i}" for i in ids[:10]) + f"\n ... (共 {count} 个)" + + def _expand_template(self, template: str) -> str: + result = [] + i = 0 + while i < len(template): + if template[i] == '{' and i + 2 < len(template) and template[i + 2] == '}': + token = template[i + 1] + if token == 'U': + result.append(random.choice(string.ascii_uppercase)) + elif token == 'L': + result.append(random.choice(string.ascii_lowercase)) + elif token == 'D': + result.append(random.choice(string.digits)) + elif token == 'X': + result.append(random.choice('0123456789ABCDEF')) + elif token == 'x': + result.append(random.choice('0123456789abcdef')) + elif token == 'A': + result.append(random.choice(string.ascii_letters)) + else: + result.append(template[i:i + 3]) + i += 3 + else: + result.append(template[i]) + i += 1 + return ''.join(result) + + def _op_ulid(self, args: List[str]) -> str: + opts = self._parse_opts(args) + count = int(opts.get("n", 1)) + count = min(count, 1000) + + # ULID: 48位时间戳 + 80位随机数, Crockford Base32 + alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + + ids = [] + for _ in range(count): + ts = int(time.time() * 1000) + # 时间部分 (10 chars) + t_chars = [] + t = ts + for _ in range(10): + t_chars.append(alphabet[t & 0x1F]) + t >>= 5 + t_chars.reverse() + # 随机部分 (16 chars) + r_chars = [random.choice(alphabet) for _ in range(16)] + ids.append(''.join(t_chars) + ''.join(r_chars)) + + header = f" ULID ({count} 个):\n" + if count <= 20: + return header + "\n".join(f" {i}" for i in ids) + else: + return header + "\n".join(f" {i}" for i in ids[:10]) + f"\n ... (共 {count} 个)" + + def _op_timestamp(self, args: List[str]) -> str: + now = time.time() + return (f"⏱ 当前时间戳:\n" + f" 秒: {int(now)}\n" + f" 毫秒: {int(now * 1000)}\n" + f" 微秒: {int(now * 1000000)}\n" + f" ISO: {datetime.now().isoformat()}") + + def _op_from_name(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: from_name <命名空间> <名称>\n命名空间: dns, url, oid, x500" + ns_map = {"dns": uuid.NAMESPACE_DNS, "url": uuid.NAMESPACE_URL, + "oid": uuid.NAMESPACE_OID, "x500": uuid.NAMESPACE_X500} + ns = ns_map.get(args[0].lower(), uuid.NAMESPACE_DNS) + name = args[1] + result = uuid.uuid5(ns, name) + return f" 确定性 UUID v5:\n 命名空间: {args[0]}\n 名称: {name}\n UUID: {result}" + + def _op_batch(self, args: List[str]) -> str: + if not args: + return "错误:请提供类型 (uuid/nanoid/short/snowflake/ulid)" + id_type = args[0] + opts = self._parse_opts(args[1:]) + count = int(opts.get("n", 10)) + output = opts.get("o", "") + + # 生成对应类型的ID + gen_args = [f"-n", str(count)] + handlers = { + "uuid": self._op_uuid, + "nanoid": self._op_nanoid, + "short": self._op_short, + "snowflake": self._op_snowflake, + "ulid": self._op_ulid, + } + handler = handlers.get(id_type) + if not handler: + return f"错误:不支持的类型 '{id_type}'" + + result = handler(gen_args) + + if output: + # 提取纯ID保存到文件 + ids = [line.strip().lstrip(' ') for line in result.split('\n') + if line.strip() and not any(c in line for c in [' ', '️', '', '❄️', ''])] + try: + with open(output, 'w', encoding='utf-8') as f: + f.write('\n'.join(ids)) + return f"{result}\n\n 已保存到: {output}" + except Exception as e: + return f"{result}\n\n 保存失败: {e}" + + return result + + +def test_plugin(): + plugin = Plugin() + print("ID 生成器测试:") + print(plugin.handle('uuid')) + print(plugin.handle('uuid -v 1 -n 3')) + print(plugin.handle('nanoid -s 16')) + print(plugin.handle('short -s 6 -n 5')) + print(plugin.handle('snowflake -n 3')) + print(plugin.handle('ulid')) + print(plugin.handle('timestamp')) + print(plugin.handle('custom "ORD-{D}{D}{D}{D}-{U}{U}{U}" -n 5')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/json_toolkit.py b/plugins/json_toolkit.py new file mode 100644 index 0000000..ced8273 --- /dev/null +++ b/plugins/json_toolkit.py @@ -0,0 +1,386 @@ +""" +JSON 工具箱插件 - JSON 格式化、校验、查询、转换 +提供 JSON 美化、压缩、验证、路径查询、类型转换等功能 +""" + +import json +import os +import re +from typing import List + + +class Plugin: + """JSON 工具箱 - 格式化/校验/查询/转换""" + + def __init__(self): + self.usage = """JSON 工具箱 + +操作: + format <文件或JSON> - 美化 JSON(自动缩进) + minify <文件或JSON> - 压缩 JSON(去除空白) + validate <文件或JSON> - 校验 JSON 是否合法 + query <文件或JSON> <路径> - 按路径查询值(如 users[0].name) + keys <文件或JSON> - 列出所有键(扁平化) + diff <文件1> <文件2> - 比较两个 JSON 的差异 + to_csv <文件或JSON> - 将 JSON 数组转为 CSV + merge <文件1> <文件2> - 合并两个 JSON 对象 + sample <文件或JSON> [-n 数量] - 从数组中随机采样 + schema <文件或JSON> - 推断 JSON Schema + +示例: + json_toolkit format data.json + json_toolkit validate '{"name": "test"}' + json_toolkit query config.json "database.host" + json_toolkit keys data.json + json_toolkit diff a.json b.json +""" + self.cli = None + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "json_toolkit", + "description": "JSON 工具箱 - 格式化、压缩、校验、路径查询、键提取、差异比较、转换", + "keywords": ["json", "格式化", "校验", "查询", "格式", "转换", "美化", "压缩", "validate", "format"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "json_toolkit", + "description": "JSON 工具箱,支持格式化、压缩、校验、路径查询、键提取、差异比较", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["format", "minify", "validate", "query", "keys", + "diff", "to_csv", "merge", "sample", "schema"], + "description": "操作类型" + }, + "input": {"type": "string", "description": "JSON 字符串或文件路径"}, + "path": {"type": "string", "description": "查询路径(query 操作用)"}, + "input2": {"type": "string", "description": "第二个 JSON 输入(diff/merge 用)"}, + "count": {"type": "integer", "description": "采样数量(sample 操作用)"} + }, + "required": ["operation", "input"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + inp = arguments.get("input", "") + path = arguments.get("path", "") + parts = [op, inp] + if path: + parts.append(path) + return " ".join(parts) + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return "错误:请提供操作类型" + + operation = parts[0].lower() + handlers = { + "format": self._op_format, + "minify": self._op_minify, + "validate": self._op_validate, + "query": self._op_query, + "keys": self._op_keys, + "diff": self._op_diff, + "to_csv": self._op_to_csv, + "merge": self._op_merge, + "sample": self._op_sample, + "schema": self._op_schema, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'\n支持: {', '.join(handlers.keys())}" + return handler(parts[1:]) + + except Exception as e: + return f"JSON 工具错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _load_json(self, text: str): + """加载 JSON(支持文件路径或 JSON 字符串)""" + if os.path.isfile(text): + with open(text, 'r', encoding='utf-8') as f: + return json.load(f), f"文件: {text}" + try: + return json.loads(text), "JSON 字符串" + except json.JSONDecodeError as e: + raise ValueError(f"无效的 JSON: {e}") + + def _op_format(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, source = self._load_json(args[0]) + formatted = json.dumps(data, indent=2, ensure_ascii=False) + return f"✅ 格式化结果 ({source}):\n{formatted}" + + def _op_minify(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, source = self._load_json(args[0]) + minified = json.dumps(data, separators=(',', ':'), ensure_ascii=False) + original_len = len(json.dumps(data, indent=2, ensure_ascii=False)) + saved = original_len - len(minified) + return f"✅ 压缩结果 ({source}):\n{minified}\n\n📊 大小: {len(minified)} 字符 (节省 {saved} 字符, {saved/original_len*100:.1f}%)" + + def _op_validate(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + text = args[0] + try: + if os.path.isfile(text): + with open(text, 'r', encoding='utf-8') as f: + data = json.load(f) + source = f"文件: {text}" + else: + data = json.loads(text) + source = "JSON 字符串" + + depth = self._get_depth(data) + keys_count = self._count_keys(data) + + result = f"✅ JSON 合法 ({source})\n" + result += f" 类型: {type(data).__name__}\n" + result += f" 嵌套深度: {depth}\n" + result += f" 键总数: {keys_count}" + return result + except json.JSONDecodeError as e: + return f"❌ JSON 无效:\n 行 {e.lineno}, 列 {e.colno}: {e.msg}" + except Exception as e: + return f"❌ 验证失败: {e}" + + def _op_query(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: query <路径>\n示例: query data.json users[0].name" + data, _ = self._load_json(args[0]) + path = args[1] + + result = data + parts = re.findall(r'\w+|\[\d+\]', path) + current_path = "" + + for part in parts: + if part.startswith('['): + idx = int(part[1:-1]) + current_path += part + if isinstance(result, list) and 0 <= idx < len(result): + result = result[idx] + else: + return f"❌ 路径 '{current_path}' 无效: 索引越界或非数组" + else: + current_path += f".{part}" if current_path else part + if isinstance(result, dict) and part in result: + result = result[part] + else: + return f"❌ 路径 '{current_path}' 无效: 键不存在" + + output = json.dumps(result, indent=2, ensure_ascii=False) if isinstance(result, (dict, list)) else str(result) + return f"🔍 查询结果 ({path}):\n{output}" + + def _op_keys(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, _ = self._load_json(args[0]) + + keys = [] + self._flatten_keys(data, "", keys) + + result = f"🔑 键列表 (共 {len(keys)} 个):\n" + for k in keys: + result += f" {k}\n" + return result.rstrip() + + def _flatten_keys(self, obj, prefix, keys): + if isinstance(obj, dict): + for k, v in obj.items(): + full_key = f"{prefix}.{k}" if prefix else k + keys.append(full_key) + self._flatten_keys(v, full_key, keys) + elif isinstance(obj, list) and obj: + self._flatten_keys(obj[0], f"{prefix}[0]", keys) + + def _op_diff(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:请提供两个 JSON 文件或字符串" + data1, src1 = self._load_json(args[0]) + data2, src2 = self._load_json(args[1]) + + diffs = [] + self._compare(data1, data2, "", diffs) + + if not diffs: + return "✅ 两个 JSON 完全相同" + + result = f"🔀 差异比较 ({len(diffs)} 处不同):\n" + for d in diffs[:50]: + result += f" {d}\n" + if len(diffs) > 50: + result += f" ... 还有 {len(diffs) - 50} 处差异" + return result.rstrip() + + def _compare(self, a, b, path, diffs): + if type(a) != type(b): + diffs.append(f" 类型不同 @ {path or '根'}: {type(a).__name__} vs {type(b).__name__}") + return + if isinstance(a, dict): + all_keys = set(a.keys()) | set(b.keys()) + for k in sorted(all_keys): + p = f"{path}.{k}" if path else k + if k not in a: + diffs.append(f"➕ 新增 @ {p}: {b[k]}") + elif k not in b: + diffs.append(f"➖ 删除 @ {p}: {a[k]}") + else: + self._compare(a[k], b[k], p, diffs) + elif isinstance(a, list): + if len(a) != len(b): + diffs.append(f" 数组长度不同 @ {path}: {len(a)} vs {len(b)}") + for i in range(min(len(a), len(b))): + self._compare(a[i], b[i], f"{path}[{i}]", diffs) + else: + if a != b: + diffs.append(f" 值不同 @ {path}: {a!r} → {b!r}") + + def _op_to_csv(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, _ = self._load_json(args[0]) + + if not isinstance(data, list) or not data: + return "错误:需要非空 JSON 数组" + if not isinstance(data[0], dict): + return "错误:数组元素必须是对象" + + headers = list(data[0].keys()) + lines = [",".join(headers)] + for item in data: + row = [] + for h in headers: + val = str(item.get(h, "")) + if "," in val or '"' in val or "\n" in val: + val = '"' + val.replace('"', '""') + '"' + row.append(val) + lines.append(",".join(row)) + + csv_content = "\n".join(lines) + return f"📊 CSV 转换结果 ({len(data)} 行):\n{csv_content}" + + def _op_merge(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:请提供两个 JSON 文件或字符串" + data1, _ = self._load_json(args[0]) + data2, _ = self._load_json(args[1]) + + if not isinstance(data1, dict) or not isinstance(data2, dict): + return "错误:合并仅支持 JSON 对象" + + merged = self._deep_merge(data1, data2) + return f"✅ 合并结果:\n{json.dumps(merged, indent=2, ensure_ascii=False)}" + + def _deep_merge(self, a, b): + result = a.copy() + for k, v in b.items(): + if k in result and isinstance(result[k], dict) and isinstance(v, dict): + result[k] = self._deep_merge(result[k], v) + else: + result[k] = v + return result + + def _op_sample(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, _ = self._load_json(args[0]) + count = 3 + for i, a in enumerate(args): + if a == "-n" and i + 1 < len(args): + count = int(args[i + 1]) + + if not isinstance(data, list): + return "错误:需要 JSON 数组" + + import random + sampled = random.sample(data, min(count, len(data))) + return f"🎲 随机采样 ({len(sampled)}/{len(data)}):\n{json.dumps(sampled, indent=2, ensure_ascii=False)}" + + def _op_schema(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 文件或字符串" + data, _ = self._load_json(args[0]) + schema = self._infer_schema(data) + return f"📋 推断 Schema:\n{json.dumps(schema, indent=2, ensure_ascii=False)}" + + def _infer_schema(self, obj, depth=0): + if depth > 10: + return {"type": "any"} + if isinstance(obj, dict): + props = {} + for k, v in obj.items(): + props[k] = self._infer_schema(v, depth + 1) + return {"type": "object", "properties": props} + elif isinstance(obj, list): + if obj: + return {"type": "array", "items": self._infer_schema(obj[0], depth + 1)} + return {"type": "array", "items": {}} + elif isinstance(obj, bool): + return {"type": "boolean"} + elif isinstance(obj, int): + return {"type": "integer"} + elif isinstance(obj, float): + return {"type": "number"} + else: + return {"type": "string"} + + def _get_depth(self, obj, depth=0): + if depth > 50: + return depth + if isinstance(obj, dict): + return max((self._get_depth(v, depth + 1) for v in obj.values()), default=depth) + elif isinstance(obj, list): + return max((self._get_depth(v, depth + 1) for v in obj), default=depth) + return depth + + def _count_keys(self, obj): + if isinstance(obj, dict): + return len(obj) + sum(self._count_keys(v) for v in obj.values()) + elif isinstance(obj, list): + return sum(self._count_keys(v) for v in obj) + return 0 + + +def test_plugin(): + plugin = Plugin() + print("JSON 工具箱测试:") + print(plugin.handle('validate \'{"name": "test", "age": 25}\'')) + print(plugin.handle('format \'{"name":"test","items":[1,2,3]}\'')) + print(plugin.handle('keys \'{"database": {"host": "localhost", "port": 3306}}\'')) + print(plugin.handle('query \'{"users": [{"name": "Alice"}]}\' users[0].name')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/markdown_tool.py b/plugins/markdown_tool.py new file mode 100644 index 0000000..8669fd2 --- /dev/null +++ b/plugins/markdown_tool.py @@ -0,0 +1,540 @@ +""" +Markdown 工具插件 - Markdown 渲染、目录生成、格式转换 +支持 TOC 生成、标题提取、表格生成、格式化、统计 +""" + +import re +import json +import os +from typing import List, Tuple + + +class Plugin: + """Markdown 工具 - TOC/渲染/格式转换/统计""" + + def __init__(self): + self.usage = """Markdown 工具 + +操作: + toc <文件或Markdown> - 生成目录 (Table of Contents) + headings <文件或Markdown> - 提取所有标题 + stats <文件或Markdown> - 统计信息(字数/行数/段落数等) + table <数据> - 从 JSON 数组生成 Markdown 表格 + checklist <文件> - 提取所有待办事项 + links <文件或Markdown> - 提取所有链接 + images <文件或Markdown> - 提取所有图片 + frontmatter <文件> - 提取 YAML frontmatter + format <文件或Markdown> - 格式化(标题层级、列表缩进等) + wordcount <文件或Markdown> - 详细字数统计 + outline <文件或Markdown> - 生成大纲视图 + lint <文件> - Markdown 语法检查 + +示例: + markdown_tool toc README.md + markdown_tool stats article.md + markdown_tool table '[{"name":"Alice","age":25},{"name":"Bob","age":30}]' + markdown_tool checklist TODO.md + markdown_tool links README.md + markdown_tool lint README.md +""" + self.cli = None + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "markdown_tool", + "description": "Markdown 工具 - TOC 生成、标题提取、统计、表格生成、语法检查", + "keywords": ["markdown", "md", "目录", "toc", "标题", "表格", "统计", "链接", "检查"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "markdown_tool", + "description": "Markdown 工具,支持 TOC、统计、表格生成、链接提取、语法检查", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["toc", "headings", "stats", "table", "checklist", + "links", "images", "frontmatter", "format", + "wordcount", "outline", "lint"], + "description": "操作类型" + }, + "input": {"type": "string", "description": "Markdown 文件路径或内容"}, + "data": {"type": "string", "description": "JSON 数据(table 操作用)"} + }, + "required": ["operation", "input"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + inp = arguments.get("input", "") + return f"{op} {inp}".strip() + + def _load_content(self, text: str) -> Tuple[str, str]: + """加载内容(文件路径或文本)""" + if os.path.isfile(text): + with open(text, 'r', encoding='utf-8') as f: + return f.read(), f"文件: {text}" + return text, "文本输入" + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return "错误:请提供操作类型" + + operation = parts[0].lower() + handlers = { + "toc": self._op_toc, + "headings": self._op_headings, + "stats": self._op_stats, + "table": self._op_table, + "checklist": self._op_checklist, + "links": self._op_links, + "images": self._op_images, + "frontmatter": self._op_frontmatter, + "format": self._op_format, + "wordcount": self._op_wordcount, + "outline": self._op_outline, + "lint": self._op_lint, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"Markdown 工具错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _extract_headings(self, content: str) -> List[Tuple[int, str, str]]: + """提取标题: [(level, text, slug)]""" + headings = [] + in_code = False + for line in content.split('\n'): + if line.strip().startswith('```'): + in_code = not in_code + continue + if in_code: + continue + match = re.match(r'^(#{1,6})\s+(.+)', line) + if match: + level = len(match.group(1)) + text = match.group(2).strip() + # 生成 slug + slug = re.sub(r'[^\w\s-]', '', text.lower()) + slug = re.sub(r'\s+', '-', slug.strip()) + headings.append((level, text, slug)) + return headings + + def _op_toc(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + headings = self._extract_headings(content) + + if not headings: + return "📭 未找到标题" + + result = f" 目录 ({source}):\n" + result += "─" * 40 + "\n" + for level, text, slug in headings: + indent = " " * (level - 1) + result += f"{indent}- [{text}](#{slug})\n" + return result.rstrip() + + def _op_headings(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + headings = self._extract_headings(content) + + if not headings: + return "📭 未找到标题" + + result = f" 标题列表 ({len(headings)} 个):\n" + for level, text, _ in headings: + prefix = "#" * level + result += f" {prefix:8s} {text}\n" + return result.rstrip() + + def _op_stats(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + + lines = content.split('\n') + total_lines = len(lines) + blank_lines = sum(1 for l in lines if not l.strip()) + code_blocks = content.count('```') // 2 + headings = len(self._extract_headings(content)) + links = len(re.findall(r'\[([^\]]+)\]\([^)]+\)', content)) + images = len(re.findall(r'!\[([^\]]*)\]\([^)]+\)', content)) + # 中文字符 + chinese = len(re.findall(r'[\u4e00-\u9fff]', content)) + # 英文单词 + english = len(re.findall(r'[a-zA-Z]+', content)) + + # 段落数(连续非空行) + paragraphs = 0 + in_para = False + for l in lines: + if l.strip(): + if not in_para: + paragraphs += 1 + in_para = True + else: + in_para = False + + result = f"📊 Markdown 统计 ({source})\n" + result += "─" * 35 + "\n" + result += f" 总行数: {total_lines}\n" + result += f" 空行: {blank_lines}\n" + result += f" 段落: {paragraphs}\n" + result += f" 代码块: {code_blocks}\n" + result += f" 标题: {headings}\n" + result += f" 链接: {links}\n" + result += f" 图片: {images}\n" + result += f" 中文字: {chinese}\n" + result += f" 英文词: {english}\n" + result += f" 字符数: {len(content)}\n" + return result.rstrip() + + def _op_table(self, args: List[str]) -> str: + if not args: + return "错误:请提供 JSON 数组" + data_str = args[0] + try: + data = json.loads(data_str) + except json.JSONDecodeError as e: + return f"❌ JSON 解析错误: {e}" + + if not isinstance(data, list) or not data: + return "❌ 需要非空 JSON 数组" + if not isinstance(data[0], dict): + return "❌ 数组元素必须是对象" + + headers = list(data[0].keys()) + # 计算列宽 + widths = {h: len(h) for h in headers} + for row in data: + for h in headers: + widths[h] = max(widths[h], len(str(row.get(h, "")))) + + # 生成表格 + header_line = "| " + " | ".join(h.ljust(widths[h]) for h in headers) + " |" + sep_line = "| " + " | ".join("-" * widths[h] for h in headers) + " |" + rows = [] + for row in data: + rows.append("| " + " | ".join(str(row.get(h, "")).ljust(widths[h]) for h in headers) + " |") + + result = f" Markdown 表格 ({len(data)} 行):\n\n" + result += header_line + "\n" + sep_line + "\n" + "\n".join(rows) + return result + + def _op_checklist(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件" + content, source = self._load_content(args[0]) + + todos = [] + done = [] + for i, line in enumerate(content.split('\n'), 1): + match = re.match(r'^(\s*)[-*]\s*\[([ xX])\]\s*(.+)', line) + if match: + indent = match.group(1) + checked = match.group(2).lower() == 'x' + text = match.group(3).strip() + if checked: + done.append((i, text)) + else: + todos.append((i, text)) + + result = f" 待办事项 ({source}):\n" + result += "─" * 45 + "\n" + result += f" ☑ 已完成: {len(done)} 个\n" + result += f" 未完成: {len(todos)} 个\n" + result += "─" * 45 + "\n" + + if todos: + result += "\n 未完成:\n" + for line_no, text in todos[:30]: + result += f" L{line_no:4d} {text}\n" + if done: + result += "\n 已完成:\n" + for line_no, text in done[:20]: + result += f" L{line_no:4d} {text}\n" + return result.rstrip() + + def _op_links(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + + links = re.findall(r'\[([^\]]+)\]\(([^)]+)\)', content) + if not links: + return f"📭 未找到链接 ({source})" + + result = f" 链接列表 ({len(links)} 个, {source}):\n" + result += "─" * 50 + "\n" + seen = set() + for text, url in links: + if url not in seen: + seen.add(url) + result += f" {text[:30]:30s} → {url}\n" + return result.rstrip() + + def _op_images(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + + images = re.findall(r'!\[([^\]]*)\]\(([^)]+)\)', content) + if not images: + return f"📭 未找到图片 ({source})" + + result = f" ️ 图片列表 ({len(images)} 个, {source}):\n" + result += "─" * 50 + "\n" + seen = set() + for alt, url in images: + if url not in seen: + seen.add(url) + result += f" {alt[:25]:25s} → {url}\n" + return result.rstrip() + + def _op_frontmatter(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件" + content, source = self._load_content(args[0]) + + match = re.match(r'^---\n(.*?)\n---', content, re.DOTALL) + if not match: + return f"📭 未找到 frontmatter ({source})" + + fm = match.group(1) + result = f" Frontmatter ({source}):\n" + result += "─" * 35 + "\n" + result += fm + return result + + def _op_format(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + lines = content.split('\n') + formatted = [] + prev_blank = False + + for line in lines: + stripped = line.strip() + # 确保标题前后有空行 + if re.match(r'^#{1,6}\s', stripped): + if formatted and not prev_blank: + formatted.append('') + formatted.append(stripped) + prev_blank = False + continue + # 确保列表项格式一致 + if re.match(r'^[-*]\s', stripped): + formatted.append(re.sub(r'^[-*]\s', '- ', stripped)) + prev_blank = False + continue + # 去除多余空行 + if not stripped: + if not prev_blank: + formatted.append('') + prev_blank = True + continue + + prev_blank = False + formatted.append(line.rstrip()) + + # 确保文件末尾有换行 + while formatted and not formatted[-1]: + formatted.pop() + formatted.append('') + + return f"✅ 格式化完成 ({source}):\n" + '\n'.join(formatted) + + def _op_wordcount(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + + # 移除代码块 + no_code = re.sub(r'```[\s\S]*?```', '', content) + # 移除行内代码 + no_code = re.sub(r'`[^`]+`', '', no_code) + # 移除链接标记 + no_code = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', no_code) + # 移除图片标记 + no_code = re.sub(r'!\[([^\]]*)\]\([^)]+\)', '', no_code) + # 移除标题标记 + no_code = re.sub(r'^#{1,6}\s+', '', no_code, flags=re.MULTILINE) + # 移除加粗/斜体标记 + no_code = re.sub(r'\*{1,2}([^*]+)\*{1,2}', r'\1', no_code) + + chinese = len(re.findall(r'[\u4e00-\u9fff]', no_code)) + english_words = re.findall(r'[a-zA-Z]+', no_code) + english = len(english_words) + total_chars = len(no_code.strip()) + + # 阅读时间估算(中文 300字/分钟, 英文 200词/分钟) + read_time_cn = chinese / 300 + read_time_en = english / 200 + read_time = read_time_cn + read_time_en + + result = f" 详细字数统计 ({source}):\n" + result += "─" * 35 + "\n" + result += f" 中文字数: {chinese}\n" + result += f" 英文单词: {english}\n" + result += f" 总字符数: {total_chars}\n" + result += f" 预计阅读: {read_time:.1f} 分钟\n" + return result.rstrip() + + def _op_outline(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件或内容" + content, source = self._load_content(args[0]) + headings = self._extract_headings(content) + + if not headings: + return "📭 未找到标题" + + result = f" 大纲视图 ({source}):\n" + result += "═" * 40 + "\n" + for level, text, _ in headings: + if level == 1: + result += f"\n {text}\n" + result += "─" * 30 + "\n" + elif level == 2: + result += f" {text}\n" + elif level == 3: + result += f" {text}\n" + else: + result += f" {'·' * (level - 3)} {text}\n" + return result.rstrip() + + def _op_lint(self, args: List[str]) -> str: + if not args: + return "错误:请提供 Markdown 文件" + file_path = args[0] + if not os.path.isfile(file_path): + return f"❌ 文件不存在: {file_path}" + + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + lines = content.split('\n') + + issues = [] + in_code = False + prev_blank = False + heading_levels = [] + + for i, line in enumerate(lines, 1): + stripped = line.strip() + + # 代码块跟踪 + if stripped.startswith('```'): + in_code = not in_code + continue + if in_code: + continue + + # 检查标题格式 + heading_match = re.match(r'^(#{1,6})\s+(.+)', line) + if heading_match: + level = len(heading_match.group(1)) + text = heading_match.group(2) + # 检查标题前后空行 + if i > 1 and lines[i - 2].strip() and not lines[i - 2].startswith('#'): + issues.append(f" L{i:4d} 标题前缺少空行") + # 检查标题层级跳跃 + if heading_levels and level > heading_levels[-1] + 1: + issues.append(f" L{i:4d} 标题层级跳跃 ({heading_levels[-1]} → {level})") + heading_levels.append(level) + # 检查标题末尾空格 + if text != text.rstrip(): + issues.append(f" L{i:4d} 标题末尾有多余空格") + else: + # 检查行尾空格 + if line != line.rstrip() and stripped: + trailing = len(line) - len(line.rstrip()) + if trailing >= 2: # 2个以上空格是换行符 + pass # 这是合法的 Markdown 换行 + elif trailing > 0: + issues.append(f" L{i:4d} 行尾有多余空格") + + # 检查连续空行 + if not stripped: + if prev_blank: + issues.append(f" L{i:4d} 连续空行") + prev_blank = True + else: + prev_blank = False + + # 检查列表格式 + list_match = re.match(r'^(\s*)[-*+]\s', line) + if list_match: + indent = len(list_match.group(1)) + if indent % 2 != 0 and indent > 0: + issues.append(f" L{i:4d} 列表缩进不是2的倍数") + + # 检查链接格式 + if re.search(r'\[[^\]]*\]\(\s*\)', line): + issues.append(f" L{i:4d} 链接URL为空") + + # 检查重复词 + words = stripped.lower().split() + for j in range(len(words) - 1): + if len(words[j]) > 2 and words[j] == words[j + 1]: + issues.append(f" L{i:4d} 重复词: '{words[j]}'") + + result = f" ️ Markdown 语法检查 ({file_path}):\n" + result += "─" * 45 + "\n" + + if not issues: + result += " 未发现问题!\n" + else: + result += f" 发现 {len(issues)} 个问题:\n\n" + for issue in issues[:50]: + result += f" {issue}\n" + if len(issues) > 50: + result += f"\n ... 还有 {len(issues) - 50} 个问题\n" + + return result.rstrip() + + +def test_plugin(): + plugin = Plugin() + print("Markdown 工具测试:") + print(plugin.handle('headings "# Hello\\n## World\\n### Sub"')) + print(plugin.handle('stats "# Hello\\n\\nThis is a test."')) + print(plugin.handle('table \'[{"name":"Alice","age":25},{"name":"Bob","age":30}]\'')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/regex_tester.py b/plugins/regex_tester.py new file mode 100644 index 0000000..75be488 --- /dev/null +++ b/plugins/regex_tester.py @@ -0,0 +1,442 @@ +""" +正则表达式测试器插件 - 测试、调试、管理正则表达式 +支持匹配测试、提取、替换、常用模式库 +""" + +import re +import json +import os +from typing import List + + +class Plugin: + """正则表达式测试器 - 测试、调试、管理正则""" + + def __init__(self): + self.usage = """正则表达式测试器 + +操作: + test <正则> <文本> [-f flags] - 测试正则是否匹配 + match <正则> <文本> [-f flags] - 提取所有匹配项 + replace <正则> <文本> <替换> - 测试替换 + groups <正则> <文本> - 显示分组信息 + explain <正则> - 解释正则含义 + flags <正则> <文本> - 测试不同标志组合 + pattern [名称] - 列出/查看常用模式 + save <名称> <正则> [-d 描述] - 保存自定义模式 + delete_pattern <名称> - 删除自定义模式 + bench <正则> <文本> [-n 次数] - 性能测试 + +标志 (flags): + i - 忽略大小写 m - 多行模式 s - 点匹配换行 + x - 详细模式 a - ASCII模式 + +示例: + regex_tester test "\\d+" "abc123def456" + regex_tester match "[a-z]+@[a-z]+\\.[a-z]+" "test@example.com foo@bar.org" + regex_tester groups "(\\w+)@(\\w+)\\.(\\w+)" "user@domain.com" + regex_tester explain "[a-z]+@\\w+\\.\\w+" + regex_tester pattern email +""" + self.cli = None + self.patterns_file = "regex_patterns.json" + self.custom_patterns = self._load_patterns() + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "regex_tester", + "description": "正则表达式测试器 - 测试匹配、提取、替换、解释、常用模式库", + "keywords": ["正则", "regex", "匹配", "表达式", "测试", "替换", "提取", "模式"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "regex_tester", + "description": "正则表达式测试器,支持匹配测试、提取、替换、解释、性能测试", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["test", "match", "replace", "groups", "explain", + "flags", "pattern", "save", "bench"], + "description": "操作类型" + }, + "pattern": {"type": "string", "description": "正则表达式"}, + "text": {"type": "string", "description": "测试文本"}, + "replacement": {"type": "string", "description": "替换文本"}, + "flags": {"type": "string", "description": "标志 (i/m/s/x)"}, + "name": {"type": "string", "description": "模式名称"}, + "description": {"type": "string", "description": "模式描述"}, + "count": {"type": "integer", "description": "测试次数"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + pattern = arguments.get("pattern", "") + text = arguments.get("text", "") + return f"{op} {pattern} {text}".strip() + + def _load_patterns(self): + try: + if os.path.exists(self.patterns_file): + with open(self.patterns_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception: + pass + return self._default_patterns() + + def _save_patterns(self): + try: + with open(self.patterns_file, 'w', encoding='utf-8') as f: + json.dump(self.custom_patterns, f, ensure_ascii=False, indent=2) + except Exception: + pass + + def _default_patterns(self): + return { + "email": {"pattern": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "desc": "邮箱地址"}, + "phone_cn": {"pattern": r"1[3-9]\d{9}", "desc": "中国手机号"}, + "url": {"pattern": r"https?://[^\s<>\"']+|www\.[^\s<>\"']+", "desc": "URL链接"}, + "ipv4": {"pattern": r"\b(?:\d{1,3}\.){3}\d{1,3}\b", "desc": "IPv4地址"}, + "date": {"pattern": r"\d{4}[-/]\d{1,2}[-/]\d{1,2}", "desc": "日期格式"}, + "time": {"pattern": r"\d{2}:\d{2}(:\d{2})?", "desc": "时间格式"}, + "chinese": {"pattern": r"[\u4e00-\u9fff]+", "desc": "中文字符"}, + "id_card": {"pattern": r"\d{17}[\dXx]", "desc": "身份证号"}, + "html_tag": {"pattern": r"<[^>]+>", "desc": "HTML标签"}, + "ip_port": {"pattern": r"\b(?:\d{1,3}\.){3}\d{1,3}:\d{1,5}\b", "desc": "IP:端口"}, + "mac_addr": {"pattern": r"([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}", "desc": "MAC地址"}, + "hex_color": {"pattern": r"#[0-9A-Fa-f]{3,8}", "desc": "十六进制颜色"}, + "number": {"pattern": r"-?\d+\.?\d*", "desc": "数字(整数/浮点)"}, + "word": {"pattern": r"\b\w+\b", "desc": "单词"}, + } + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return "错误:请提供操作类型" + + operation = parts[0].lower() + handlers = { + "test": self._op_test, + "match": self._op_match, + "replace": self._op_replace, + "groups": self._op_groups, + "explain": self._op_explain, + "flags": self._op_flags, + "pattern": self._op_pattern, + "save": self._op_save, + "delete_pattern": self._op_delete_pattern, + "bench": self._op_bench, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"正则测试错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _parse_flags(self, flags_str: str) -> int: + flags = 0 + for c in flags_str.lower(): + if c == 'i': flags |= re.IGNORECASE + elif c == 'm': flags |= re.MULTILINE + elif c == 's': flags |= re.DOTALL + elif c == 'x': flags |= re.VERBOSE + return flags + + def _op_test(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: test <正则> <文本> [-f 标志]" + pattern_str, text = args[0], args[1] + flags = 0 + for i, a in enumerate(args): + if a == "-f" and i + 1 < len(args): + flags = self._parse_flags(args[i + 1]) + + try: + pattern = re.compile(pattern_str, flags) + match = pattern.search(text) + if match: + return (f"✅ 匹配成功!\n" + f" 正则: {pattern_str}\n" + f" 位置: {match.start()}-{match.end()}\n" + f" 匹配: '{match.group()}'\n" + f" 全文: {pattern.findall(text)}") + else: + return f"❌ 无匹配\n 正则: {pattern_str}\n 文本: {text[:100]}" + except re.error as e: + return f"❌ 正则语法错误: {e}" + + def _op_match(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: match <正则> <文本>" + pattern_str, text = args[0], args[1] + flags = 0 + for i, a in enumerate(args): + if a == "-f" and i + 1 < len(args): + flags = self._parse_flags(args[i + 1]) + + try: + pattern = re.compile(pattern_str, flags) + matches = pattern.finditer(text) + + results = [] + for m in matches: + results.append(f" [{m.start():3d}-{m.end():3d}] '{m.group()}'") + + if results: + return f" 找到 {len(results)} 个匹配:\n" + "\n".join(results) + else: + return "❌ 未找到匹配" + except re.error as e: + return f"❌ 正则语法错误: {e}" + + def _op_replace(self, args: List[str]) -> str: + if len(args) < 3: + return "错误:格式: replace <正则> <文本> <替换>" + pattern_str, text, replacement = args[0], args[1], args[2] + + try: + result = re.sub(pattern_str, replacement, text) + count = len(re.findall(pattern_str, text)) + return (f" 替换结果 ({count} 处):\n" + f" 原文: {text}\n" + f" 结果: {result}") + except re.error as e: + return f"❌ 正则语法错误: {e}" + + def _op_groups(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: groups <正则> <文本>" + pattern_str, text = args[0], args[1] + + try: + pattern = re.compile(pattern_str) + match = pattern.search(text) + if not match: + return "❌ 无匹配" + + result = f" 分组信息:\n" + result += f" 完整匹配: '{match.group()}' (span: {match.span()})\n" + for i, group in enumerate(match.groups(), 1): + result += f" 分组 {i}: '{group}' (span: {match.span(i) if group else 'N/A'})\n" + if match.groupdict(): + result += f" 命名分组:\n" + for name, value in match.groupdict().items(): + result += f" {name}: '{value}'\n" + return result.rstrip() + except re.error as e: + return f"❌ 正则语法错误: {e}" + + def _op_explain(self, args: List[str]) -> str: + if not args: + return "错误:请提供正则表达式" + pattern_str = args[0] + + explanations = [] + i = 0 + while i < len(pattern_str): + ch = pattern_str[i] + + if ch == '\\': + if i + 1 < len(pattern_str): + next_ch = pattern_str[i + 1] + escape_map = { + 'd': '数字 [0-9]', 'D': '非数字', 'w': '字母/数字/下划线', + 'W': '非字母数字下划线', 's': '空白字符', 'S': '非空白字符', + 'b': '单词边界', 'B': '非单词边界', 'n': '换行', 't': '制表符', + } + if next_ch in escape_map: + explanations.append(('\\' + next_ch, escape_map[next_ch])) + else: + explanations.append(('\\' + next_ch, f'转义字符 {next_ch}')) + i += 2 + continue + + if ch == '.': + explanations.append(('.', '任意字符(除换行)')) + elif ch == '^': + explanations.append(('^', '行/字符串开头')) + elif ch == '$': + explanations.append(('$', '行/字符串结尾')) + elif ch == '*': + explanations.append(('*', '0次或多次')) + elif ch == '+': + explanations.append(('+', '1次或多次')) + elif ch == '?': + explanations.append(('?', '0次或1次')) + elif ch == '{': + end = pattern_str.find('}', i) + if end != -1: + quantifier = pattern_str[i:end + 1] + explanations.append((quantifier, f'量词: {quantifier}')) + i = end + 1 + continue + elif ch == '[': + end = pattern_str.find(']', i) + if end != -1: + char_class = pattern_str[i:end + 1] + explanations.append((char_class, f'字符类: {char_class}')) + i = end + 1 + continue + elif ch == '(': + if i + 1 < len(pattern_str) and pattern_str[i + 1] == '?': + if i + 2 < len(pattern_str): + special = pattern_str[i + 2] + if special == ':': + explanations.append(('(?:', '非捕获组')) + elif special == '=': + explanations.append(('(?=', '正向前瞻')) + elif special == '!': + explanations.append(('(?!', '负向前瞻')) + i += 3 + continue + explanations.append(('(', '捕获组开始')) + elif ch == ')': + explanations.append((')', '组结束')) + elif ch == '|': + explanations.append(('|', '或 (交替)')) + + i += 1 + + result = f" 正则解释: {pattern_str}\n" + result += "─" * 45 + "\n" + for token, meaning in explanations: + result += f" {token:12s} → {meaning}\n" + return result.rstrip() + + def _op_flags(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: flags <正则> <文本>" + pattern_str, text = args[0], args[1] + + flag_combos = [ + ("", 0, "默认"), + ("i", re.IGNORECASE, "忽略大小写"), + ("m", re.MULTILINE, "多行模式"), + ("s", re.DOTALL, "点匹配换行"), + ("im", re.IGNORECASE | re.MULTILINE, "忽略大小写+多行"), + ] + + result = f" ️ 标志测试: {pattern_str}\n" + result += "─" * 45 + "\n" + for flag_str, flag_val, desc in flag_combos: + try: + matches = re.findall(pattern_str, text, flag_val) + count = len(matches) if isinstance(matches, list) else 1 + result += f" [{flag_str or '-':3s}] {desc:12s} → {count} 个匹配: {matches[:3]}\n" + except re.error as e: + result += f" [{flag_str or '-':3s}] {desc:12s} → 错误: {e}\n" + return result.rstrip() + + def _op_pattern(self, args: List[str]) -> str: + if args: + name = args[0] + if name in self.custom_patterns: + p = self.custom_patterns[name] + return f" 模式 '{name}':\n 正则: {p['pattern']}\n 描述: {p.get('desc', '')}" + return f"❌ 未找到模式: {name}" + + result = " 常用正则模式:\n" + result += "─" * 50 + "\n" + for name, p in self.custom_patterns.items(): + result += f" {name:15s} {p['pattern'][:35]:35s} {p.get('desc', '')}\n" + return result.rstrip() + + def _op_save(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: save <名称> <正则> [-d 描述]" + name, pattern_str = args[0], args[1] + desc = "" + for i, a in enumerate(args): + if a == "-d" and i + 1 < len(args): + desc = args[i + 1] + + # 验证正则 + try: + re.compile(pattern_str) + except re.error as e: + return f"❌ 正则语法错误: {e}" + + self.custom_patterns[name] = {"pattern": pattern_str, "desc": desc} + self._save_patterns() + return f"✅ 已保存模式 '{name}': {pattern_str}" + + def _op_delete_pattern(self, args: List[str]) -> str: + if not args: + return "错误:请提供模式名称" + name = args[0] + if name in self.custom_patterns: + del self.custom_patterns[name] + self._save_patterns() + return f" ️ 已删除模式: {name}" + return f"❌ 未找到模式: {name}" + + def _op_bench(self, args: List[str]) -> str: + if len(args) < 2: + return "错误:格式: bench <正则> <文本> [-n 次数]" + pattern_str, text = args[0], args[1] + count = 10000 + for i, a in enumerate(args): + if a == "-n" and i + 1 < len(args): + count = int(args[i + 1]) + + try: + import time + pattern = re.compile(pattern_str) + + start = time.perf_counter() + for _ in range(count): + pattern.findall(text) + elapsed = time.perf_counter() - start + + per_op = elapsed / count * 1_000_000 # 微秒 + return (f"⏱ 性能测试:\n" + f" 正则: {pattern_str}\n" + f" 次数: {count}\n" + f" 总耗时: {elapsed:.3f}s\n" + f" 每次: {per_op:.2f}μs\n" + f" 吞吐: {count / elapsed:.0f} ops/s") + except re.error as e: + return f"❌ 正则语法错误: {e}" + + +def test_plugin(): + plugin = Plugin() + print("正则表达式测试器测试:") + print(plugin.handle('test "\\d+" "abc123def456"')) + print(plugin.handle('match "[a-z]+" "hello world foo bar"')) + print(plugin.handle('groups "(\\w+)@(\\w+)\\.(\\w+)" "user@domain.com"')) + print(plugin.handle('explain "\\d{4}-\\d{2}-\\d{2}"')) + + +if __name__ == "__main__": + test_plugin() diff --git a/plugins/snippet_manager.py b/plugins/snippet_manager.py new file mode 100644 index 0000000..8c62558 --- /dev/null +++ b/plugins/snippet_manager.py @@ -0,0 +1,428 @@ +""" +代码片段管理器插件 - 保存、搜索、复用常用代码片段 +支持分类、标签、搜索、导出,JSON 持久化存储 +""" + +import json +import os +import time +import hashlib +from datetime import datetime +from typing import List, Optional + + +class Plugin: + """代码片段管理器 - 保存和复用代码片段""" + + def __init__(self): + self.usage = """代码片段管理器 + +操作: + add <名称> -l <语言> -c "代码" [-t 标签1,标签2] [-d 描述] + - 添加代码片段 + list [-l 语言] [-t 标签] - 列出片段(可按语言/标签过滤) + get <名称或ID> - 获取片段详情 + search <关键词> - 搜索片段(名称/描述/代码/标签) + delete <名称或ID> - 删除片段 + edit <名称或ID> -c "新代码" - 更新片段代码 + export [文件] - 导出所有片段 + import <文件> - 导入片段 + tags - 列出所有标签 + stats - 统计信息 + copy <名称或ID> - 复制到剪贴板(如可用) + recent [-n 数量] - 最近添加的片段 + +示例: + snippet_manager add quick_sort -l python -c "def qsort(arr): ..." -t 排序,算法 + snippet_manager list -l python + snippet_manager search 排序 + snippet_manager get quick_sort +""" + self.cli = None + self.snippets_file = "snippets.json" + self.snippets = [] + self._load_snippets() + + def set_cli(self, cli): + self.cli = cli + + def get_tool_info(self): + return { + "name": "snippet_manager", + "description": "代码片段管理器 - 保存、搜索、复用常用代码片段,支持分类和标签", + "keywords": ["代码", "片段", "snippet", "模板", "复用", "保存", "搜索", "代码片段"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "snippet_manager", + "description": "代码片段管理器,支持添加、搜索、获取、删除代码片段", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["add", "list", "get", "search", "delete", + "edit", "export", "import", "tags", "stats", "recent"], + "description": "操作类型" + }, + "name": {"type": "string", "description": "片段名称或ID"}, + "language": {"type": "string", "description": "编程语言"}, + "code": {"type": "string", "description": "代码内容"}, + "tags": {"type": "string", "description": "标签(逗号分隔)"}, + "description": {"type": "string", "description": "描述"}, + "keyword": {"type": "string", "description": "搜索关键词"}, + "file": {"type": "string", "description": "文件路径"}, + "count": {"type": "integer", "description": "数量"} + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + name = arguments.get("name", "") + return f"{op} {name}".strip() + + def _load_snippets(self): + try: + if os.path.exists(self.snippets_file): + with open(self.snippets_file, 'r', encoding='utf-8') as f: + self.snippets = json.load(f) + except Exception: + self.snippets = [] + + def _save_snippets(self): + try: + with open(self.snippets_file, 'w', encoding='utf-8') as f: + json.dump(self.snippets, f, ensure_ascii=False, indent=2) + return None + except Exception as e: + return f"保存失败: {e}" + + def _gen_id(self, name: str) -> str: + return hashlib.md5(f"{name}{time.time()}".encode()).hexdigest()[:8] + + def handle(self, args: str) -> str: + try: + parts = self._parse_args(args) + if not parts: + return self._op_list([]) + + operation = parts[0].lower() + handlers = { + "add": self._op_add, + "list": self._op_list, + "get": self._op_get, + "search": self._op_search, + "delete": self._op_delete, + "edit": self._op_edit, + "export": self._op_export, + "import": self._op_import, + "tags": self._op_tags, + "stats": self._op_stats, + "recent": self._op_recent, + "copy": self._op_copy, + } + + handler = handlers.get(operation) + if not handler: + return f"错误:不支持的操作 '{operation}'" + return handler(parts[1:]) + + except Exception as e: + return f"片段管理错误: {str(e)}" + + def _parse_args(self, args: str) -> List[str]: + parts, current, in_quotes, qc = [], "", False, None + for ch in args: + if ch in ('"', "'") and not in_quotes: + in_quotes, qc = True, ch + elif ch == qc and in_quotes: + in_quotes, qc = False, None + elif ch == ' ' and not in_quotes: + if current: + parts.append(current) + current = "" + else: + current += ch + if current: + parts.append(current) + return parts + + def _op_add(self, args: List[str]) -> str: + if not args: + return "错误:请提供片段名称" + name = args[0] + language, code, tags, desc = "", "", [], "" + + i = 1 + while i < len(args): + if args[i] == "-l" and i + 1 < len(args): + language = args[i + 1]; i += 1 + elif args[i] == "-c" and i + 1 < len(args): + code = args[i + 1]; i += 1 + elif args[i] == "-t" and i + 1 < len(args): + tags = [t.strip() for t in args[i + 1].split(",") if t.strip()]; i += 1 + elif args[i] == "-d" and i + 1 < len(args): + desc = args[i + 1]; i += 1 + i += 1 + + if not code: + return "错误:请用 -c 提供代码内容" + + # 检查重名 + for s in self.snippets: + if s["name"] == name: + return f"错误:片段 '{name}' 已存在,用 edit 更新" + + snippet = { + "id": self._gen_id(name), + "name": name, + "language": language, + "code": code, + "tags": tags, + "description": desc, + "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "usage_count": 0 + } + self.snippets.append(snippet) + err = self._save_snippets() + if err: + return err + return f"✅ 已添加片段 '{name}' (ID: {snippet['id']}, 语言: {language or '未指定'})" + + def _op_list(self, args: List[str]) -> str: + lang_filter, tag_filter = None, None + i = 0 + while i < len(args): + if args[i] == "-l" and i + 1 < len(args): + lang_filter = args[i + 1].lower(); i += 1 + elif args[i] == "-t" and i + 1 < len(args): + tag_filter = args[i + 1].lower(); i += 1 + i += 1 + + filtered = self.snippets + if lang_filter: + filtered = [s for s in filtered if s.get("language", "").lower() == lang_filter] + if tag_filter: + filtered = [s for s in filtered if any(tag_filter in t.lower() for t in s.get("tags", []))] + + if not filtered: + return "📭 没有找到代码片段" + + result = f"\n📝 代码片段列表 (共 {len(filtered)} 个)\n" + result += "─" * 55 + "\n" + for s in filtered: + lang = s.get("language", "未知") + tags = ", ".join(s.get("tags", [])) if s.get("tags") else "" + code_preview = s.get("code", "")[:60].replace("\n", "↵") + result += f" [{s['id']}] {s['name']} ({lang})\n" + if tags: + result += f" ️ {tags}\n" + result += f" {code_preview}...\n" + result += f" ⏱ {s.get('created_at', '未知')}\n\n" + + result += "─" * 55 + return result + + def _op_get(self, args: List[str]) -> str: + if not args: + return "错误:请提供片段名称或ID" + key = args[0] + snippet = self._find_snippet(key) + if not snippet: + return f"❌ 未找到片段: {key}" + + snippet["usage_count"] = snippet.get("usage_count", 0) + 1 + self._save_snippets() + + result = f"\n 片段: {snippet['name']}\n" + result += "─" * 50 + "\n" + result += f" ID: {snippet['id']}\n" + result += f" 语言: {snippet.get('language', '未指定')}\n" + if snippet.get("tags"): + result += f" ️ 标签: {', '.join(snippet['tags'])}\n" + if snippet.get("description"): + result += f" 描述: {snippet['description']}\n" + result += f" 创建: {snippet.get('created_at', '未知')}\n" + result += f" 使用: {snippet.get('usage_count', 0)} 次\n" + result += "─" * 50 + "\n" + result += f" 代码:\n{snippet['code']}\n" + result += "─" * 50 + return result + + def _op_search(self, args: List[str]) -> str: + if not args: + return "错误:请提供搜索关键词" + keyword = " ".join(args).lower() + + results = [] + for s in self.snippets: + searchable = f"{s['name']} {s.get('description', '')} {s.get('code', '')} {' '.join(s.get('tags', []))} {s.get('language', '')}".lower() + if keyword in searchable: + results.append(s) + + if not results: + return f" 未找到匹配 '{keyword}' 的片段" + + output = f" 搜索结果 ({len(results)} 个):\n" + output += "─" * 50 + "\n" + for s in results: + lang = s.get("language", "") + output += f" [{s['id']}] {s['name']} ({lang}) - {s.get('description', '')[:40]}\n" + return output.rstrip() + + def _op_delete(self, args: List[str]) -> str: + if not args: + return "错误:请提供片段名称或ID" + key = args[0] + for i, s in enumerate(self.snippets): + if s["name"] == key or s["id"] == key: + removed = self.snippets.pop(i) + self._save_snippets() + return f" ️ 已删除片段: {removed['name']}" + return f"❌ 未找到片段: {key}" + + def _op_edit(self, args: List[str]) -> str: + if not args: + return "错误:请提供片段名称或ID" + key = args[0] + snippet = self._find_snippet(key) + if not snippet: + return f"❌ 未找到片段: {key}" + + new_code = None + new_desc = None + i = 1 + while i < len(args): + if args[i] == "-c" and i + 1 < len(args): + new_code = args[i + 1]; i += 1 + elif args[i] == "-d" and i + 1 < len(args): + new_desc = args[i + 1]; i += 1 + i += 1 + + if new_code: + snippet["code"] = new_code + if new_desc: + snippet["description"] = new_desc + snippet["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + self._save_snippets() + return f"✅ 已更新片段: {snippet['name']}" + + def _op_export(self, args: List[str]) -> str: + output_file = args[0] if args else "snippets_export.json" + try: + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(self.snippets, f, ensure_ascii=False, indent=2) + return f"✅ 已导出 {len(self.snippets)} 个片段到: {output_file}" + except Exception as e: + return f"❌ 导出失败: {e}" + + def _op_import(self, args: List[str]) -> str: + if not args: + return "错误:请提供导入文件路径" + file_path = args[0] + if not os.path.isfile(file_path): + return f"❌ 文件不存在: {file_path}" + try: + with open(file_path, 'r', encoding='utf-8') as f: + imported = json.load(f) + if not isinstance(imported, list): + return "❌ 文件格式错误:需要 JSON 数组" + + existing_names = {s["name"] for s in self.snippets} + added = 0 + for s in imported: + if isinstance(s, dict) and s.get("name") and s["name"] not in existing_names: + s["id"] = self._gen_id(s["name"]) + self.snippets.append(s) + existing_names.add(s["name"]) + added += 1 + + self._save_snippets() + return f"✅ 已导入 {added} 个片段 (跳过 {len(imported) - added} 个重复)" + except Exception as e: + return f"❌ 导入失败: {e}" + + def _op_tags(self, args: List[str]) -> str: + tag_count = {} + for s in self.snippets: + for t in s.get("tags", []): + tag_count[t] = tag_count.get(t, 0) + 1 + + if not tag_count: + return " ️ 暂无标签" + + result = " ️ 标签列表:\n" + for tag, count in sorted(tag_count.items(), key=lambda x: -x[1]): + result += f" {tag} ({count})\n" + return result.rstrip() + + def _op_stats(self, args: List[str]) -> str: + total = len(self.snippets) + if not total: + return "📊 暂无片段" + + lang_count = {} + total_usage = 0 + for s in self.snippets: + lang = s.get("language", "未知") + lang_count[lang] = lang_count.get(lang, 0) + 1 + total_usage += s.get("usage_count", 0) + + result = f"📊 片段统计\n" + result += "─" * 30 + "\n" + result += f" 总数: {total}\n" + result += f" 总使用: {total_usage} 次\n" + result += f" 语言分布:\n" + for lang, count in sorted(lang_count.items(), key=lambda x: -x[1]): + result += f" {lang}: {count}\n" + return result.rstrip() + + def _op_recent(self, args: List[str]) -> str: + count = 5 + for i, a in enumerate(args): + if a == "-n" and i + 1 < len(args): + count = int(args[i + 1]) + + recent = sorted(self.snippets, key=lambda s: s.get("created_at", ""), reverse=True)[:count] + if not recent: + return "📭 暂无片段" + + result = f"⏰ 最近 {len(recent)} 个片段:\n" + for s in recent: + result += f" [{s['id']}] {s['name']} - {s.get('created_at', '')}\n" + return result.rstrip() + + def _op_copy(self, args: List[str]) -> str: + if not args: + return "错误:请提供片段名称或ID" + snippet = self._find_snippet(args[0]) + if not snippet: + return f"❌ 未找到片段: {args[0]}" + + snippet["usage_count"] = snippet.get("usage_count", 0) + 1 + self._save_snippets() + return f" 代码内容 ({snippet['name']}):\n{snippet['code']}" + + def _find_snippet(self, key: str) -> Optional[dict]: + for s in self.snippets: + if s["name"] == key or s["id"] == key: + return s + return None + + +def test_plugin(): + plugin = Plugin() + print("代码片段管理器测试:") + print(plugin.handle('add hello -l python -c "print(\'Hello, World!\')" -t 示例,入门 -d 最简单的Python程序')) + print(plugin.handle('add fib -l python -c "def fib(n): return n if n < 2 else fib(n-1)+fib(n-2)" -t 算法,递归')) + print(plugin.handle('list')) + print(plugin.handle('search python')) + + +if __name__ == "__main__": + test_plugin()