Add ProfileSyncMiddleware to keep user profile data updated in the database
- Integrated ProfileSyncMiddleware to ensure that user profile information (username, first_name, last_name) remains current in the database. - Updated get_enhanced_user_statistics to use timezone-aware datetime for accurate SQL queries, preventing naive/aware comparison issues. - Clarified comments in the user statistics function for better understanding of active user metrics and subscription handling.
This commit is contained in:
@@ -0,0 +1,2 @@
|
|||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
import logging
|
||||||
|
from typing import Dict
|
||||||
|
|
||||||
|
from aiogram import Bot, Dispatcher
|
||||||
|
from aiogram.enums import ParseMode
|
||||||
|
from aiogram.client.default import DefaultBotProperties
|
||||||
|
from aiogram.fsm.storage.memory import MemoryStorage
|
||||||
|
from sqlalchemy.orm import sessionmaker
|
||||||
|
|
||||||
|
from config.settings import Settings
|
||||||
|
from bot.middlewares.db_session import DBSessionMiddleware
|
||||||
|
from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance
|
||||||
|
from bot.middlewares.ban_check_middleware import BanCheckMiddleware
|
||||||
|
from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
|
||||||
|
from bot.middlewares.profile_sync import ProfileSyncMiddleware
|
||||||
|
from bot.routers import build_root_router
|
||||||
|
|
||||||
|
|
||||||
|
def build_dispatcher(settings: Settings, async_session_factory: sessionmaker) -> tuple[Dispatcher, Bot, Dict]:
|
||||||
|
storage = MemoryStorage()
|
||||||
|
default_props = DefaultBotProperties(parse_mode=ParseMode.HTML)
|
||||||
|
bot = Bot(token=settings.BOT_TOKEN, default=default_props)
|
||||||
|
|
||||||
|
dp = Dispatcher(storage=storage, settings=settings, bot_instance=bot)
|
||||||
|
|
||||||
|
i18n_instance = get_i18n_instance(path="locales", default=settings.DEFAULT_LANGUAGE)
|
||||||
|
|
||||||
|
dp["i18n_instance"] = i18n_instance
|
||||||
|
dp["async_session_factory"] = async_session_factory
|
||||||
|
|
||||||
|
dp.update.outer_middleware(DBSessionMiddleware(async_session_factory))
|
||||||
|
dp.update.outer_middleware(I18nMiddleware(i18n=i18n_instance, settings=settings))
|
||||||
|
dp.update.outer_middleware(ProfileSyncMiddleware())
|
||||||
|
dp.update.outer_middleware(BanCheckMiddleware(settings=settings, i18n_instance=i18n_instance))
|
||||||
|
dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings))
|
||||||
|
|
||||||
|
dp.include_router(build_root_router(settings))
|
||||||
|
|
||||||
|
return dp, bot, {"i18n_instance": i18n_instance}
|
||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,55 @@
|
|||||||
|
from aiogram import Bot
|
||||||
|
from sqlalchemy.orm import sessionmaker
|
||||||
|
from typing import Tuple
|
||||||
|
|
||||||
|
from config.settings import Settings
|
||||||
|
from bot.services.yookassa_service import YooKassaService
|
||||||
|
from bot.services.panel_api_service import PanelApiService
|
||||||
|
from bot.services.subscription_service import SubscriptionService
|
||||||
|
from bot.services.referral_service import ReferralService
|
||||||
|
from bot.services.promo_code_service import PromoCodeService
|
||||||
|
from bot.services.stars_service import StarsService
|
||||||
|
from bot.services.tribute_service import TributeService
|
||||||
|
from bot.services.crypto_pay_service import CryptoPayService
|
||||||
|
from bot.services.panel_webhook_service import PanelWebhookService
|
||||||
|
|
||||||
|
|
||||||
|
def build_core_services(settings: Settings, bot: Bot, async_session_factory: sessionmaker):
|
||||||
|
panel_service = PanelApiService(settings)
|
||||||
|
subscription_service = SubscriptionService(settings, panel_service, bot)
|
||||||
|
referral_service = ReferralService(settings, subscription_service, bot)
|
||||||
|
promo_code_service = PromoCodeService(settings, subscription_service, bot)
|
||||||
|
stars_service = StarsService(bot, settings, None, subscription_service, referral_service)
|
||||||
|
cryptopay_service = CryptoPayService(
|
||||||
|
settings.CRYPTOPAY_TOKEN,
|
||||||
|
settings.CRYPTOPAY_NETWORK,
|
||||||
|
bot,
|
||||||
|
settings,
|
||||||
|
None,
|
||||||
|
async_session_factory,
|
||||||
|
subscription_service,
|
||||||
|
referral_service,
|
||||||
|
)
|
||||||
|
tribute_service = TributeService(
|
||||||
|
bot,
|
||||||
|
settings,
|
||||||
|
None,
|
||||||
|
async_session_factory,
|
||||||
|
panel_service,
|
||||||
|
subscription_service,
|
||||||
|
referral_service,
|
||||||
|
)
|
||||||
|
panel_webhook_service = PanelWebhookService(bot, settings, None, async_session_factory)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"panel_service": panel_service,
|
||||||
|
"subscription_service": subscription_service,
|
||||||
|
"referral_service": referral_service,
|
||||||
|
"promo_code_service": promo_code_service,
|
||||||
|
"stars_service": stars_service,
|
||||||
|
"cryptopay_service": cryptopay_service,
|
||||||
|
"tribute_service": tribute_service,
|
||||||
|
"panel_webhook_service": panel_webhook_service,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,2 @@
|
|||||||
|
|
||||||
|
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
import asyncio
|
||||||
|
import logging
|
||||||
|
from aiohttp import web
|
||||||
|
from aiogram import Bot, Dispatcher
|
||||||
|
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
|
||||||
|
from sqlalchemy.orm import sessionmaker
|
||||||
|
|
||||||
|
from config.settings import Settings
|
||||||
|
|
||||||
|
|
||||||
|
async def build_and_start_web_app(
|
||||||
|
dp: Dispatcher,
|
||||||
|
bot: Bot,
|
||||||
|
settings: Settings,
|
||||||
|
async_session_factory: sessionmaker,
|
||||||
|
):
|
||||||
|
app = web.Application()
|
||||||
|
app["bot"] = bot
|
||||||
|
app["dp"] = dp
|
||||||
|
app["settings"] = settings
|
||||||
|
app["async_session_factory"] = async_session_factory
|
||||||
|
|
||||||
|
setup_application(app, dp, bot=bot)
|
||||||
|
|
||||||
|
telegram_uses_webhook_mode = bool(settings.WEBHOOK_BASE_URL)
|
||||||
|
|
||||||
|
if telegram_uses_webhook_mode:
|
||||||
|
telegram_webhook_path = f"/{settings.BOT_TOKEN}"
|
||||||
|
app.router.add_post(telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot))
|
||||||
|
logging.info(
|
||||||
|
f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
|
||||||
|
)
|
||||||
|
|
||||||
|
from bot.services.tribute_service import tribute_webhook_route
|
||||||
|
from bot.services.crypto_pay_service import cryptopay_webhook_route
|
||||||
|
from bot.services.panel_webhook_service import panel_webhook_route
|
||||||
|
|
||||||
|
tribute_path = settings.tribute_webhook_path
|
||||||
|
if tribute_path.startswith("/"):
|
||||||
|
app.router.add_post(tribute_path, tribute_webhook_route)
|
||||||
|
logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
|
||||||
|
|
||||||
|
cp_path = settings.cryptopay_webhook_path
|
||||||
|
if cp_path.startswith("/"):
|
||||||
|
app.router.add_post(cp_path, cryptopay_webhook_route)
|
||||||
|
logging.info(f"CryptoPay webhook route configured at: [POST] {cp_path}")
|
||||||
|
|
||||||
|
panel_path = settings.panel_webhook_path
|
||||||
|
if panel_path.startswith("/"):
|
||||||
|
app.router.add_post(panel_path, panel_webhook_route)
|
||||||
|
logging.info(f"Panel webhook route configured at: [POST] {panel_path}")
|
||||||
|
|
||||||
|
web_app_runner = web.AppRunner(app)
|
||||||
|
await web_app_runner.setup()
|
||||||
|
site = web.TCPSite(
|
||||||
|
web_app_runner,
|
||||||
|
host=settings.WEB_SERVER_HOST,
|
||||||
|
port=settings.WEB_SERVER_PORT,
|
||||||
|
)
|
||||||
|
|
||||||
|
await site.start()
|
||||||
|
logging.info(
|
||||||
|
f"AIOHTTP server started on http://{settings.WEB_SERVER_HOST}:{settings.WEB_SERVER_PORT}"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Run until cancelled
|
||||||
|
await asyncio.Event().wait()
|
||||||
|
|
||||||
|
|
||||||
@@ -20,6 +20,7 @@ from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n
|
|||||||
from bot.middlewares.db_session import DBSessionMiddleware
|
from bot.middlewares.db_session import DBSessionMiddleware
|
||||||
from bot.middlewares.ban_check_middleware import BanCheckMiddleware
|
from bot.middlewares.ban_check_middleware import BanCheckMiddleware
|
||||||
from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
|
from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
|
||||||
|
from bot.middlewares.profile_sync import ProfileSyncMiddleware
|
||||||
|
|
||||||
from bot.routers import build_root_router
|
from bot.routers import build_root_router
|
||||||
|
|
||||||
@@ -314,6 +315,8 @@ async def run_bot(settings_param: Settings):
|
|||||||
dp.update.outer_middleware(
|
dp.update.outer_middleware(
|
||||||
I18nMiddleware(i18n=i18n_instance, settings=settings_param)
|
I18nMiddleware(i18n=i18n_instance, settings=settings_param)
|
||||||
)
|
)
|
||||||
|
# Keep profile data in DB fresh (username/first_name/last_name)
|
||||||
|
dp.update.outer_middleware(ProfileSyncMiddleware())
|
||||||
dp.update.outer_middleware(
|
dp.update.outer_middleware(
|
||||||
BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance)
|
BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -0,0 +1,47 @@
|
|||||||
|
import logging
|
||||||
|
from typing import Callable, Dict, Any, Awaitable, Optional
|
||||||
|
|
||||||
|
from aiogram import BaseMiddleware
|
||||||
|
from aiogram.types import Update, User as TgUser
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
|
from db.dal import user_dal
|
||||||
|
|
||||||
|
|
||||||
|
class ProfileSyncMiddleware(BaseMiddleware):
|
||||||
|
|
||||||
|
async def __call__(
|
||||||
|
self,
|
||||||
|
handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
|
||||||
|
event: Update,
|
||||||
|
data: Dict[str, Any],
|
||||||
|
) -> Any:
|
||||||
|
session: AsyncSession = data.get("session")
|
||||||
|
tg_user: Optional[TgUser] = data.get("event_from_user")
|
||||||
|
|
||||||
|
if session and tg_user:
|
||||||
|
try:
|
||||||
|
db_user = await user_dal.get_user_by_id(session, tg_user.id)
|
||||||
|
if db_user:
|
||||||
|
update_payload: Dict[str, Any] = {}
|
||||||
|
if db_user.username != tg_user.username:
|
||||||
|
update_payload["username"] = tg_user.username
|
||||||
|
if db_user.first_name != tg_user.first_name:
|
||||||
|
update_payload["first_name"] = tg_user.first_name
|
||||||
|
if db_user.last_name != tg_user.last_name:
|
||||||
|
update_payload["last_name"] = tg_user.last_name
|
||||||
|
|
||||||
|
if update_payload:
|
||||||
|
await user_dal.update_user(session, tg_user.id, update_payload)
|
||||||
|
logging.info(
|
||||||
|
f"ProfileSyncMiddleware: Updated user {tg_user.id} profile fields: {list(update_payload.keys())}"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logging.error(
|
||||||
|
f"ProfileSyncMiddleware: Failed to sync profile for user {getattr(tg_user, 'id', 'N/A')}: {e}",
|
||||||
|
exc_info=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
return await handler(event, data)
|
||||||
|
|
||||||
|
|
||||||
+6
-7
@@ -93,9 +93,10 @@ async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]:
|
|||||||
|
|
||||||
async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
|
async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
|
||||||
"""Get comprehensive user statistics including active users, trial users, etc."""
|
"""Get comprehensive user statistics including active users, trial users, etc."""
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
now = datetime.utcnow()
|
# Use timezone-aware UTC to avoid naive/aware comparison issues in SQL queries
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||||
|
|
||||||
# Total users
|
# Total users
|
||||||
@@ -106,13 +107,11 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
|
|||||||
banned_users_stmt = select(func.count(User.user_id)).where(User.is_banned == True)
|
banned_users_stmt = select(func.count(User.user_id)).where(User.is_banned == True)
|
||||||
banned_users = (await session.execute(banned_users_stmt)).scalar() or 0
|
banned_users = (await session.execute(banned_users_stmt)).scalar() or 0
|
||||||
|
|
||||||
# Active users today (users with login activity - for now using registration as proxy)
|
# Active users today (proxy: registered today)
|
||||||
active_today_stmt = select(func.count(User.user_id)).where(
|
active_today_stmt = select(func.count(User.user_id)).where(User.registration_date >= today_start)
|
||||||
User.registration_date >= today_start
|
|
||||||
)
|
|
||||||
active_today = (await session.execute(active_today_stmt)).scalar() or 0
|
active_today = (await session.execute(active_today_stmt)).scalar() or 0
|
||||||
|
|
||||||
# Users with active paid subscriptions
|
# Users with active paid subscriptions (non-trial providers only)
|
||||||
paid_subs_stmt = (
|
paid_subs_stmt = (
|
||||||
select(func.count(func.distinct(Subscription.user_id)))
|
select(func.count(func.distinct(Subscription.user_id)))
|
||||||
.join(User, Subscription.user_id == User.user_id)
|
.join(User, Subscription.user_id == User.user_id)
|
||||||
|
|||||||
Reference in New Issue
Block a user