116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
import json
|
||
from time import perf_counter
|
||
from fastapi import FastAPI, HTTPException, Header, Request
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from loguru import logger
|
||
from parser import *
|
||
|
||
app = FastAPI()
|
||
|
||
# 添加 CORS 支持
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"], # 可以改成你的前端域名，例如 ["https://yourdomain.com"]
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# batch 接口单次请求的条目上限（防滥用）
|
||
BATCH_MAX_ITEMS = 1000
|
||
|
||
|
||
def _check_auth(Authorization: str):
|
||
if Authorization != "3FFA6A4B073CF065969630692331A873":
|
||
raise HTTPException(status_code=403, detail="未授权的操作")
|
||
|
||
|
||
def _parse_urlencoded_content(body: bytes) -> str:
|
||
"""手动解析 x-www-form-urlencoded 的 content 字段：绕过 starlette 1MB 表单限制"""
|
||
from urllib.parse import parse_qs
|
||
params = parse_qs(body.decode("utf-8"), keep_blank_values=True)
|
||
return params.get("content", [""])[0]
|
||
|
||
|
||
@app.post("/convert/start")
|
||
async def convert_str(
|
||
request: Request,
|
||
Authorization: str = Header(None)
|
||
):
|
||
"""
|
||
带调试信息的转换接口 (支持 x-www-form-urlencoded)
|
||
手动解析表单体：绕过 starlette 1MB 表单限制（族谱全文可达数 MB）
|
||
"""
|
||
|
||
_check_auth(Authorization)
|
||
|
||
try:
|
||
content = _parse_urlencoded_content(await request.body())
|
||
res = convert_text(content)
|
||
logger.info(f"转换结果: {json.dumps(res, ensure_ascii=False, indent=2)}")
|
||
return JSONResponse(content={"code": 0, "result": res})
|
||
except Exception as e:
|
||
logger.error(f"转换异常: {str(e)}")
|
||
return JSONResponse(
|
||
content={"code": 500, "result": f"转换服务异常: {str(e)}"},
|
||
status_code=500
|
||
)
|
||
|
||
|
||
@app.post("/convert/batch")
|
||
async def convert_batch(
|
||
request: Request,
|
||
Authorization: str = Header(None)
|
||
):
|
||
"""
|
||
批量转换接口：短文本多任务场景，一次请求处理 N 条 content。
|
||
请求: {"items": [{"id": "row1", "content": "..."}, ...]}（application/json）
|
||
响应: {"code": 0, "result": [{"id": "row1", "result": [...]}, ...]}
|
||
- id 透传用于前端回填，缺省按数组顺序编号
|
||
- 单条解析失败不影响其他条目（该条返回 error 字段）
|
||
- 每条 content 独立解析，位置语义与单接口一致（从 0 起）
|
||
"""
|
||
_check_auth(Authorization)
|
||
|
||
# 请求体解析与入参校验失败 → 400（客户端错误）
|
||
try:
|
||
payload = json.loads((await request.body()).decode("utf-8"))
|
||
except Exception as e:
|
||
logger.warning(f"batch 请求体解析失败: {str(e)}")
|
||
raise HTTPException(status_code=400, detail="请求体必须是合法 JSON")
|
||
items = payload.get("items") if isinstance(payload, dict) else None
|
||
if not isinstance(items, list) or not items:
|
||
raise HTTPException(status_code=400, detail="items 必须是非空数组")
|
||
if len(items) > BATCH_MAX_ITEMS:
|
||
raise HTTPException(status_code=400, detail=f"items 超过上限 {BATCH_MAX_ITEMS} 条")
|
||
|
||
try:
|
||
t0 = perf_counter()
|
||
results = []
|
||
for i, item in enumerate(items):
|
||
item_id = item.get("id") if isinstance(item, dict) else None
|
||
content = item.get("content", "") if isinstance(item, dict) else ""
|
||
if item_id is None:
|
||
item_id = str(i)
|
||
try:
|
||
results.append({"id": item_id, "result": convert_text(content)})
|
||
except Exception as e:
|
||
logger.error(f"batch 条目 {item_id} 解析异常: {str(e)}")
|
||
results.append({"id": item_id, "error": str(e)})
|
||
dt = perf_counter() - t0
|
||
logger.info(f"批量转换: {len(results)} 条, 耗时 {dt * 1000:.0f}ms")
|
||
return JSONResponse(content={"code": 0, "result": results})
|
||
except Exception as e:
|
||
logger.error(f"批量转换异常: {str(e)}")
|
||
return JSONResponse(
|
||
content={"code": 500, "result": f"批量转换服务异常: {str(e)}"},
|
||
status_code=500
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
|
||
uvicorn.run(app, host="0.0.0.0", port=16661, log_level="info")
|