add user management cli

pyproject.toml

@@ -36,6 +36,7 @@ show-user = "wordlinator.app:sync_show_user"
 show-missing = "wordlinator.app:sync_show_missing"
 tweet-missing = "wordlinator.app:sync_tweet_missing"
 db-load = "wordlinator.app:load_db_scores"
+add-user = "wordlinator.app:sync_add_user"
 
 [tool.mypy]
 ignore_missing_imports = true

wordlinator/app/__init__.py

@@ -197,6 +197,62 @@ def load_db_scores():
     _save_db_scores(wordle_day, scores)
 
 
+def _add_user_args():
+    parser = argparse.ArgumentParser("add-user")
+    parser.add_argument("username", help="The user twitter handle or name.")
+    parser.add_argument(
+        "--no-check-twitter",
+        dest="check_twitter",
+        action="store_false",
+        default=True,
+        help="don't check Twitter for user's scores",
+    )
+    parser.add_argument(
+        "-g", "--games", nargs="*", help="The game/round number(s) to enroll the user."
+    )
+    parser.add_argument(
+        "-u",
+        "--unenroll-games",
+        nargs="*",
+        help="Game/round number(s) to unenroll the user.",
+    )
+    args = parser.parse_args()
+    return args
+
+
+async def add_user(args=None):
+    args = args or _add_user_args()
+    db = wordlinator.db.pg.WordleDb()
+
+    user = db.get_user(args.username)
+    if not user:
+        rich.print(f"[green]Creating user {args.username}")
+        user_id = None
+        check_twitter = args.check_twitter
+        if check_twitter:
+            twitter = wordlinator.twitter.TwitterClient()
+            user_id = await twitter.get_user_twitter_id(args.username)
+            if not user_id:
+                check_twitter = False
+                rich.print(
+                    f"[yellow]No twitter ID found for {args.username}, "
+                    "disabling twitter check"
+                )
+        user_id = user_id or f"{args.username}-NA"
+        user = db.add_user(args.username, user_id, check_twitter=check_twitter)
+    for round in args.games or []:
+        rich.print(f"[green]Adding {args.username} to round {round}")
+        db.add_user_to_round(args.username, round)
+    for round in args.unenroll_games or []:
+        rich.print(f"[green]Removing {args.username} from round {round}")
+        db.remove_user_from_round(args.username, round)
+
+
+def sync_add_user():
+    args = _add_user_args()
+    asyncio.run(add_user(args))
+
+
 def sync_main():
     wordle_day = _get_day()
     asyncio.run(main(wordle_day=wordle_day))

wordlinator/db/pg.py

@@ -85,8 +85,10 @@ class WordleDb:
         user = self.get_user(username)
         return user.twitter_id if user else None
 
-    def add_user(self, username, user_id):
-        return User.create(username=username, twitter_id=user_id)
+    def add_user(self, username, user_id, check_twitter=True):
+        return User.create(
+            username=username, twitter_id=user_id, check_twitter=check_twitter
+        )
 
     def get_rounds(self):
         return list(sorted(Game.select(), key=lambda d: d.start_date))
@@ -118,6 +120,30 @@ class WordleDb:
         for hole_no in range(1, 19):
             self.get_or_create_hole(round_no, hole_no)
 
+    def get_or_create_player_round(self, user_id, game_id):
+        try:
+            return Player.get(user_id=user_id, game_id=game_id)
+        except peewee.DoesNotExist:
+            return Player.create(user_id=user_id, game_id=game_id)
+
+    def add_user_to_round(self, username, round_no):
+        user = self.get_user(username)
+        if not user:
+            raise ValueError(f"No user found with username {username}")
+        round = self.get_or_create_round(round_no)
+        return self.get_or_create_player_round(user.user_id, round.game_id)
+
+    def remove_user_from_round(self, username, round_no):
+        user = self.get_user(username)
+        if not user:
+            raise ValueError(f"No user found with username {username}")
+        round = self.get_or_create_round(round_no)
+        try:
+            player = Player.get(user_id=user.user_id, game_id=round.game_id)
+            player.delete_instance()
+        except peewee.DoesNotExist:
+            return
+
     def add_score(self, username, game, hole, score):
         if not score:
             return

wordlinator/twitter/__init__.py

@@ -133,15 +133,19 @@ class TwitterClient(httpx.AsyncClient):
     async def get_user_by(self, username: str):
         return await self.get(self.USER_PATH.format(username=username))
 
+    async def get_user_twitter_id(self, username: str):
+        user_id = None
+        twitter_user = await self.get_user_by(username)
+        if twitter_user.is_success:
+            user_id = twitter_user.json().get("data", {}).get("id", None)
+        return user_id
+
     async def get_user_id(self, username: str):
         db_user = self.db.get_user(username)
         if db_user:
             return db_user.twitter_id if db_user.check_twitter else False
         else:
-            twitter_user = await self.get_user_by(username)
-            user_id = None
-            if twitter_user.is_success:
-                user_id = twitter_user.json().get("data", {}).get("id", None)
+            user_id = await self.get_user_twitter_id(username)
             if user_id:
                 self.db.add_user(username, user_id)
             return user_id