Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- '''
- Бот для автопостинга картинок и текста в гартик
- v0.1
- FAQ:
- Q: как запустить бота?
- A: 1) устанавливаете python версии не ниже 3.6
- 2) устанавливаете библиотеки через консоль командой: python -m pip install websocket_client requests pillow
- 3) запускаете этот файл (если не работает через даблклик, то через консоль командой: python autogartic.pyw)
- 4) вставляете в бота ссылку на игру, никнейм/куки, путь к файлу с фразами и к папке с картинками, нажимаете "подключиться". Бот сам наченет постить при старте игры
- Q: что за файл с фразами?
- A: это строки, которые будут использоваться как предложения в начале игры и при описании рисунков. Они вставляются в игру в случайном порядке, как и картинки из папки
- Q: куда сохраняются настройки
- A: в файл autogartic_settings.json в той же папке, где находится бот
- Q: где взять куки для логина через твич?
- A: 1) в браузере открываете сетевой лог (в firefox F12, в хроме и прочих ctrl-shift-J)
- 2) создаете свое лобби в гартике с авторизацией через твич
- 3) в сетевых запросах ищете запрос на домен id.twitch.tv
- 4) в запросе ищете куки с названием persistent, должно быть что-то типа 701234671::safnk4jtn3nbgieru656hosd7, копируете эту строчку в бота
- '''
- from PIL import Image
- import websocket
- import requests
- import tkinter as tk
- from tkinter import RIGHT, ttk
- import tkinter.scrolledtext as scrolledtext
- from tkinter import filedialog
- import random
- import time
- import threading
- import json
- import re
- import datetime
- import os
- # global vars
- logBox = None
- imagesList = None
- phrasesList = None
- playerUuid = None
- def log(s):
- t = datetime.datetime.now().strftime("%H:%M:%S")
- logBox.insert(tk.END, f'[{t}] {s}\n')
- logBox.see(tk.END)
- def getRandomUuid():
- s = []
- for i in range(36):
- s.append(random.choice('af'))
- s[8] = '-'
- s[13] = '-'
- s[18] = '-'
- s[23] = '-'
- return ''.join(s)
- class ImagesList:
- def __init__(self, dir) -> None:
- if not os.path.isdir(dir):
- raise Exception('указана неверная папка с картинками')
- self.filenames = [os.path.join(dir, fname) for fname in os.listdir(dir) \
- if (fname.lower().endswith('.png') or fname.lower().endswith('.jpg'))]
- if len(self.filenames) < 1:
- raise Exception('в указанной папке нет картинок')
- random.shuffle(self.filenames)
- self.imageNum = 0
- def getNext(self):
- fname = self.filenames[self.imageNum]
- self.imageNum += 1
- if self.imageNum == len(self.filenames):
- self.imageNum = 0
- return fname
- class PhrasesList:
- def __init__(self, path) -> None:
- if not os.path.isfile(path):
- raise Exception('указан неверный файл с фразами')
- with open(path, encoding='utf8') as f:
- self.phrases = f.read().split('\n')
- random.shuffle(self.phrases)
- self.phraseNum = 0
- def getNext(self):
- phrase = self.phrases[self.phraseNum]
- self.phraseNum += 1
- if self.phraseNum == len(self.phrases):
- self.phraseNum = 0
- return phrase
- def rgb2hex(r, g, b):
- return '#{:02x}{:02x}{:02x}'.format(r, g, b)
- def imgToMoves(turnNum, shakalValue, filename):
- colorsDict = {}
- img = Image.open(filename)
- img = img.resize((758, 424), Image.ANTIALIAS)
- for x in range(img.width):
- for y in range(img.height):
- if shakalValue > 1 and (x % shakalValue != 0 or y % shakalValue != 0):
- continue
- pixel = img.getpixel((x,y))
- colorHex = rgb2hex(pixel[0], pixel[1], pixel[2])
- if not colorsDict.get(colorHex):
- colorsDict[colorHex] = f'[8,-1,["{colorHex}",1],{x},{y},{shakalValue},{shakalValue}'
- else:
- colorsDict[colorHex] += f',{x},{y},{shakalValue},{shakalValue}'
- img.close()
- movesArray = [colorsDict[i] for i in colorsDict]
- random.shuffle(movesArray)
- moves = ['42[2,7,{"t":'+str(turnNum)+',"d":1,"v":'+move+']}]' for move in movesArray]
- return moves
- class WebsocketClient(threading.Thread):
- def __init__(self, svServerHost, sid, userId) -> None:
- threading.Thread.__init__(self)
- self.isCloseRequest = False
- self.svServerHost = svServerHost
- self.sid = sid
- self.userId = userId
- def on_message(self, ws, message):
- if type(message) != str or message == '3' or message == '3probe':
- return
- if message.startswith('42[2,5'):
- log('Игра началась')
- elif message.startswith('42[2,11'):
- turnNum = json.loads(message[2:])[2]['turnNum']
- screen = json.loads(message[2:])[2]['screen']
- if screen == 3 or screen == 4:
- phrase = phrasesList.getNext()
- log(f'Ход {turnNum}, отправляем фразу "{phrase}"')
- ws.send('42[2,6,{"t":'+str(turnNum)+',"v":"'+phrase+'"}]')
- elif screen == 5:
- try:
- filename = imagesList.getNext()
- log(f'Ход {turnNum}, отправляем картинку "{os.path.basename(filename)}"')
- for shakalValue in [4,2]:
- movesArray = imgToMoves(turnNum, shakalValue, filename)
- for move in movesArray:
- ws.send(move)
- except Exception as e:
- log(f'Ошибка при выгрузке картинки: {e}')
- ws.send('42[2,15,true]') # сообщаем о готовности
- elif message.startswith('42[2,15'):
- pass # статус готовности других игроков
- elif message.startswith('42[2,24'):
- log('Игра закончена, игроки смотрят альбомы')
- elif message.startswith('42[2,23') or message.startswith('42[2,12') or message.startswith('42[2,9'):
- pass # просмотр ходов в альбоме
- elif message.startswith('42[2,20]'):
- log('Альбомы просмотрены, выходим в лобби')
- elif message.startswith('42[2,14'):
- id = json.loads(message[2:])[2]
- if id != self.userId:
- log(f'Игрок с id {id} был забанен')
- else:
- log(f'Бот был забанен, отключаемся и создаем новый ID')
- global playerUuid
- playerUuid = getRandomUuid()
- self.isCloseRequest = True
- else:
- # log(f'Неизвестный тип сообщения: {message}')
- pass
- def on_error(self, ws, error):
- log(f'WebSocket error: {error}')
- def on_close(self, ws, close_status_code, close_msg):
- self.isCloseRequest = True
- log(f'WebSocket closed: status: {close_status_code}, msg: {close_msg}')
- def on_open(self, ws):
- log('WebSocket opened, starting new thread')
- def pingFunc():
- ws.send("2probe")
- time.sleep(3)
- ws.send("5")
- self.onConnected()
- time.sleep(3)
- while True:
- time.sleep(3)
- if self.isCloseRequest:
- break
- ws.send("2")
- ws.close()
- self.pingThread = threading.Thread(target=pingFunc)
- self.pingThread.start()
- def setHandlers(self, onConnected, onDisconnected):
- self.onConnected = onConnected
- self.onDisconnected = onDisconnected
- def stop(self):
- self.isCloseRequest = True
- def run(self):
- ws = websocket.WebSocketApp(f"wss://{self.svServerHost}/socket.io/?EIO=3&transport=websocket&sid={self.sid}",
- on_open=self.on_open,
- on_message=self.on_message,
- on_error=self.on_error,
- on_close=self.on_close,
- cookie=f'io={self.sid}')
- ws.run_forever()
- self.pingThread.join()
- log('WebSocket threads terminated')
- self.onDisconnected()
- # method: 'anon' or 'twitch'
- def authentication(inviteLink, method, twitchCookie = None, nickname = None):
- global playerUuid
- if method == 'anon' and not nickname:
- raise Exception('пустой ник')
- elif method == 'twitch' and not twitchCookie:
- raise Exception('отсутствуют cookie')
- elif method != 'twitch' and method != 'anon':
- raise Exception('неизвестный метод аутентификации')
- avatarNum = random.choice([31,32,33,40,16,39])
- inviteCode = re.search('.*/\?c=(.*)', inviteLink)
- if not inviteCode:
- raise Exception('не удалось найти код приглашения в ссылке')
- inviteCode = inviteCode.groups()[0]
- garticSession = requests.Session()
- if method == 'twitch':
- r = garticSession.get(f'https://garticphone.com/api/auth/providers')
- # 1) get csrf
- r = garticSession.get(f'https://garticphone.com/api/auth/csrf')
- csrfToken = json.loads(r.content.decode('utf8'))['csrfToken']
- # 2) get twitch auth url
- r = garticSession.post('https://garticphone.com/api/auth/signin/twitch',
- data={
- 'callbackUrl': f'https://garticphone.com/ru/redirecting?avatar={avatarNum}&c={inviteCode}',
- 'csrfToken': csrfToken,
- 'json': 'true'
- }
- )
- twitchAuthUrl = json.loads(r.content.decode('utf8'))['url']
- # 3) twitch auth
- r = requests.get(twitchAuthUrl, cookies={
- 'persistent':twitchCookie,
- })
- content = r.content.decode('utf8')
- nextUrl = re.search('.*URL=\'(.*)\'" />.*', content).groups()[0]
- nextUrl = nextUrl.replace('&', '&')
- nextUrl = nextUrl.replace('%3A', ':')
- # 4) get session token from gartic
- garticSession.get(nextUrl)
- # get server
- r = garticSession.get(f'https://garticphone.com/api/server?code={inviteCode}')
- svServerHost = r.content.decode('utf8')
- # get sid
- r = garticSession.get(f'{svServerHost}/socket.io/?EIO=3&transport=polling&t=1')
- sid = re.search('.*sid":"(.*)","upgrades.*', r.content.decode('utf8')).groups()[0]
- # register in lobby
- if method == 'twitch':
- r = garticSession.get('https://garticphone.com/api/auth/session')
- twitchJwt = json.loads(r.content.decode('utf8'))['code']
- data = f'42[1,"{playerUuid}",null,"{avatarNum}","ru",false,"{inviteCode}","{twitchJwt}",null]'
- r = garticSession.post(f'{svServerHost}/socket.io/?EIO=3&transport=polling&t=2&sid={sid}',
- data=f'{len(data)}:{data}'.encode('utf-8'),
- cookies={'io':sid})
- else:
- data = f'42[1,"{playerUuid}","{nickname}","{avatarNum}","ru",false,"{inviteCode}",null,null]'
- r = requests.post(f'{svServerHost}/socket.io/?EIO=3&transport=polling&t=2&sid={sid}',
- data=f'{len(data)}:{data}'.encode('utf-8'))
- if r.status_code != 200:
- raise Exception(f'anon register status_code={r.status_code}')
- # get lobby object
- r = requests.get(f'{svServerHost}/socket.io/?EIO=3&transport=polling&t=3&sid={sid}',
- cookies={'io':sid})
- if r.status_code != 200:
- raise Exception(f'get lobby obj status_code={r.status_code}')
- s = r.content.decode('utf8')
- lobbyObj = json.loads(s[s.find('['):])
- garticError = lobbyObj[1].get('error')
- if garticError == 1:
- raise Exception('некорректный ник')
- elif garticError == 2:
- playerUuid = getRandomUuid()
- if method == 'anon':
- raise Exception('вы забанены в этом лобби. Новый ID создан, переподключитесь')
- else:
- raise Exception('вы забанены в этом лобби. Используйте другой аккаунт')
- elif garticError == 3:
- raise Exception('лобби не существует')
- elif garticError == 4:
- raise Exception('в лобби уже максимальное число игроков')
- elif garticError == 6:
- raise Exception('вы уже находитесь в этом лобби')
- elif garticError == 8:
- raise Exception('cookie expired')
- elif garticError == 9:
- raise Exception('неверный метод аутентификации')
- elif garticError:
- raise Exception(f'ошибка при входе в лобби: {garticError}')
- garticUser = lobbyObj[1].get('user')
- if not garticUser:
- raise Exception(f'неизвестная ошибка при входе в лобби. Данные лобби: {s}')
- return WebsocketClient(svServerHost[8:], sid, garticUser['id'])
- def start():
- global logBox
- global playerUuid
- playerUuid = getRandomUuid()
- window = tk.Tk()
- window.title('Autogartic v0.1 (by 2ch anon)')
- window.geometry('700x800')
- logBox = scrolledtext.ScrolledText(window)
- ttk.Label(window, text='Ссылка на приглашение в игру:').pack(padx=5, pady=5)
- inviteLinkEntry = ttk.Entry(window, width=70)
- inviteLinkEntry.pack(padx=5, pady=5)
- AUTH_METHOD_ANON_STR = 'Анонимно'
- AUTH_METHOD_TWITCH_STR = 'Через twitch'
- ttk.Label(window, text='Метод аутентификации:').pack(padx=5, pady=5)
- authMethodStr = tk.StringVar()
- authMethodCombobox = ttk.Combobox(window, textvariable=authMethodStr)
- authMethodCombobox['values'] = [AUTH_METHOD_ANON_STR, AUTH_METHOD_TWITCH_STR]
- authMethodCombobox['state'] = 'readonly'
- authMethodCombobox.pack(padx=5, pady=5)
- authFrame = ttk.Frame(window)
- authFrame.pack()
- nickLabel = ttk.Label(authFrame, text='Ник:')
- nickEntry = ttk.Entry(authFrame, width=30)
- twitchCookieLabel = ttk.Label(authFrame, text='Cookie (см. readme):')
- twitchCookieEntry = ttk.Entry(authFrame, width=50)
- def authMethodChanged(event):
- if authMethodStr.get() == AUTH_METHOD_ANON_STR:
- nickLabel.pack(padx=5, pady=1, side=tk.LEFT)
- nickEntry.pack(padx=5, pady=1, side=tk.LEFT)
- twitchCookieLabel.pack_forget()
- twitchCookieEntry.pack_forget()
- elif authMethodStr.get() == AUTH_METHOD_TWITCH_STR:
- twitchCookieLabel.pack(padx=5, pady=1, side=tk.LEFT)
- twitchCookieEntry.pack(padx=5, pady=1, side=tk.LEFT)
- nickLabel.pack_forget()
- nickEntry.pack_forget()
- authMethodCombobox.bind('<<ComboboxSelected>>', authMethodChanged)
- ttk.Label(window, text='Путь к файлу с фразами:').pack(padx=5, pady=1)
- phrasesPathEntry = ttk.Entry(window, width=70)
- phrasesPathEntry.pack(padx=5, pady=1)
- def onChoosePhrases(event=None):
- filePathSelected = filedialog.askopenfilename()
- phrasesPathEntry.delete(0, tk.END)
- phrasesPathEntry.insert(0, filePathSelected)
- choosePhrasesButton = ttk.Button(window, text="Открыть...", command=onChoosePhrases)
- choosePhrasesButton.pack(padx=5, pady=5)
- ttk.Label(window, text='Путь к папке с картинками:').pack(padx=5, pady=1)
- imagesDirEntry = ttk.Entry(window, width=70)
- imagesDirEntry.pack(padx=5, pady=1)
- def onChooseImages(event=None):
- dirPathSelected = filedialog.askdirectory()
- imagesDirEntry.delete(0, tk.END)
- imagesDirEntry.insert(0, dirPathSelected)
- chooseImagesButton = ttk.Button(window, text="Открыть...", command=onChooseImages)
- chooseImagesButton.pack(padx=5, pady=5)
- startButton = None
- isConnected = False
- websocketClient = None
- def onConnected():
- nonlocal isConnected
- log('Подключено!')
- startButton.config(text='Отключиться')
- startButton["state"] = "enabled"
- isConnected = True
- def onDisconnected():
- nonlocal isConnected
- log('Отключено!')
- startButton.config(text='Подключиться')
- startButton["state"] = "enabled"
- isConnected = False
- def onConnectClick(event=None):
- nonlocal isConnected
- nonlocal websocketClient
- inviteLink = inviteLinkEntry.get()
- startButton["state"] = "disabled"
- if not isConnected:
- try:
- global phrasesList
- phrasesList = PhrasesList(phrasesPathEntry.get())
- global imagesList
- imagesList = ImagesList(imagesDirEntry.get())
- if authMethodStr.get() == AUTH_METHOD_ANON_STR:
- websocketClient = authentication(inviteLink, 'anon', nickname=nickEntry.get())
- elif authMethodStr.get() == AUTH_METHOD_TWITCH_STR:
- websocketClient = authentication(inviteLink, 'twitch', twitchCookie=twitchCookieEntry.get())
- else:
- raise Exception('не выбран метод аутентификации')
- websocketClient.setHandlers(onConnected, onDisconnected)
- websocketClient.start()
- except Exception as e:
- log(f'Ошибка при подключении: {e}')
- startButton.config(text='Подключиться')
- startButton["state"] = "enabled"
- else:
- if websocketClient:
- websocketClient.stop()
- else:
- startButton.config(text='Подключиться')
- startButton["state"] = "enabled"
- isConnected = False
- def onSaveSettingsClick(event=None):
- try:
- with open('autogartic_settings.json', 'w', encoding='utf8') as f:
- settings = {
- 'inviteLink':inviteLinkEntry.get(),
- 'authMethod':authMethodStr.get(),
- 'nick':nickEntry.get(),
- 'twitchCookie':twitchCookieEntry.get(),
- 'phrasesPath':phrasesPathEntry.get(),
- 'imagesDir':imagesDirEntry.get(),
- }
- json.dump(settings, f)
- log('Настройки сохранены')
- except Exception as e:
- log(f'Ошибка при сохранении настроек: {e}')
- saveSettingsButton = ttk.Button(window, text="Сохранить настройки", command=onSaveSettingsClick)
- saveSettingsButton.pack(padx=5, pady=(15, 5))
- startButton = ttk.Button(window, text="Подключиться", command=onConnectClick)
- startButton.pack(padx=5, pady=5)
- ttk.Label(window, text='Лог:').pack()
- logBox.pack(padx=5, pady=5)
- settings = {
- 'inviteLink':'',
- 'authMethod':AUTH_METHOD_ANON_STR,
- 'nick':'',
- 'twitchCookie':'',
- 'phrasesPath':'',
- 'imagesDir':'',
- }
- try:
- with open('autogartic_settings.json', 'r', encoding='utf8') as f:
- loadedSettings = json.load(f)
- for k in settings.keys():
- settings[k] = loadedSettings[k]
- log('Настройки загружены')
- except FileNotFoundError:
- pass
- except Exception as e:
- log(f'Ошибка при загрузке настроек: {e}')
- # set loaded settings
- inviteLinkEntry.delete(0, tk.END)
- inviteLinkEntry.insert(0, settings['inviteLink'])
- authMethodCombobox.set(settings['authMethod'])
- authMethodChanged(None)
- nickEntry.delete(0, tk.END)
- nickEntry.insert(0, settings['nick'])
- twitchCookieEntry.delete(0, tk.END)
- twitchCookieEntry.insert(0, settings['twitchCookie'])
- phrasesPathEntry.delete(0, tk.END)
- phrasesPathEntry.insert(0, settings['phrasesPath'])
- imagesDirEntry.delete(0, tk.END)
- imagesDirEntry.insert(0, settings['imagesDir'])
- window.mainloop()
- start()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement