110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
|
"""
|
||
|
|
收视分析 API — 提供收视走势和指标卡数据
|
||
|
|
|
||
|
|
端点：
|
||
|
|
GET /api/analytics/years → 有收视数据的年份列表（去重降序）
|
||
|
|
GET /api/analytics/episodes?year=2026 → 指定年份所有期次的收视数据 + 年度目标
|
||
|
|
"""
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends, Query
|
||
|
|
from sqlalchemy import extract
|
||
|
|
from sqlalchemy import distinct
|
||
|
|
from sqlmodel import Session, select
|
||
|
|
|
||
|
|
from app.core.deps import require_role
|
||
|
|
from app.db.session import get_session
|
||
|
|
from app.models.episode import Episode
|
||
|
|
from app.models.yearly_target import YearlyTarget
|
||
|
|
from app.models.user import UserRole
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/analytics", tags=["收视分析"])
|
||
|
|
|
||
|
|
|
||
|
|
def _require_read():
|
||
|
|
"""三角色都可读"""
|
||
|
|
return require_role(UserRole.zhipianren, UserRole.zebian, UserRole.biandao)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/years")
|
||
|
|
def get_available_years(
|
||
|
|
session: Session = Depends(get_session),
|
||
|
|
current_user=Depends(_require_read()),
|
||
|
|
):
|
||
|
|
"""返回有收视数据的年份列表（去重，降序）。"""
|
||
|
|
statement = (
|
||
|
|
select(distinct(extract("year", Episode.air_date).label("year")))
|
||
|
|
.where(Episode.audience_share.is_not(None))
|
||
|
|
.order_by(extract("year", Episode.air_date).desc())
|
||
|
|
)
|
||
|
|
result = session.exec(statement).all()
|
||
|
|
# extract 返回 float，转 int
|
||
|
|
return [int(y) for y in result]
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/episodes")
|
||
|
|
def get_analytics_episodes(
|
||
|
|
year: int | None = Query(None, description="年份，不传则取最近有数据的年份"),
|
||
|
|
session: Session = Depends(get_session),
|
||
|
|
current_user=Depends(_require_read()),
|
||
|
|
):
|
||
|
|
"""返回指定年份所有期次的收视数据（按 air_date 升序），附带该年年度目标。
|
||
|
|
|
||
|
|
返回格式：
|
||
|
|
{
|
||
|
|
"year": 2026,
|
||
|
|
"yearly_target": { "base_target": 0.6448, "stretch_target": 0.8989 } | null,
|
||
|
|
"episodes": [ { id, episode_number, program_name, air_date, editor_name_snapshot,
|
||
|
|
audience_share, audience_rating }, ... ]
|
||
|
|
}
|
||
|
|
"""
|
||
|
|
# 如果没传 year，找最近有收视数据的年份
|
||
|
|
if year is None:
|
||
|
|
year_stmt = (
|
||
|
|
select(extract("year", Episode.air_date).label("y"))
|
||
|
|
.where(Episode.audience_share.is_not(None))
|
||
|
|
.order_by(extract("year", Episode.air_date).desc())
|
||
|
|
.limit(1)
|
||
|
|
)
|
||
|
|
latest_year = session.exec(year_stmt).first()
|
||
|
|
if latest_year is None:
|
||
|
|
return {"year": None, "yearly_target": None, "episodes": []}
|
||
|
|
year = int(latest_year)
|
||
|
|
|
||
|
|
# 查询该年份的期次（按 air_date 升序）
|
||
|
|
ep_stmt = (
|
||
|
|
select(Episode)
|
||
|
|
.where(extract("year", Episode.air_date) == year)
|
||
|
|
.order_by(Episode.air_date.asc())
|
||
|
|
)
|
||
|
|
episodes = session.exec(ep_stmt).all()
|
||
|
|
|
||
|
|
# 查询该年份的年度目标
|
||
|
|
target_stmt = select(YearlyTarget).where(YearlyTarget.year == year)
|
||
|
|
target = session.exec(target_stmt).first()
|
||
|
|
|
||
|
|
# 组装返回
|
||
|
|
ep_list = [
|
||
|
|
{
|
||
|
|
"id": ep.id,
|
||
|
|
"episode_number": ep.episode_number,
|
||
|
|
"program_name": ep.program_name,
|
||
|
|
"air_date": ep.air_date.isoformat() if ep.air_date else None,
|
||
|
|
"editor_name_snapshot": ep.editor_name_snapshot,
|
||
|
|
"audience_share": ep.audience_share,
|
||
|
|
"audience_rating": ep.audience_rating,
|
||
|
|
}
|
||
|
|
for ep in episodes
|
||
|
|
]
|
||
|
|
|
||
|
|
yearly_target = None
|
||
|
|
if target:
|
||
|
|
yearly_target = {
|
||
|
|
"base_target": target.base_target,
|
||
|
|
"stretch_target": target.stretch_target,
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
"year": year,
|
||
|
|
"yearly_target": yearly_target,
|
||
|
|
"episodes": ep_list,
|
||
|
|
}
|