#Alerts autorization API #https://www.donationalerts.com/oauth/authorize?redirect_uri=http://127.0.0.1/logined&response_type=code&scope=oauth-donation-index&client_id=12552 #http://127.0.0.1/logined?code=def5020000c8c717bd3993ac0bc29b903eb910fcabac108c2ac9dcaf4c3858b8ba61057915c4a2dd2599b83bebe509a2c771e2a3f4f2581d1db3a6ef5f01331aefdd8956610115e0ec0a6b1292cce0e285c5d38400ff83f79f02f402c2c73f2bde29214728229f2a2580fd36cc269bdbd75d8d2bade73eb8baa04844e0e437e5a1b856abfd4309580b6999d92d4529774289b409a72d81aada4ce4f3164febdab347372c2f9ea6a2038d9fbfef6c2942328b6187c4e6967eaa52530a6183cb1529bcee61279e43c176fcfff291d6f5d410c3648bf2f7497370ccf1bbf21298148f42e93fbcf9821ed307b97070cda02942c4ee7a552e61d3501718d8895da0631745cb3988d37fd73770d9112e56e84b431e80463efa21d6b73a59b5c176ad4adb55a85ff24e9981765ae48c2431e068f77b2982327b4e7dc9535c3045fb8ad6abab2167fd1dcf7144f5752dc517392822d509d7f342266ec24acfac1f0b6c3cb3beb9769168a5a4167cf067c27eabebf0fe79bdd6e59cf6 #https://www.donationalerts.com/oauth/token?grant_type=authorization_code&client_id=12552&client_secret=EwOBapII6syZy0k9z7AkEVhtUNDRe1VzTHW4AcmZ&redirect_uri=http://127.0.0.1/logined?code=def5020000c8c717bd3993ac0bc29b903eb910fcabac108c2ac9dcaf4c3858b8ba61057915c4a2dd2599b83bebe509a2c771e2a3f4f2581d1db3a6ef5f01331aefdd8956610115e0ec0a6b1292cce0e285c5d38400ff83f79f02f402c2c73f2bde29214728229f2a2580fd36cc269bdbd75d8d2bade73eb8baa04844e0e437e5a1b856abfd4309580b6999d92d4529774289b409a72d81aada4ce4f3164febdab347372c2f9ea6a2038d9fbfef6c2942328b6187c4e6967eaa52530a6183cb1529bcee61279e43c176fcfff291d6f5d410c3648bf2f7497370ccf1bbf21298148f42e93fbcf9821ed307b97070cda02942c4ee7a552e61d3501718d8895da0631745cb3988d37fd73770d9112e56e84b431e80463efa21d6b73a59b5c176ad4adb55a85ff24e9981765ae48c2431e068f77b2982327b4e7dc9535c3045fb8ad6abab2167fd1dcf7144f5752dc517392822d509d7f342266ec24acfac1f0b6c3cb3beb9769168a5a4167cf067c27eabebf0fe79bdd6e59cf6 import configparser import random import keyboard import json import os import sys import threading import asyncio import socket import emoji from datetime import date, datetime, timedelta import time import tzlocal from zoneinfo import * from http.server import BaseHTTPRequestHandler, HTTPServer import pytchat import telegram from donationalerts import DonationAlertsAPI, Scopes from urllib import parse sys.path.append('twitchchatirc') import twitch # Reading Configs config = configparser.ConfigParser() config.read("config.ini") alerts = DonationAlertsAPI( config['Alerts']['app_id'], config['Alerts']['api_key'], "http://127.0.0.1:8008/login", [ Scopes.USER_SHOW, ] ) # some functions to parse json date class DateTimeEncoder(json.JSONEncoder): def default(self, o): if isinstance(o, datetime): return o.isoformat() if isinstance(o, bytes): return list(o) return json.JSONEncoder.default(self, o) # chat = pytchat.create(video_id=config['Youtube']['video_id']) # twitchConn = twitch_chat_irc.TwitchChatIRC() TGenabled = config['Telegram']['enabled'] if TGenabled: bot = telegram.Bot(config['Telegram']['bot_secret']) #our nice Telegram eikichiBot all_comments = [] all_old_comments = [] is_changed = False yt_comments = [] tg_comments = [] tw_comments = [] alerts_comments = [] ytJSON = [] overallcount = 0 access_token = '' #alerts_access_token LOCAL_TIMEZONE = tzlocal.get_localzone_name() tz = ZoneInfo('Asia/Yekaterinburg') #gett DonationAlerts def get_alerts(): global access_token global alerts_comments alerts_comments = [] if not access_token: print('Waiting for Alerts auth...') return repr(access_token) al = alerts.donations.get(access_token.access_token) #print(al.items) for item in al.items: u = item.username if item.username else 'Аноним' m = item.message if item.message else ' --- ' amount = format(item.amount, '.2f') if item.amount != int(item.amount) else int(item.amount) #Если без десятичных, то удаляем их amount = f"
{amount} {item.currency}" #dt = item.created_at(ZoneInfo('UTC')) if item.is_shown: dt = datetime.fromisoformat(item.shown_at) + timedelta(hours=5) else: continue dtM = datetime.now() - timedelta(days=30) if dt > dtM: #Донаты за последний месяц... dt = dt.replace().astimezone(tz) comm = dict({'id': item.id, 'type': 'donate', 'amount': amount, 'date': dt , 'sendr': u, 'msg': emoji.emojize(m)}) alerts_comments.append(comm) #Get telegram comments... async def get_tg(): global tg_comments tg_comments = [] async with bot: try: updates = await bot.get_updates(allowed_updates=['message', 'edited_message'], timeout=None) except telegram.error.TimedOut: print('TG connection timeout 2') time.sleep(5) except: print('TG connection timeout...') else: for upd in updates: msg = upd.message if upd.edited_message: msg = upd.edited_message #tg_comments = list(filter(lambda x: (hasattr(msg,'message_id') & x['id'] != msg.message_id), tg_comments)) #если это сообщение редактирования, то удаляем все предыдущие if not hasattr(msg, 'text'): continue if not msg.text: continue sendr = msg.from_user.first_name if msg.from_user.last_name: sendr = sendr + ' ' + msg.from_user.last_name if sendr == "Group": sendr = 'Админ' #corricting TG time +timezone!!!!3 netdatetime = msg.date.replace(tzinfo=tz) #print (repr(dt)) comm = dict({'id': msg.message_id, 'type': 'tg', 'date': netdatetime, 'sendr': sendr, 'msg': msg.text}) tg_comments.append(comm) #Get youtube new comments... def get_yt(): global yt_comments global chat itms = chat.get() if not hasattr(itms, 'items'): print('YT has no items attribute! (Empty?). Reconnecting...') #Catching YT empty object exception time.sleep(5) chat = pytchat.create(video_id=config['Youtube']['video_id']) #Reconnect YT return for c in itms.items: dt = datetime.fromisoformat(c.datetime).replace(tzinfo=tz) comm = dict({'id': c.id, 'type': 'yt', 'date': dt , 'sendr': c.author.name, 'msg': emoji.emojize(c.message)}) yt_comments.append(comm) def makeJSONObject(): #we are parsing TG and YT global all_comments global yt_comments global tg_comments global tw_comments global alerts_comments global all_old_comments global overallcount global twitch_socket tw_comments = twitch_socket.all_messages # get_yt() #Get YouTube try: get_alerts() #Get Donations except: print('ALERTS TIMEOUNT BLYAT!') if TGenabled: #& int(time.time()) % 4 == 0: try: asyncio.run(get_tg()) #Get Telegram Comments #print('') except: print('TG TIMEOUNT BLYAT!') time.sleep(2) if len(yt_comments) > 15: yt_comments = yt_comments[-15:] if len(tg_comments) > 15: tg_comments = tg_comments[-15:] if len(tw_comments) > 15: tw_comments = tw_comments[-15:] dt = datetime.now(ZoneInfo('UTC')) dt = dt.replace().astimezone(tz) hello_txt = dict({'id': random.randint(100000, 999999), 'type': 'hello', 'date': dt, 'sendr': 'Eikichi-bot', 'msg': '🔥 Спасибо всем на стриме за интерес к переводу и поддержку! 🔥'}) if int(time.time()) % 600 == 0: #Выдавать сообщение каждые 45 минут all_comments.append(hello_txt) updateAllComments(hello_txt) overallcount += 1 print(datetime.now().strftime('%H:%M:%S') + ' TG And YouTube checked.... ' + str(len(all_comments)) + ' elements (' + str(overallcount) + ')') #------------------------------------------------------------ def sortdate(e): return e['date'] def updateAllComments(hello): global all_comments global tg_comments global yt_comments global tw_comments global alerts_comments for i in tg_comments: if not i in all_comments: all_comments.append(i) for i in yt_comments: if not i in all_comments: all_comments.append(i) for i in tw_comments: if not i in all_comments: all_comments.append(i) for i in alerts_comments: if not i in all_comments: all_comments.append(i) all_comments.sort(key=sortdate) #Sort objects by time all_comments = all_comments[-150:] #Пул сообщений 150 # searchAndAddHello(hello) def searchAndAddHello(hello): global all_comments for i in all_comments: if i['type'] == 'hello': return True all_comments.append(hello) return False def printAllComments(): global all_comments print('----------') print(all_comments) print('----------') class Server(BaseHTTPRequestHandler): global access_token def _set_headers(self): self.send_response(200) self.send_header('Content-type', 'application/json') self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() def do_HEAD(self): self._set_headers() def log_message(self, format, *args): #Disable server log!!! return def do_GET(self): global access_token if self.path == '/alert_auth': self.send_response(301) self.send_header('Location', alerts.authorize.login()) self.end_headers() if self.path.startswith('/login'): url = dict(parse.parse_qsl(parse.urlsplit(self.path).query)) code = url['code'] access_token = alerts.authorize.get_access_token(code) print("Access token Success!") self.send_response(200) self.send_header('Content-type','text/html') self.end_headers() self.wfile.write(b'Access token success!') return self._set_headers() self.wfile.write(json.dumps(all_comments, cls=DateTimeEncoder).encode('utf-8')) def run(server_class=HTTPServer, handler_class=Server, port=8008): server_address = ('', port) httpd = server_class(server_address, handler_class) print('Starting httpd on http://127.0.0.1:%d ...' % port) print() try: httpd.serve_forever() except KeyboardInterrupt: pass httpd.server_close() sys.exit(1) print() print('--- TelegramYoutubeTwitch Simple Chat parser --- Sergey Shemet (C) 2025 ---') print() print('Please, autorize your DonationAlerts at http://127.0.0.1:8008/alert_auth') print('...Hold Ctrl-Alt-Shift-C for exit and C to clear console...') twitch_socket = twitch.TwitchChatIRC(config['Twitch']['channel']) # init the thread as daemon for HTTPServer d = threading.Thread(target=run, name='Daemon') d.daemon = True d.start() # init twitch socket reader tw = threading.Thread(target=twitch_socket.listen, name='twitchirc') tw.daemon = True tw.start() print('Starting YT DA TG subsystems...') polTime = 2 # Secs to run check routine while True: if keyboard.is_pressed('Ctrl+Shift+Alt+c'): sys.exit(1) if keyboard.is_pressed('Ctrl+Shift+Alt+z'): printAllComments() if keyboard.is_pressed('c'): os.system("cls") thisSecond = int(time.time()) if thisSecond % polTime == 0: # every X seconds makeJSONObject() time.sleep(1)