FazziCLAY
31602e96d0
All checks were successful
Build and Run VSTU TG Poster / build_and_run (push) Successful in 46s
132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
import os
|
||
import json
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
|
||
import aio_pika
|
||
from aiogram import Bot
|
||
from aiogram.client.default import DefaultBotProperties
|
||
from aiogram.enums import ParseMode
|
||
from aiohttp_socks import ProxyConnector
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
# --- КОНФИГУРАЦИЯ ---
|
||
RABBITMQ_URL = os.getenv("RABBITMQ_URL")
|
||
EXCHANGE_NAME = os.getenv("EXCHANGE_NAME", "vstu_schedule")
|
||
TG_TOKEN = os.getenv("TG_TOKEN")
|
||
TG_CHANNEL_ID = os.getenv("TG_CHANNEL_ID")
|
||
PROXY_URL = os.getenv("PROXY_URL")
|
||
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger("TGPublisher")
|
||
|
||
# Глобальный объект бота
|
||
bot = None
|
||
|
||
async def init_bot():
|
||
"""Инициализация бота с поддержкой прокси"""
|
||
global bot
|
||
connector = ProxyConnector.from_url(PROXY_URL)
|
||
# Используем сессию с прокси коннектором
|
||
bot = Bot(
|
||
token=TG_TOKEN,
|
||
connector=connector,
|
||
default=DefaultBotProperties(parse_mode=ParseMode.HTML) # Используем HTML для надежности
|
||
)
|
||
logger.info(f"[*] Бот инициализирован через прокси {PROXY_URL.split('@')[-1] if '@' in PROXY_URL else 'local'}")
|
||
|
||
async def send_to_tg(text: str) -> bool:
|
||
"""Отправляет сообщение в канал. Возвращает True при успехе."""
|
||
try:
|
||
await bot.send_message(chat_id=TG_CHANNEL_ID, text=text)
|
||
logger.info("[+] Сообщение опубликовано в Telegram.")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"[!] Ошибка Telegram API: {e}")
|
||
return False
|
||
|
||
async def process_message(message: aio_pika.IncomingMessage):
|
||
"""Обработка события от SLS с гарантией доставки (Manual Ack)"""
|
||
try:
|
||
# Мы НЕ используем context manager 'async with message.process()',
|
||
# так как нам нужен ручной ack только после успешного TG-поста
|
||
|
||
payload = json.loads(message.body.decode())
|
||
shm = payload.get("humanlike_message", "")
|
||
|
||
if not shm:
|
||
logger.warning("Получено пустое сообщение, скип.")
|
||
await message.ack()
|
||
return
|
||
|
||
# Формируем текст
|
||
facultet = payload.get("facultet", "news").upper()
|
||
# Экранируем HTML символы на всякий случай
|
||
shm_safe = shm.replace("<", "<").replace(">", ">")
|
||
|
||
tg_text = f"<b>#{facultet}</b>\n\n{shm_safe}"
|
||
|
||
ai = payload.get("ai_summary")
|
||
if ai and isinstance(ai, dict) and 'wide_review' in ai:
|
||
tg_text += f"\n\n🤖 <b>AI Резюме:</b>\n<i>{ai['wide_review']}</i>"
|
||
|
||
tg_text += "\n\n🔗 <a href='https://fazziclay.com'>fazziclay.com</a>"
|
||
|
||
# Пытаемся отправить
|
||
success = await send_to_tg(tg_text)
|
||
|
||
if success:
|
||
await message.ack() # Удаляем из RabbitMQ только после успеха в TG
|
||
else:
|
||
logger.warning("Повтор попытки через 15 секунд...")
|
||
await asyncio.sleep(15)
|
||
await message.nack(requeue=True) # Возвращаем в очередь
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error("Критическая ошибка: невалидный JSON. Удаление.")
|
||
await message.ack()
|
||
except Exception as e:
|
||
logger.error(f"Непредвиденная ошибка воркера: {e}")
|
||
await message.nack(requeue=True)
|
||
|
||
async def main():
|
||
if not TG_TOKEN or not TG_CHANNEL_ID:
|
||
logger.error("Кредиты Telegram (Token/ChannelID) не найдены!")
|
||
return
|
||
|
||
# Инициализируем бота
|
||
await init_bot()
|
||
|
||
# Коннект к RabbitMQ
|
||
connection = await aio_pika.connect_robust(RABBITMQ_URL)
|
||
channel = await connection.channel()
|
||
|
||
# Рекомендуется ограничить количество сообщений, обрабатываемых за раз
|
||
await channel.set_qos(prefetch_count=1)
|
||
|
||
exchange = await channel.declare_exchange(EXCHANGE_NAME, aio_pika.ExchangeType.TOPIC, durable=True)
|
||
|
||
# Создаем независимую очередь для Telegram
|
||
queue = await channel.declare_queue("tg_publisher_queue", durable=True)
|
||
await queue.bind(exchange, routing_key="schedule_logging_service.event.excel.#")
|
||
|
||
logger.info(f"[*] TG Publisher готов к работе. Канал: {TG_CHANNEL_ID}")
|
||
|
||
await queue.consume(process_message)
|
||
|
||
try:
|
||
await asyncio.Future() # Вечный цикл
|
||
finally:
|
||
await connection.close()
|
||
# Не забываем закрыть сессию бота
|
||
if bot:
|
||
await bot.session.close()
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
asyncio.run(main())
|
||
except KeyboardInterrupt:
|
||
pass |