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()