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

Refactor admin sync process for enhanced user and subscription tracking

Browse Source 
- Updated the `perform_sync` function to improve tracking of users and subscriptions during synchronization.
- Introduced additional counters for detailed logging, including users without Telegram IDs and those not found in the database.
- Enhanced error handling and logging to provide clearer insights into the synchronization process and outcomes.
- Improved the summary details returned after synchronization, offering a comprehensive overview of the sync results.
This commit is contained in:
machka-pasla
2025-08-06 17:21:25 +03:00
parent 990b08cfdc
commit 9f7171a5c3
1 changed files with 138 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/handlers/admin/sync_admin.py
+133 -56
View File
@@ -21,11 +21,19 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
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
""" """
users_processed_count = 0 panel_records_checked = 0
users_synced_successfully = 0 users_found_in_db = 0
users_updated = 0
subscriptions_synced_count = 0 subscriptions_synced_count = 0
sync_errors = [] sync_errors = []
# Additional counters for detailed logging
users_without_telegram_id = 0
users_not_found_in_db = 0
users_uuid_updated = 0
subscriptions_created = 0
subscriptions_updated = 0
try: try:
panel_users_data = await panel_service.get_all_panel_users() panel_users_data = await panel_service.get_all_panel_users()
@@ -49,74 +57,119 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
for panel_user_dict in panel_users_data: for panel_user_dict in panel_users_data:
try: try:
users_processed_count += 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}")
continue continue
# Sync user data # Track users without telegram ID
if not telegram_id_from_panel:
users_without_telegram_id += 1
# Try to find existing user in local DB
existing_user = None existing_user = None
# 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:
logging.debug(f"Found user by telegramId {telegram_id_from_panel}")
if existing_user: # If not found by telegram ID, try to find by panel UUID
if existing_user.panel_user_uuid != panel_uuid: if not existing_user:
existing_user.panel_user_uuid = panel_uuid existing_user = await user_dal.get_user_by_panel_uuid(session, panel_uuid)
users_synced_successfully += 1 if existing_user:
logging.info(f"Updated panel UUID for user {telegram_id_from_panel}") 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
if 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}")
# Sync subscription data if user exists if not existing_user:
if existing_user: users_not_found_in_db += 1
panel_expire_at_iso = panel_user_dict.get("expireAt") if telegram_id_from_panel:
panel_status = panel_user_dict.get("status", "UNKNOWN") logging.debug(f"Panel user with telegramId {telegram_id_from_panel} and UUID {panel_uuid} not found in local DB")
else:
logging.debug(f"Panel user with UUID {panel_uuid} (no telegramId) not found in local DB")
continue
if panel_expire_at_iso: # User found in local DB
try: users_found_in_db += 1
panel_expire_at = datetime.fromisoformat( user_was_updated = False
panel_expire_at_iso.replace("Z", "+00:00")
)
# Update or create subscription # Get the actual user_id for subscription operations
active_sub = await subscription_dal.get_active_subscription_by_user_id( actual_user_id = existing_user.user_id
session, telegram_id_from_panel, panel_uuid
)
if active_sub: # Update panel UUID if different
if active_sub.end_date != panel_expire_at: if existing_user.panel_user_uuid != panel_uuid:
await subscription_dal.update_subscription_end_date( existing_user.panel_user_uuid = panel_uuid
session, active_sub.subscription_id, panel_expire_at user_was_updated = True
) users_uuid_updated += 1
subscriptions_synced_count += 1 logging.info(f"Updated panel UUID for user {actual_user_id}: {panel_uuid}")
else:
# Create or update subscription record
# Use actual subscription UUID from panel if available, fallback to user UUID
subscription_uuid_to_use = panel_subscription_uuid or panel_uuid
if panel_subscription_uuid: # Sync subscription data
logging.info(f"Using panel subscriptionUuid {panel_subscription_uuid} for user {telegram_id_from_panel}") panel_expire_at_iso = panel_user_dict.get("expireAt")
else: panel_status = panel_user_dict.get("status", "UNKNOWN")
logging.info(f"No subscriptionUuid from panel, using panel_uuid {panel_uuid} for user {telegram_id_from_panel}")
sub_payload = { if panel_expire_at_iso:
"user_id": telegram_id_from_panel, try:
"panel_user_uuid": panel_uuid, panel_expire_at = datetime.fromisoformat(
"panel_subscription_uuid": subscription_uuid_to_use, panel_expire_at_iso.replace("Z", "+00:00")
"start_date": datetime.now(timezone.utc), )
"end_date": panel_expire_at,
"duration_months": 1, # Default # Update or create subscription
"is_active": panel_status == "ACTIVE", active_sub = await subscription_dal.get_active_subscription_by_user_id(
"status_from_panel": panel_status, session, actual_user_id, panel_uuid
"traffic_limit_bytes": settings.user_traffic_limit_bytes, )
}
await subscription_dal.upsert_subscription(session, sub_payload) if active_sub:
# Check if subscription needs update
if (active_sub.end_date != panel_expire_at or
active_sub.status_from_panel != panel_status or
active_sub.is_active != (panel_status == "ACTIVE")):
await subscription_dal.update_subscription_end_date(
session, active_sub.subscription_id, panel_expire_at
)
# Update status fields
active_sub.status_from_panel = panel_status
active_sub.is_active = (panel_status == "ACTIVE")
subscriptions_synced_count += 1 subscriptions_synced_count += 1
subscriptions_updated += 1
user_was_updated = True
logging.info(f"Updated subscription for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}")
else:
# Create new subscription record
subscription_uuid_to_use = panel_subscription_uuid or panel_uuid
except Exception as e: logging.info(f"Creating new subscription for user {actual_user_id} with UUID {subscription_uuid_to_use}")
sync_errors.append(f"Error syncing subscription for user {telegram_id_from_panel}: {str(e)}")
sub_payload = {
"user_id": actual_user_id,
"panel_user_uuid": panel_uuid,
"panel_subscription_uuid": subscription_uuid_to_use,
"start_date": datetime.now(timezone.utc),
"end_date": panel_expire_at,
"duration_months": 1, # Default
"is_active": panel_status == "ACTIVE",
"status_from_panel": panel_status,
"traffic_limit_bytes": settings.user_traffic_limit_bytes,
}
await subscription_dal.upsert_subscription(session, sub_payload)
subscriptions_synced_count += 1
subscriptions_created += 1
user_was_updated = True
except Exception as e:
sync_errors.append(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:
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)}")
@@ -124,20 +177,44 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
# Update sync status # Update sync status
status = "completed_with_errors" if sync_errors else "completed" status = "completed_with_errors" if sync_errors else "completed"
details = f"Synced {users_synced_successfully}/{users_processed_count} users, {subscriptions_synced_count} subscriptions" details = (f"📊 Статистика синхронизации:\n"
f"🔍 Проверено записей панели: {panel_records_checked}\n"
f"👥 Найдено пользователей в БД: {users_found_in_db}\n"
f"🔄 Пользователей обновлено: {users_updated}\n"
f"📋 Подписок синхронизировано: {subscriptions_synced_count}\n"
f" ├── Создано новых: {subscriptions_created}\n"
f" └── Обновлено существующих: {subscriptions_updated}")
if users_without_telegram_id > 0:
details += f"\n⚠️ Записей без telegramId: {users_without_telegram_id}"
if users_not_found_in_db > 0:
details += f"\n❌ Не найдено в БД: {users_not_found_in_db}"
if sync_errors: if sync_errors:
details += f", {len(sync_errors)} errors" details += f"\n🚫 Ошибок: {len(sync_errors)}"
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, status, details, users_processed_count, subscriptions_synced_count session, status, details, panel_records_checked, subscriptions_synced_count
) )
await session.commit() await session.commit()
# Detailed logging summary
logging.info(f"Sync completed - Summary:")
logging.info(f" Panel records checked: {panel_records_checked}")
logging.info(f" Users without telegramId: {users_without_telegram_id}")
logging.info(f" Users not found in local DB: {users_not_found_in_db}")
logging.info(f" Users found in local DB: {users_found_in_db}")
logging.info(f" Users with UUID updated: {users_uuid_updated}")
logging.info(f" Users updated overall: {users_updated}")
logging.info(f" Subscriptions total synced: {subscriptions_synced_count}")
logging.info(f" Subscriptions created: {subscriptions_created}")
logging.info(f" Subscriptions updated: {subscriptions_updated}")
logging.info(f" Sync errors: {len(sync_errors)}")
return { return {
"status": status, "status": status,
"details": details, "details": details,
"users_processed": users_processed_count, "users_processed": panel_records_checked,
"users_synced": users_synced_successfully, "users_synced": users_found_in_db,
"subs_synced": subscriptions_synced_count, "subs_synced": subscriptions_synced_count,
"errors": sync_errors "errors": sync_errors
} }
@@ -148,7 +225,7 @@ async def perform_sync(panel_service: PanelApiService, session: AsyncSession,
error_detail = f"Unexpected error during sync: {str(e_sync_global)[:200]}" error_detail = f"Unexpected error during sync: {str(e_sync_global)[:200]}"
await panel_sync_dal.update_panel_sync_status( await panel_sync_dal.update_panel_sync_status(
session, "failed", error_detail, users_processed_count, 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)]}