File size: 4,641 Bytes
8ffb677 cc826a1 8ffb677 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | 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()
|