浅念
193 lines
7.0 KiB
Python
193 lines
7.0 KiB
Python
"""GET /api/analytics/* —— 数据分析板块端组(前缀分组=未来角色权限落点)。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any, Literal
|
|
|
|
import sqlalchemy as sa
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.analytics import health as health_svc
|
|
from app.analytics import metrics as metrics_svc
|
|
from app.analytics.sensitive import hits as hits_svc
|
|
from app.analytics.sensitive import words as words_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()]
|
|
# 看板 v2(2026-09-07):overview 默认全平台——"all" 在 router 层翻译成
|
|
# metrics 的 channel=None,不把哨兵字符串传进 SQL
|
|
AllChannelParam = Annotated[Literal["all", "wechat", "whatsapp"], Query()]
|
|
ScopeParam = Annotated[Literal["today", "week", "month", "last30"], 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: AllChannelParam = "all",
|
|
scope: ScopeParam = "today",
|
|
) -> dict[str, Any]:
|
|
"""A1 经营总览:卡片+趋势+异常警示一次返回(单往返)。
|
|
|
|
channel=all(默认)=全平台口径:卡片合计+by_channel 拆分+trend_by_channel
|
|
双渠道趋势序列;单渠道行为不变。
|
|
cards 不含 data_as_of——数据截至走 /health(data_as_of 为 ms-epoch int
|
|
与 entities[].last_seen 为 ISO 字符串的双格式)。
|
|
"""
|
|
return await metrics_svc.overview(
|
|
session, channel=None if channel == "all" else channel, scope=scope
|
|
)
|
|
|
|
|
|
@router.get("/churn")
|
|
async def churn_list(
|
|
session: SessionDep,
|
|
channel: ChannelParam = "wechat",
|
|
type: Literal["real", "inactive", "handover"] = "real",
|
|
tier: int = Query(30, ge=30, le=90),
|
|
min_msgs: int = Query(5, ge=1, le=100),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
) -> dict[str, Any]:
|
|
"""A3 流失预警名单(三分法)。tier 只认 30/60/90。"""
|
|
if tier not in (30, 60, 90):
|
|
raise HTTPException(422, "tier must be 30/60/90")
|
|
return await metrics_svc.churn(
|
|
session, channel=channel, kind=type, tier=tier,
|
|
min_msgs=min_msgs, page=page, page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.get("/unreplied")
|
|
async def unreplied_list(
|
|
session: SessionDep,
|
|
channel: ChannelParam = "wechat",
|
|
hours: int = Query(2, ge=1, le=168),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
) -> dict[str, Any]:
|
|
"""B2 该回未回清单(spec §3.4)。hours=超时阈值(默认 2h)。"""
|
|
return await metrics_svc.unreplied(
|
|
session, channel=channel, hours=hours, page=page, page_size=page_size
|
|
)
|
|
|
|
|
|
class WordIn(BaseModel):
|
|
word: str
|
|
category: str | None = None
|
|
min_len: int = 2
|
|
exclude_words: list[str] | None = None
|
|
|
|
|
|
class WordPreviewIn(BaseModel):
|
|
word: str
|
|
exclude_words: list[str] | None = None
|
|
|
|
|
|
@router.get("/sensitive/words")
|
|
async def list_words(session: SessionDep) -> list[dict]:
|
|
"""C1 词库清单(enabled=false 的词 scanner 跳过)。"""
|
|
return [
|
|
{"id": w.id, "word": w.word, "category": w.category, "enabled": w.enabled,
|
|
"min_len": w.min_len, "exclude_words": w.exclude_words or []}
|
|
for w in await words_svc.list_words(session)
|
|
]
|
|
|
|
|
|
@router.post("/sensitive/words")
|
|
async def create_word(session: SessionDep, body: WordIn) -> dict:
|
|
"""加词(重复 409)。词内容是敏感数据,错误信息只回 id 不回词。"""
|
|
try:
|
|
w = await words_svc.create_word(
|
|
session, word=body.word, category=body.category,
|
|
min_len=body.min_len, exclude_words=body.exclude_words,
|
|
)
|
|
await session.commit()
|
|
return {"id": w.id, "word": w.word, "enabled": w.enabled}
|
|
except ValueError as exc:
|
|
raise HTTPException(409, str(exc)) from exc
|
|
except sa.exc.IntegrityError as exc:
|
|
# 并发撞 word UNIQUE(两请求同词,预检都未见)→ 409,不裸 500。
|
|
raise HTTPException(409, "词已存在") from exc
|
|
|
|
|
|
@router.post("/sensitive/words/preview")
|
|
async def preview_word(session: SessionDep, body: WordPreviewIn) -> dict:
|
|
"""加词前命中量预览(count 为 MySQL 上限语义;warn=超阈值建议收窄)。
|
|
|
|
MySQL 单字词拒扫(400)——LIKE 前导通配全扫 ~16s,加词后从命中列表观察。
|
|
"""
|
|
try:
|
|
n = await words_svc.preview_count(session, body.word, body.exclude_words)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
return {"count": n, "warn": n >= words_svc.PREVIEW_WARN_THRESHOLD}
|
|
|
|
|
|
@router.patch("/sensitive/words/{word_id}")
|
|
async def update_word(session: SessionDep, word_id: int, body: dict) -> dict:
|
|
"""改词(启停/分类/排除词);字段 None=不改。"""
|
|
try:
|
|
w = await words_svc.update_word(session, word_id, **body)
|
|
await session.commit()
|
|
return {"id": w.id, "enabled": w.enabled, "category": w.category}
|
|
except LookupError as exc:
|
|
raise HTTPException(404, str(exc)) from exc
|
|
|
|
|
|
@router.delete("/sensitive/words/{word_id}")
|
|
async def delete_word(session: SessionDep, word_id: int) -> dict:
|
|
"""删词=级联删命中(UI 二次确认)。"""
|
|
try:
|
|
await words_svc.delete_word(session, word_id)
|
|
await session.commit()
|
|
return {"ok": True}
|
|
except LookupError as exc:
|
|
raise HTTPException(404, str(exc)) from exc
|
|
|
|
|
|
class DisposeIn(BaseModel):
|
|
# 巨 IN 列表会撞 MySQL max_allowed_packet 裸 500 → pydantic 层拦成干净 422
|
|
ids: list[int] = Field(max_length=1000)
|
|
disposition: str
|
|
|
|
|
|
DispositionParam = Literal["pending", "confirmed", "false_positive", "ignored"]
|
|
|
|
|
|
@router.get("/sensitive/hits")
|
|
async def list_hits(
|
|
session: SessionDep,
|
|
disposition: DispositionParam | None = None,
|
|
word_id: int | None = None,
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
) -> dict:
|
|
"""C1 命中列表——不传 disposition 默认只回 pending+confirmed(治理闭环:
|
|
处置完即出队列);word_id 供词详情下钻。disposition 拼错 422 不静默空列表。"""
|
|
return await hits_svc.list_hits(
|
|
session, disposition=disposition, word_id=word_id,
|
|
page=page, page_size=page_size,
|
|
)
|
|
|
|
|
|
@router.patch("/sensitive/hits")
|
|
async def dispose_hits(session: SessionDep, body: DisposeIn) -> dict:
|
|
"""批量处置(pending 合法——重开误报);UPDATE 不写日志(词/snippet 红线)。"""
|
|
try:
|
|
n = await hits_svc.dispose(session, body.ids, body.disposition)
|
|
await session.commit()
|
|
return {"updated": n}
|
|
except ValueError as exc:
|
|
raise HTTPException(422, str(exc)) from exc
|