Артуризм Гора

## Планы на будущее ## Сейчас в разработке несколько серьёзных обновлений: 1. Умные напоминания - Контекстные напоминания с приоритетами - Автоматическая группировка связанных событий 3. Миграция на SQL базу данных - Переход с JSON на полноценную БД - Оптимизация запросов - Улучшение производительности при масштабировании Технические уроки проекта 1. Модульная архитектура себя оправдала Разделение на независимые модули позволило легко добавлять новый функционал и модифицировать существующий. 2. Важность логирования Подробное логирование каждого действия сильно упростило отладку, особенно в продакшене: try: # Действие logger.info("Действие выполнено успешно") except Exception as e: logger.error(f"Ошибка при выполнении действия: {e}", exc_info=True) 3. Обработка ошибок Каждое действие должно быть обёрнуто в try-except, особенно при работе с внешними API и асинхронными операциями. ## Заключение ## Проект получился гораздо интереснее, чем планировался изначально. Особенно приятно, что удалось интегрировать его с уже существующими системами (как в случае с погодным сервисом) и создать действительно полезный инструмент для корпоративного использования. Главный вывод: не бойтесь усложнять проект, если усложнение оправдано. Иногда то, что кажется излишним на первый взгляд (как система контекстных шуток в погодном модуле), становится одной из самых популярных функций.

Система верификации: сложный, но необходимый модуль Одним из самых сложных для реализации стал модуль верификации пользователей. Задача была непростой: нужно было обеспечить доступ к определённым функциям бота только сотрудникам организации, при этом сохранив часть функционала доступной для всех. 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)

Анатомия корпоративного Telegram-бота: разбор по модулям Привет, разработчики! Сегодня хочу поделиться опытом создания корпоративного Telegram-бота, который из простого справочника контактов вырос в полноценного цифрового ассистента. Особое внимание уделю архитектуре и отдельным модулям, а также расскажу о самых интересных технических вызовах, с которыми пришлось столкнуться. Архитектура проекта Сразу решил строить бота по модульному принципу. Каждый компонент функциональности - отдельный менеджер, который отвечает за свою область. Базовая структура выглядит так: class TelegramBot: def __init__(self): self.config = Config(str(DATA_DIR / 'config.json')) self.poll_manager = PollManager(self.config) self.department_manager = DepartmentManager(self.config) self.contact_manager = ContactManager(self.config) self.moderator_manager = ModeratorManager(self.config) self.notification_manager = NotificationManager(self.config) self.weather_manager = WeatherManager(self.config) Система контактов и подразделений Первым делом реализовал базовый функционал - справочник контактов. Здесь интересным решением стала двухуровневая структура: подразделения и контакты внутри них. Для каждого контакта можно указать не только стандартные поля вроде должности, ФИО и телефона, но и дополнительную информацию: async def show_department_contacts_view(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: dept_id = update.callback_query.data.replace("view_dept_contacts_", "") department = next( (d for d in self.config.data.get('departments', []) if d['id'] == dept_id), None ) text = f"📍 <b>{department['name']}</b>\n" for contact in department['contacts']: text += "═══════════════════\n\n" text += f"👔 {contact['position']}\n" text += f"👤 {contact['name']}\n" text += f"📞 {contact['phone']}\n" if 'hours' in contact: text += f"🕒 {contact['hours']}\n" Погодный модуль: интеграция с существующей системой Особенно горжусь модулем погоды. До этого у меня уже был опыт создания сайта погоды (gdnmeteo.ru), который собирает данные с моей личной метеостанции на Харасавэе и из открытых API для других локаций. Было логично интегрировать эту систему и в бота. def get_city_weather(self, city: str) -> Dict[str, Any]: try: prefix = self.cities[city]['prefix'] query = f''' from(bucket: "data") |> range(start: -15m) |> filter(fn: (r) => r["_measurement"] == "measur") |> filter(fn: (r) => r["_field"] =~ /^{prefix}.*/ ) |> last() ''' tables = self.influx_client.query_api().query(query) weather_data = {} Интересной особенностью стала система контекстных шуток о погоде. В зависимости от погодных условий бот выдаёт разные комментарии, учитывая специфику работы на производственных объектах: def get_weather_joke(temp: float, wind_speed: float, humidity: float, pressure: float, city: str) -> str: if temp <= -45 and city in field_sites: jokes.extend([ "🥶 На приборах сегодня температура '404 - тепло не найдено'!", "❄️ Инструменты на морозе так сжались, что размеры уменьшились на номер!" ])

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) ## Планы на будущее Сейчас в разработке несколько серьёзных обновлений: 1. Система мониторинга СИЗ - Автоматическое отслеживание сроков годности - Уведомления о необходимости замены - Интеграция со складским учётом

