silence warning when resolving team for self-damage

main.py

@@ -7,6 +7,7 @@ from pathlib import Path
 import argparse
 import json
 import logging
+import re
 import requests
 import sqlite3
 import sys
@@ -53,7 +54,7 @@ def team(player: str) -> str:
     if player in TEAMS:
         return TEAMS[player]
     else:
-        if not player in _logged_teams:
+        if not player in _logged_teams and player:
             LOG.warning(f"missing team mapping for {player}")
         _logged_teams.append(player)
         return player
@@ -100,7 +101,12 @@ def _init_db(conn: sqlite3.Connection):
         CREATE TABLE IF NOT EXISTS games(
             id, p1, p2, format, uploadtime, winner,
             UNIQUE(id)
-        )
+        );
+        -- No good way to ensure idempotence for damage; just re-build it.
+        DROP TABLE IF EXISTS damage;
+        CREATE TABLE damage(game, player, name, value);
+        DROP TABLE IF EXISTS indirect_damage;
+        CREATE TABLE indirect_damage(game, player, name, value);
         """
     )
 
@@ -110,6 +116,7 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
 
     turn = 0
     players = {}
+    hp = {}
 
     # ("p2a: Edward", "p1a: Meteo")
     # memorises the user of the move that causes environment setting or status,
@@ -136,8 +143,10 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
         match chunks:
             case ["player", id, username, *rest]:
                 players[id] = username
+
             case ["turn", turn]:
                 turn = int(turn)
+
             case ["move", user, move, target]:
                 last_move = (user, target)
                 player, user = resolve_mon(user)
@@ -150,7 +159,13 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
                     """,
                     (game, turn, team(player), move, user, target),
                 )
-            case ["switch", name, specie, *rest]:
+
+            case ["drag", name, specie, status, *rest]:
+                hp[name] = int(status.split("/")[0])
+
+            case ["switch", name, specie, status, *rest]:
+                hp[name] = int(status.split("/")[0])
+
                 player, name = resolve_mon(name)
                 conn.execute(
                     """
@@ -168,6 +183,7 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
                     """,
                     (game, team(player), name, specie.split(", ")[0]),
                 )
+
             case ["faint", mon]:
                 player, mon = resolve_mon(mon)
                 conn.execute(
@@ -178,6 +194,7 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
                     """,
                     (game, turn, team(player), mon),
                 )
+
             case ["win", player]:
                 conn.execute(
                     """
@@ -187,51 +204,97 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
                     """,
                     (team(player), game),
                 )
+
             case ["-sidestart", side, env]:
                 if not last_move:
                     LOG.warning(f"missing previous move for {line}")
                     continue
                 LOG.debug(f"{line} <- {last_move}")
                 last_env_set[(side[0:1], env.replace("move: ", ""))] = last_move[0]
+
             case ["-status", mon, cond]:
                 if not last_move or last_move[1] != mon:
                     LOG.warning(f"missing previous move for {line}")
                     continue
                 LOG.debug(f"{line} <- {last_move}")
                 last_status_set[(mon, cond)] = last_move[0]
-            case ["-damage", mon, *rest]:
-                # rest is new_hp and sometimes a source (if not from a move)
-                # in a knockout, new_hp is "0 fnt"
-                if rest[0] == "0 fnt" and len(rest) > 1:
-                    LOG.debug(f"tracing source for {line}")
-                    source = rest[1].replace("[from] ", "")
-                    source_user = None
 
-                    test_hazard = last_env_set.get((mon[0:1], source))
+            case ["-damage", mon, status]:
-                    if test_hazard:
+                # mon takes direct (non-hazard/condition) damage
-                        source_user = test_hazard
+                # status can be a percentage 70/100 with or without condition,
-                        LOG.debug(f"identified hazard source {source_user}")
+                # or "0 fnt"
+                new_hp = int(re.split("[/ ]", status)[0])
+                LOG.debug(f"{mon} dropped to {new_hp} from {hp[mon]}")
+                LOG.debug(f"source: {last_move}")
 
-                    test_status = last_status_set.get((mon, source))
+                # resolve to damage source
-                    if test_status:
+                if last_move[1] != mon:
-                        source_user = test_status
+                    LOG.warn(
-                        LOG.debug(f"identified move source {source_user}")
+                        f"{mon} took direct damage but last move was not targeted at them"
+                    )
+                    continue
+                user = last_move[0]
+                source_player, source_mon = resolve_mon(user)
 
-                    if source == "Recoil" or source.startswith("item: "):
+                conn.execute(
-                        LOG.debug(f"identified special source {source}")
+                    """
-                        source = source.replace("item: ", "")
+                    INSERT INTO damage(game, player, name, value)
-                        source_user = "self"
+                    VALUES(?, ?, ?, ?)
+                    ON CONFLICT DO NOTHING
+                    """,
+                    (game, team(source_player), source_mon, hp[mon] - new_hp),
+                )
 
-                    if not source_user:
+                hp[mon] = new_hp
-                        LOG.error(f"missing source for {line}")
-                        continue
 
-                    player, mon = resolve_mon(mon)
+            case ["-damage", mon, status, from_]:
-                    if source_user.startswith("p1") or source_user.startswith("p2"):
+                # mon takes indirect damage
-                        source_player, source_user = resolve_mon(source_user)
+                # status can be a percentage 70/100 with or without condition,
-                    else:
+                # or "0 fnt"
-                        source_player = None
+                # mon has fainted from an indirect damage source
+                #
+                new_hp = int(re.split("[/ ]", status)[0])
+                LOG.debug(f"{mon} dropped to {new_hp} from {from_}")
 
+                LOG.debug(f"tracing source for {line}")
+                source = from_.replace("[from] ", "")
+                source_user = None
+
+                test_hazard = last_env_set.get((mon[0:1], source))
+                if test_hazard:
+                    source_user = test_hazard
+                    LOG.debug(f"identified hazard source {source_user}")
+
+                test_status = last_status_set.get((mon, source))
+                if test_status:
+                    source_user = test_status
+                    LOG.debug(f"identified move source {source_user}")
+
+                if source == "Recoil" or source.startswith("item: "):
+                    LOG.debug(f"identified special source {source}")
+                    source = source.replace("item: ", "")
+                    source_user = "self"
+
+                if not source_user:
+                    LOG.error(f"missing source for {line}")
+                    continue
+
+                player, pkmn = resolve_mon(mon)
+                if source_user.startswith("p1") or source_user.startswith("p2"):
+                    source_player, source_mon = resolve_mon(source_user)
+                else:
+                    source_player = None
+
+                if source_player:
+                    conn.execute(
+                        """
+                        INSERT INTO indirect_damage(game, player, name, value)
+                        VALUES(?, ?, ?, ?)
+                        """,
+                        (game, team(source_player), source_mon, hp[mon] - new_hp),
+                    )
+
+                if status == "0 fnt":
                     conn.execute(
                         """
                         INSERT INTO indirect_knockouts(game, turn, player, name, source, source_user, source_player)
@@ -242,12 +305,16 @@ def parse_log(game: str, log: str, into: sqlite3.Connection):
                             game,
                             turn,
                             team(player),
-                            mon,
+                            pkmn,
                             source,
-                            source_user,
+                            source_mon,
                             team(source_player),
                         ),
                     )
+
+            case ["-heal", mon, status, *rest]:
+                hp[mon] = int(status.split("/")[0])
+
             case _:
                 # LOG.debug(f"unhandled message {chunks[0]}")
                 pass