import os import logging import asyncio import traceback from dotenv import load_dotenv from telegram import Bot from telegram.error import TelegramError, TimedOut from telegram.request import HTTPXRequest from telegram_relay import get_ptb_base_urls, get_relay_url # Загрузка переменных окружения 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 # Таймаут для медиа (на будущее) ) # Создание экземпляра бота через internal_relay base_url, base_file_url = get_ptb_base_urls() bot = Bot( token=POSTER_BOT_TOKEN, request=request, base_url=base_url, base_file_url=base_file_url, ) logger.info(f"Telegram Bot API через ретранслятор: {get_relay_url()}") # Определяем путь к изображению в том же каталоге, что и скрипт script_dir = os.path.dirname(os.path.abspath(__file__)) image_path = os.path.join(script_dir, 'testimage.jpg') # Проверяем наличие файла изображения if not os.path.exists(image_path): logger.error(f"Файл изображения не найден: {image_path}") return False # Публикация первоначального сообщения с изображением initial_text = "Это отладочное сообщение" logger.info(f"Публикация сообщения с изображением: {initial_text}") logger.info(f"Путь к изображению: {image_path}") # Повторные попытки публикации при таймауте (до 10 раз) message = None message_id = None max_retries = 10 retry_interval = 5 for attempt in range(1, max_retries + 1): try: with open(image_path, 'rb') as photo_file: message = await bot.send_photo( chat_id=CHANNEL_ID, photo=photo_file, caption=initial_text ) message_id = message.message_id logger.info(f"Сообщение успешно опубликовано. Message ID: {message_id}") break # Успешная публикация, выходим из цикла except TimedOut as e: if attempt < max_retries: logger.warning(f"Таймаут при публикации сообщения (попытка {attempt}/{max_retries}). Повтор через {retry_interval} секунд...") logger.warning(f"Детали таймаута: {type(e).__name__}: {e}") await asyncio.sleep(retry_interval) else: logger.error(f"Таймаут при публикации сообщения после {max_retries} попыток") logger.error(f"Детали ошибки: {type(e).__name__}: {e}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") return False 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 if message_id is None: logger.error("Не удалось опубликовать сообщение после всех попыток") 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}") # Повторные попытки редактирования при таймауте (до 10 раз) edit_success = False for attempt in range(1, max_retries + 1): try: await bot.edit_message_caption( chat_id=CHANNEL_ID, message_id=message_id, caption=edited_text ) logger.info(f"Сообщение успешно отредактировано") edit_success = True break # Успешное редактирование, выходим из цикла except TimedOut as e: if attempt < max_retries: logger.warning(f"Таймаут при редактировании сообщения (попытка {attempt}/{max_retries}). Повтор через {retry_interval} секунд...") logger.warning(f"Детали таймаута: {type(e).__name__}: {e}") await asyncio.sleep(retry_interval) else: logger.error(f"Таймаут при редактировании сообщения после {max_retries} попыток") logger.error(f"Детали ошибки: {type(e).__name__}: {e}") logger.error(f"Трассировка стека:\n{traceback.format_exc()}") # Сообщение уже опубликовано, поэтому считаем это частичным успехом return True 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)