From 359a3c46a4c205942cfe6a42354b5766db74ab2c Mon Sep 17 00:00:00 2001 From: machka-pasla Date: Wed, 6 Aug 2025 16:46:52 +0300 Subject: [PATCH] Implement panel synchronization functionality in admin handler - Added a new `perform_sync` function to handle the synchronization of users and subscriptions from the admin panel. - Enhanced error handling and logging during the sync process, providing detailed feedback on the synchronization status. - Updated the `sync_command_handler` to utilize the new `perform_sync` function, improving code organization and readability. - Improved messaging for sync results, including success and error details, to enhance admin user experience. --- bot/handlers/admin/sync_admin.py | 416 ++++++++++++------------------- 1 file changed, 160 insertions(+), 256 deletions(-) diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index 4beca0f..dce869e 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -15,6 +15,136 @@ from bot.middlewares.i18n import JsonI18n router = Router(name="admin_sync_router") +async def perform_sync(panel_service: PanelApiService, session: AsyncSession, + settings: Settings, i18n_instance: JsonI18n) -> dict: + """ + Perform panel synchronization and return results + Returns dict with status, details, and sync statistics + """ + users_processed_count = 0 + users_synced_successfully = 0 + subscriptions_synced_count = 0 + sync_errors = [] + + try: + panel_users_data = await panel_service.get_all_panel_users() + + if panel_users_data is None: + error_msg = "Failed to fetch users from panel or panel API issue." + sync_errors.append(error_msg) + await panel_sync_dal.update_panel_sync_status(session, "failed", error_msg) + await session.commit() + return {"status": "failed", "details": error_msg, "errors": sync_errors} + + if not panel_users_data: + status_msg = "No users found in the panel to sync." + await panel_sync_dal.update_panel_sync_status( + session, "success", status_msg, 0, 0 + ) + await session.commit() + return {"status": "success", "details": status_msg, "users_synced": 0, "subs_synced": 0} + + total_panel_users = len(panel_users_data) + logging.info(f"Starting sync for {total_panel_users} panel users.") + + for panel_user_dict in panel_users_data: + try: + users_processed_count += 1 + panel_uuid = panel_user_dict.get("uuid") + telegram_id_from_panel = panel_user_dict.get("telegramId") + + if not panel_uuid: + sync_errors.append(f"Panel user missing UUID: {panel_user_dict}") + continue + + # Sync user data + existing_user = None + if telegram_id_from_panel: + existing_user = await user_dal.get_user_by_id(session, telegram_id_from_panel) + + if existing_user: + if existing_user.panel_user_uuid != panel_uuid: + existing_user.panel_user_uuid = panel_uuid + users_synced_successfully += 1 + logging.info(f"Updated panel UUID for user {telegram_id_from_panel}") + + # Sync subscription data if user exists + if existing_user: + panel_expire_at_iso = panel_user_dict.get("expireAt") + panel_status = panel_user_dict.get("status", "UNKNOWN") + + if panel_expire_at_iso: + try: + panel_expire_at = datetime.fromisoformat( + panel_expire_at_iso.replace("Z", "+00:00") + ) + + # Update or create subscription + active_sub = await subscription_dal.get_active_subscription_by_user_id( + session, telegram_id_from_panel, panel_uuid + ) + + if active_sub: + if active_sub.end_date != panel_expire_at: + await subscription_dal.update_subscription_end_date( + session, active_sub.subscription_id, panel_expire_at + ) + subscriptions_synced_count += 1 + else: + # Create new subscription record if none exists + sub_payload = { + "user_id": telegram_id_from_panel, + "panel_user_uuid": panel_uuid, + "panel_subscription_uuid": panel_uuid, # Using same UUID + "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 + + except Exception as e: + sync_errors.append(f"Error syncing subscription for user {telegram_id_from_panel}: {str(e)}") + + except Exception as 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}") + + # Update sync status + status = "completed_with_errors" if sync_errors else "completed" + details = f"Synced {users_synced_successfully}/{users_processed_count} users, {subscriptions_synced_count} subscriptions" + if sync_errors: + details += f", {len(sync_errors)} errors" + + await panel_sync_dal.update_panel_sync_status( + session, status, details, users_processed_count, subscriptions_synced_count + ) + await session.commit() + + return { + "status": status, + "details": details, + "users_processed": users_processed_count, + "users_synced": users_synced_successfully, + "subs_synced": subscriptions_synced_count, + "errors": sync_errors + } + + except Exception as e_sync_global: + await session.rollback() + logging.error(f"Global error during sync: {e_sync_global}", exc_info=True) + error_detail = f"Unexpected error during sync: {str(e_sync_global)[:200]}" + + await panel_sync_dal.update_panel_sync_status( + session, "failed", error_detail, users_processed_count, subscriptions_synced_count + ) + + return {"status": "failed", "details": error_detail, "errors": [str(e_sync_global)]} + + @router.message(Command("sync")) async def sync_command_handler( message_event: Union[types.Message, types.CallbackQuery], @@ -52,265 +182,39 @@ async def sync_command_handler( logging.info(f"Admin ({message_event.from_user.id}) triggered panel sync.") - users_processed_count = 0 - users_synced_successfully = 0 - subscriptions_synced_count = 0 - sync_errors = [] - + # Use the extracted perform_sync function try: - panel_users_data = await panel_service.get_all_panel_users() - - if panel_users_data is None: - error_msg = "Failed to fetch users from panel or panel API issue." - sync_errors.append(error_msg) - await panel_sync_dal.update_panel_sync_status(session, "failed", error_msg) - await session.commit() - await bot.send_message(target_chat_id, _("sync_failed", details=error_msg)) - return - - if not panel_users_data: - status_msg = "No users found in the panel to sync." - await panel_sync_dal.update_panel_sync_status( - session, "success", status_msg, 0, 0 + sync_result = await perform_sync(panel_service, session, settings, i18n) + + status = sync_result.get("status") + details = sync_result.get("details", "No details available") + errors = sync_result.get("errors", []) + + if status == "failed": + await bot.send_message(target_chat_id, _("sync_failed", details=details)) + elif status == "completed_with_errors": + error_preview = "; ".join(errors[:3]) # Show first 3 errors + final_message = _( + "sync_completed_with_errors_details", + total_checked=sync_result.get("users_processed", 0), + users_synced=sync_result.get("users_synced", 0), + subs_synced=sync_result.get("subs_synced", 0), + errors_count=len(errors), + error_details_preview=error_preview[:200] + "..." if len(error_preview) > 200 else error_preview ) - await session.commit() - await bot.send_message( - target_chat_id, - _("sync_completed", status="Success", details=status_msg), - ) - return - - total_panel_users = len(panel_users_data) - logging.info(f"Starting sync for {total_panel_users} panel users.") - - for panel_user_dict in panel_users_data: - users_processed_count += 1 - panel_uuid = panel_user_dict.get("uuid") - telegram_id_from_panel_str = panel_user_dict.get("telegramId") - panel_username = panel_user_dict.get("username") - - if not panel_uuid: - logging.warning( - f"Sync: Panel user data missing 'uuid'. Data: {str(panel_user_dict)[:200]}. Skipping." - ) - sync_errors.append( - f"Panel user data (username: {panel_username or 'N/A'}) missing UUID." - ) - continue - - telegram_id_from_panel: Optional[int] = None - if telegram_id_from_panel_str: - try: - telegram_id_from_panel = int(telegram_id_from_panel_str) - except ValueError: - logging.warning( - f"Sync: Panel user {panel_uuid} (username: {panel_username}) has invalid 'telegramId': {telegram_id_from_panel_str}. Skipping TG ID based sync." - ) - - if not telegram_id_from_panel: - - logging.info( - f"Sync: Panel user {panel_uuid} (username: {panel_username}) has no valid 'telegramId'. Skipping full sync for this user." - ) - - continue - - bot_user = await user_dal.get_user_by_id(session, telegram_id_from_panel) - if not bot_user: - user_data_to_create = { - "user_id": telegram_id_from_panel, - "username": panel_username, - "panel_user_uuid": panel_uuid, - "language_code": settings.DEFAULT_LANGUAGE, - "registration_date": ( - datetime.fromisoformat( - panel_user_dict["createdAt"].replace("Z", "+00:00") - ) - if panel_user_dict.get("createdAt") - else datetime.now(timezone.utc) - ), - } - bot_user = await user_dal.create_user(session, user_data_to_create) - logging.info( - f"Sync: Created new local user {telegram_id_from_panel} from panel data {panel_uuid}." - ) - else: - if bot_user.panel_user_uuid != panel_uuid: - if bot_user.panel_user_uuid is not None: - logging.warning( - f"Sync: Local user {telegram_id_from_panel} was linked to {bot_user.panel_user_uuid}, panel now gives {panel_uuid}. Updating." - ) - - conflicting_user = await user_dal.get_user_by_panel_uuid( - session, panel_uuid - ) - if ( - conflicting_user - and conflicting_user.user_id != telegram_id_from_panel - ): - sync_errors.append( - f"Panel UUID {panel_uuid} for TG {telegram_id_from_panel} already linked to another TG user {conflicting_user.user_id}." - ) - logging.error(sync_errors[-1]) - continue - - await user_dal.update_user( - session, - telegram_id_from_panel, - {"panel_user_uuid": panel_uuid, "username": panel_username}, - ) - logging.info( - f"Sync: Updated panel_uuid for local user {telegram_id_from_panel} to {panel_uuid}." - ) - - panel_sub_link_id = panel_user_dict.get( - "subscriptionUuid" - ) or panel_user_dict.get("shortUuid") - if panel_sub_link_id: - end_date_str = panel_user_dict.get("expireAt") - start_date_str = panel_user_dict.get("createdAt") - - if end_date_str: - try: - end_date_obj = datetime.fromisoformat( - end_date_str.replace("Z", "+00:00") - ) - start_date_obj = ( - datetime.fromisoformat( - start_date_str.replace("Z", "+00:00") - ) - if start_date_str - else datetime.now(timezone.utc) - ) - - status_from_panel = panel_user_dict.get( - "status", "UNKNOWN" - ).upper() - is_active_flag = ( - 1 - if status_from_panel == "ACTIVE" - and end_date_obj > datetime.now(timezone.utc) - else 0 - ) - - sub_payload = { - "user_id": telegram_id_from_panel, - "panel_user_uuid": panel_uuid, - "panel_subscription_uuid": panel_sub_link_id, - "start_date": start_date_obj, - "end_date": end_date_obj, - "is_active": is_active_flag, - "status_from_panel": status_from_panel, - "traffic_limit_bytes": panel_user_dict.get( - "trafficLimitBytes" - ), - "traffic_used_bytes": panel_user_dict.get( - "usedTrafficBytes" - ), - } - - await subscription_dal.deactivate_other_active_subscriptions( - session, panel_uuid, panel_sub_link_id - ) - await subscription_dal.upsert_subscription(session, sub_payload) - subscriptions_synced_count += 1 - users_synced_successfully += 1 - except ValueError as e_date: - logging.warning( - f"Sync: Bad date format for panel user {panel_uuid} (TG ID: {telegram_id_from_panel}). Sub data: {str(panel_user_dict)[:100]}. Error: {e_date}" - ) - sync_errors.append( - f"Bad date for panel user {panel_uuid} (TG ID: {telegram_id_from_panel})." - ) - except Exception as e_sub_sync: - logging.error( - f"Sync: Error syncing subscription for panel user {panel_uuid} (TG ID: {telegram_id_from_panel}): {e_sub_sync}", - exc_info=True, - ) - sync_errors.append( - f"Sub sync error for panel user {panel_uuid} (TG ID: {telegram_id_from_panel})." - ) - else: - logging.warning( - f"Sync: Panel user {panel_uuid} (TG ID: {telegram_id_from_panel}) has sub link but no expireAt date. Skipping subscription sync." - ) - else: - - await subscription_dal.deactivate_other_active_subscriptions( - session, panel_uuid, None - ) - logging.info( - f"Sync: Panel user {panel_uuid} (TG ID: {telegram_id_from_panel}) has no subscription link on panel. Deactivated local subs if any." - ) - users_synced_successfully += 1 - - if users_processed_count % 20 == 0: - logging.info( - f"Sync progress: {users_processed_count}/{total_panel_users} users processed from panel." - ) - - panel_uuid_set = {u.get("uuid") for u in panel_users_data if u.get("uuid")} - local_users_with_uuid = await user_dal.get_all_users_with_panel_uuid(session) - for local_user in local_users_with_uuid: - if local_user.panel_user_uuid not in panel_uuid_set: - await subscription_dal.deactivate_other_active_subscriptions( - session, local_user.panel_user_uuid, None - ) - logging.info( - f"Sync: Local user {local_user.user_id} with panel UUID {local_user.panel_user_uuid} not found on panel. Deactivated local subs." - ) - - status_msg_key = "sync_completed_details" - final_status_type = "success" - - if sync_errors: - final_status_type = "partial_success" - status_msg_key = "sync_completed_with_errors_details" - error_preview = "\n".join(sync_errors[:3]) - details_for_db = f"Users processed: {users_processed_count}. Subs synced: {subscriptions_synced_count}. Errors: {len(sync_errors)}. First few: {error_preview}" + await bot.send_message(target_chat_id, final_message) else: - details_for_db = f"Successfully processed {users_processed_count} users. Synced {subscriptions_synced_count} subscriptions." - - await panel_sync_dal.update_panel_sync_status( - session, - final_status_type, - details_for_db, - users_processed_count, - subscriptions_synced_count, - ) - await session.commit() - - final_user_message = _( - status_msg_key, - total_checked=total_panel_users, - users_synced=users_synced_successfully, - subs_synced=subscriptions_synced_count, - errors_count=len(sync_errors), - error_details_preview=( - error_preview if sync_errors else _("no_errors_placeholder") - ), - ) - await bot.send_message(target_chat_id, final_user_message) - + final_message = _( + "sync_completed_details", + total_checked=sync_result.get("users_processed", 0), + users_synced=sync_result.get("users_synced", 0), + subs_synced=sync_result.get("subs_synced", 0) + ) + await bot.send_message(target_chat_id, _("sync_completed", status="Success", details=final_message)) + except Exception as e_sync_global: - await session.rollback() - logging.error( - f"Global error during /sync command: {e_sync_global}", exc_info=True - ) - error_detail_for_db = ( - f"An unexpected error occurred during sync: {str(e_sync_global)[:200]}" - ) - await panel_sync_dal.update_panel_sync_status( - session, - "failed", - error_detail_for_db, - users_processed_count, - subscriptions_synced_count, - ) - - await bot.send_message( - target_chat_id, _("sync_failed", details=error_detail_for_db) - ) + logging.error(f"Global error during /sync command: {e_sync_global}", exc_info=True) + await bot.send_message(target_chat_id, _("sync_failed", details=str(e_sync_global))) @router.message(Command("syncstatus")) @@ -350,4 +254,4 @@ async def sync_status_command_handler( else: response_text = _("admin_sync_status_never_run") - await message.answer(response_text, parse_mode="HTML") + await message.answer(response_text, parse_mode="HTML") \ No newline at end of file