shuiyuPyLab/xiaoli-cli
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat: 新增 7 个实用插件

Browse Source 
- 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 定义。
This commit is contained in:
shuiyu1123 committed 2026-04-30 03:01:11 +08:00
1 parent bef3fe560d
commit 78b74d1bb1
7 files changed
+3127
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

plugins/bookmark_manager.py
+489
View File
@@ -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 <URL> -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 <URL> - 获取网站 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()
plugins/color_picker.py
+459
View File
@@ -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 <RGB或HSL> - 转换为 HEX
rgb <HEX或HSL> - 转换为 RGB
hsl <HEX或RGB> - 转换为 HSL
hsv <HEX或RGB> - 转换为 HSV
cmyk <HEX或RGB> - 转换为 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()
plugins/id_generator.py
+383
View File
@@ -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()
plugins/json_toolkit.py
+386
View File
@@ -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 <JSON或文件> <路径>\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()
plugins/markdown_tool.py
+540
View File
@@ -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()
plugins/regex_tester.py
+442
View File
@@ -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()
plugins/snippet_manager.py
+428
View File
@@ -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()