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.
This commit is contained in:
machka-pasla
+160
-256
@@ -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")
|
||||
Reference in New Issue
Block a user