message / plugins /text /api.py
hunian
refactor(plugins): 插件短名并统一 MCP tool 为 {plugin}-{tool}
cc826a1
Raw
History Blame Contribute Delete
4.64 kB
import logging
import re
from fastapi import APIRouter, Form, HTTPException, UploadFile
logger = logging.getLogger(__name__)
router = APIRouter()
plugin = None
def set_plugin_instance(plugin_instance):
global plugin
plugin = plugin_instance
@router.get("/status")
async def get_status():
if plugin is None:
return {
"name": "text",
"enabled": False,
"message": "插件未加载",
}
return plugin.get_status()
@router.post("/process")
async def process_text(text: str = Form(""), mode: str = Form("standard")):
if plugin is None or not plugin.enabled:
raise HTTPException(status_code=400, detail="插件未启用")
if not text:
raise HTTPException(status_code=400, detail="文本内容不能为空")
cleaned_text = clean_text(text, mode)
return {
"original_text": text,
"cleaned_text": cleaned_text,
"original_char_count": len(text),
"original_line_count": len(text.split("\n")),
"cleaned_char_count": len(cleaned_text),
"cleaned_line_count": len(cleaned_text.split("\n")),
}
@router.post("/upload")
async def upload_file(file: UploadFile):
if plugin is None or not plugin.enabled:
raise HTTPException(status_code=400, detail="插件未启用")
filename = file.filename.lower()
if not (filename.endswith(".md") or filename.endswith(".txt")):
raise HTTPException(status_code=400, detail="只支持 .md 和 .txt 文件")
content_bytes = await file.read()
try:
content = content_bytes.decode("utf-8")
except UnicodeDecodeError:
raise HTTPException(status_code=400, detail="文件编码不支持,请使用 UTF-8 编码")
return {
"filename": file.filename,
"content": content,
}
def clean_text(text: str, mode: str = "standard") -> str:
if mode == "standard":
lines = text.split("\n")
cleaned_lines = []
prev_line_type = None
for line in lines:
stripped = line.strip()
if stripped:
# 检查当前行类型
is_heading = re.match(r"^#+\s", stripped) is not None
is_numbered = re.match(r"^\d+[\.\)]\s+", stripped) is not None
is_bulleted = re.match(r"^[\-\*]\s+", stripped) is not None
current_line_type = (
"heading"
if is_heading
else (
"numbered"
if is_numbered
else ("bulleted" if is_bulleted else "other")
)
)
# 判断是否需要添加空行
if prev_line_type is None:
# 第一行,不需要空行
cleaned_lines.append(line)
elif prev_line_type == "heading" and current_line_type in [
"numbered",
"bulleted",
"other",
]:
# 标题后直接跟内容,不需要空行
cleaned_lines.append(line)
elif current_line_type in [
"numbered",
"bulleted",
] and prev_line_type in ["numbered", "bulleted"]:
# 列表项之间,不需要空行
cleaned_lines.append(line)
else:
# 其他情况,添加空行
cleaned_lines.append(line)
prev_line_type = current_line_type
else:
# 空行,跳过
pass
result = "\n".join(cleaned_lines)
elif mode == "compact":
result = re.sub(r"\r\n+", " ", text)
result = re.sub(r"\r+", " ", result)
result = re.sub(r"\n+", " ", result)
result = re.sub(r"[ \t\u00A0\u200B\u200C\u200D\u2060]+", " ", result)
result = re.sub(r" +", " ", result)
else:
result = text
result = re.sub(r"[,,]", ",", result)
result = re.sub(r"(?<=\d)\.", "\x00", result)
result = re.sub(r"[。.](?![a-zA-Z0-9])", "。", result)
result = re.sub(r"\x00", ".", result)
result = re.sub(r"[!!]", "!", result)
result = re.sub(r"[??]", "?", result)
result = re.sub(r"[;;]", ";", result)
result = re.sub(r"[::]", ":", result)
result = re.sub(r"[((]", "(", result)
result = re.sub(r"[))]", ")", result)
result = re.sub(r"[【\[]", "【", result)
result = re.sub(r"[】\]]", "】", result)
return result.strip()