Найти в Дзене

Система верификации: сложный, но необходимый модуль


Одним из самых сложных для реализации стал модуль верификации пользователей. Задача была непростой: нужно было обеспечить доступ к определённым функциям бота только сотрудникам организации, при этом сохранив часть функционала доступной для всех.

class AccessControl:
def __init__(self, config):
self.config = config
self.pending_invites = {}
self.bot = None
self.load_invites()

async def is_authorized(self, user_id: int) -> bool:
try:
# Проверяем список авторизованных пользователей
if user_id in self.config.data.get('authorized_users', []):
return True

# Проверяем членство в группах
if not self.bot:
logger.error("Bot instance not set in AccessControl")
return False

for group_id in self.config.data.get('groups', []):
try:
member = await self.bot.get_chat_member(group_id, user_id)
if member.status in ['member', 'administrator', 'creator']:
self.add_authorized_user(user_id)
return True
except Exception as e:
logger.warning(f"Ошибка при проверке членства пользователя {user_id} в группе {group_id}: {e}")
continue

return False
except Exception as e:
logger.error(f"Ошибка при проверке авторизации пользователя {user_id}: {e}")
return False

Реализовал систему одноразовых приглашений и автоматическую верификацию через членство в корпоративных Telegram-группах. Это позволило сделать процесс подключения новых пользователей максимально удобным, сохранив при этом необходимый уровень безопасности.

Система оповещений: самый амбициозный модуль

Система оповещений оказалась, пожалуй, самым сложным модулем в проекте. Основные вызовы:

1. Необходимость восстановления запланированных оповещений после перезапуска бота
2. Разные типы периодичности (разовые, ежедневные, еженедельные, ежемесячные)

async def schedule_notification(self, notification_data: Dict[str, Any], context) -> bool:
try:
notification_time = datetime.strptime(notification_data['time'], "%H:%M").time()
tz = self._get_timezone()

local_dt = datetime.now(tz).replace(
hour=notification_time.hour,
minute=notification_time.minute,
second=0,
microsecond=0
)
utc_time = local_dt.astimezone(pytz.UTC).time()

trigger = CronTrigger(
hour=utc_time.hour,
minute=utc_time.minute,
timezone=pytz.UTC
)

job = self.job_queue.scheduler.add_job(
self.send_scheduled_notification,
trigger=trigger,
id=job_name,
name=job_name,
replace_existing=True,
kwargs={
'context': context,
'notification_data': notification_data
}
)

Особое внимание пришлось уделить обработке ошибок и восстановлению состояния после сбоев. Каждое оповещение сохраняется в конфигурации и может быть восстановлено:

async def restore_scheduled_notifications(self) -> None:
try:
scheduled_notifications = self.config.data.get('notifications', {}).get('scheduled', [])
restored_count = 0
failed_count = 0

for notification in scheduled_notifications:
try:
if await self.schedule_notification(notification, self.context):
restored_count += 1
else:
failed_count += 1
except Exception as e:
failed_count += 1
logger.error(f"Ошибка при восстановлении оповещения: {e}")

logger.info(f"Восстановлено {restored_count} оповещений, не удалось восстановить {failed_count}")
except Exception as e:
logger.error(f"Ошибка при восстановлении оповещений: {e}", exc_info=True)
2 минуты