diff --git a/yunke-server/app/analytics/__init__.py b/yunke-server/app/analytics/__init__.py new file mode 100644 index 0000000..d4800ef --- /dev/null +++ b/yunke-server/app/analytics/__init__.py @@ -0,0 +1,5 @@ +"""分析服务层——所有数据分析指标的口径收敛处(spec 2026-09-04 §2)。 + +指标函数是纯查询(只依赖 AsyncSession,无 HTTP/请求上下文),E2 报表订阅 +(二批)直接复用。口径速查表:spec §8;函数 docstring 与之同步。 +""" diff --git a/yunke-server/app/analytics/health.py b/yunke-server/app/analytics/health.py new file mode 100644 index 0000000..c304008 --- /dev/null +++ b/yunke-server/app/analytics/health.py @@ -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 diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index 0a24e6b..6971bb3 100644 --- a/yunke-server/app/api/main.py +++ b/yunke-server/app/api/main.py @@ -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") diff --git a/yunke-server/app/api/routers/analytics.py b/yunke-server/app/api/routers/analytics.py new file mode 100644 index 0000000..f73fea9 --- /dev/null +++ b/yunke-server/app/api/routers/analytics.py @@ -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)} diff --git a/yunke-server/tests/test_analytics_api.py b/yunke-server/tests/test_analytics_api.py new file mode 100644 index 0000000..0779716 --- /dev/null +++ b/yunke-server/tests/test_analytics_api.py @@ -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