expand local DB to store more data

pyproject.toml

@@ -27,6 +27,7 @@ wordlinator = "wordlinator.app:sync_main"
 update = "wordlinator.app:sync_update"
 show-user = "wordlinator.app:sync_show_user"
 show-missing = "wordlinator.app:sync_show_missing"
+db-load = "wordlinator.app:load_db_scores"
 
 [tool.mypy]
 ignore_missing_imports = true

wordlinator/app/__init__.py

@@ -4,6 +4,7 @@ import asyncio
 import rich
 import rich.table
 
+import wordlinator.db
 import wordlinator.sheets
 import wordlinator.twitter
 import wordlinator.utils
@@ -56,6 +57,19 @@ def print_score_table(wordle_day, scores):
     rich.print(table)
 
 
+def _save_db_scores(wordle_day: wordlinator.utils.WordleDay, scores: dict):
+    db = wordlinator.db.WordleDb()
+    hole_data = wordle_day.golf_hole
+    if not hole_data:
+        return
+    game_no = hole_data.game_no
+    for user, score_list in scores.items():
+        if not db.get_user(user):
+            continue
+        for day, score_entry in enumerate(score_list, start=1):
+            db.add_score(user, game_no, day, score_entry)
+
+
 async def main_update(
     wordle_day: wordlinator.utils.WordleDay = wordlinator.utils.WORDLE_TODAY,
 ):
@@ -65,7 +79,9 @@ async def main_update(
     if not any((s is not None for s in today_scores.values())):
         raise ValueError("No scores pulled!")
 
-    sheets_client.update_scores(today_scores)
+    updated_scores = sheets_client.update_scores(today_scores)
+
+    _save_db_scores(wordle_day, updated_scores)
 
     print_score_table(wordle_day, today_scores)
 
@@ -108,6 +124,13 @@ def _get_day():
     return wordle_day
 
 
+def load_db_scores():
+    wordle_day = _get_day()
+    client = wordlinator.sheets.SheetsClient(wordle_day=wordle_day)
+    scores = client.get_scores()
+    _save_db_scores(wordle_day, scores)
+
+
 def sync_main():
     wordle_day = _get_day()
     asyncio.run(main(wordle_day=wordle_day))

wordlinator/db.py

@@ -0,0 +1,126 @@
+import sqlite3
+
+
+class WordleDb:
+    def __init__(self):
+        self.con = sqlite3.connect("wordle.db")
+        cur = self.con.cursor()
+        cur.execute(
+            """CREATE TABLE IF NOT EXISTS user
+            (id INTEGER PRIMARY KEY,
+             username varchar(50) NOT NULL,
+             user_id varchar(20) NOT NULL)"""
+        )
+        cur.execute(
+            """CREATE TABLE IF NOT EXISTS game (
+                id INTEGER PRIMARY KEY,
+                game INTEGER NOT NULL)"""
+        )
+        cur.execute(
+            """CREATE TABLE IF NOT EXISTS hole
+            (id INTEGER PRIMARY KEY,
+             hole INTEGER NOT NULL,
+             game_id INTEGER NOT NULL,
+             FOREIGN KEY (game_id)
+               REFERENCES game (id)
+             )"""
+        )
+        cur.execute(
+            """CREATE TABLE IF NOT EXISTS score
+           (id INTEGER PRIMARY KEY,
+            score INTEGER NOT NULL,
+            user_id INTEGER NOT NULL,
+            game_id INTEGER NOT NULL,
+            hole_id INTEGER NOT NULL,
+            FOREIGN KEY (game_id)
+              REFERENCES game (id),
+            FOREIGN KEY (user_id)
+              REFERENCES user (id),
+            FOREIGN KEY (hole_id)
+              REFERENCES hole (id),
+            UNIQUE(user_id, game_id, hole_id)
+            )"""
+        )
+        self.con.commit()
+
+    def get_user(self, username):
+        cur = self.con.cursor()
+        res = list(cur.execute(f"SELECT * FROM user WHERE username = '{username}'"))
+        return res[0] if res else None
+
+    def get_user_id(self, username):
+        user = self.get_user(username)
+        if not user:
+            return None
+        return user[2]
+
+    def add_user(self, username, user_id):
+        cur = self.con.cursor()
+        cur.execute(
+            f"INSERT INTO user (username, user_id) VALUES ('{username}', '{user_id}')"
+        )
+        self.con.commit()
+
+    def get_or_create_round(self, round_no):
+        cur = self.con.cursor()
+        res = list(cur.execute(f"SELECT * FROM game WHERE game = {round_no}"))
+        if not res:
+            list(cur.execute(f"INSERT INTO game (game) VALUES ({round_no})"))
+            self.con.commit()
+            res = list(cur.execute(f"SELECT * FROM game WHERE game = {round_no}"))
+        return res[0]
+
+    def get_or_create_hole(self, round_no, hole_no):
+        round_id = self.get_or_create_round(round_no)[0]
+        cur = self.con.cursor()
+        res = list(
+            cur.execute(
+                f"SELECT * FROM hole WHERE hole = {hole_no} AND game_id = {round_id}"
+            )
+        )
+        if not res:
+            cur.execute(
+                f"INSERT INTO hole (hole, game_id) VALUES ({hole_no}, {round_id})"
+            )
+            res = list(
+                cur.execute(
+                    "SELECT * FROM hole "
+                    f"WHERE hole = {hole_no} AND game_id = {round_id}"
+                )
+            )
+        self.con.commit()
+        return res[0]
+
+    def create_round_holes(self, round_no):
+        for hole_no in range(1, 19):
+            self.get_or_create_hole(round_no, hole_no)
+
+    def add_score(self, username, game, hole, score):
+        if not score:
+            return
+        user = self.get_user(username)
+        if not user:
+            raise ValueError("No such user!")
+        user_id = user[0]
+
+        hole = self.get_or_create_hole(game, hole)
+        _, hole_id, game_id = hole
+
+        cur = self.con.cursor()
+        res = list(
+            cur.execute(
+                f"""SELECT score FROM score
+            WHERE user_id = {user_id} AND game_id = {game_id} AND hole_id = {hole_id}"""
+            )
+        )
+        if res:
+            cmd = f"""UPDATE score
+                SET score = {score}
+                WHERE user_id = {user_id}
+                  AND game_id = {game_id}
+                  AND hole_id = {hole_id}"""
+        else:
+            cmd = f"""INSERT INTO score (score, user_id, game_id, hole_id)
+                VALUES ({score}, {user_id}, {game_id}, {hole_id})"""
+        cur.execute(cmd)
+        self.con.commit()

wordlinator/sheets/__init__.py

@@ -145,6 +145,7 @@ class SheetsClient:
                     current_row[day_idx] = score_val
             current_scores[name] = current_row
         self.write_scores(current_scores)
+        return current_scores
 
 
 def main():

wordlinator/twitter/__init__.py

@@ -4,13 +4,13 @@ import datetime
 import enum
 import os
 import re
-import sqlite3
 
 import authlib.integrations.httpx_client
 import dateutil.parser
 import httpx
 import rich
 
+import wordlinator.db
 import wordlinator.utils
 
 BASE_URL = "https://api.twitter.com/2"
@@ -90,26 +90,6 @@ class WordleTweet:
         )
 
 
