diff --git a/backend/bot/app/web/admin_api_impl/broadcast.py b/backend/bot/app/web/admin_api_impl/broadcast.py index 7daae95..3c7c094 100644 --- a/backend/bot/app/web/admin_api_impl/broadcast.py +++ b/backend/bot/app/web/admin_api_impl/broadcast.py @@ -1,5 +1,84 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 +from .common import _panel_user_connection_activity + +import asyncio +from collections import defaultdict + + +BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED = "active_never_connected" +BROADCAST_TARGETS = { + "all", + "active", + "inactive", + "expired", + "never", + BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED, +} +PANEL_ACTIVITY_LOOKUP_CONCURRENCY = 10 + + +def _resolve_panel_service(request: web.Request) -> Any: + subscription_service = request.app.get("subscription_service") + return getattr(subscription_service, "panel_service", None) + + +async def _active_subscription_panel_uuids_by_user( + session: AsyncSession, +) -> Dict[int, List[str]]: + now = datetime.now(timezone.utc) + stmt = ( + select(Subscription.user_id, Subscription.panel_user_uuid) + .join(User, Subscription.user_id == User.user_id) + .where( + User.is_banned == False, + Subscription.is_active == True, + Subscription.end_date > now, + Subscription.panel_user_uuid.is_not(None), + Subscription.panel_user_uuid != "", + ) + .order_by(Subscription.user_id.asc(), Subscription.end_date.desc()) + ) + result = await session.execute(stmt) + + grouped: Dict[int, List[str]] = defaultdict(list) + seen: Dict[int, set[str]] = defaultdict(set) + for user_id, panel_uuid in result.all(): + user_id_int = int(user_id) + panel_uuid_str = str(panel_uuid or "").strip() + if panel_uuid_str and panel_uuid_str not in seen[user_id_int]: + grouped[user_id_int].append(panel_uuid_str) + seen[user_id_int].add(panel_uuid_str) + return dict(grouped) + + +async def _panel_connection_status(panel_service: Any, panel_uuid: str) -> str: + try: + panel_user = await panel_service.get_user_by_uuid(panel_uuid) + except Exception as exc: + logger.warning("Failed to fetch panel user activity uuid=%s: %s", panel_uuid, exc) + return "unknown" + activity = _panel_user_connection_activity(panel_user) + return str(activity.get("status") or "unknown") + + +async def _user_ids_with_active_subscription_never_connected( + session: AsyncSession, + panel_service: Any, +) -> List[int]: + panel_uuids_by_user = await _active_subscription_panel_uuids_by_user(session) + semaphore = asyncio.Semaphore(PANEL_ACTIVITY_LOOKUP_CONCURRENCY) + + async def lookup(panel_uuid: str) -> str: + async with semaphore: + return await _panel_connection_status(panel_service, panel_uuid) + + user_ids: List[int] = [] + for user_id, panel_uuids in panel_uuids_by_user.items(): + statuses = await asyncio.gather(*(lookup(panel_uuid) for panel_uuid in panel_uuids)) + if statuses and all(status == "never" for status in statuses): + user_ids.append(user_id) + return user_ids async def admin_broadcast_route(request: web.Request) -> web.Response: @@ -9,7 +88,7 @@ async def admin_broadcast_route(request: web.Request) -> web.Response: target = str(payload.get("target") or "all").strip().lower() if not text: return _error(400, "empty_text") - if target not in {"all", "active", "inactive", "expired", "never"}: + if target not in BROADCAST_TARGETS: target = "all" queue_manager = get_queue_manager() @@ -18,7 +97,15 @@ async def admin_broadcast_route(request: web.Request) -> web.Response: async_session_factory: sessionmaker = request.app["async_session_factory"] async with async_session_factory() as session: - if target == "active": + if target == BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED: + panel_service = _resolve_panel_service(request) + if panel_service is None: + return _error(503, "panel_service_unavailable") + user_ids = await _user_ids_with_active_subscription_never_connected( + session, + panel_service, + ) + elif target == "active": user_ids = await user_dal.get_user_ids_with_active_subscription(session) elif target == "inactive": user_ids = await user_dal.get_user_ids_without_active_subscription(session) @@ -64,12 +151,21 @@ async def admin_broadcast_audience_counts_route(request: web.Request) -> web.Res async_session_factory: sessionmaker = request.app["async_session_factory"] async with async_session_factory() as session: + panel_service = _resolve_panel_service(request) counts = { "all": len(await user_dal.get_all_active_user_ids_for_broadcast(session)), "active": len(await user_dal.get_user_ids_with_active_subscription(session)), "inactive": len(await user_dal.get_user_ids_without_active_subscription(session)), "expired": len(await user_dal.get_user_ids_with_expired_subscription(session)), "never": len(await user_dal.get_user_ids_without_any_subscription(session)), + BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED: None, } + if panel_service is not None: + counts[BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED] = len( + await _user_ids_with_active_subscription_never_connected( + session, + panel_service, + ) + ) return _ok({"counts": counts}) diff --git a/backend/bot/app/web/admin_api_impl/common.py b/backend/bot/app/web/admin_api_impl/common.py index 907914d..bfc8653 100644 --- a/backend/bot/app/web/admin_api_impl/common.py +++ b/backend/bot/app/web/admin_api_impl/common.py @@ -22,6 +22,165 @@ async def _read_json(request: web.Request) -> Dict[str, Any]: return {} +_PANEL_LAST_CONNECTED_KEYS = ( + "onlineAt", + "online_at", + "lastSeenAt", + "last_seen_at", + "lastConnectedAt", + "last_connected_at", + "lastConnectionAt", + "last_connection_at", +) +_PANEL_CONNECTION_MARKER_KEYS = ( + *_PANEL_LAST_CONNECTED_KEYS, + "firstConnectedAt", + "first_connected_at", + "lastConnectedNodeUuid", + "last_connected_node_uuid", +) +_PANEL_CONNECTION_MARKER_OBJECT_KEYS = ("lastConnectedNode", "last_connected_node") +_PANEL_TRAFFIC_OBJECT_KEYS = ("userTraffic", "user_traffic", "traffic", "trafficStats") +_PANEL_TRAFFIC_USED_KEYS = ( + "lifetimeUsedTrafficBytes", + "lifetime_used_traffic_bytes", + "usedTrafficBytes", + "used_traffic_bytes", + "trafficUsedBytes", + "traffic_used_bytes", + "downloadBytes", + "download_bytes", + "uploadBytes", + "upload_bytes", +) + + +def _panel_user_payload(panel_user_data: Any) -> Dict[str, Any]: + if not isinstance(panel_user_data, dict): + return {} + response = panel_user_data.get("response") + if isinstance(response, dict) and not any( + key in panel_user_data + for key in ("uuid", "shortUuid", "subscriptionUrl", "userTraffic", "status") + ): + return response + return panel_user_data + + +def _coerce_panel_datetime(value: Any) -> Optional[str]: + if value is None or value is False: + return None + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, (int, float)): + if value <= 0: + return None + seconds = float(value) / 1000.0 if value > 10_000_000_000 else float(value) + try: + return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat() + except (OSError, OverflowError, ValueError): + return None + text = str(value).strip() + if not text or text.lower() in {"0", "null", "none", "never"}: + return None + if text.isdigit(): + return _coerce_panel_datetime(int(text)) + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + return None + return parsed.isoformat() + + +def _coerce_panel_int(value: Any) -> Optional[int]: + try: + if value is None or value == "": + return None + return int(float(value)) + except (TypeError, ValueError): + return None + + +def _panel_nested_dicts(panel_user: Dict[str, Any], keys: Tuple[str, ...]) -> List[Dict[str, Any]]: + out: List[Dict[str, Any]] = [] + for key in keys: + value = panel_user.get(key) + if isinstance(value, dict): + out.append(value) + return out + + +def _panel_user_last_connected_at(panel_user_data: Any) -> Optional[str]: + panel_user = _panel_user_payload(panel_user_data) + if not panel_user: + return None + containers = [ + panel_user, + *_panel_nested_dicts(panel_user, _PANEL_CONNECTION_MARKER_OBJECT_KEYS), + ] + for container in containers: + for key in _PANEL_LAST_CONNECTED_KEYS: + connected_at = _coerce_panel_datetime(container.get(key)) + if connected_at: + return connected_at + return None + + +def _panel_user_positive_traffic_bytes(panel_user: Dict[str, Any]) -> bool: + containers = [panel_user, *_panel_nested_dicts(panel_user, _PANEL_TRAFFIC_OBJECT_KEYS)] + for container in containers: + for key in _PANEL_TRAFFIC_USED_KEYS: + value = _coerce_panel_int(container.get(key)) + if value is not None and value > 0: + return True + return False + + +def _panel_user_has_connection_marker(panel_user: Dict[str, Any]) -> bool: + for key in _PANEL_CONNECTION_MARKER_KEYS: + if key in panel_user: + return True + for key in _PANEL_CONNECTION_MARKER_OBJECT_KEYS: + if key in panel_user: + return True + return False + + +def _panel_user_has_connected_marker_value(panel_user: Dict[str, Any]) -> bool: + containers = [ + panel_user, + *_panel_nested_dicts(panel_user, _PANEL_CONNECTION_MARKER_OBJECT_KEYS), + ] + for container in containers: + for key in (*_PANEL_LAST_CONNECTED_KEYS, "firstConnectedAt", "first_connected_at"): + if _coerce_panel_datetime(container.get(key)): + return True + for key in ("lastConnectedNodeUuid", "last_connected_node_uuid"): + if str(container.get(key) or "").strip(): + return True + for key in _PANEL_CONNECTION_MARKER_OBJECT_KEYS: + marker = panel_user.get(key) + if isinstance(marker, dict) and any(str(value or "").strip() for value in marker.values()): + return True + if marker and not isinstance(marker, dict): + return True + return False + + +def _panel_user_connection_activity(panel_user_data: Any) -> Dict[str, Any]: + panel_user = _panel_user_payload(panel_user_data) + last_connected_at = _panel_user_last_connected_at(panel_user) + if not panel_user: + return {"status": "unknown", "last_connected_at": None} + if last_connected_at or _panel_user_positive_traffic_bytes(panel_user): + return {"status": "connected", "last_connected_at": last_connected_at} + if _panel_user_has_connected_marker_value(panel_user): + return {"status": "connected", "last_connected_at": last_connected_at} + if _panel_user_has_connection_marker(panel_user): + return {"status": "never", "last_connected_at": None} + return {"status": "unknown", "last_connected_at": None} + + def _serialize_user(user: User) -> Dict[str, Any]: return { "user_id": int(user.user_id), diff --git a/frontend/src/lib/admin/stores/broadcastStore.js b/frontend/src/lib/admin/stores/broadcastStore.js index 3a1af9f..e431be1 100644 --- a/frontend/src/lib/admin/stores/broadcastStore.js +++ b/frontend/src/lib/admin/stores/broadcastStore.js @@ -14,6 +14,14 @@ export function createBroadcastStore({ api, onToast, at }) { { value: "active", label: at("broadcast_target_active", {}, "С подпиской") }, { value: "inactive", label: at("broadcast_target_inactive", {}, "Без подписки") }, { value: "expired", label: at("broadcast_target_expired", {}, "Expired subscription") }, + { + value: "active_never_connected", + label: at( + "broadcast_target_active_never_connected", + {}, + "С подпиской, но без VPN-подключений" + ), + }, { value: "never", label: at("broadcast_target_never", {}, "Без подписки и без истории"), diff --git a/locales/en.json b/locales/en.json index f234711..19bce94 100644 --- a/locales/en.json +++ b/locales/en.json @@ -1200,6 +1200,7 @@ "admin_broadcast_target_active": "With subscription", "admin_broadcast_target_inactive": "No subscription", "admin_broadcast_target_expired": "Expired subscription", + "admin_broadcast_target_active_never_connected": "With subscription, no VPN connections", "admin_broadcast_target_never": "No subscription, no history", "admin_expired_at": "Expired {date}", "admin_expired_badge": "Expired {date}", diff --git a/locales/ru.json b/locales/ru.json index 6900916..031d215 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -1200,6 +1200,7 @@ "admin_broadcast_target_active": "С подпиской", "admin_broadcast_target_inactive": "Без подписки", "admin_broadcast_target_expired": "С просроченной подпиской", + "admin_broadcast_target_active_never_connected": "С подпиской, но без VPN-подключений", "admin_broadcast_target_never": "Без подписки и без истории", "admin_expired_at": "Истекла {date}", "admin_expired_badge": "Expired {date}", diff --git a/tests/test_admin_panel_activity.py b/tests/test_admin_panel_activity.py new file mode 100644 index 0000000..be9205b --- /dev/null +++ b/tests/test_admin_panel_activity.py @@ -0,0 +1,100 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from bot.app.web.admin_api_impl import broadcast as broadcast_module +from bot.app.web.admin_api_impl import common as common_module + + +class FakeResult: + def __init__(self, rows=None): + self._rows = rows or [] + + def all(self): + return self._rows + + +class AdminPanelActivityTests(unittest.IsolatedAsyncioTestCase): + def test_panel_activity_detects_connected_and_never_connected_users(self): + self.assertEqual( + common_module._panel_user_connection_activity( + {"onlineAt": "2026-06-05T12:00:00Z"} + ), + { + "status": "connected", + "last_connected_at": "2026-06-05T12:00:00+00:00", + }, + ) + self.assertEqual( + common_module._panel_user_connection_activity( + { + "onlineAt": None, + "firstConnectedAt": None, + "lastConnectedNode": None, + "userTraffic": {"lifetimeUsedTrafficBytes": 0}, + } + ), + {"status": "never", "last_connected_at": None}, + ) + self.assertEqual( + common_module._panel_user_connection_activity( + {"userTraffic": {"lifetimeUsedTrafficBytes": 1024}} + ), + {"status": "connected", "last_connected_at": None}, + ) + + async def test_active_never_connected_audience_uses_panel_status(self): + session = SimpleNamespace( + execute=AsyncMock( + return_value=FakeResult( + [ + (1, "never-panel"), + (2, "connected-panel"), + (3, "missing-panel"), + (4, "also-never-panel"), + (4, "also-connected-panel"), + ] + ) + ) + ) + + async def get_user_by_uuid(panel_uuid): + return { + "never-panel": { + "onlineAt": None, + "firstConnectedAt": None, + "lastConnectedNode": None, + }, + "connected-panel": {"onlineAt": "2026-06-05T12:00:00Z"}, + "also-never-panel": { + "onlineAt": None, + "firstConnectedAt": None, + "lastConnectedNode": None, + }, + "also-connected-panel": { + "userTraffic": {"lifetimeUsedTrafficBytes": 1}, + }, + }.get(panel_uuid) + + panel_service = SimpleNamespace(get_user_by_uuid=AsyncMock(side_effect=get_user_by_uuid)) + + result = await broadcast_module._user_ids_with_active_subscription_never_connected( + session, + panel_service, + ) + + self.assertEqual(result, [1]) + self.assertEqual( + [call.args[0] for call in panel_service.get_user_by_uuid.await_args_list], + [ + "never-panel", + "connected-panel", + "missing-panel", + "also-never-panel", + "also-connected-panel", + ], + ) + + +if __name__ == "__main__": + unittest.main()