File size: 4,538 Bytes
fbd9d3d e5e756a fbd9d3d cc826a1 fbd9d3d e5e756a fbd9d3d e5e756a fbd9d3d e5e756a cc826a1 e5e756a fbd9d3d e5e756a fbd9d3d | 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 147 148 149 150 151 152 153 154 155 156 157 | import logging
from typing import Any
from fastapi import APIRouter, HTTPException, Request, Query
from app.plugins.run_log import get_run_log_service, RunStatus
logger = logging.getLogger(__name__)
router = APIRouter()
plugin = None
core = None
def set_plugin_instance(plugin_instance):
"""设置插件实例"""
global plugin
plugin = plugin_instance
def set_core_instance(core_instance):
"""设置核心逻辑实例"""
global core
core = core_instance
@router.get("/status")
async def get_status():
"""获取插件状态"""
if plugin is None:
return {
"name": "content",
"enabled": False,
"message": "插件未加载",
}
return plugin.get_status()
@router.post("/extract")
async def extract_content(request: Request, wait: bool = Query(False, description="是否同步等待提取完成")):
"""
提取内容
默认异步返回 run_id，设置 wait=true 同步等待结果。
"""
if plugin is None or not plugin.enabled:
raise HTTPException(status_code=400, detail="插件未启用")
if core is None:
raise HTTPException(status_code=500, detail="核心逻辑未初始化")
payload = await _parse_extract_payload(request)
url = str(payload.get("url") or "")
include_ocr = _parse_bool(payload.get("include_ocr", True))
if not url or not url.strip():
raise HTTPException(status_code=400, detail="链接不能为空")
# 创建 run
run_service = get_run_log_service()
run = run_service.create_run("content")
# 写入上传事件
run_service.add_event(
run_id=run.run_id,
stage="upload",
message=f"开始提取: {url}",
detail=f"include_ocr={include_ocr}",
)
try:
# 执行提取
result = await core.extract_with_run(
url=url.strip(),
run_id=run.run_id,
include_ocr=include_ocr,
)
if result["success"]:
# 完成 run
run_service.finish_run(
run_id=run.run_id,
status=RunStatus.SUCCEEDED,
result={
"title": result["title"],
"content": result["content"],
"raw_html": result["raw_html"],
"raw_text": result["raw_text"],
"normalized_content": result["normalized_content"],
"images": result["images"],
"images_text": result["images_text"],
"source_type": result["source_type"],
},
)
else:
# 完成 run（失败）
run_service.finish_run(
run_id=run.run_id,
status=RunStatus.FAILED,
error=result["error"],
)
# 如果同步等待，直接返回结果
if wait:
return {
"run_id": run.run_id,
"status": "succeeded" if result["success"] else "failed",
**result,
}
# 异步返回 run_id
return {
"run_id": run.run_id,
"status": "running",
"source_type": result["source_type"],
}
except Exception as e:
# 写入异常事件
run_service.add_event(
run_id=run.run_id,
stage="extract",
message=f"提取异常: {str(e)}",
level="error",
)
run_service.finish_run(
run_id=run.run_id,
status=RunStatus.FAILED,
error=str(e),
)
raise HTTPException(status_code=500, detail=f"提取失败: {str(e)}")
async def _parse_extract_payload(request: Request) -> dict[str, Any]:
"""解析提取请求，兼容JSON和表单提交"""
content_type = request.headers.get("content-type", "").lower()
if "application/json" in content_type:
try:
data = await request.json()
except Exception as e:
logger.error(f"解析内容提取JSON请求失败: {e}")
raise HTTPException(status_code=400, detail="请求体不是有效JSON") from e
return data if isinstance(data, dict) else {}
form = await request.form()
return dict(form)
def _parse_bool(value: Any) -> bool:
"""解析布尔参数"""
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.lower() in {"1", "true", "yes", "on"}
return bool(value)
|