requirements.txt

requirements.txt

@@ -1,3 +1,2 @@
 twitchAPI==2.5.7.1
-termcolor
+telebot
-pyTelegramBotAPI==4.8.0

src/log_msg.py

@@ -0,0 +1,47 @@
+import logging
+from logging.handlers import TimedRotatingFileHandler
+import sys
+log_format = logging.Formatter('%(asctime)s %(levelname)s:%(message)s')
+log_file = 'output.log'
+
+
+class CustomFormatter(logging.Formatter):
+
+    grey = "\x1b[38;20m"
+    yellow = "\x1b[33;20m"
+    red = "\x1b[31;20m"
+    bold_red = "\x1b[31;1m"
+    reset = "\x1b[0m"
+    format = "%(asctime)s %(levelname)s - %(message)s"
+
+    FORMATS = {
+        logging.DEBUG: grey + format + reset,
+        logging.INFO: grey + format + reset,
+        logging.WARNING: yellow + format + reset,
+        logging.ERROR: red + format + reset,
+        logging.CRITICAL: bold_red + format + reset
+    }
+
+    def format(self, record):
+        log_fmt = self.FORMATS.get(record.levelno)
+        formatter = logging.Formatter(log_fmt)
+        return formatter.format(record)
+
+
+class MyLog:
+
+    def __init__(self):
+        self.logger = logging.getLogger("main")
+        self.logger.setLevel(logging.DEBUG)
+        # Console logging
+        console_handler = logging.StreamHandler(sys.stdout)
+        console_handler.setFormatter(log_format)
+
+        console = console_handler
+        console.setFormatter(CustomFormatter())
+        self.logger.addHandler(console)
+
+        file_handler = TimedRotatingFileHandler(log_file, when='midnight')
+        file_handler.setFormatter(log_format)
+        self.logger.addHandler(file_handler)
+        self.logger.propagate = False

src/parse_config.py

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

src/twitch-tgbot

@@ -1,12 +1,8 @@
 #!/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
@@ -14,10 +10,8 @@ from datetime import datetime
 import requests
 import shutil
 
-from parse_config import Config, load_config
+from parse_config import Config
-
+from log_msg import MyLog
-log_format = logging.Formatter('%(asctime)s %(levelname)s:%(message)s')
-log_file = 'output.log'
 
 
 def stream_message_worker(data: dict):
@@ -118,12 +112,7 @@ def check_alive():
     # Получаем инфо о стримере, если не получается, выходим с ошибкой
     resolved_id = twitch_client.get_users(logins=[config_python.streamer])
     if not resolved_id['data']:
-        log.error(
+        log.error("Аккаунт " + config_python.streamer + " не найден")
-            colored(
-                "Аккаунт " + config_python.streamer + " не найден",
-                'red',
-            )
-        )
     # Достаем ID стримера из инфо
     user_id = resolved_id['data'][0]['id']
     # Получаем стримы
@@ -136,34 +125,13 @@ def check_alive():
         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 = MyLog().logger
     log.info("Запущен")
-    config_python = load_config()
+    config_python = Config.load()
     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)
+        time.sleep(config_python.check_period)

twitch-tgbot.py

@@ -1,173 +0,0 @@
-#!/usr/bin/python3
-import time
-import os
-import sys
-import logging
-
-sys.path.append(os.path.abspath(os.path.curdir))
-import config_python
-
-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
-
-streamer = config_python.streamer
-app_id = config_python.appid
-app_secret = config_python.appsecret
-
-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/'+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.tgchat, 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.tgchat, msgid, reply_markup=keyboard)
-                # Редактируем текст
-                edit_msg_caption = bot.edit_message_caption(message, config_python.tgchat, msgid, reply_markup=keyboard,
-                                                            parse_mode='MarkdownV2')
-            except telebot.apihelper.ApiTelegramException as e:
-                log.error(e)
-
-
-# Если стрим не идет
-def no_stream_msg_worker():
-    log.info(streamer + " Не стримит")
-    # Проверяем наличие файла
-    if os.path.exists('msgid'):
-        # Если файл существует - читаем
-        with open('msgid', 'r') as f:
-            msgid = f.read()
-        if msgid == '':
-            # Если пустой - удаляем
-            os.remove('msgid')
-        else:
-            if config_python.delete_msg:
-                # Если не пустой, то удаляем сообщение по айдишнику
-                try:
-                    # Удаляем сообщение
-                    edit_msg = bot.delete_message(config_python.tgchat, 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=[streamer])
-    if not resolved_id['data']:
-        log.error(
-            colored(
-                "Аккаунт " + 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("Запущен")
-    twitch_client = Twitch(app_id, app_secret)
-    bot = telebot.TeleBot(config_python.tgtoken)
-
-    while True:
-        check_alive()
-        time.sleep(config_python.period)