demand_server/app/api/routes.py

from datetime import datetime
from io import BytesIO
from zoneinfo import ZoneInfo

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import StreamingResponse

from app.core.config import settings
from app.scheduler.manager import scheduler
from app.services.demand_pool_service import (
    export_demand_pool_records,
    query_demand_pool_records,
    query_strategy_options,
)
from app.services.element_search_service import (
    query_monthly_element_demands,
    query_same_period_last_year_element_demands,
    query_same_period_last_year_lunar_element_demands,
    query_video_decode_url2_for_today,
)
from app.utils.excel_export import build_content_disposition, rows_to_excel_bytes

router = APIRouter()

SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")

DEMAND_POOL_EXPORT_COLUMNS: list[tuple[str, str]] = [
    ("ID", "id"),
    ("策略名", "strategy"),
    ("需求名称", "demand_name"),
    ("需求类型", "type"),
    ("权重", "weight"),
    ("视频数量", "video_count"),
    ("日期", "dt"),
]

ELEMENT_DEMAND_EXPORT_COLUMNS: list[tuple[str, str]] = [
    ("策略", "strategy"),
    ("特征点名称", "demand_name"),
    ("权重", "weight"),
    ("视频数", "video_count"),
    ("视频列表", "video_list"),
]

MONTHLY_ELEMENT_DEMAND_EXPORT_COLUMNS: list[tuple[str, str]] = [
    *ELEMENT_DEMAND_EXPORT_COLUMNS,
    ("月份列表", "month_list"),
    ("频次", "frequency"),
]


def _export_timestamp() -> str:
    return datetime.now(SHANGHAI_TZ).strftime("%Y%m%d_%H%M%S")


def _excel_streaming_response(
    rows: list[dict[str, object]],
    columns: list[tuple[str, str]],
    *,
    sheet_name: str,
    filename: str,
) -> StreamingResponse:
    content = rows_to_excel_bytes(rows, columns, sheet_name=sheet_name)
    return StreamingResponse(
        BytesIO(content),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": build_content_disposition(filename)},
    )


@router.get("/health")
async def health_check() -> dict[str, str]:
    return {"status": "ok", "service": settings.app_name, "env": settings.app_env}


@router.get("/scheduler/status")
async def scheduler_status() -> dict[str, object]:
    jobs = scheduler.get_jobs()
    return {
        "running": scheduler.running,
        "job_count": len(jobs),
        "jobs": [job.id for job in jobs],
    }


@router.get("/demand-pool")
async def query_demand_pool(
    strategy: list[str] | None = Query(default=None, description="策略，支持多选"),
    start_dt: str | None = Query(default=None, description="开始日期: yyyymmdd 或 yyyy-mm-dd"),
    end_dt: str | None = Query(default=None, description="结束日期: yyyymmdd 或 yyyy-mm-dd"),
    demand_name: str | None = Query(
        default=None,
        description="需求名称包含该子串则保留；空或未传则不筛选",
    ),
    min_weight: float | None = Query(default=None, description="最小权重"),
    max_weight: float | None = Query(default=None, description="最大权重"),
    sort_by: str | None = Query(default="weight", description="排序字段"),
    sort_order: str | None = Query(default="desc", description="排序方向: asc 或 desc"),
    page: int = Query(default=1, ge=1, description="页码，从 1 开始"),
    page_size: int = Query(default=20, ge=1, le=200, description="每页条数"),
) -> dict[str, object]:
    return query_demand_pool_records(
        strategies=strategy,
        start_dt=start_dt,
        end_dt=end_dt,
        demand_name=demand_name,
        min_weight=min_weight,
        max_weight=max_weight,
        sort_by=sort_by,
        sort_order=sort_order,
        page=page,
        page_size=page_size,
    )


@router.get("/demand-pool/export")
async def export_demand_pool(
    strategy: list[str] | None = Query(default=None, description="策略，支持多选"),
    start_dt: str | None = Query(default=None, description="开始日期: yyyymmdd 或 yyyy-mm-dd"),
    end_dt: str | None = Query(default=None, description="结束日期: yyyymmdd 或 yyyy-mm-dd"),
    demand_name: str | None = Query(
        default=None,
        description="需求名称包含该子串则保留；空或未传则不筛选",
    ),
    min_weight: float | None = Query(default=None, description="最小权重"),
    max_weight: float | None = Query(default=None, description="最大权重"),
    sort_by: str | None = Query(default="weight", description="排序字段"),
    sort_order: str | None = Query(default="desc", description="排序方向: asc 或 desc"),
) -> StreamingResponse:
    items = export_demand_pool_records(
        strategies=strategy,
        start_dt=start_dt,
        end_dt=end_dt,
        demand_name=demand_name,
        min_weight=min_weight,
        max_weight=max_weight,
        sort_by=sort_by,
        sort_order=sort_order,
    )
    filename = f"需求池_{_export_timestamp()}.xlsx"
    return _excel_streaming_response(
        items,
        DEMAND_POOL_EXPORT_COLUMNS,
        sheet_name="需求明细",
        filename=filename,
    )


