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.hot_content_source_service import fetch_hot_content_source_detail
from app.services.hot_content_demand_export_service import (
    export_hot_content_demand_exports,
    query_hot_content_demand_exports,
)
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"),
]

HOT_CONTENT_DEMAND_EXPORT_COLUMNS: list[tuple[str, str]] = [
    ("来源", "source"),
    ("热点标题", "hot_title"),
    ("类型", "point_category"),
    ("需求类型", "item_type_label"),
    ("匹配需求", "matched_demand"),
    ("是否成为需求", "is_as_demand_label"),
    ("创建时间", "record_created_at"),
    ("贡献分", "contribution_score"),
    ("最高微信指数词", "wxindex_keyword"),
    ("待选微信指数词", "all_hot_keywords"),
    ("微信指数热度", "wxindex_latest_score"),
    ("微信指数趋势", "wxindex_trend"),
]


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/hot-content-source")
async def get_demand_pool_hot_content_source(
    demand_name: str = Query(..., min_length=1, description="需求名称"),
    demand_type: str = Query(..., min_length=1, description="需求类型，如 短语 / 特征点"),
    dt: str = Query(
        ...,
        min_length=1,
        description="分区日期，yyyymmdd 或 yyyy-mm-dd",
    ),
    strategy: str | None = Query(default=None, description="策略名，默认新热事件"),
) -> dict[str, object]:
    try:
        return fetch_hot_content_source_detail(
            demand_name=demand_name,
            demand_type=demand_type,
            partition_dt=dt,
            strategy=strategy,
        )
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except LookupError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@router.get("/hot-content/demand-exports")
async def get_hot_content_demand_exports(
    start_dt: str | None = Query(
        default=None,
        description="开始日期（热点记录创建时间）: yyyymmdd 或 yyyy-mm-dd，默认当天",
    ),
    end_dt: str | None = Query(
        default=None,
        description="结束日期（热点记录创建时间）: yyyymmdd 或 yyyy-mm-dd，默认当天",
    ),
    is_as_demand: int | None = Query(
        default=None,
        description="是否成为需求：0 否，1 是；不传表示全部",
    ),
    item_type: str | None = Query(
        default=None,
        description="需求类型：词（特征点）、点（短语）；不传表示全部",
    ),
    min_wxindex_latest_score: float | None = Query(
        default=None,
        ge=0,
        description="微信指数热度下限（wxindex_latest_score >= 该值）；不传表示不限制",
    ),
    page: int = Query(default=1, ge=1, description="页码，从 1 开始"),
    page_size: int = Query(default=20, ge=1, le=200, description="每页条数"),
) -> dict[str, object]:
    try:
        return query_hot_content_demand_exports(
            start_dt=start_dt,
            end_dt=end_dt,
            is_as_demand=is_as_demand,
            item_type=item_type,
            min_wxindex_latest_score=min_wxindex_latest_score,
            page=page,
            page_size=page_size,
        )
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


@router.get("/hot-content/demand-exports/export")
async def export_hot_content_demand_exports_api(
    start_dt: str | None = Query(default=None, description="开始日期"),
    end_dt: str | None = Query(default=None, description="结束日期"),
    is_as_demand: int | None = Query(default=None, description="是否成为需求：0/1"),
    item_type: str | None = Query(default=None, description="需求类型：词/点"),
    min_wxindex_latest_score: float | None = Query(
        default=None,
        ge=0,
        description="微信指数热度下限",
    ),
) -> StreamingResponse:
    try:
        items = export_hot_content_demand_exports(
            start_dt=start_dt,
            end_dt=end_dt,
            is_as_demand=is_as_demand,
            item_type=item_type,
            min_wxindex_latest_score=min_wxindex_latest_score,
        )
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    filename = f"新热事件查询_{_export_timestamp()}.xlsx"
    return _excel_streaming_response(
        items,
        HOT_CONTENT_DEMAND_EXPORT_COLUMNS,
        sheet_name="新热事件",
        filename=filename,
    )


@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,
    )