From 76e6659c00f5865fcdcee3733bd4ad4700afac7d Mon Sep 17 00:00:00 2001 From: root Date: Fri, 7 Aug 2026 05:08:46 +0500 Subject: [PATCH] refactor: add ConfigError and Lolz error handling --- config.py | 34 ++++++-- main.py | 11 ++- modules/lolz.py | 169 +++++++++++++++++++++++++----------- modules/notification_bot.py | 53 +++++++---- modules/stars_bot.py | 24 +++-- 5 files changed, 208 insertions(+), 83 deletions(-) diff --git a/config.py b/config.py index 470ce1d..e2bcd6d 100644 --- a/config.py +++ b/config.py @@ -7,6 +7,11 @@ from dotenv import load_dotenv logger = logging.getLogger(__name__) +class ConfigError(Exception): + """Ошибка валидации конфигурации""" + pass + + class Config: def __init__(self): self._config_data = self._load() @@ -53,18 +58,25 @@ class Config: # Notification bot (aiogram) bot_token = os.getenv(key="BOT_TOKEN", default=None) - admin_id = ( - os.getenv(key="ADMIN_ID", default=None) - if not os.getenv(key="ADMIN_ID", default=None) - else int(os.getenv(key="ADMIN_ID", default=None)) - ) + admin_id_str = os.getenv(key="ADMIN_ID", default=None) + admin_id = int(admin_id_str) if admin_id_str else None notify_admin = ( os.getenv(key="NOTIFY_ADMIN", default="true") ).lower() == "true" - # Lolz + # Lolz - required fields lolz_token = os.getenv(key="LOLZ_TOKEN", default=None) - lolz_thread_id = int(os.getenv(key="LOLZ_THREAD_ID", default=None)) + if not lolz_token: + raise ConfigError("LOLZ_TOKEN обязателен, но не установлен") + + lolz_thread_id_str = os.getenv(key="LOLZ_THREAD_ID", default=None) + if not lolz_thread_id_str: + raise ConfigError("LOLZ_THREAD_ID обязателен, но не установлен") + + try: + lolz_thread_id = int(lolz_thread_id_str) + except (ValueError, TypeError) as e: + raise ConfigError(f"LOLZ_THREAD_ID должен быть целым числом: {lolz_thread_id_str}") from e # Stars stars_count = int(os.getenv(key="STARS_COUNT", default=1)) @@ -118,8 +130,14 @@ class Config: "bump_check_interval": bump_check_interval, } - except: + except ConfigError: raise + except (ValueError, TypeError) as e: + logger.error(f"Invalid configuration value: {e}") + raise ConfigError(f"Invalid configuration value: {e}") from e + except Exception as e: + logger.error(f"Unexpected error loading configuration: {e}") + raise ConfigError(f"Failed to load configuration: {e}") from e def __getattr__(self, name: str) -> Any: return self._config_data.get( diff --git a/main.py b/main.py index 6d19def..7cf398b 100644 --- a/main.py +++ b/main.py @@ -2,8 +2,9 @@ import os import asyncio import logging -from config import Config +from config import Config, ConfigError from modules.stars_bot import TelegramStarsBot +from modules.lolz import LolzAPIError logger = logging.getLogger(__name__) @@ -26,6 +27,14 @@ async def main(): await stars_bot.start() + except ConfigError as e: + logger.critical(f"Ошибка конфигурации: {e}") + raise + + except LolzAPIError as e: + logger.critical(f"Ошибка API Lolz: {e}") + raise + except (ValueError, TypeError) as e: logger.critical(f"Ошибка в конфигурации или при запуске: {e}") raise diff --git a/modules/lolz.py b/modules/lolz.py index 447a515..6b54477 100644 --- a/modules/lolz.py +++ b/modules/lolz.py @@ -7,8 +7,26 @@ from typing import Dict, Any, Optional, Union, List logger = logging.getLogger(__name__) +class LolzAPIError(Exception): + """Base exception for Lolz API errors""" + pass + + +class LolzNetworkError(LolzAPIError): + """Network-related errors""" + pass + + +class LolzResponseError(LolzAPIError): + """API returned invalid or empty response""" + pass + + class Lolz: def __init__(self, token: str): + if not token: + raise ValueError("LOLZ_TOKEN обязателен") + try: self.client = Forum(token=token, timeout=15) @@ -16,50 +34,78 @@ class Lolz: # ! ибо логи занимают очень много места в экране при этом логгере # self.client.settings.logger.enable() - except: - raise + except Exception as e: + logger.error(f"Не удалось инициализировать Lolz клиент: {e}") + raise LolzAPIError(f"Не удалось инициализировать Lolz клиент: {e}") from e async def get_thread(self, thread_id: Union[str, int]) -> Dict[str, Any]: logger.debug(f"Запрашиваю данные темы {thread_id}...") - response = await self.client.threads.get(thread_id=thread_id) - thread = (response.json()).get("thread", {}) + try: + response = await self.client.threads.get(thread_id=thread_id) + thread = (response.json()).get("thread", {}) - if thread == {}: - logger.warning(f"Тема {thread_id} не найдена или вернула пустой ответ.") + if not thread: + raise LolzResponseError(f"Тема {thread_id} не найдена или вернула пустой ответ") - return thread + return thread + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при получении темы {thread_id}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при получении темы {thread_id}") from e + except Exception as e: + logger.error(f"Ошибка при получении темы {thread_id}: {e}") + raise LolzAPIError(f"Не удалось получить тему {thread_id}") from e async def can_bump( self, thread_id: Optional[int] = None, thread: Dict[str, Any] = None ) -> bool: if not thread: if not thread_id: - thread = {} - else: - thread = await self.get_thread(thread_id=thread_id) + raise ValueError("Необходимо указать thread_id или thread") + thread = await self.get_thread(thread_id=thread_id) - return thread["permissions"]["bump"]["can"] + try: + return thread.get("permissions", {}).get("bump", {}).get("can", False) + except (KeyError, AttributeError, TypeError) as e: + logger.error(f"Неверная структура темы для проверки поднятия: {e}") + return False async def bump_thread(self, thread_id: Union[str, int]) -> bool: logger.debug(f"Поднимаю тему {thread_id}...") - response = await self.client.threads.bump(thread_id=thread_id) + try: + response = await self.client.threads.bump(thread_id=thread_id) - if response.status_code == 200: - logger.info(f"Тема {thread_id} успешно поднята.") - return True - else: - logger.error(f"Не удалось поднять тему {thread_id}.") - return False + if response.status_code == 200: + logger.info(f"Тема {thread_id} успешно поднята.") + return True + else: + logger.error(f"Не удалось поднять тему {thread_id}. Статус: {response.status_code}") + return False + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при поднятии темы {thread_id}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при поднятии темы {thread_id}") from e + except Exception as e: + logger.error(f"Ошибка при поднятии темы {thread_id}: {e}") + raise LolzAPIError(f"Не удалось поднять тему {thread_id}") from e async def get_post(self, post_id: Union[str, int]) -> Dict[str, Any]: logger.debug(f"Запрашиваю данные поста {post_id}...") - response = await self.client.posts.get(post_id=post_id) - post = (response.json()).get("post", {}) + try: + response = await self.client.posts.get(post_id=post_id) + post = (response.json()).get("post", {}) - if post == {}: - logger.warning(f"Пост {post_id} не найден или вернул пустой ответ.") + if not post: + raise LolzResponseError(f"Пост {post_id} не найден или вернул пустой ответ") - return post + return post + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при получении поста {post_id}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при получении поста {post_id}") from e + except Exception as e: + logger.error(f"Ошибка при получении поста {post_id}: {e}") + raise LolzAPIError(f"Не удалось получить пост {post_id}") from e async def get_thread_posts( self, thread_id: Union[str, int], start_page: int = 1 @@ -74,20 +120,28 @@ class Lolz: last_page = start_page while True: - response = await self.client.posts.list(thread_id=thread_id, page=page) - posts = (response.json()).get("posts", []) + try: + response = await self.client.posts.list(thread_id=thread_id, page=page) + posts = (response.json()).get("posts", []) - if len(posts) == 0: - break + if len(posts) == 0: + break - all_posts.extend(posts) - last_page = page - logger.debug( - f"Получено {len(posts)} постов из темы {thread_id} на странице {page}." - ) + all_posts.extend(posts) + last_page = page + logger.debug( + f"Получено {len(posts)} постов из темы {thread_id} на странице {page}." + ) - page += 1 - await asyncio.sleep(1) + page += 1 + await asyncio.sleep(1) + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при получении постов из темы {thread_id}, страница {page}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при получении постов из темы {thread_id}") from e + except Exception as e: + logger.error(f"Ошибка при получении постов из темы {thread_id}, страница {page}: {e}") + raise LolzAPIError(f"Не удалось получить посты из темы {thread_id}") from e if all_posts: logger.info(f"Всего получено {len(all_posts)} постов из темы {thread_id}.") @@ -104,31 +158,46 @@ class Lolz: async def get_post_comments(self, post_id: int) -> List[Dict[str, Any]]: logger.debug(f"Запрашиваю комментарии к посту {post_id}...") - response = await self.client.posts.comments.list(post_id=post_id) - comments = (response.json()).get("comments", []) - logger.debug(f"Найдено {len(comments)} комментариев к посту {post_id}.") - return comments + try: + response = await self.client.posts.comments.list(post_id=post_id) + comments = (response.json()).get("comments", []) + logger.debug(f"Найдено {len(comments)} комментариев к посту {post_id}.") + return comments + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при получении комментариев к посту {post_id}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при получении комментариев к посту {post_id}") from e + except Exception as e: + logger.error(f"Ошибка при получении комментариев к посту {post_id}: {e}") + raise LolzAPIError(f"Не удалось получить комментарии к посту {post_id}") from e async def has_comments( self, post_id: Optional[int] = None, post: Dict[str, Any] = None ) -> bool: if not post: if not post_id: - post = {} - else: - post = await self.get_post(post_id=post_id) + raise ValueError("Необходимо указать post_id или post") + post = await self.get_post(post_id=post_id) return post.get("post_comment_count", 0) > 0 async def create_comment(self, post_id: int, comment_body: str) -> bool: logger.info(f"Публикую комментарий к посту {post_id}...") - response = await self.client.posts.comments.create( - post_id=post_id, comment_body=comment_body - ) + try: + response = await self.client.posts.comments.create( + post_id=post_id, comment_body=comment_body + ) - if response.status_code == 200: - logger.info(f"Комментарий к посту {post_id} успешно опубликован.") - return True - else: - logger.error(f"Не удалось опубликовать комментарий к посту {post_id}.") - return False + if response.status_code == 200: + logger.info(f"Комментарий к посту {post_id} успешно опубликован.") + return True + else: + logger.error(f"Не удалось опубликовать комментарий к посту {post_id}. Статус: {response.status_code}") + return False + + except asyncio.TimeoutError as e: + logger.error(f"Превышено время ожидания при создании комментария к посту {post_id}: {e}") + raise LolzNetworkError(f"Превышено время ожидания при создании комментария к посту {post_id}") from e + except Exception as e: + logger.error(f"Ошибка при создании комментария к посту {post_id}: {e}") + raise LolzAPIError(f"Не удалось создать комментарий к посту {post_id}") from e diff --git a/modules/notification_bot.py b/modules/notification_bot.py index 8c7e3cb..b224645 100644 --- a/modules/notification_bot.py +++ b/modules/notification_bot.py @@ -4,9 +4,10 @@ from aiogram import Bot from aiogram.enums import ParseMode from aiogram.client.default import DefaultBotProperties from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup +from aiogram.exceptions import TelegramAPIError, TelegramBadRequest, TelegramNetworkError from config import Config -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional logger = logging.getLogger(__name__) @@ -28,11 +29,11 @@ class NotificationBot: "Бот для уведомлений инициализирован (admin_id=%s)", config.admin_id ) - except: - logger.exception("Не удалось инициализировать Бот для уведомлений") + except (TelegramAPIError, ValueError, TypeError) as e: + logger.error(f"Не удалось инициализировать Бот для уведомлений: {e}") raise - async def _notify(self, message: str, reply_to: int = None, link: str = None): + async def _notify(self, message: str, reply_to: int = None, link: str = None) -> Optional[int]: try: logger.debug("Отправка уведомления (reply_to=%s, link=%s)", reply_to, link) keyboard = None @@ -58,10 +59,20 @@ class NotificationBot: ) return sent_message.message_id + except TelegramBadRequest as e: + logger.error("Bad request sending notification: %s", e) + return None + except TelegramNetworkError as e: + logger.error("Network error sending notification: %s", e) + return None + except TelegramAPIError as e: + logger.error("Telegram API error sending notification: %s", e) + return None except Exception as e: - logger.error("Ошибка отправки уведомления: %s", e) + logger.error("Unexpected error sending notification: %s", e) + return None - async def new_post(self, post: Dict[str, Any], links: List[str]): + async def new_post(self, post: Dict[str, Any], links: List[str]) -> Optional[int]: try: thread = post.get("thread") @@ -99,11 +110,14 @@ class NotificationBot: ) return post_message_id - except: - logger.exception("Ошибка в new_post") - raise + except (KeyError, AttributeError, TypeError) as e: + logger.error(f"Неверная структура поста в new_post: {e}") + return None + except Exception as e: + logger.error(f"Неожиданная ошибка в new_post: {e}") + return None - async def success(self, link: str, post_message_id: int): + async def success(self, link: str, post_message_id: int) -> Optional[int]: try: logger.info( "Звезда успешно отправлена: link=%s, stars=%s", @@ -124,13 +138,13 @@ class NotificationBot: ) return success_message_id - except: - logger.exception("Ошибка в success") - raise + except Exception as e: + logger.error(f"Неожиданная ошибка в success: {e}") + return None - async def failure(self, post: Dict[str, Any], reason: str, post_message_id: int): + async def failure(self, post: Dict[str, Any], reason: str, post_message_id: int) -> Optional[int]: try: - permalink = post.get("links").get("permalink") + permalink = post.get("links", {}).get("permalink") logger.warning( "Не удалось отправить звезду: reason=%r, link=%s", reason, permalink ) @@ -147,6 +161,9 @@ class NotificationBot: ) return failure_message_id - except: - logger.exception("Ошибка в failure") - raise + except (KeyError, AttributeError, TypeError) as e: + logger.error(f"Неверная структура поста в failure: {e}") + return None + except Exception as e: + logger.error(f"Неожиданная ошибка в failure: {e}") + return None diff --git a/modules/stars_bot.py b/modules/stars_bot.py index 033497b..242c67a 100644 --- a/modules/stars_bot.py +++ b/modules/stars_bot.py @@ -325,7 +325,7 @@ class TelegramStarsBot: except Exception as e: logger.exception(f"Критическая ошибка в цикле поднятия темы: {e}") - await asyncio.sleep(self.config.check_interval) + await asyncio.sleep(self.config.bump_check_interval) async def start(self): is_first_login = not os.path.exists(f"{self.SESSION_NAME}.session") @@ -341,9 +341,15 @@ class TelegramStarsBot: api_hash=self.config.api_hash, ) - await self.client.start() - await self.parse_existing_posts() - await self.client.stop() + try: + await self.client.start() + await self.parse_existing_posts() + except Exception as e: + logger.error(f"Ошибка при парсинге существующих постов: {e}") + raise + finally: + if self.client and self.client.is_connected: + await self.client.stop() logger.info( "Парсинг завершен. Все существующие посты добавлены в обработанные." @@ -372,11 +378,17 @@ class TelegramStarsBot: logger.info("Бот в работе. Для остановки нажмите Ctrl+C.") # _main_loop - Основной цикл для остлеживания темы - # _bump_look - Цикл для авто-поднятия темы + # _bump_loop - Цикл для авто-поднятия темы await asyncio.gather(self._main_loop(), self._bump_loop()) + except Exception as e: + logger.error(f"Ошибка в главном процессе бота: {e}") + raise finally: if self.client and self.client.is_connected: - await self.client.stop() + try: + await self.client.stop() + except Exception as e: + logger.error(f"Ошибка при остановке клиента: {e}") logger.info("Бот остановлен.")