Прощай, мерзкий config_python.py
This commit is contained in:
parent
bcca39152c
commit
d8a9efffbe
1
.gitignore
vendored
1
.gitignore
vendored
@ -6,3 +6,4 @@ msgid
|
||||
tmppic.jpg
|
||||
venv
|
||||
__pycache__/
|
||||
twitch-tgbot.cfg
|
||||
|
||||
@ -1,6 +0,0 @@
|
||||
streamer = "1n7er"
|
||||
appid="addid"
|
||||
appsecret="addsecret"
|
||||
tgtoken = "token"
|
||||
tgchat = "chat"
|
||||
period = 300
|
||||
36
src/parse_config.py
Normal file
36
src/parse_config.py
Normal file
@ -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
|
||||
@ -1,12 +1,9 @@
|
||||
#!/usr/bin/env python3.10
|
||||
#!/usr/bin/env 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
|
||||
@ -17,9 +14,7 @@ from datetime import datetime
|
||||
import requests
|
||||
import shutil
|
||||
|
||||
streamer = config_python.streamer
|
||||
app_id = config_python.appid
|
||||
app_secret = config_python.appsecret
|
||||
from parse_config import Config, load_config
|
||||
|
||||
log_format = logging.Formatter('%(asctime)s %(levelname)s:%(message)s')
|
||||
log_file = 'output.log'
|
||||
@ -92,7 +87,7 @@ def stream_message_worker(data: dict):
|
||||
|
||||
# Если стрим не идет
|
||||
def no_stream_msg_worker():
|
||||
log.info(streamer + " Не стримит")
|
||||
log.info(config_python.streamer + " Не стримит")
|
||||
# Проверяем наличие файла
|
||||
if os.path.exists('msgid'):
|
||||
# Если файл существует - читаем
|
||||
@ -121,11 +116,11 @@ def check_alive():
|
||||
"""
|
||||
|
||||
# Получаем инфо о стримере, если не получается, выходим с ошибкой
|
||||
resolved_id = twitch_client.get_users(logins=[streamer])
|
||||
resolved_id = twitch_client.get_users(logins=[config_python.streamer])
|
||||
if not resolved_id['data']:
|
||||
log.error(
|
||||
colored(
|
||||
"Аккаунт " + streamer + " не найден",
|
||||
"Аккаунт " + config_python.streamer + " не найден",
|
||||
'red',
|
||||
)
|
||||
)
|
||||
@ -165,7 +160,8 @@ def get_logger(logger_name):
|
||||
if __name__ == '__main__':
|
||||
log = get_logger("main")
|
||||
log.info("Запущен")
|
||||
twitch_client = Twitch(app_id, app_secret)
|
||||
config_python = load_config()
|
||||
twitch_client = Twitch(config_python.app_id, config_python.app_secret)
|
||||
bot = telebot.TeleBot(config_python.tgtoken)
|
||||
|
||||
while True:
|
||||
|
||||
173
twitch-tgbot.py
Normal file
173
twitch-tgbot.py
Normal file
@ -0,0 +1,173 @@
|
||||
#!/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)
|
||||
Loading…
Reference in New Issue
Block a user