# Анатомия корпоративного Telegram-бота: разбор по модулям Привет, разработчики! Сегодня хочу поделиться опытом создания корпоративного Telegram-бота, который из простого справочника контактов вырос в полноценного цифрового ассистента. Особое внимание уделю архитектуре и отдельным модулям, а также расскажу о самых интересных технических вызовах, с которыми пришлось столкнуться. ## Архитектура проекта Сразу решил строить бота по модульному принципу. Каждый компонент функциональности - отдельный менеджер, который отвечает за свою область. Базовая структура выглядит так: class TelegramBot: def __init__(self): self.config = Config(str(DATA_DIR / 'config.json')) self.poll_manager = PollManager(self.config) self.department_manager = DepartmentManager(self.config) self.contact_manager = ContactManager(self.config) self.moderator_manager = ModeratorManager(self.config) self.notification_manager = NotificationManager(self.config) self.weather_manager = WeatherManager(self.config) ## Система контактов и подразделений Первым делом реализовал базовый функционал - справочник контактов. Здесь интересным решением стала двухуровневая структура: подразделения и контакты внутри них. Для каждого контакта можно указать не только стандартные поля вроде должности, ФИО и телефона, но и дополнительную информацию: async def show_department_contacts_view(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: dept_id = update.callback_query.data.replace("view_dept_contacts_", "") department = next( (d for d in self.config.data.get('departments', []) if d['id'] == dept_id), None ) text = f"📍 <b>{department['name']}</b>\n" for contact in department['contacts']: text += "═══════════════════\n\n" text += f"👔 {contact['position']}\n" text += f"👤 {contact['name']}\n" text += f"📞 {contact['phone']}\n" if 'hours' in contact: text += f"🕒 {contact['hours']}\n" ## Погодный модуль: интеграция с существующей системой Особенно горжусь модулем погоды. До этого у меня уже был опыт создания сайта погоды (gdnmeteo.ru), который собирает данные с моей личной метеостанции на Харасавэе и из открытых API для других локаций. Было логично интегрировать эту систему и в бота. def get_city_weather(self, city: str) -> Dict[str, Any]: try: prefix = self.cities[city]['prefix'] query = f''' from(bucket: "data") |> range(start: -15m) |> filter(fn: (r) => r["_measurement"] == "measur") |> filter(fn: (r) => r["_field"] =~ /^{prefix}.*/ ) |> last() ''' tables = self.influx_client.query_api().query(query) weather_data = {} Интересной особенностью стала система контекстных шуток о погоде. В зависимости от погодных условий бот выдаёт разные комментарии, учитывая специфику работы на производственных объектах: def get_weather_joke(temp: float, wind_speed: float, humidity: float, pressure: float, city: str) -> str: if temp <= -45 and city in field_sites: jokes.extend([ "🥶 На приборах сегодня температура '404 - тепло не найдено'!", "❄️ Инструменты на морозе так сжались, что размеры уменьшились на номер!" ]) ## Система верификации: сложный, но необходимый модуль Одним из самых сложных для реализации стал модуль верификации пользователей. Задача была непростой: нужно было обеспечить доступ к определённым функциям бота только сотрудникам организации, при этом сохранив часть функционала доступной для всех.