From 5cf3f2f76eb2cd93282f845ea0850f6707bfcc09 Mon Sep 17 00:00:00 2001 From: Fringg Date: Sun, 8 Feb 2026 21:49:42 +0300 Subject: [PATCH] feat: add traffic usage enrichment endpoint with devices, spending, dates, last node Add GET /admin/traffic/enrichment that returns per-user enrichment data (connected devices, total spending, subscription dates, last connected node) via bulk panel API calls with 5-min server-side cache. --- app/cabinet/routes/admin_traffic.py | 145 +++++++++++++++++++++++++++- app/cabinet/schemas/traffic.py | 12 +++ app/external/remnawave_api.py | 10 ++ 3 files changed, 165 insertions(+), 2 deletions(-) diff --git a/app/cabinet/routes/admin_traffic.py b/app/cabinet/routes/admin_traffic.py index 204403a0..a35327e5 100644 --- a/app/cabinet/routes/admin_traffic.py +++ b/app/cabinet/routes/admin_traffic.py @@ -12,20 +12,22 @@ from aiogram.client.default import DefaultBotProperties from aiogram.enums import ParseMode from aiogram.types import BufferedInputFile from fastapi import APIRouter, Depends, HTTPException, Query, status -from sqlalchemy import select +from sqlalchemy import and_, func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.config import settings -from app.database.models import Subscription, User +from app.database.models import Subscription, Transaction, TransactionType, User from app.services.remnawave_service import RemnaWaveService from ..dependencies import get_cabinet_db, get_current_admin_user from ..schemas.traffic import ( ExportCsvRequest, ExportCsvResponse, + TrafficEnrichmentResponse, TrafficNodeInfo, TrafficUsageResponse, + UserTrafficEnrichment, UserTrafficItem, ) @@ -346,6 +348,145 @@ async def get_traffic_usage( ) +# ============== Enrichment endpoint ============== + +_enrichment_cache: dict[str, tuple[float, dict[int, UserTrafficEnrichment]]] = {} +_ENRICHMENT_CACHE_TTL = 300 # 5 minutes +_enrichment_lock = asyncio.Lock() + + +async def _get_bulk_spending(db: AsyncSession, user_ids: list[int]) -> dict[int, int]: + """Get total spent kopeks for multiple users in a single query.""" + if not user_ids: + return {} + result = await db.execute( + select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0)) + .where( + and_( + Transaction.user_id.in_(user_ids), + Transaction.is_completed.is_(True), + Transaction.type == TransactionType.SUBSCRIPTION_PAYMENT.value, + ) + ) + .group_by(Transaction.user_id) + ) + return {row[0]: int(row[1]) for row in result.all()} + + +@router.get('/enrichment', response_model=TrafficEnrichmentResponse) +async def get_traffic_enrichment( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +): + """Return enrichment data: device counts, spending, dates, last node.""" + cache_key = 'enrichment' + now = time.time() + + cached = _enrichment_cache.get(cache_key) + if cached and (now - cached[0]) < _ENRICHMENT_CACHE_TTL: + return TrafficEnrichmentResponse(data=cached[1]) + + async with _enrichment_lock: + now = time.time() + cached = _enrichment_cache.get(cache_key) + if cached and (now - cached[0]) < _ENRICHMENT_CACHE_TTL: + return TrafficEnrichmentResponse(data=cached[1]) + + user_map = await _load_user_map(db) + + # Build uuid -> user_id and short_uuid -> user_id maps + uuid_to_user_id: dict[str, int] = {} + short_uuid_to_user_id: dict[str, int] = {} + for uuid, user in user_map.items(): + uuid_to_user_id[uuid] = user.id + if user.subscription and user.subscription.remnawave_short_uuid: + short_uuid_to_user_id[user.subscription.remnawave_short_uuid] = user.id + + service = RemnaWaveService() + devices_by_user: dict[int, int] = {} + last_node_uuid_by_user: dict[int, str] = {} + node_uuid_to_name: dict[str, str] = {} + + if service.is_configured: + try: + async with service.get_api_client() as api: + devices_task = api.get_all_hwid_devices() + subs_task = api.get_all_panel_subscriptions() + nodes_task = api.get_all_nodes() + devices_data, subs_data, nodes_list = await asyncio.gather(devices_task, subs_task, nodes_task) + except Exception: + logger.warning('Failed to fetch enrichment data from panel', exc_info=True) + devices_data = {'devices': []} + subs_data = [] + nodes_list = [] + + # Build node name map + for node in nodes_list: + node_uuid_to_name[node.uuid] = node.name + + # Count devices per user + for device in devices_data.get('devices', []): + user_uuid = device.get('userUuid', '') + uid = uuid_to_user_id.get(user_uuid) + if uid is not None: + devices_by_user[uid] = devices_by_user.get(uid, 0) + 1 + + # Extract last connected node from panel subscriptions + if isinstance(subs_data, list): + for sub in subs_data: + sub_user = sub.get('user') or sub + short_uuid = sub_user.get('shortUuid', '') + uid = short_uuid_to_user_id.get(short_uuid) + if uid is None: + user_uuid = sub_user.get('uuid', '') + uid = uuid_to_user_id.get(user_uuid) + if uid is not None: + user_traffic = sub_user.get('userTraffic') or {} + last_node = user_traffic.get('lastConnectedNodeUuid') + if last_node: + last_node_uuid_by_user[uid] = last_node + + # Bulk spending stats + all_user_ids = [u.id for u in user_map.values()] + spending_map = await _get_bulk_spending(db, all_user_ids) + + # Build enrichment data + enrichment: dict[int, UserTrafficEnrichment] = {} + for uuid, user in user_map.items(): + uid = user.id + sub = user.subscription + + start_date = None + end_date = None + if sub: + if sub.start_date: + start_date = sub.start_date.isoformat() + if sub.end_date: + end_date = sub.end_date.isoformat() + + last_node_name = None + last_uuid = last_node_uuid_by_user.get(uid) + if last_uuid: + last_node_name = node_uuid_to_name.get(last_uuid) + + enrichment[uid] = UserTrafficEnrichment( + devices_connected=devices_by_user.get(uid, 0), + total_spent_kopeks=spending_map.get(uid, 0), + subscription_start_date=start_date, + subscription_end_date=end_date, + last_node_name=last_node_name, + ) + + _enrichment_cache[cache_key] = (now, enrichment) + + # Evict expired + expired = [k for k, (ts, _) in _enrichment_cache.items() if (now - ts) >= _ENRICHMENT_CACHE_TTL] + for k in expired: + del _enrichment_cache[k] + + return TrafficEnrichmentResponse(data=enrichment) + + @router.post('/export-csv', response_model=ExportCsvResponse) async def export_traffic_csv( request: ExportCsvRequest, diff --git a/app/cabinet/schemas/traffic.py b/app/cabinet/schemas/traffic.py index 0742780d..8522f83a 100644 --- a/app/cabinet/schemas/traffic.py +++ b/app/cabinet/schemas/traffic.py @@ -33,6 +33,18 @@ class TrafficUsageResponse(BaseModel): available_statuses: list[str] +class UserTrafficEnrichment(BaseModel): + devices_connected: int = 0 + total_spent_kopeks: int = 0 + subscription_start_date: str | None = None + subscription_end_date: str | None = None + last_node_name: str | None = None + + +class TrafficEnrichmentResponse(BaseModel): + data: dict[int, UserTrafficEnrichment] + + class ExportCsvRequest(BaseModel): period: int = Field(30, ge=1, le=30) start_date: str | None = None diff --git a/app/external/remnawave_api.py b/app/external/remnawave_api.py index f5a76904..3bfbcb48 100644 --- a/app/external/remnawave_api.py +++ b/app/external/remnawave_api.py @@ -999,6 +999,16 @@ class RemnaWaveAPI: uuid=data['uuid'], name=data['name'], view_position=data['viewPosition'], config=data.get('config') ) + async def get_all_hwid_devices(self) -> dict[str, Any]: + """GET /api/hwid/devices — all devices for all users.""" + response = await self._make_request('GET', '/api/hwid/devices') + return response.get('response', {'devices': [], 'total': 0}) + + async def get_all_panel_subscriptions(self) -> list[dict[str, Any]]: + """GET /api/subscriptions — all panel subscriptions.""" + response = await self._make_request('GET', '/api/subscriptions') + return response.get('response') or [] + async def get_user_devices(self, user_uuid: str) -> dict[str, Any]: try: response = await self._make_request('GET', f'/api/hwid/devices/{user_uuid}')