smolkik-code
7c7c88621d
- FastAPI REST API with JWT auth - aiogram 3 Telegram bot with admin middleware - APScheduler daily tasks (expiry, reminders, revoke, sync) - SQLAlchemy 2 async ORM with Alembic migrations - Jinja2 admin panel (Dashboard, Users, Payments, Servers, Tariffs) - VPN provider abstraction with MockProvider - Stats service with revenue/subscription analytics - Docker Compose (PostgreSQL + Redis + app) - Healthcheck endpoint
72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from loguru import logger
|
|
|
|
from app.providers.base import VPNProvider, VPNUserInfo
|
|
|
|
|
|
class MockProvider(VPNProvider):
|
|
def __init__(self) -> None:
|
|
self._store: dict[str, VPNUserInfo] = {}
|
|
|
|
async def create_user(
|
|
self,
|
|
internal_id: int,
|
|
server_id: int,
|
|
name: str,
|
|
) -> VPNUserInfo:
|
|
vpn_id = str(uuid.uuid4())
|
|
info = VPNUserInfo(
|
|
vpn_id=vpn_id,
|
|
internal_id=internal_id,
|
|
server_id=server_id,
|
|
name=name,
|
|
is_enabled=True,
|
|
access_config=(
|
|
f"mock://{vpn_id}/server/{server_id}/user/{internal_id}"
|
|
),
|
|
bytes_used=0,
|
|
data_limit=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
self._store[vpn_id] = info
|
|
logger.debug(
|
|
"[MockProvider] User created: vpn_id={} internal_id={} server={}",
|
|
vpn_id, internal_id, server_id,
|
|
)
|
|
return info
|
|
|
|
async def delete_user(self, vpn_id: str) -> bool:
|
|
if vpn_id in self._store:
|
|
del self._store[vpn_id]
|
|
logger.debug("[MockProvider] User deleted: vpn_id={}", vpn_id)
|
|
return True
|
|
logger.warning("[MockProvider] User not found for delete: {}", vpn_id)
|
|
return False
|
|
|
|
async def enable_user(self, vpn_id: str) -> bool:
|
|
info = self._store.get(vpn_id)
|
|
if info is None:
|
|
logger.warning("[MockProvider] User not found for enable: {}", vpn_id)
|
|
return False
|
|
info.is_enabled = True
|
|
logger.debug("[MockProvider] User enabled: vpn_id={}", vpn_id)
|
|
return True
|
|
|
|
async def disable_user(self, vpn_id: str) -> bool:
|
|
info = self._store.get(vpn_id)
|
|
if info is None:
|
|
logger.warning("[MockProvider] User not found for disable: {}", vpn_id)
|
|
return False
|
|
info.is_enabled = False
|
|
logger.debug("[MockProvider] User disabled: vpn_id={}", vpn_id)
|
|
return True
|
|
|
|
async def get_user(self, vpn_id: str) -> VPNUserInfo | None:
|
|
info = self._store.get(vpn_id)
|
|
if info is None:
|
|
logger.debug("[MockProvider] User not found: vpn_id={}", vpn_id)
|
|
return None
|
|
return info
|