Прощай, мерзкий config_python.py

.gitignore

@@ -6,3 +6,4 @@ msgid
 tmppic.jpg
 venv
 __pycache__/
+twitch-tgbot.cfg

config_python.py.tpl

@@ -1,6 +0,0 @@
-streamer = "1n7er"
-appid="addid"
-appsecret="addsecret"
-tgtoken = "token"
-tgchat = "chat"
-period = 300

src/parse_config.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass, asdict
+import configparser
+
+@dataclass(slots=True, frozen=True)
+class Config:
+    streamer: str
+    app_id: str
+    app_secret: str
+    target_tg_chat: str
+    tg_token: str
+    check_period: int
+    cleanup_msg: bool
+
+def load_config(config_path: str = "twitch-tgbot.cfg") -> Config:
+    """
+    Эта функция либо читает существующий конфиг, либо создает новый.
+    Возвращает объект конфига (configparser.ConfigParser())
+    """
+    config_file = configparser.ConfigParser()
+    # Читаем конфиг, если пустой - заполняем
+    if not config_file.read(config_path) or not config_file.has_section('twitch'):
+        config_file.add_section('twitch')
+    twitch = config_file['twitch']
+    config = Config(
+        streamer=twitch.get('streamer',fallback='the_viox'),
+        app_id=twitch.get('app_id',fallback=''),
+        app_secret=twitch.get('app_secret',fallback=''),
+        target_tg_chat=twitch.get('target_tg_chat',fallback=''),
+        tg_token=twitch.get('tg_token',fallback=''),
+        check_period=twitch.getint('check_period',fallback=10),
+        cleanup_msg=twitch.getboolean('cleanup_msg',fallback=True)
+    )
+    with open(config_path, 'w') as cfg_file:
+        config_file.read_dict({'twitch': asdict(config)})
+        config_file.write(cfg_file)
+    return config