-class UserDb:
-    def __init__(self):
-        self.con = sqlite3.connect("users.db")
-        cur = self.con.cursor()
-        cur.execute(
-            """CREATE TABLE IF NOT EXISTS users (username text, user_id text)"""
-        )
-        self.con.commit()
-
-    def get_user(self, username):
-        cur = self.con.cursor()
-        res = list(cur.execute(f"SELECT * from users WHERE username = '{username}'"))
-        return res[0] if res else None
-
-    def add_user(self, username, user_id):
-        cur = self.con.cursor()
-        cur.execute(f"INSERT INTO users VALUES ('{username}', '{user_id}')")
-        self.con.commit()
-
-
 class TwitterClient(httpx.AsyncClient):
     SEARCH_PATH = "tweets/search/recent"
     USER_PATH = "users/by/username/{username}"
@@ -125,7 +105,7 @@ class TwitterClient(httpx.AsyncClient):
             auth = authlib.integrations.httpx_client.OAuth1Auth(**oauth_creds)
             kwargs["auth"] = auth
         super().__init__(base_url=BASE_URL, **kwargs)
-        self.db = UserDb()
+        self.db = wordlinator.db.WordleDb()
         self.wordle_day = wordle_day
         if not oauth_creds:
             self.headers["Authorization"] = f"Bearer {TOKEN}"
@@ -145,9 +125,9 @@ class TwitterClient(httpx.AsyncClient):
         return await self.get(self.USER_PATH.format(username=username))
 
     async def get_user_id(self, username: str):
-        db_user = self.db.get_user(username)
-        if db_user:
-            return db_user[1]
+        db_user_id = self.db.get_user_id(username)
+        if db_user_id:
+            return db_user_id
         else:
             twitter_user = await self.get_user_by(username)
             user_id = None