Zilant2025/test_tg_poster.py

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)