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