2569718930@qq.com
37494a7192
将 web/routes.py 拆分为模块化 router + service 架构
- 新增 web/app_factory.py 集中注册 7 个域名 router
- 新增 web/routers/ 薄壳路由层（auth/city/system/scan/ops/payments/analytics）
- 新增 web/services/ 业务函数下沉（每域独立 service 文件）
- web/routes.py 缩减为 city_runtime 的兼容重导出 facade
- analysis_service.py/app.py 适配新入口并清理冗余导入
Scope-risk: LOW — 全量 170 测试通过，router 注册顺序与原路由一致
Tested: python -m pytest -q (170 passed), ruff check . (All checks passed)
@
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
"""Analytics API service functions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
from src.database.db_manager import DBManager
|
|
from web.core import AnalyticsEventRequest
|
|
import web.routes as legacy_routes
|
|
|
|
|
|
def track_analytics_event(request: Request, body: AnalyticsEventRequest) -> Dict[str, Any]:
|
|
legacy_routes._bind_optional_supabase_identity(request)
|
|
event_type = str(body.event_type or "").strip().lower()
|
|
if event_type not in legacy_routes.TRACKABLE_ANALYTICS_EVENTS:
|
|
raise HTTPException(status_code=400, detail="unsupported_event_type")
|
|
|
|
payload = body.payload if isinstance(body.payload, dict) else {}
|
|
normalized_payload = {
|
|
key: value
|
|
for key, value in payload.items()
|
|
if isinstance(key, str) and len(key) <= 64
|
|
}
|
|
|
|
db = DBManager()
|
|
db.append_app_analytics_event(
|
|
event_type,
|
|
normalized_payload,
|
|
user_id=getattr(request.state, "auth_user_id", None),
|
|
client_id=body.client_id,
|
|
session_id=body.session_id,
|
|
)
|
|
return {"ok": True}
|