vioms2tg/vioms2tg.py

#!/usr/bin/env python3

# vioms2tg: telegram bot for posting news from VIOMS API
# https://notabug.org/granthin/vioms2tg
# Aliaksandr Hrankin - 2021-2022
# version: 1.0

import asyncio
import sys
import os
from dotenv import load_dotenv
load_dotenv()
from aiogram import Bot, Dispatcher, executor, types, utils
from news_api import News
import config
from telegraph_post import make_telegraph_post
from logger import logging

# Это телеграм бот, который работает с новостными рассылками ВИОМС,
# пересылает их в телеграм канал или пользователю
# На данный момент может получать новости с одного заданного списка рассылки
# через API для мобильного приложения ВИОМС
# Основные настройки в файле config.py

vioms = News(config.VIOMS_CHANNEL_ID)
DELAY = config.CHECK_NEWS_DELAY * 60
is_admin_connected = None

try:
    bot = Bot(token=config.BOT_TOKEN)
except utils.exceptions.ValidationError:
    logging.error("Ошибка валидации токена")
    print("Произошла ошибка валидации токена. Возможно, вам нужно указать \
    правильный бот токен в файле .env.\n=== Завершение программы ===")
    sys.exit()

dp = Dispatcher(bot)

async def check_admin_connection():
    """Проверяет возможность отправки сообщений админу.
    Отправляет оповещение о запуске бота"""
    global is_admin_connected
    try:
         await bot.send_message(
                 config.ADMIN_ID,
                 "Проверяем связь с админом. Вы будете получать важные сообщения"
        )
    except:
        print('Ошибка отправки сообщения админу. Чтобы получать экстренные сообщения, \
вам необходимо добавить свой айди в переменные окружения. Например, в файл .env. \
А затем перезапустить бота.')
        logging.error("Ошибка отправки сообщения админу")
        is_admin_connected = False
    else:
        is_admin_connected = True

async def get_news():
    """Асинхронная функция, которая запускается периодически для проверки новостей
    и отправки сообщений"""
    if is_admin_connected is None:
        await check_admin_connection()
    result = vioms.check_news()

    if 'error' in result:
        logging.error(result['error'])
    elif result["news_count"]:
        news = vioms.get_news_items()[::-1]
        for n in news:
            try:
                post = make_telegraph_post(
                    n["title"],
                    n["link"],
                    n["html"] 
                )
            except Exception as err:
                return await bot.send_message(config.ADMIN_ID, f"Произошла ошибка! {err=}, {type(err)=}")
            try:
                await bot.send_message(config.CHAT_ID, f"{n['title']} 👇\n{post['url']}")
            except utils.exceptions.ChatNotFound:
                logging.error("Ошибка отправки сообщения в функции get_news")
                if is_admin_connected:
                    await bot.send_message(
                            config.ADMIN_ID,
                            "Ошибка отправки сообщения в канал. \
Возможно неправильно указан его айди в файле .env или переменных окружения\nБот отключен")
                print("Ошибка отправки сообщения в канал. Возможно неправильно указан \
его айди в файле .env или переменных окружения\n=== Завершение программы ===")
                sys.exit()
            except KeyError:
                logging.error(f"Произошла ошибка при отправке поста в телеграф: {post}")
            except Exception as err:
                await bot.send_message(config.ADMIN_ID, f"Произошла ошибка! {err=}, {type(err)=}")

    else:
        pass


def repeat(coro, loop):
    asyncio.ensure_future(coro(), loop=loop)
    loop.call_later(DELAY, repeat, coro, loop)


@dp.message_handler(commands=['start', 'help'])
async def send_welcome(message: types.Message):
    """Обрабатывает команды `/start` и `/help`"""
    if message.from_user.id == int(config.ADMIN_ID):
        await message.reply("Бот работает. Вы админ")
    else:
        await message.reply(f"Бот работает.")

# чтобы переслать новость из другой рассылки ВИОМС, нужно послать комманду
# /add_new [channel_id] [post_id]
@dp.message_handler(lambda message: message.text.startswith('/add_new '))
async def send_new_post(message: types.Message):
    post_ids = message.text.replace('/add_new ', '').split()
    try:
        channel_id, post_id = [int(item) for item in post_ids]
    except:
        return
    news_item = vioms.get_post_by_id(channel_id, post_id)    
    try:
        post = make_telegraph_post(
            news_item["title"],
            news_item["link"],
            news_item["html"]
        )
    
    except Exception as err:
        return await bot.send_message(config.ADMIN_ID, f"Произошла ошибка! {err=}, {type(err)=}")
    await bot.send_message(config.CHAT_ID, post['url'])


# дополнительный функционал для взаимодействия с другим моим ботом
# вы можете убрать или закомментировать до строки из равно
@dp.message_handler(content_types=['audio'])
async def resend_audio(message: types.Message):
    try:
        AGENT_ID = int(os.getenv('AGENT_ID') or 0)
        AUDIO_CHAT_ID = int(os.getenv('AUDIO_CHAT_ID') or 0)
    except:
        await bot.send_message(
                config.ADMIN_ID,
                "Функция пересылки аудио не работает, потому что \
неправильно определены айди источника и/или цели. Или же бот не добавлен в нужный чат")
        return 
    if message.from_user.id == AGENT_ID:
        audio_id = message.audio.file_id
        artist = message.audio.performer
        title = message.audio.title
        try:
            await bot.send_audio(
                    AUDIO_CHAT_ID, audio_id,
                    f"{artist}\n{title}\nЛекции @newjaipur"
                )
        except utils.exceptions.ChatNotFound:    
            await bot.send_message(
                    config.ADMIN_ID,
                    "Не удалось отправить аудио. Возможно, бот не добавлен в нужный чат")


# ========================================================================


if __name__ == "__main__":
    logging.warning(
        f'Начинаем следить за новостями с VIOMS ID {config.VIOMS_CHANNEL_ID} \
каждые {config.CHECK_NEWS_DELAY} минут')
    loop = asyncio.get_event_loop()
    loop.call_later(DELAY, repeat, get_news, loop)
    executor.start_polling(dp, loop=loop)