autogartic.pyw

'''
Бот для автопостинга картинок и текста в гартик
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: ссылка не вставляется через ctrl-V
A: переключи раскладку на английскую

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)
        if not nextUrl:
            raise Exception(f'не удалось пройти аутентификацию через твич')
        nextUrl = nextUrl.groups()[0]
        nextUrl = nextUrl.replace('&amp;', '&')
        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')
    content = r.content.decode('utf8')
    sid = re.search('.*sid":"(.*)","upgrades.*', content)
    if not sid:
        raise Exception(f'не удалось найти sid: {content}')
    sid = sid.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()