SupplyAgent/api/app.py

"""FastAPI application — category tree API on port 8080."""
from __future__ import annotations

from contextlib import asynccontextmanager
from pathlib import Path

from typing import Literal

from fastapi import FastAPI, HTTPException, Query, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from sqlalchemy import text
from starlette.exceptions import HTTPException as StarletteHTTPException

from api.auth_middleware import AuthenticationMiddleware
from api.routers.auth import router as auth_router
from api.routers.business_harness import router as business_harness_router
from api.routers.pipeline import router as pipeline_router
from api.schemas.demand_feedback import CreateDemandFeedbackBody
from api.services.agent_catalog import (
    get_agent_detail,
    update_agent_document_injection,
)
from api.services.auth import ensure_bootstrap_admin
from api.services.category_tree import build_category_tree
from api.services.demand_belong_category import list_demand_belong_categories
from api.services.demand_grade import list_demand_grades
from api.services.demand_grade_videos import list_videos_for_demand_grade
from api.services.demand_feedback import (
    FeedbackRequestConflictError,
    FeedbackTargetConflictError,
    FeedbackTargetNotFoundError,
    create_demand_feedback,
    list_demand_feedback,
)
from api.services.demand_videos import list_videos_for_demand_belong
from api.services.oss_logs import list_agent_oss_logs, list_demand_belong_oss_logs
from api.services.scheduler import (
    get_scheduler_job_run,
    list_triggerable_jobs,
    run_scheduler_job,
    run_supply_pipeline,
    scheduler_status as get_persistent_scheduler_status,
)
from api.services.video_discovery import (
    get_video_discovery_demand,
    list_video_discovery_demands,
)
from api.services.video_discovery_records import (
    get_video_discovery_run,
    list_video_discovery_candidates,
    list_video_discovery_runs,
    list_video_discovery_searches,
)
from supply_infra.config import get_infra_settings
from supply_infra.db import dispose_engine, get_session, init_db
from supply_infra.pipeline.health import pipeline_health_snapshot


class SPAStaticFiles(StaticFiles):
    """Serve index.html for browser routes handled by Vue Router."""

    async def get_response(self, path: str, scope):
        try:
            return await super().get_response(path, scope)
        except StarletteHTTPException as exc:
            is_backend_path = (
                path == "api"
                or path.startswith("api/")
                or path == "health"
                or path.startswith("health/")
            )
            if exc.status_code != 404 or is_backend_path or Path(path).suffix:
                raise
            return await super().get_response("index.html", scope)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    if get_infra_settings().database_auto_create:
        init_db()
    ensure_bootstrap_admin()
    yield
    dispose_engine()


app = FastAPI(title="SupplyAgent API", version="0.1.0", lifespan=lifespan)
app.include_router(auth_router)
app.include_router(pipeline_router)
app.include_router(business_harness_router)

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:5173",
        "http://127.0.0.1:5173",
        "http://localhost:4173",
        "http://127.0.0.1:4173",
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(AuthenticationMiddleware)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/health/live")
def health_live() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/health/ready")
def health_ready() -> dict:
    try:
        with get_session() as session:
            session.execute(text("SELECT 1"))
        runtime = pipeline_health_snapshot()
    except Exception as exc:
        raise HTTPException(status_code=503, detail="database unavailable") from exc
    if not runtime["runtime_components_ready"]:
        raise HTTPException(
            status_code=503,
            detail={
                "message": "pipeline runtime components unavailable",
                "scheduler_running": runtime["scheduler_running"],
                "worker_active_count": runtime["worker_active_count"],
                "worker_expected_count": runtime["worker_expected_count"],
                "reconciler_running": runtime["reconciler_running"],
            },
        )
    return {
        "status": "ready",
        "scheduler_running": runtime["scheduler_running"],
        "worker_active_count": runtime["worker_active_count"],
        "reconciler_running": runtime["reconciler_running"],
    }


@app.get("/api/scheduler/status")
def scheduler_status() -> dict:
    """Deprecated compatibility view backed by MySQL pipeline state."""
    return get_persistent_scheduler_status()


class RunPipelineBody(BaseModel):
    biz_dt: str | None = Field(
        default=None,
        pattern=r"^\d{8}$",
        description="业务日 YYYYMMDD；省略则取当天",
    )


class TriggerSchedulerJobBody(BaseModel):
    biz_dt: str | None = Field(
        default=None,
        pattern=r"^\d{8}$",
        description="业务日 YYYYMMDD；省略则取当天",
    )


