wordlebot/bot.py

import collections
import datetime
import json
import os
import pathlib
import time
import sys

import requests
from selenium import webdriver
from selenium.webdriver.chrome.options import Options as ChromeOptions
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys

import b2


EMOJI = {
    'absent': ':black_large_square:',
    'present': ':large_yellow_square:',
    'correct': ':large_green_square:',
}


def filter_word(word, conditions):
    # first, filter on correct
    for char, idx in conditions['correct']:
        if word[idx] != char:
            return False

    # we only want to consider characters in positions that haven't been spoken for already
    positions = [0, 1, 2, 3, 4]
    for _, i in conditions['correct']:
        positions.remove(i)

    # now filter on present-but-not-correct
    if conditions['present']:
        required_chars = set(c[0] for c in conditions['present'])
        actual_chars = set(word[p] for p in positions)
        overlap = required_chars & actual_chars
        if len(overlap) == 0:
            return False

        to_remove = []
        for cond in conditions['present']:
            for p in positions:
                char = word[p]
                if cond[0] == char:
                    if cond[1] == p:
                        return False # this might break everything?
                    else:
                        to_remove.append(p)

        # remove the characters that met our criteria above
        for p in to_remove:
            positions.remove(p)

    # filter on absence
    for p in positions:
        char = word[p]
        if char in conditions['absent']:
            return False

    return True


def slack_request(method, **kwargs):
    kwargs['headers'] = {'Authorization': 'Bearer ' + os.environ['SLACK_TOKEN']}

    r = requests.post(
        f'https://slack.com/api/{method}',
        **kwargs
    )

    resp = r.json()
    if not resp['ok']:
        print('Slack API responded with error:')
        print(resp)
        sys.exit(1)
    return resp


class State:
    def __init__(self, driver):
        self.driver = driver
        self.b2_client = b2.Client(os.environ['B2_KEY_ID'], os.environ['B2_APP_KEY'])
        self.bucket_id = os.environ['B2_BUCKET_ID']

    def restore(self):
        # use the latest state that isn't from today
        files = self.b2_client.list_objects(self.bucket_id, 'wordle/')
        state = None
        for file in reversed(files):
            path = file['fileName']
            filename = path.split('/')[-1]
            if filename < f'state_{datetime.date.today()}.json':
                data = self.b2_client.get_object('cupboard', path)
                state = json.loads(data.decode('utf-8'))
                break

        if state:
            self.driver.execute_script(
                "window.localStorage.setItem('gameState', arguments[0]);",
                state['gameState']
            )
            self.driver.execute_script(
                "window.localStorage.setItem('statistics', arguments[0]);",
                state['statistics']
            )
            print('Restored state: ', filename)

    def save(self):
        game = self.driver.execute_script("return window.localStorage.getItem('gameState')")
        stats = self.driver.execute_script("return window.localStorage.getItem('statistics')")
        state = {'gameState': game, 'statistics': stats} # yes, these are going to be double-encoded

        filename = f'state_{datetime.date.today()}.json'
        self.b2_client.put_object(
            self.bucket_id,
            f'wordle/{filename}',
            json.dumps(state).encode('utf-8')
        )
        print('Saved state: ', filename)


class Solver:
    def __init__(self):
        print('Launching web browser')
        self.driver = self.setup_driver()
        self.state = State(self.driver)
        print('Navigating to page')
        self.driver.get('https://www.powerlanguage.co.uk/wordle/')
        time.sleep(2)
        self.body = self.driver.find_element(By.TAG_NAME, 'body')
        self.body.click()
        self.words = pathlib.Path('dictionary.txt').read_text().splitlines()

    def setup_driver(self):
        options = ChromeOptions()
        options.headless = True
        options.add_argument('--no-sandbox')
        options.add_argument('--remote-debugging-port=9222')
        return webdriver.Chrome(options=options)

    def input_word(self, word):
        for char in word:
            self.body.send_keys(char)
        self.body.send_keys(Keys.ENTER)

    def get_rows(self):
        return self.driver.execute_script(
            """return Array.from(
                document
                    .querySelector('game-app')
                    .shadowRoot
                    .querySelectorAll('game-row')
            ).map(e => e.shadowRoot.querySelectorAll('game-tile'))"""
        )

    def read_row(self, n):
        result = {
            'present': [],
            'correct': [],
            'absent': [],
        }
        rows = self.get_rows()
        for i, tile in enumerate(rows[n]):
            char = tile.get_attribute('letter')
            status = tile.get_attribute('evaluation')
            result[status].append((char, i))
        
        result['absent'] = set(c[0] for c in result['absent'])
        return result

    def get_history(self, row_count):
        rows = self.get_rows()
        history = []
        for i in range(row_count):
            row_hist = []
            for tile in rows[i]:
                status = tile.get_attribute('evaluation')
                row_hist.append(EMOJI[status])
            history.append(row_hist)
        return history

    def solve(self):
        print('Attempting to solve')
        for i in range(6):
            print('Trying word:', self.words[0])
            self.input_word(self.words[0])
            time.sleep(2)

            conditions = self.read_row(i)
            if len(conditions['correct']) == 5:
                time.sleep(3) # wait for the message that comes up when you win
                self.body.click() # dismiss it
                time.sleep(0.5) # wait for overlay to clear
                return {
                    'word': self.words[0],
                    'history': self.get_history(i + 1),
                    'iterations': i + 1
                }
            
            self.words = [w for w in self.words if filter_word(w, conditions)]
        return None

    def capture_board(self):
        board = self.driver.execute_script("return document.querySelector('game-app').shadowRoot.querySelector('#board')")
        filename = f'result_{datetime.date.today()}.png'
        board.screenshot(filename)
        return filename

    def capture_stats(self):
        btn = self.driver.execute_script("return document.querySelector('game-app').shadowRoot.querySelector('game-icon[icon=statistics]')")
        btn.click()
        time.sleep(0.5)
        stats = self.driver.execute_script("return document.querySelector('game-app').shadowRoot.querySelector('game-stats')")
        filename = f'stats_{datetime.date.today()}.png'
        stats.screenshot(filename)
        return filename


if __name__ == '__main__':
    solver = Solver()
    try:
        solver.state.restore()
        result = solver.solve()
        solver.state.save()
        board_path = solver.capture_board()
        stats_path = solver.capture_stats()
    finally:
        solver.driver.close()

    if result:
        print(f'Success!')
    else:
        print('Failed to find the word.')

    wordle_num = (datetime.date.today() - datetime.date(2022, 1, 17)).days + 212
    lines = [f"Wordle {wordle_num}: {result['iterations']}/6"]
    lines = lines + [''.join(row) for row in result['history']]
    
    channel = os.environ['SLACK_CHANNEL']
    msg = slack_request('chat.postMessage', json={'channel': channel, 'text': '\n'.join(lines)})
    slack_request(
        'files.upload',
        params={'thread_ts': msg['ts'], 'channels': channel},
        files={'file': (board_path, open(board_path, 'rb').read())},
    )
    slack_request(
        'files.upload',
        params={'thread_ts': msg['ts'], 'channels': channel},
        files={'file': (stats_path, open(stats_path, 'rb').read())},
    )