@router.get("/element-demands/solar-calendar")
async def get_element_demands_solar_calendar(
    period_days: int = Query(
        ...,
        ge=0,
        description="区间天数（含去年阳历今日）；0 表示仅当日分区",
    ),
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="按原始元素分组后的平均 ROV 下限（HAVING）",
    ),
) -> dict[str, object]:
    items = query_same_period_last_year_element_demands(
        period_days=period_days,
        view_pv_count=view_pv_count,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
    )
    return {"items": items}


@router.get("/element-demands/solar-calendar/export")
async def export_element_demands_solar_calendar(
    period_days: int = Query(
        ...,
        ge=0,
        description="区间天数（含去年阳历今日）；0 表示仅当日分区",
    ),
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="按原始元素分组后的平均 ROV 下限（HAVING）",
    ),
) -> StreamingResponse:
    items = query_same_period_last_year_element_demands(
        period_days=period_days,
        view_pv_count=view_pv_count,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
    )
    filename = f"去年同期阳历特征点_{_export_timestamp()}.xlsx"
    return _excel_streaming_response(
        items,
        ELEMENT_DEMAND_EXPORT_COLUMNS,
        sheet_name="特征点明细",
        filename=filename,
    )


@router.get("/element-demands/lunar-calendar")
async def get_element_demands_lunar_calendar(
    period_days: int = Query(
        ...,
        ge=0,
        description="区间天数（含去年阴历今日）；0 表示仅当日分区",
    ),
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="按原始元素分组后的平均 ROV 下限（HAVING）",
    ),
) -> dict[str, object]:
    items = query_same_period_last_year_lunar_element_demands(
        period_days=period_days,
        view_pv_count=view_pv_count,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
    )
    return {"items": items}


@router.get("/element-demands/lunar-calendar/export")
async def export_element_demands_lunar_calendar(
    period_days: int = Query(
        ...,
        ge=0,
        description="区间天数（含去年阴历今日）；0 表示仅当日分区",
    ),
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="按原始元素分组后的平均 ROV 下限（HAVING）",
    ),
) -> StreamingResponse:
    items = query_same_period_last_year_lunar_element_demands(
        period_days=period_days,
        view_pv_count=view_pv_count,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
    )
    filename = f"去年同期阴历特征点_{_export_timestamp()}.xlsx"
    return _excel_streaming_response(
        items,
        ELEMENT_DEMAND_EXPORT_COLUMNS,
        sheet_name="特征点明细",
        filename=filename,
    )


@router.get("/element-demands/monthly")
async def get_element_demands_monthly(
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column 单日行）",
    ),
    month_total_pv_threshold: float = Query(
        ...,
        ge=0,
        description="视频单月累计分发曝光 PV 和的下限（严格大于该值才保留）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="元素单月平均 ROV 下限（按月聚合后再汇总）",
    ),
    min_frequency: int = Query(
        ...,
        ge=0,
        description="元素在回溯窗口内满足条件的月份数下限",
    ),
) -> dict[str, object]:
    items = query_monthly_element_demands(
        view_pv_count=view_pv_count,
        month_total_pv_threshold=month_total_pv_threshold,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
        min_frequency=min_frequency,
    )
    return {"items": items}


@router.get("/element-demands/monthly/export")
async def export_element_demands_monthly(
    view_pv_count: int = Query(
        ...,
        ge=0,
        description="当日分发曝光 pv 下限（video_dimension_detail_add_column 单日行）",
    ),
    month_total_pv_threshold: float = Query(
        ...,
        ge=0,
        description="视频单月累计分发曝光 PV 和的下限（严格大于该值才保留）",
    ),
    min_contribution_score: float = Query(
        ...,
        description="贡献分下限（dwd_topic_decode_result_detail_di）",
    ),
    rov_avg: float = Query(
        ...,
        description="元素单月平均 ROV 下限（按月聚合后再汇总）",
    ),
    min_frequency: int = Query(
        ...,
        ge=0,
        description="元素在回溯窗口内满足条件的月份数下限",
    ),
) -> StreamingResponse:
    items = query_monthly_element_demands(
        view_pv_count=view_pv_count,
        month_total_pv_threshold=month_total_pv_threshold,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
        min_frequency=min_frequency,
    )
    filename = f"逐月特征点_{_export_timestamp()}.xlsx"
    return _excel_streaming_response(
        items,
        MONTHLY_ELEMENT_DEMAND_EXPORT_COLUMNS,
        sheet_name="特征点明细",
        filename=filename,
    )


@router.get("/videos/decode-url")
async def get_video_decode_page_url(
    vid: str = Query(
        ...,
        min_length=1,
        max_length=128,
        description="视频 id，对应 dwd_topic_decode_result_di.vid（当天上海分区）",
    ),
) -> dict[str, str | None]:
    try:
        url2 = query_video_decode_url2_for_today(vid)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return {"url2": url2}


@router.get("/demand-pool/strategies")
async def get_demand_pool_strategies(
    start_dt: str | None = Query(default=None, description="开始日期: yyyymmdd 或 yyyy-mm-dd"),
    end_dt: str | None = Query(default=None, description="结束日期: yyyymmdd 或 yyyy-mm-dd"),
    min_weight: float | None = Query(default=None, description="最小权重"),
    max_weight: float | None = Query(default=None, description="最大权重"),
) -> dict[str, object]:
    return query_strategy_options(
        start_dt=start_dt,
        end_dt=end_dt,
        min_weight=min_weight,
        max_weight=max_weight,
    )