vent_agent_2.0/api/business_routes.py

"""业务目录及持久化任务 API，管理和提交使用单独的服务凭证。"""
import asyncio
import hmac
import json
from typing import Any, Literal

from fastapi import APIRouter, Depends, Header, HTTPException, Query
from pydantic import BaseModel, ConfigDict, Field, ValidationError

import config
from db import business_store
from extensions.registry import load_extensions
from runtime import business
from runtime.business_models import RemoteFile
from runtime.business_tasks import get_controller, public_job, submit
from runtime.result_models import PlatformResultPayload

router = APIRouter(prefix="/business-agents", tags=["业务智能体"])


def spec_of(name):
    try:
        return load_extensions().business(name)
    except ValueError as exc:
        raise HTTPException(404, str(exc)) from None


async def service_auth(authorization: str = Header(default="")):
    token = config.get_str("BUSINESS_API_TOKEN")
    if not token:
        raise HTTPException(503, "业务任务 API 未配置 BUSINESS_API_TOKEN")
    if not hmac.compare_digest(authorization.encode(), f"Bearer {token}".encode()):
        raise HTTPException(401, "业务任务 API 认证失败")


class JobRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    input: dict[str, Any]
    user_name: str = Field(default="default", min_length=1, max_length=128)
    request_id: str | None = Field(default=None, min_length=1, max_length=128)
    files: list[RemoteFile] = Field(default_factory=list, max_length=20)
    publish: bool = False


@router.get("")
async def list_agents():
    return {"items": [{"name": s.name, "display_name": s.display_name,
                       "description": load_extensions().agents[s.name].description,
                       "max_concurrency": s.max_concurrency,
                       "gateway_configured": bool(s.gateway.agent_code),
                       "result_platform_configured": bool(s.result_binding.agent_code) and config.get_bool("RESULT_PLATFORM_ENABLED", False)}
                      for s in load_extensions().business_agents.values()]}


@router.get("/{name}/schema")
async def schemas(name: str):
    spec = spec_of(name)
    return {"input": spec.input_schema.model_json_schema(), "result": spec.result_schema.model_json_schema(),
            "platform_result": PlatformResultPayload.model_json_schema()}


@router.post("/main_fan/mine-catalog", status_code=201, dependencies=[Depends(service_auth)])
async def refresh_main_fan_mines():
    """固定首步：读取部门树，冻结仅含正常生产煤矿的目录。"""
    from extensions.main_fan import mine_catalog
    from extensions.shaanxi.client import GatewayError
    spec_of('main_fan')
    try:
        return await mine_catalog.refresh()
    except (GatewayError, mine_catalog.MineCatalogError) as exc:
        raise HTTPException(502, str(exc)) from None
    except RuntimeError as exc:
        raise HTTPException(503, str(exc)) from None


@router.get("/main_fan/mine-catalog", dependencies=[Depends(service_auth)])
async def latest_main_fan_mines():
    """查看最近一次成功筛选的煤矿清单。"""
    return await _main_fan_mines()


@router.get("/main_fan/mine-catalog/{snapshot_id}", dependencies=[Depends(service_auth)])
async def main_fan_mines(snapshot_id: str):
    """按快照编号重读本批次固定输入。"""
    return await _main_fan_mines(snapshot_id)


async def _main_fan_mines(snapshot_id=None):
    from extensions.main_fan import mine_catalog
    spec_of('main_fan')
    try:
        return await asyncio.to_thread(mine_catalog.load, snapshot_id)
    except (FileNotFoundError, ValueError, KeyError):
        raise HTTPException(404, '煤矿清单不存在，请先刷新目录') from None


from extensions.main_fan.batch_models import BatchRequest


@router.post('/main_fan/batches', status_code=202, dependencies=[Depends(service_auth)])
async def create_main_fan_batch(request: BatchRequest):
    """显式选矿的结果批次；始终落本地，按总开关选择是否进入推送队列。"""
    from extensions import get_service
    try:
        return get_service('main_fan_batches').submit(request)
    except RuntimeError as exc:
        raise HTTPException(503, '本地主通风机批次尚未启用') from exc
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from None


@router.get('/main_fan/batches/{batch_id}', dependencies=[Depends(service_auth)])
async def get_main_fan_batch(batch_id: str):
    from extensions import get_service
    try:
        return get_service('main_fan_batches').get(batch_id)
    except RuntimeError as exc:
        raise HTTPException(503, '本地主通风机批次尚未启用') from exc
    except (ValueError, FileNotFoundError):
        raise HTTPException(404, '批次不存在') from None