src/twitch-tgbot

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+import time
+import os
+import sys
+import logging
+
+from twitchAPI.twitch import Twitch
+from logging.handlers import TimedRotatingFileHandler
+from termcolor import colored
+import telebot
+from telebot.types import InputFile
+from telebot import types
+from datetime import datetime
+import requests
+import shutil
+
+from parse_config import Config, load_config
+
+log_format = logging.Formatter('%(asctime)s %(levelname)s:%(message)s')
+log_file = 'output.log'
+
+
+def stream_message_worker(data: dict):
+    # Если это стрим, а не повтор, значит все в порядке
+    if data['type'] == 'live':
+        # Для удобства выносим в переменные
+        game_name = telebot.formatting.escape_markdown(data['game_name'])
+        stream_title = telebot.formatting.escape_markdown(data['title'])
+        stream_viewers = str(data['viewer_count'])
+
+        # Получаем скрин стрима
+        stream_pic_url = data['thumbnail_url'].format(width='1920',height='1080')
+        picture_file = requests.get(stream_pic_url, stream=True)
+        picture_file.raw.decode_content = True
+        # Увы, но я не придумал ничего лучше, чем сохранять фотку во временный файл и потом читать его
+        with open('tmppic.jpg', 'wb') as f:
+            picture_file.raw.decode_content = True
+            shutil.copyfileobj(picture_file.raw, f)
+
+        # Считаем длительность стрима
+        # '2022-12-14T17:26:47Z'
+        stream_started_at = str(data['started_at'])
+        time_started = datetime.strptime(stream_started_at, "%Y-%m-%dT%H:%M:%SZ")
+        time_now = datetime.utcnow().replace(microsecond=0)
+        stream_duration = time_now - time_started
+        # Готовим сообщение
+        message = "*"+stream_title+"*" + '\n' + '\n' + \
+                "Игра: " + game_name + '\n' + \
+                "Зрителей: " + stream_viewers + '\n' + \
+                "Длительность: " + str(stream_duration)
+
+        # Готовим кнопку
+        stream_button = types.InlineKeyboardButton('Открыть стрим', url='https://twitch.tv/'+config_python.streamer)
+        keyboard = types.InlineKeyboardMarkup()
+        keyboard.add(stream_button)
+
+        log.info(message)
+        # Если файл с айди существует, читаем его и редактируем сообщение из него
+        if os.path.exists('msgid'):
+            with open('msgid', 'r') as f:
+                msgid = f.read()
+            if msgid == '':
+                msgid = 0
+        else:
+            msgid = 0
+
+        if msgid == 0:
+            # Отправляем новое сообщение
+            send_message = bot.send_photo(config_python.target_tg_chat, InputFile('tmppic.jpg'), message, reply_markup=keyboard,
+                                          parse_mode='MarkdownV2')
+            print(send_message)
+            msgid = send_message.message_id
+            with open('msgid', 'w') as f:
+                f.write(str(msgid))
+        else:
+            try:
+                # Готовим медиа для отправки
+                media = telebot.types.InputMediaPhoto(InputFile('tmppic.jpg'))
+                # Редактируем фотку
+                edit_msg_media = bot.edit_message_media(media, config_python.target_tg_chat, msgid, reply_markup=keyboard)
+                # Редактируем текст
+                edit_msg_caption = bot.edit_message_caption(message, config_python.target_tg_chat, msgid, reply_markup=keyboard,
+                                                            parse_mode='MarkdownV2')
+            except telebot.apihelper.ApiTelegramException as e:
+                log.error(e)
+
+
+# Если стрим не идет
+def no_stream_msg_worker():
+    log.info(config_python.streamer + " Не стримит")
+    # Проверяем наличие файла
+    if os.path.exists('msgid'):
+        # Если файл существует - читаем
+        with open('msgid', 'r') as f:
+            msgid = f.read()
+        if msgid == '':
+            # Если пустой - удаляем
+            os.remove('msgid')
+        else:
+            if config_python.cleanup_msg:
+                # Если не пустой, то удаляем сообщение по айдишнику
+                try:
+                    # Удаляем сообщение
+                    edit_msg = bot.delete_message(config_python.target_tg_chat, int(msgid))
+                except telebot.apihelper.ApiTelegramException as e:
+                    log.error(e)
+            # В конце удаляем айди сообщения
+            os.remove('msgid')
+
+
+def check_alive():
+    """
+    1. Проверка на наличие стрима
+    1.1 Если нет - удалить lock файл, если он есть
+    1.2 Если есть - создать lock файл, запустить записывалку
+    """
+
+    # Получаем инфо о стримере, если не получается, выходим с ошибкой
+    resolved_id = twitch_client.get_users(logins=[config_python.streamer])
+    if not resolved_id['data']:
+        log.error(
+            colored(
+                "Аккаунт " + config_python.streamer + " не найден",
+                'red',
+            )
+        )
+    # Достаем ID стримера из инфо
+    user_id = resolved_id['data'][0]['id']
+    # Получаем стримы
+    user_stream = twitch_client.get_streams(user_id=user_id)
+    # Если стрим/повтор идет, то идем дальше
+    if user_stream['data']:
+        stream_message_worker(user_stream['data'][0])
+    else:
+        # Если стрим не идет, то запускаем функцию без параметров
+        no_stream_msg_worker()
+
+
+def get_console_handler():
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(log_format)
+    return console_handler
+
+
+def get_file_handler():
+    file_handler = TimedRotatingFileHandler(log_file, when='midnight')
+    file_handler.setFormatter(log_format)
+    return file_handler
+
+
+def get_logger(logger_name):
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(logging.DEBUG)
+    logger.addHandler(get_console_handler())
+    logger.addHandler(get_file_handler())
+    logger.propagate = False
+    return logger
+
+
+if __name__ == '__main__':
+    log = get_logger("main")
+    log.info("Запущен")
+    config_python = load_config()
+    twitch_client = Twitch(config_python.app_id, config_python.app_secret)
+    bot = telebot.TeleBot(config_python.tg_token)
+
+    while True:
+        check_alive()
+        time.sleep(config_python.check_period)