class AgentDocumentInjectionBody(BaseModel):
    content: str = Field(
        default="",
        max_length=100_000,
        description="追加到 Agent System Prompt 后的业务文档",
    )
    enabled: bool = Field(
        default=True,
        description="是否在新建 Agent 时启用该文档注入",
    )


@app.post("/api/pipeline/run", status_code=202)
def run_pipeline(
    body: RunPipelineBody | None = None,
    biz_dt: str | None = Query(
        default=None,
        pattern=r"^\d{8}$",
        description="业务日 YYYYMMDD（与 body 二选一，body 优先）",
    ),
) -> dict:
    """
    异步一键执行供给数据全流程，立即返回 run_id，后台串行执行：

    全局树同步 → 需求池同步 → 需求分级 → 视频点位拓展 → find_agent 找视频 → AIGC 发布。
    """
    resolved_biz_dt = body.biz_dt if body and body.biz_dt is not None else biz_dt
    return run_supply_pipeline(biz_dt=resolved_biz_dt)


@app.get("/api/scheduler/jobs")
def scheduler_jobs() -> dict:
    """Return jobs that can be triggered manually."""
    return {"items": list_triggerable_jobs()}


@app.post("/api/scheduler/jobs/{job_id}/run")
def trigger_scheduler_job(
    job_id: str,
    body: TriggerSchedulerJobBody | None = None,
    biz_dt: str | None = Query(
        default=None,
        pattern=r"^\d{8}$",
        description="业务日 YYYYMMDD（与 body 二选一，body 优先）",
    ),
) -> dict:
    """Manually trigger a scheduler job."""
    resolved_biz_dt = body.biz_dt if body and body.biz_dt is not None else biz_dt

    try:
        return run_scheduler_job(job_id, biz_dt=resolved_biz_dt)
    except KeyError:
        raise HTTPException(status_code=404, detail=f"scheduler job not found: {job_id}") from None


@app.get("/api/scheduler/runs/{run_id}")
def scheduler_job_run(run_id: str) -> dict:
    """Return durable pipeline run status."""
    result = get_scheduler_job_run(run_id)
    if result is None:
        raise HTTPException(status_code=404, detail="scheduler run not found")
    return result


@app.get("/api/category-tree")
def category_tree(
    biz_dt: str | None = Query(
        default=None,
        description="业务日 YYYYMMDD；省略则取 category_tree_weight 最新一日",
    ),
) -> dict:
    """Return nested global_tree_category with per-dim avg score."""
    return build_category_tree(biz_dt=biz_dt)


@app.get("/api/demand-belong-category")
def demand_belong_category() -> dict:
    """Return all active demand_belong_category rows in one response."""
    items = list_demand_belong_categories()
    return {"items": items}


@app.get("/api/demand-belong-category/{belong_id}/videos")
def demand_belong_videos(belong_id: int) -> dict:
    """Return videos linked to a demand_belong_category row (vid + title + points JSON)."""
    result = list_videos_for_demand_belong(belong_id)
    if result is None:
        raise HTTPException(status_code=404, detail="demand_belong_category not found")
    return result


@app.get("/api/demand-grade")
def demand_grade(
    biz_dt: str | None = Query(
        default=None,
        description="业务日 YYYYMMDD；省略则取 demand_grade 最新一日",
    ),
) -> dict:
    """Return demand_grade rows (one per category_id) for the given/latest biz_dt."""
    items = list_demand_grades(biz_dt=biz_dt)
    return {"items": items}


@app.get("/api/demand-grade/{demand_grade_id}/videos")
def demand_grade_videos(demand_grade_id: int) -> dict:
    """Return expansion videos/points for a demand_grade row (by its biz_dt)."""
    result = list_videos_for_demand_grade(demand_grade_id)
    if result is None:
        raise HTTPException(status_code=404, detail="demand_grade not found")
    return result


@app.get("/api/video-discovery/demands")
def video_discovery_demands(
    biz_dt: str | None = Query(
        default=None,
        description="业务日 YYYYMMDD；省略则取 demand_grade 最新一日",
    ),
) -> dict:
    """Return demand-first video discovery cards for the selected/latest day."""
    return list_video_discovery_demands(biz_dt=biz_dt)