@router.get("/{name}/status", dependencies=[Depends(service_auth)])
async def status(name: str):
    spec_of(name)
    state, metrics = await asyncio.gather(asyncio.to_thread(business_store.state, name),
                                         asyncio.to_thread(business_store.metrics, name))
    return {"name": name, **state, **metrics}


@router.post("/{name}/manage", dependencies=[Depends(service_auth)])
async def manage(name: str, action: Literal["start", "stop", "restart"]):
    spec_of(name)
    if action in {"stop", "restart"}:
        await business.stop(name)
    if action in {"start", "restart"}:
        await business.start(name)
    return await status(name)


@router.post("/{name}/jobs", status_code=202, dependencies=[Depends(service_auth)])
async def create_job(name: str, payload: JobRequest):
    spec_of(name)
    try:
        get_controller()
        job = await asyncio.to_thread(submit, name, payload.input, user_name=payload.user_name,
            files=[f.model_dump() for f in payload.files], request_id=payload.request_id, publish=payload.publish)
    except business.BusinessStopped as exc:
        raise HTTPException(409, str(exc)) from None
    except ValidationError as exc:
        raise HTTPException(422, exc.errors(include_input=False, include_context=False)) from None
    except ValueError as exc:
        raise HTTPException(422, str(exc)) from None
    except RuntimeError as exc:
        raise HTTPException(503, str(exc)) from None
    return public_job(job)


@router.get("/{name}/jobs", dependencies=[Depends(service_auth)])
async def jobs(name: str, user_name: str = "default", limit: int = Query(50, ge=1, le=200)):
    spec_of(name)
    return {"items": [public_job(j) for j in await asyncio.to_thread(business_store.list_jobs, name, user_name, limit)]}


async def owned_job(name, job_id, user_name):
    spec_of(name)
    job = await asyncio.to_thread(business_store.get, job_id)
    if not job or job["agent_name"] != name or job["user_name"] != user_name:
        raise HTTPException(404, "业务任务不存在")
    return job


@router.get("/{name}/jobs/{job_id}", dependencies=[Depends(service_auth)])
async def detail(name: str, job_id: str, user_name: str = "default"):
    return public_job(await owned_job(name, job_id, user_name))


@router.get("/{name}/jobs/{job_id}/result", dependencies=[Depends(service_auth)])
async def exported_result(name: str, job_id: str, user_name: str = "default"):
    job = await owned_job(name, job_id, user_name)
    result = json.loads(job["result_json"]) if job.get("result_json") else {}
    if not result.get("platform_payload"):
        raise HTTPException(409, "当前任务还没有新格式的结果报文")
    return PlatformResultPayload.model_validate(result["platform_payload"]).model_dump(mode="json")


@router.post("/{name}/jobs/{job_id}/publish", status_code=202, dependencies=[Depends(service_auth)])
async def publish(name: str, job_id: str, user_name: str = "default"):
    job = await owned_job(name, job_id, user_name)
    try:
        await get_controller().queue_publish(job)
    except ValueError as exc:
        raise HTTPException(409, str(exc)) from None
    except RuntimeError as exc:
        raise HTTPException(503, str(exc)) from None
    return public_job(await asyncio.to_thread(business_store.get, job_id))


@router.get("/main_fan/analyses/{analysis_id}/result", dependencies=[Depends(service_auth)])
async def exported_analysis(analysis_id: str, session_id: str, user_name: str = "default"):
    from db import main_fan_store
    try:
        result = await asyncio.to_thread(main_fan_store.canonical_result,
                                        {"analysis_id": analysis_id}, user_name, session_id)
    except ValueError:
        raise HTTPException(404, "研判结果不存在或不属于当前会话") from None
    if not result.get("platform_payload"):
        raise HTTPException(409, "历史结果没有新格式报文，请重新执行研判")
    return PlatformResultPayload.model_validate(result["platform_payload"]).model_dump(mode="json")


@router.post("/{name}/jobs/{job_id}/retry", dependencies=[Depends(service_auth)])
async def retry(name: str, job_id: str, user_name: str = "default"):
    job = await owned_job(name, job_id, user_name)
    if job["status"] == "completed" and job["push_status"] == "failed":
        return await publish(name, job_id, user_name)
    elif not await asyncio.to_thread(business_store.retry, job_id):
        raise HTTPException(409, "任务当前状态不支持重试")
    return await detail(name, job_id, user_name)