wordlebot/bot.py

import collections
import datetime
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 Solver:
    def __init__(self):
        print('Launching web browser')
        self.driver = self.setup_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()
        self.b2_client = b2.Client(os.environ['B2_KEY_ID'], os.environ['B2_APP_KEY'])

    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 restore_state(self):
        data = self.b2_client.get_object('cupboard', 'wordle_state.json')
        if data:
            self.driver.execute_script(
                "window.localStorage.setItem('statistics', arguments[1]);",
                data.decode('utf-8')
            )

    def save_state(self):
        state = driver.execute_script("return window.localStorage.getItem('statistics')")
        self.b2_client.put_object(
            os.environ['B2_BUCKET_ID'],
            'wordle_state.json',
            state.encode('utf-8')
        )

    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 capture_board(self):
        board = self.driver.execute_script("return document.querySelector('game-app').shadowRoot.querySelector('#board')")
        filename = f'wordlebot_{datetime.datetime.now().strftime("%Y-%m-%d")}.png'
        board.screenshot(filename)
        return filename

    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


if __name__ == '__main__':
    solver = Solver()
    try:
        if result := solver.solve():
            filename = solver.capture_board()
    finally:
        solver.driver.close()

    if result:
        print(f'Success!')
        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': (filename, open(filename, 'rb').read())},
        )

    else:
        print('Failed to find the word.')