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)