@app.get("/api/video-discovery/runs")
def video_discovery_runs(
    biz_dt: str | None = Query(default=None, pattern=r"^\d{8}$"),
    status: Literal["running", "finished", "failed"] | None = None,
    keyword: str | None = Query(default=None, max_length=256),
    limit: int = Query(default=20, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
) -> dict:
    """Return paged find-agent runs for the records workspace."""
    return list_video_discovery_runs(
        biz_dt=biz_dt,
        status=status,
        keyword=keyword,
        limit=limit,
        offset=offset,
    )


@app.get("/api/video-discovery/runs/{run_id}")
def video_discovery_run(run_id: str) -> dict:
    """Return one find-agent run and its aggregate counts."""
    result = get_video_discovery_run(run_id)
    if result is None:
        raise HTTPException(status_code=404, detail="video discovery run not found")
    return result


@app.get("/api/video-discovery/runs/{run_id}/searches")
def video_discovery_searches(
    run_id: str,
    keyword: str | None = Query(default=None, max_length=256),
    limit: int = Query(default=20, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
) -> dict:
    """Return paged search-page records for one find-agent run."""
    result = list_video_discovery_searches(
        run_id,
        keyword=keyword,
        limit=limit,
        offset=offset,
    )
    if result is None:
        raise HTTPException(status_code=404, detail="video discovery run not found")
    return result


@app.get("/api/video-discovery/runs/{run_id}/candidates")
def video_discovery_candidates(
    run_id: str,
    bucket: str | None = Query(default=None, max_length=24),
    keyword: str | None = Query(default=None, max_length=256),
    limit: int = Query(default=20, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
) -> dict:
    """Return paged candidate records for one find-agent run."""
    result = list_video_discovery_candidates(
        run_id,
        bucket=bucket,
        keyword=keyword,
        limit=limit,
        offset=offset,
    )
    if result is None:
        raise HTTPException(status_code=404, detail="video discovery run not found")
    return result


@app.get("/api/video-discovery/demands/{demand_grade_id}")
def video_discovery_demand(demand_grade_id: int) -> dict:
    """Return one demand and its videos/points resolved from the source tables."""
    result = get_video_discovery_demand(demand_grade_id)
    if result is None:
        raise HTTPException(status_code=404, detail="demand_grade not found")
    return result


@app.post(
    "/api/video-discovery/feedback",
    status_code=status.HTTP_201_CREATED,
)
def submit_video_discovery_feedback(
    body: CreateDemandFeedbackBody,
    request: Request,
) -> dict:
    """Append feedback for one demand, video or hit-content record."""
    try:
        return create_demand_feedback(body, request.state.current_user)
    except FeedbackTargetNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (FeedbackTargetConflictError, FeedbackRequestConflictError) as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@app.get("/api/video-discovery/feedback")
def video_discovery_feedback_history(
    target_type: Literal["demand", "video", "hit_content"],
    demand_grade_id: int = Query(gt=0),
    video_id: str | None = Query(default=None, max_length=64),
    demand_video_expansion_id: int | None = Query(default=None, gt=0),
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
) -> dict:
    """Return feedback records and feedback people for one target."""
    try:
        return list_demand_feedback(
            target_type=target_type,
            demand_grade_id=demand_grade_id,
            video_id=video_id,
            demand_video_expansion_id=demand_video_expansion_id,
            limit=limit,
            offset=offset,
        )
    except FeedbackTargetNotFoundError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except FeedbackTargetConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@app.get("/api/demand-belong-oss-logs")
def demand_belong_oss_logs() -> dict:
    """Return demand_belong_category_agent oss_logs ordered by create_time desc."""
    items = list_demand_belong_oss_logs()
    return {"items": items}


@app.get("/api/agent-oss-logs")
def agent_oss_logs(
    agent_name: str | None = Query(
        default=None,
        description="Agent 名称；省略则返回全部 Agent 日志",
    ),
) -> dict:
    """Return Agent oss_logs and the available Agent names."""
    return list_agent_oss_logs(agent_name=agent_name)


@app.get("/api/agents/{agent_name}")
def agent_detail(agent_name: str) -> dict:
    """Return one business Agent's current system prompt and tools."""
    result = get_agent_detail(agent_name)
    if result is None:
        raise HTTPException(status_code=404, detail=f"agent not found: {agent_name}")
    return result


@app.put("/api/agents/{agent_name}/document-injection")
def agent_document_injection(
    agent_name: str,
    body: AgentDocumentInjectionBody,
) -> dict:
    """Create or update the web-managed document injected into an Agent prompt."""
    result = update_agent_document_injection(
        agent_name,
        content=body.content,
        enabled=body.enabled,
    )
    if result is None:
        raise HTTPException(status_code=404, detail=f"agent not found: {agent_name}")
    return result


_web_dist = Path(__file__).resolve().parent.parent / "web" / "dist"
if _web_dist.is_dir():
    app.mount("/", SPAStaticFiles(directory=_web_dist, html=True), name="web")