austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

Refactor perform_sync function for readability

Browse Source
This commit is contained in:
orryxvpn
2025-11-13 21:30:15 +05:00
committed by GitHub
parent 183bef070d
commit eb6e343e4c
1 changed files with 180 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/handlers/admin/sync_admin.py
+180 -67
View File
@@ -3,6 +3,7 @@ from aiogram import Router, types, Bot
from aiogram.filters import Command from aiogram.filters import Command
from typing import Optional, Union from typing import Optional, Union
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import update, or_
from datetime import datetime, timezone from datetime import datetime, timezone
from config.settings import Settings from config.settings import Settings
@@ -10,14 +11,19 @@ from bot.services.panel_api_service import PanelApiService
from bot.services.notification_service import NotificationService from bot.services.notification_service import NotificationService
from db.dal import user_dal, subscription_dal, panel_sync_dal from db.dal import user_dal, subscription_dal, panel_sync_dal
from db.models import Subscription
from bot.middlewares.i18n import JsonI18n from bot.middlewares.i18n import JsonI18n
router = Router(name="admin_sync_router") router = Router(name="admin_sync_router")
async def perform_sync(panel_service: PanelApiService, session: AsyncSession, async def perform_sync(
settings: Settings, i18n_instance: JsonI18n) -> dict: panel_service: PanelApiService,
session: AsyncSession,
settings: Settings,
i18n_instance: JsonI18n,
) -> dict:
""" """
Perform panel synchronization and return results Perform panel synchronization and return results
Returns dict with status, details, and sync statistics Returns dict with status, details, and sync statistics
@@ -27,7 +33,7 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
users_updated = 0 users_updated = 0
subscriptions_synced_count = 0 subscriptions_synced_count = 0
sync_errors = [] sync_errors = []
# Additional counters for detailed logging # Additional counters for detailed logging
users_without_telegram_id = 0 users_without_telegram_id = 0
users_not_found_in_db = 0 users_not_found_in_db = 0
@@ -52,7 +58,12 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
session, "success", status_msg, 0, 0 session, "success", status_msg, 0, 0
) )
await session.commit() await session.commit()
return {"status": "success", "details": status_msg, "users_synced": 0, "subs_synced": 0} return {
"status": "success",
"details": status_msg,
"users_synced": 0,
"subs_synced": 0,
}
total_panel_users = len(panel_users_data) total_panel_users = len(panel_users_data)
logging.info(f"Starting sync for {total_panel_users} panel users.") logging.info(f"Starting sync for {total_panel_users} panel users.")
@@ -61,12 +72,16 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
try: try:
panel_records_checked += 1 panel_records_checked += 1
panel_uuid = panel_user_dict.get("uuid") panel_uuid = panel_user_dict.get("uuid")
panel_subscription_uuid = panel_user_dict.get("subscriptionUuid") or panel_user_dict.get("shortUuid") panel_subscription_uuid = panel_user_dict.get("subscriptionUuid") or panel_user_dict.get(
"shortUuid"
)
telegram_id_from_panel = panel_user_dict.get("telegramId") telegram_id_from_panel = panel_user_dict.get("telegramId")
if not panel_uuid: if not panel_uuid:
sync_errors.append(f"Panel user missing UUID: {panel_user_dict}") sync_errors.append(f"Panel user missing UUID: {panel_user_dict}")
logging.warning(f"Skipping panel user without UUID: {panel_user_dict}") logging.warning(
f"Skipping panel user without UUID: {panel_user_dict}"
)
continue continue
# Track users without telegram ID # Track users without telegram ID
@@ -75,22 +90,35 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
# Try to find existing user in local DB # Try to find existing user in local DB
existing_user = None existing_user = None
# First, try to find by telegram ID if available # First, try to find by telegram ID if available
if telegram_id_from_panel: if telegram_id_from_panel:
existing_user = await user_dal.get_user_by_id(session, telegram_id_from_panel) existing_user = await user_dal.get_user_by_id(
session, telegram_id_from_panel
)
if existing_user: if existing_user:
logging.debug(f"Found user by telegramId {telegram_id_from_panel}") logging.debug(
f"Found user by telegramId {telegram_id_from_panel}"
)
# If not found by telegram ID, try to find by panel UUID # If not found by telegram ID, try to find by panel UUID
if not existing_user: if not existing_user:
existing_user = await user_dal.get_user_by_panel_uuid(session, panel_uuid) existing_user = await user_dal.get_user_by_panel_uuid(
session, panel_uuid
)
if existing_user: if existing_user:
logging.info(f"Found user by panel UUID {panel_uuid}, telegramId: {existing_user.user_id}") logging.info(
f"Found user by panel UUID {panel_uuid}, telegramId: {existing_user.user_id}"
)
# Update telegram ID if it was missing in panel data but we have local user # Update telegram ID if it was missing in panel data but we have local user
if telegram_id_from_panel and existing_user.user_id != telegram_id_from_panel: if (
logging.warning(f"TelegramId mismatch: panel={telegram_id_from_panel}, local={existing_user.user_id}") telegram_id_from_panel
and existing_user.user_id != telegram_id_from_panel
):
logging.warning(
f"TelegramId mismatch: panel={telegram_id_from_panel}, local={existing_user.user_id}"
)
if not existing_user: if not existing_user:
users_not_found_in_db += 1 users_not_found_in_db += 1
if telegram_id_from_panel: if telegram_id_from_panel:
@@ -100,26 +128,36 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
"user_id": telegram_id_from_panel, "user_id": telegram_id_from_panel,
"username": None, # Username will be updated when user interacts with bot "username": None, # Username will be updated when user interacts with bot
"first_name": None, # Panel doesn't provide this info "first_name": None, # Panel doesn't provide this info
"last_name": None, # Panel doesn't provide this info "last_name": None, # Panel doesn't provide this info
"language_code": "ru", # Default language "language_code": "ru", # Default language
"panel_user_uuid": panel_uuid, "panel_user_uuid": panel_uuid,
"is_banned": False, "is_banned": False,
"referred_by_id": None "referred_by_id": None,
} }
new_user, was_created = await user_dal.create_user(session, user_data) new_user, was_created = await user_dal.create_user(
session, user_data
)
if was_created: if was_created:
users_created += 1 users_created += 1
logging.info(f"Created new user {telegram_id_from_panel} from panel sync with UUID {panel_uuid}") logging.info(
f"Created new user {telegram_id_from_panel} from panel sync with UUID {panel_uuid}"
)
existing_user = new_user existing_user = new_user
except Exception as e_create: except Exception as e_create:
sync_errors.append(f"Error creating user {telegram_id_from_panel}: {str(e_create)}") sync_errors.append(
logging.error(f"Error creating user {telegram_id_from_panel}: {e_create}") f"Error creating user {telegram_id_from_panel}: {str(e_create)}"
)
logging.error(
f"Error creating user {telegram_id_from_panel}: {e_create}"
)
continue continue
else: else:
logging.debug(f"Panel user with UUID {panel_uuid} (no telegramId) not found in local DB - skipping") logging.debug(
f"Panel user with UUID {panel_uuid} (no telegramId) not found in local DB - skipping"
)
continue continue
# User found in local DB # User found in local DB
@@ -134,20 +172,29 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
existing_user.panel_user_uuid = panel_uuid existing_user.panel_user_uuid = panel_uuid
user_was_updated = True user_was_updated = True
users_uuid_updated += 1 users_uuid_updated += 1
logging.info(f"Updated panel UUID for user {actual_user_id}: {panel_uuid}") logging.info(
f"Updated panel UUID for user {actual_user_id}: {panel_uuid}"
)
# Ensure panel description contains Telegram fields # Ensure panel description contains Telegram fields
try: try:
if panel_uuid and existing_user: if panel_uuid and existing_user:
description_text = "\n".join([ description_text = "\n".join(
existing_user.username or "", [
existing_user.first_name or "", existing_user.username or "",
existing_user.last_name or "", existing_user.first_name or "",
]) existing_user.last_name or "",
]
)
# Update description only when it differs from the current one on panel # Update description only when it differs from the current one on panel
current_panel_description = (panel_user_dict.get("description") or "").strip() current_panel_description = (
panel_user_dict.get("description") or ""
).strip()
desired_description = description_text.strip() desired_description = description_text.strip()
if desired_description and desired_description != current_panel_description: if (
desired_description
and desired_description != current_panel_description
):
await panel_service.update_user_details_on_panel( await panel_service.update_user_details_on_panel(
panel_uuid, {"description": description_text} panel_uuid, {"description": description_text}
) )
@@ -159,13 +206,13 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
# Sync subscription data # Sync subscription data
panel_expire_at_iso = panel_user_dict.get("expireAt") panel_expire_at_iso = panel_user_dict.get("expireAt")
panel_status = panel_user_dict.get("status", "UNKNOWN") panel_status = panel_user_dict.get("status", "UNKNOWN")
if panel_expire_at_iso: if panel_expire_at_iso:
try: try:
panel_expire_at = datetime.fromisoformat( panel_expire_at = datetime.fromisoformat(
panel_expire_at_iso.replace("Z", "+00:00") panel_expire_at_iso.replace("Z", "+00:00")
) )
# Prefer syncing by concrete subscription UUID (shortUuid/subscriptionUuid) # Prefer syncing by concrete subscription UUID (shortUuid/subscriptionUuid)
subscription_uuid_from_panel = ( subscription_uuid_from_panel = (
panel_user_dict.get("subscriptionUuid") panel_user_dict.get("subscriptionUuid")
@@ -173,6 +220,27 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
) )
if subscription_uuid_from_panel: if subscription_uuid_from_panel:
# Если панель говорит, что подписка ACTIVE — сначала деактивируем все другие активные
if panel_status == "ACTIVE":
await session.execute(
update(Subscription)
.where(
Subscription.panel_user_uuid == panel_uuid,
Subscription.is_active.is_(True),
or_(
Subscription.panel_subscription_uuid
!= subscription_uuid_from_panel,
Subscription.panel_subscription_uuid.is_(
None
),
),
)
.values(
is_active=False,
status_from_panel="INACTIVE",
)
)
# Try to find subscription by its panel_subscription_uuid first (idempotent) # Try to find subscription by its panel_subscription_uuid first (idempotent)
existing_sub_by_uuid = ( existing_sub_by_uuid = (
await subscription_dal.get_subscription_by_panel_subscription_uuid( await subscription_dal.get_subscription_by_panel_subscription_uuid(
@@ -197,7 +265,8 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
subscriptions_updated += 1 subscriptions_updated += 1
user_was_updated = True user_was_updated = True
logging.info( logging.info(
f"Synced existing subscription {existing_sub_by_uuid.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" f"Synced existing subscription {existing_sub_by_uuid.subscription_id} "
f"for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}"
) )
else: else:
# Create a new subscription only when we have a concrete subscription UUID # Create a new subscription only when we have a concrete subscription UUID
@@ -221,12 +290,15 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
subscriptions_created += 1 subscriptions_created += 1
user_was_updated = True user_was_updated = True
logging.info( logging.info(
f"Created subscription {created_sub.subscription_id} for user {actual_user_id} by panel_sub_uuid {subscription_uuid_from_panel}" f"Created subscription {created_sub.subscription_id} "
f"for user {actual_user_id} by panel_sub_uuid {subscription_uuid_from_panel}"
) )
else: else:
# No subscription UUID from panel: only update an already active subscription for this user/panel UUID # No subscription UUID from panel: only update an already active subscription for this user/panel UUID
active_sub = await subscription_dal.get_active_subscription_by_user_id( active_sub = (
session, actual_user_id, panel_uuid await subscription_dal.get_active_subscription_by_user_id(
session, actual_user_id, panel_uuid
)
) )
if active_sub: if active_sub:
await subscription_dal.update_subscription( await subscription_dal.update_subscription(
@@ -242,23 +314,30 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
subscriptions_updated += 1 subscriptions_updated += 1
user_was_updated = True user_was_updated = True
logging.info( logging.info(
f"Updated active subscription {active_sub.subscription_id} for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" f"Updated active subscription {active_sub.subscription_id} "
f"for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}"
) )
else: else:
# Without a concrete subscription UUID we avoid creating new records to keep sync idempotent # Without a concrete subscription UUID we avoid creating new records to keep sync idempotent
logging.debug( logging.debug(
f"No subscriptionUuid for panel user {panel_uuid}; skipped creation for user {actual_user_id}" f"No subscriptionUuid for panel user {panel_uuid}; skipped creation for user {actual_user_id}"
) )
except Exception as e: except Exception as e:
sync_errors.append(f"Error syncing subscription for user {actual_user_id}: {str(e)}") sync_errors.append(
logging.error(f"Error syncing subscription for user {actual_user_id}: {e}") f"Error syncing subscription for user {actual_user_id}: {str(e)}"
)
logging.error(
f"Error syncing subscription for user {actual_user_id}: {e}"
)
if user_was_updated: if user_was_updated:
users_updated += 1 users_updated += 1
except Exception as e_user: except Exception as e_user:
sync_errors.append(f"Error processing panel user {panel_user_dict.get('uuid', 'unknown')}: {str(e_user)}") sync_errors.append(
f"Error processing panel user {panel_user_dict.get('uuid', 'unknown')}: {str(e_user)}"
)
logging.error(f"Error syncing user: {e_user}") logging.error(f"Error syncing user: {e_user}")
# Update sync status # Update sync status
@@ -267,14 +346,26 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
default_lang = settings.DEFAULT_LANGUAGE default_lang = settings.DEFAULT_LANGUAGE
additional_stats = "" additional_stats = ""
if users_without_telegram_id > 0: if users_without_telegram_id > 0:
additional_stats += i18n_instance.gettext(default_lang, "admin_sync_no_telegram_id", count=users_without_telegram_id) additional_stats += i18n_instance.gettext(
default_lang,
"admin_sync_no_telegram_id",
count=users_without_telegram_id,
)
if users_not_found_in_db > 0: if users_not_found_in_db > 0:
additional_stats += i18n_instance.gettext(default_lang, "admin_sync_not_found_in_db", count=users_not_found_in_db) additional_stats += i18n_instance.gettext(
default_lang,
"admin_sync_not_found_in_db",
count=users_not_found_in_db,
)
if sync_errors: if sync_errors:
additional_stats += i18n_instance.gettext(default_lang, "admin_sync_errors", count=len(sync_errors)) additional_stats += i18n_instance.gettext(
default_lang, "admin_sync_errors", count=len(sync_errors)
)
# Build full details using localization # Build full details using localization
details = i18n_instance.gettext(default_lang, "admin_sync_details", details = i18n_instance.gettext(
default_lang,
"admin_sync_details",
panel_records_checked=panel_records_checked, panel_records_checked=panel_records_checked,
users_found_in_db=users_found_in_db, users_found_in_db=users_found_in_db,
users_created=users_created, users_created=users_created,
@@ -282,11 +373,15 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
subscriptions_synced_count=subscriptions_synced_count, subscriptions_synced_count=subscriptions_synced_count,
subscriptions_created=subscriptions_created, subscriptions_created=subscriptions_created,
subscriptions_updated=subscriptions_updated, subscriptions_updated=subscriptions_updated,
additional_stats=additional_stats additional_stats=additional_stats,
) )
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, status, details, panel_records_checked, subscriptions_synced_count session,
status,
details,
panel_records_checked,
subscriptions_synced_count,
) )
await session.commit() await session.commit()
@@ -311,19 +406,27 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
"users_synced": users_found_in_db, "users_synced": users_found_in_db,
"users_created": users_created, "users_created": users_created,
"subs_synced": subscriptions_synced_count, "subs_synced": subscriptions_synced_count,
"errors": sync_errors "errors": sync_errors,
} }
except Exception as e_sync_global: except Exception as e_sync_global:
await session.rollback() await session.rollback()
logging.error(f"Global error during sync: {e_sync_global}", exc_info=True) logging.error(f"Global error during sync: {e_sync_global}", exc_info=True)
error_detail = f"Unexpected error during sync: {str(e_sync_global)}" error_detail = f"Unexpected error during sync: {str(e_sync_global)}"
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, "failed", error_detail, panel_records_checked, subscriptions_synced_count session,
"failed",
error_detail,
panel_records_checked,
subscriptions_synced_count,
) )
return {"status": "failed", "details": error_detail, "errors": [str(e_sync_global)]} return {
"status": "failed",
"details": error_detail,
"errors": [str(e_sync_global)],
}
@router.message(Command("sync")) @router.message(Command("sync"))
@@ -366,34 +469,40 @@ async def sync_command_handler(
# Use the extracted perform_sync function # Use the extracted perform_sync function
try: try:
sync_result = await perform_sync(panel_service, session, settings, i18n) sync_result = await perform_sync(panel_service, session, settings, i18n)
status = sync_result.get("status") status = sync_result.get("status")
details = sync_result.get("details", "No details available") details = sync_result.get("details", "No details available")
errors = sync_result.get("errors", []) errors = sync_result.get("errors", [])
# Simple confirmation message to admin # Simple confirmation message to admin
if status == "failed": if status == "failed":
await bot.send_message(target_chat_id, _("sync_failed_simple")) await bot.send_message(target_chat_id, _("sync_failed_simple"))
elif status == "completed_with_errors": elif status == "completed_with_errors":
await bot.send_message(target_chat_id, _("sync_errors_simple", errors_count=len(errors))) await bot.send_message(
target_chat_id,
_("sync_errors_simple", errors_count=len(errors)),
)
else: else:
await bot.send_message(target_chat_id, _("sync_success_simple")) await bot.send_message(target_chat_id, _("sync_success_simple"))
# Send notification to log channel with proper thread handling # Send notification to log channel with proper thread handling
try: try:
notification_service = NotificationService(bot, settings, i18n) notification_service = NotificationService(bot, settings, i18n)
await notification_service.notify_panel_sync( await notification_service.notify_panel_sync(
status, details, status,
details,
sync_result.get("users_processed", 0), sync_result.get("users_processed", 0),
sync_result.get("subs_synced", 0) sync_result.get("subs_synced", 0),
) )
except Exception as e_notification: except Exception as e_notification:
logging.error(f"Failed to send sync notification: {e_notification}") logging.error(f"Failed to send sync notification: {e_notification}")
except Exception as e_sync_global: except Exception as e_sync_global:
logging.error(f"Global error during /sync command: {e_sync_global}", exc_info=True) logging.error(
f"Global error during /sync command: {e_sync_global}", exc_info=True
)
await bot.send_message(target_chat_id, _("sync_critical_error")) await bot.send_message(target_chat_id, _("sync_critical_error"))
# Send notification to log channel about failure # Send notification to log channel about failure
try: try:
notification_service = NotificationService(bot, settings, i18n) notification_service = NotificationService(bot, settings, i18n)
@@ -401,7 +510,9 @@ async def sync_command_handler(
"failed", str(e_sync_global), 0, 0 "failed", str(e_sync_global), 0, 0
) )
except Exception as e_notification: except Exception as e_notification:
logging.error(f"Failed to send sync failure notification: {e_notification}") logging.error(
f"Failed to send sync failure notification: {e_notification}"
)
@router.message(Command("syncstatus")) @router.message(Command("syncstatus"))
@@ -420,7 +531,9 @@ async def sync_status_command_handler(
if status_record_model: if status_record_model:
last_time_val = status_record_model.last_sync_time last_time_val = status_record_model.last_sync_time
last_time_str = ( last_time_str = (
last_time_val.strftime("%Y-%m-%d %H:%M:%S UTC") if last_time_val else "N/A" last_time_val.strftime("%Y-%m-%d %H:%M:%S UTC")
if last_time_val
else "N/A"
) )
details_val = status_record_model.details details_val = status_record_model.details