feat(analytics): E0-lite 数据健康端点——数据截至+实体新鲜度
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
"""分析服务层——所有数据分析指标的口径收敛处(spec 2026-09-04 §2)。
|
||||
|
||||
指标函数是纯查询(只依赖 AsyncSession,无 HTTP/请求上下文),E2 报表订阅
|
||||
(二批)直接复用。口径速查表:spec §8;函数 docstring 与之同步。
|
||||
"""
|
||||
@@ -0,0 +1,48 @@
|
||||
"""E0-lite 数据健康——「数据截至」角标 + 实体新鲜度。
|
||||
|
||||
完整 E0(调度日志解析/成功率账本)二批;v1 用表新鲜度判定:
|
||||
msg_daily_stats 当日有行 = 4h 消息链路活着。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.db.models import (
|
||||
ChatGroup,
|
||||
Customer,
|
||||
Media,
|
||||
Message,
|
||||
MomentsPost,
|
||||
MsgDailyStats,
|
||||
PhoneCall,
|
||||
)
|
||||
|
||||
|
||||
async def data_as_of(session: AsyncSession) -> int | None:
|
||||
"""数据截至(ms-epoch)= MAX(msg_daily_stats.last_ts)。"""
|
||||
val = (
|
||||
await session.execute(sa.select(sa.func.max(MsgDailyStats.last_ts)))
|
||||
).scalar_one_or_none()
|
||||
return int(val) if val is not None else None
|
||||
|
||||
|
||||
async def entity_freshness(session: AsyncSession) -> list[dict]:
|
||||
"""各实体最后更新时间(UPDATED_AT 列 MAX;message 用 fetched_at)。"""
|
||||
|
||||
async def _max(col) -> dt.datetime | None:
|
||||
return (await session.execute(sa.select(sa.func.max(col)))).scalar_one_or_none()
|
||||
|
||||
out = []
|
||||
for name, col in (
|
||||
("customer", Customer.updated_at),
|
||||
("chat_group", ChatGroup.updated_at),
|
||||
("phone_call", PhoneCall.updated_at),
|
||||
("media", Media.updated_at),
|
||||
("moments_post", MomentsPost.updated_at),
|
||||
("message", Message.fetched_at),
|
||||
):
|
||||
out.append({"table": name, "last_seen": await _max(col)})
|
||||
return out
|
||||
@@ -21,6 +21,7 @@ Includes:
|
||||
* ``GET /api/search`` (see :mod:`app.api.routers.search`).
|
||||
* ``GET /api/export/chat-html`` (see :mod:`app.api.routers.export`).
|
||||
* ``GET /api/staff-stats`` (see :mod:`app.api.routers.staff_stats`).
|
||||
* ``GET /api/analytics/health`` (see :mod:`app.api.routers.analytics`).
|
||||
|
||||
CORS is configured to read explicit allowed origins from ``Settings.cors_origins``
|
||||
(comma-separated ``CORS_ORIGINS`` env var). Empty (default) falls back to the dev
|
||||
@@ -42,6 +43,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api.auth import require_auth
|
||||
from app.api.routers import (
|
||||
analytics,
|
||||
calls,
|
||||
customer,
|
||||
export,
|
||||
@@ -98,6 +100,7 @@ app.include_router(search.router)
|
||||
app.include_router(sessions.router)
|
||||
app.include_router(export.router)
|
||||
app.include_router(voice_clips.router)
|
||||
app.include_router(analytics.router)
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
"""GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Any
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.analytics import health as health_svc
|
||||
from app.api.deps import get_session
|
||||
|
||||
router = APIRouter(prefix="/api/analytics", tags=["analytics"])
|
||||
|
||||
SessionDep = Annotated[AsyncSession, Depends(get_session)]
|
||||
|
||||
|
||||
@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)}
|
||||
@@ -0,0 +1,39 @@
|
||||
"""/api/analytics/health:E0-lite 冒烟。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from app.db.models import MsgDailyStats
|
||||
|
||||
|
||||
async def test_health_empty(make_client):
|
||||
async def seed(session):
|
||||
...
|
||||
|
||||
client = await make_client(seed)
|
||||
r = await client.get("/api/analytics/health")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["data_as_of"] is None
|
||||
assert any(e["table"] == "message" for e in body["entities"])
|
||||
|
||||
|
||||
async def test_health_with_stats(make_client):
|
||||
async def seed(session):
|
||||
session.add(
|
||||
MsgDailyStats(
|
||||
channel="wechat",
|
||||
owner_account_id="wx1",
|
||||
peer_kind="friend",
|
||||
peer_id="t1",
|
||||
day=20260904,
|
||||
send_cnt=1,
|
||||
recv_cnt=2,
|
||||
last_ts=1756900000000,
|
||||
last_is_send=False,
|
||||
)
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
client = await make_client(seed)
|
||||
r = await client.get("/api/analytics/health")
|
||||
assert r.status_code == 200
|
||||
assert r.json()["data_as_of"] == 1756900000000
|
||||
Reference in New Issue
Block a user