SupplyAgent/api/app.py

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

from pathlib import Path

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

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_videos import list_videos_for_demand_belong
from api.services.oss_logs import list_demand_belong_oss_logs

app = FastAPI(title="SupplyAgent API", version="0.1.0")

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.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@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 videos linked to a demand_grade row (vid + title + points JSON)."""
    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/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}


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