import os import logging import asyncio import traceback from dotenv import load_dotenv from telegram import Bot from telegram.error import TelegramError from telegram.request import HTTPXRequest # Загрузка переменных окружения load_dotenv() # Получение переменных окружения POSTER_BOT_TOKEN = os.getenv('POSTER_BOT_TOKEN') LOG_FILE = os.getenv('LOG_FILE', 'test_tg_poster.log') CHANNEL_ID = os.getenv('CHANNEL_ID') # Настройка логирования logger = logging.getLogger('test_tg_poster') logger.setLevel(logging.INFO) # Форматтер для логов formatter = logging.Formatter('[%(asctime)s] [%(name)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S') # Обработчик для консоли console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(console_handler) # Обработчик для файла с повторными попытками при ошибках записи class RetryFileHandler(logging.FileHandler): def emit(self, record): import time for _ in range(5): try: super().emit(record) return except (IOError, PermissionError): time.sleep(0.5) print(f"Failed to write to log file after 5 attempts: {record.msg}") file_handler = RetryFileHandler(LOG_FILE, encoding='utf-8') file_handler.setFormatter(formatter) logger.addHandler(file_handler) async def make_test_post_async(): """ Асинхронная функция для публикации тестового сообщения в Telegram канал. Публикует сообщение, затем редактирует его, добавляя Message ID. """ bot = None try: # Проверка наличия необходимых переменных окружения if not POSTER_BOT_TOKEN: logger.error("Переменная окружения POSTER_BOT_TOKEN не установлена") return False if not CHANNEL_ID: logger.error("Переменная окружения CHANNEL_ID не установлена") return False logger.info("Начало публикации тестового сообщения") logger.info(f"Используется канал: {CHANNEL_ID}") # Настройка таймаутов для HTTPXRequest # Увеличиваем таймауты для избежания ошибок подключения # Используем параметры, аналогичные тем, что используются в tg_publish.py request = HTTPXRequest( read_timeout=60.0, # Таймаут на чтение (увеличен) write_timeout=60.0, # Таймаут на запись connect_timeout=30.0, # Таймаут на подключение (увеличен) pool_timeout=10.0, # Таймаут на получение соединения из пула media_write_timeout=120.0 # Таймаут для медиа (на будущее) ) # Создание экземпляра бота с кастомным Request bot = Bot(token=POSTER_BOT_TOKEN, request=request) # Публикация первоначального сообщения initial_text = "Это отладочное сообщение" logger.info(f"Публикация сообщения: {initial_text}") try: message = await bot.send_message( chat_id=CHANNEL_ID, text=initial_text ) message_id = message.message_id logger.info(f"Сообщение успешно опубликовано. Message ID: {message_id}") except TelegramError as e: logger.error(f"Ошибка при публикации сообщения в Telegram: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") return False except Exception as e: logger.error(f"Неожиданная ошибка при публикации сообщения: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") return False # Небольшая задержка перед редактированием await asyncio.sleep(1) # Редактирование сообщения: добавляем две пустые строки и Message ID edited_text = f"{initial_text}\n\n\nИдентификатор сообщения - {message_id}" logger.info(f"Редактирование сообщения, добавление Message ID: {message_id}") try: await bot.edit_message_text( chat_id=CHANNEL_ID, message_id=message_id, text=edited_text ) logger.info(f"Сообщение успешно отредактировано") except TelegramError as e: logger.error(f"Ошибка при редактировании сообщения в Telegram: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") # Сообщение уже опубликовано, поэтому считаем это частичным успехом return True except Exception as e: logger.error(f"Неожиданная ошибка при редактировании сообщения: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") # Сообщение уже опубликовано, поэтому считаем это частичным успехом return True logger.info("Тестовое сообщение успешно опубликовано и отредактировано") return True except Exception as e: logger.error(f"Критическая ошибка в make_test_post_async: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") return False def make_test_post(): """ Синхронная функция-обертка для публикации тестового сообщения. Предназначена для вызова из Flask-приложения, работающего под gunicorn. Returns: bool: True если публикация успешна, False в противном случае """ try: # Проверяем, есть ли уже запущенный event loop try: loop = asyncio.get_running_loop() # Если event loop уже запущен (например, в async контексте), # это означает, что мы в async функции, и нужно использовать другой подход # В контексте gunicorn это маловероятно, но на всякий случай logger.warning("Обнаружен запущенный event loop. Функция должна вызываться из синхронного контекста.") # Пытаемся запустить в новом потоке import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as executor: future = executor.submit(asyncio.run, make_test_post_async()) return future.result() except RuntimeError: # Event loop не запущен, создаем новый # Это нормальная ситуация для синхронного контекста (Flask/gunicorn) logger.info("Event loop не запущен, создаем новый через asyncio.run()") return asyncio.run(make_test_post_async()) except Exception as e: logger.error(f"Ошибка в make_test_post при запуске асинхронной функции: {type(e).__name__}: {e}") logger.error(f"Детали ошибки: {repr(e)}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") return False if __name__ == "__main__": """ Выполнение функции при запуске модуля из командной строки """ logger.info("Запуск модуля test_tg_poster.py из командной строки") result = make_test_post() if result: logger.info("Тестовое сообщение успешно опубликовано") exit(0) else: logger.error("Не удалось опубликовать тестовое сообщение") exit(1)