浅念
39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
"""GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any, Literal
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.analytics import health as health_svc
|
|
from app.analytics import metrics as metrics_svc
|
|
from app.api.deps import get_session
|
|
|
|
router = APIRouter(prefix="/api/analytics", tags=["analytics"])
|
|
|
|
SessionDep = Annotated[AsyncSession, Depends(get_session)]
|
|
ChannelParam = Annotated[Literal["wechat", "whatsapp"], Query()]
|
|
ScopeParam = Annotated[Literal["today", "week", "month"], Query()]
|
|
|
|
|
|
@router.get("/health")
|
|
async def analytics_health(session: SessionDep) -> dict[str, Any]:
|
|
"""E0-lite:数据截至 + 实体新鲜度。"""
|
|
as_of = await health_svc.data_as_of(session)
|
|
return {"data_as_of": as_of, "entities": await health_svc.entity_freshness(session)}
|
|
|
|
|
|
@router.get("/overview")
|
|
async def analytics_overview(
|
|
session: SessionDep,
|
|
channel: ChannelParam = "wechat",
|
|
scope: ScopeParam = "today",
|
|
) -> dict[str, Any]:
|
|
"""A1 经营总览:卡片+趋势+异常警示一次返回(单往返)。
|
|
|
|
cards 不含 data_as_of——数据截至走 /health(data_as_of 为 ms-epoch int
|
|
与 entities[].last_seen 为 ISO 字符串的双格式)。
|
|
"""
|
|
return await metrics_svc.overview(session, channel=channel, scope=scope)
|