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

# Загрузка переменных окружения
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}")
        
        # Создание экземпляра бота
        bot = Bot(token=POSTER_BOT_TOKEN)
        
        # Публикация первоначального сообщения
        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)