refactor: let NFL & NCAAF share same DB workflows.

.gitignore

@@ -21,6 +21,9 @@ data/*.db-wal
 
 node_modules/
 
+# Kept on the owner's machine only.
+tools/download_ncaaf_logos.py
+
 .claude/
 
 # `playbook/` is private by defaul.

database/columns.py

@@ -0,0 +1,102 @@
+"""The columns of each stat category, declared once for models, ingestion and the API.
+
+Each `DataColumn` ties the Excel header in the preprocessed workbook to the model
+attribute, the API payload key, and how the value is read. This one declaration
+feeds the ingestion's required-column check and row building and the API
+serializer, so a category's columns are added in a single place; a test checks
+that the models carry exactly these attributes. The NFL tables use it now and the
+college tables reuse it unchanged.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class DataColumn:
+    excel_header: str  # Column name in the preprocessed xlsx sheet.
+    attr: str  # Model attribute and ORM column name.
+    api_key: str  # Key in the /api payload.
+    kind: str  # "int", "float", or "complement" (stored as 100 - the Excel value).
+
+
+# Identity columns read from every sheet. "Team" serializes as "team" and reads
+# into team_code; "Player Id" reads into player_id.
+IDENTITY_EXCEL_HEADERS = ("Team", "Player Id", "Player", "Abbr Name")
+
+
+PASS_RUSH_DATA_COLUMNS = (
+    DataColumn("Games", "games", "games", "int"),
+    DataColumn("PR Opp", "pr_opp", "PR Opp", "int"),
+    DataColumn("TPS PR Opp", "tps_pr_opp", "TPS PR Opp", "int"),
+    DataColumn(
+        "Pass Rush Percent", "pass_cov_percent", "Pass Cov Percent", "complement"
+    ),
+    DataColumn(
+        "TPS Pass Rush Percent",
+        "tps_pass_cov_percent",
+        "TPS Pass Cov Percent",
+        "complement",
+    ),
+    DataColumn("Win Rate", "win_rate", "Win Rate", "float"),
+    DataColumn("TPS Win Rate", "tps_win_rate", "TPS Win Rate", "float"),
+    DataColumn("Pressure Rate", "pressure_rate", "Pressure Rate", "float"),
+    DataColumn("TPS Pressure Rate", "tps_pressure_rate", "TPS Pressure Rate", "float"),
+    DataColumn("Havoc Rate", "havoc_rate", "Havoc Rate", "float"),
+    DataColumn("TPS Havoc Rate", "tps_havoc_rate", "TPS Havoc Rate", "float"),
+    DataColumn("PRP", "prp", "PRP", "float"),
+    DataColumn("TPS PRP", "tps_prp", "TPS PRP", "float"),
+)
+
+
+PASS_BLOCK_DATA_COLUMNS = (
+    DataColumn("Games", "games", "games", "int"),
+    DataColumn("Non Spike PB Snaps", "non_spike_pb_snaps", "Non Spike PB Snaps", "int"),
+    DataColumn(
+        "TPS Non Spike PB Snaps",
+        "tps_non_spike_pb_snaps",
+        "TPS Non Spike PB Snaps",
+        "int",
+    ),
+    DataColumn(
+        "Allowed Pressure %", "allowed_pressure_pct", "Allowed Pressure %", "float"
+    ),
+    DataColumn(
+        "TPS Allowed Pressure %",
+        "tps_allowed_pressure_pct",
+        "TPS Allowed Pressure %",
+        "float",
+    ),
+    DataColumn("Allowed Havoc %", "allowed_havoc_pct", "Allowed Havoc %", "float"),
+    DataColumn(
+        "TPS Allowed Havoc %", "tps_allowed_havoc_pct", "TPS Allowed Havoc %", "float"
+    ),
+    DataColumn("PBE", "pbe", "PBE", "float"),
+    DataColumn("TPS PBE", "tps_pbe", "TPS PBE", "float"),
+)
+
+
+def required_excel_headers(data_columns: tuple[DataColumn, ...]) -> tuple[str, ...]:
+    """The headers a sheet must have: the identity columns plus every data column."""
+    return IDENTITY_EXCEL_HEADERS + tuple(
+        column.excel_header for column in data_columns
+    )
+
+
+def row_to_payload(row, data_columns: tuple[DataColumn, ...]) -> dict:
+    """The /api payload for one ORM row: the common identity keys plus each data key."""
+    payload = {
+        "id": row.id,
+        "season": row.season,
+        "position": row.position,
+        "team": row.team_code,
+        "player_id": row.player_id,
+        "player": row.player,
+        "abbr_name": row.abbr_name,
+    }
+
+    for column in data_columns:
+        payload[column.api_key] = getattr(row, column.attr)
+
+    return payload

database/db_ingestion.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 import argparse
 import os
+from dataclasses import dataclass
 import re
 import sys
 from datetime import date
@@ -34,9 +35,14 @@
 from sqlalchemy import delete
 from sqlalchemy.orm import Session
 from config import DYNAMIC_THRESHOLDS
+from database.columns import (
+    PASS_BLOCK_DATA_COLUMNS,
+    PASS_RUSH_DATA_COLUMNS,
+    required_excel_headers,
+)
 from database.db_models import Base, NFLPassBlockStat, NFLPassRushStat, NFLTeam
-from main import engine, SessionLocal
-from teams_reference import TEAMS
+from database.session import engine, SessionLocal
+from teams.nfl import TEAMS
 
 FRONT_7_POSITIONS = ("DI", "ED", "LB")
 OL_POSITIONS = ("T", "G", "C")
@@ -46,43 +52,6 @@
 
 SEASON_CATEGORY_DIRS = ("front_7_pass_rush", "ol_pass_block")
 
-# Columns each category's sheets must have; ingestion fails loudly without them.
-PASS_RUSH_COLUMNS = (
-    "Team",
-    "Player Id",
-    "Player",
-    "Abbr Name",
-    "Games",
-    "PR Opp",
-    "TPS PR Opp",
-    "Pass Rush Percent",
-    "TPS Pass Rush Percent",
-    "Win Rate",
-    "TPS Win Rate",
-    "Pressure Rate",
-    "TPS Pressure Rate",
-    "Havoc Rate",
-    "TPS Havoc Rate",
-    "PRP",
-    "TPS PRP",
-)
-
-PASS_BLOCK_COLUMNS = (
-    "Team",
-    "Player Id",
-    "Player",
-    "Abbr Name",
-    "Games",
-    "Non Spike PB Snaps",
-    "TPS Non Spike PB Snaps",
-    "Allowed Pressure %",
-    "TPS Allowed Pressure %",
-    "Allowed Havoc %",
-    "TPS Allowed Havoc %",
-    "PBE",
-    "TPS PBE",
-)
-
 
 def _find_project_root() -> Path:
     """Walk up from this file until we find a marker that identifies project root."""
@@ -183,128 +152,122 @@ def upsert_teams(sess: Session) -> None:
     sess.commit()
 
 
-def ingest_pass_rush(sess: Session, data_dir: Path, seasons: list[int]) -> int:
-    total = 0
-    for season in seasons:
-        path = data_dir / NFL_SUBDIR / "front_7_pass_rush" / f"{season}.xlsx"
+def _read_value(kind: str, value):
+    """Read one Excel cell the way its column's kind says."""
+    if kind == "int":
+        return _safe_int(value)
 
-        if not path.exists():
-            print(f"Skip pass rush {season}: {path} not found")
-            continue
+    if kind == "float":
+        return _safe_float(value)
 
-        sheets = pd.read_excel(path, sheet_name=None)
-        for position in FRONT_7_POSITIONS:
-            position_df = sheets.get(position)
-            if position_df is None or position_df.empty:
-                continue
+    if kind == "complement":
+        return _complement(value)
 
-            _require_columns(position_df, PASS_RUSH_COLUMNS, path, position)
-            position_df.dropna(subset=["PR Opp"], inplace=True)
+    raise ValueError(f"unknown column kind: {kind!r}")
 
-            # Wipe slice before re-insertion.
-            sess.execute(
-                delete(NFLPassRushStat).where(
-                    NFLPassRushStat.season == season,
-                    NFLPassRushStat.position == position,
-                )
-            )
 
-            rows = []
-            for _, row in position_df.iterrows():
-                rows.append(
-                    NFLPassRushStat(
-                        season=season,
-                        position=position,
-                        team_code=row["Team"],
-                        player_id=_player_id(row, path, position),
-                        player=row["Player"],
-                        abbr_name=row["Abbr Name"],
-                        games=_safe_int(row.get("Games")),
-                        pr_opp=_safe_int(row.get("PR Opp")),
-                        tps_pr_opp=_safe_int(row.get("TPS PR Opp")),
-                        pass_cov_percent=_complement(row.get("Pass Rush Percent")),
-                        tps_pass_cov_percent=_complement(
-                            row.get("TPS Pass Rush Percent")
-                        ),
-                        win_rate=_safe_float(row.get("Win Rate")),
-                        tps_win_rate=_safe_float(row.get("TPS Win Rate")),
-                        pressure_rate=_safe_float(row.get("Pressure Rate")),
-                        tps_pressure_rate=_safe_float(row.get("TPS Pressure Rate")),
-                        havoc_rate=_safe_float(row.get("Havoc Rate")),
-                        tps_havoc_rate=_safe_float(row.get("TPS Havoc Rate")),
-                        prp=_safe_float(row.get("PRP")),
-                        tps_prp=_safe_float(row.get("TPS PRP")),
-                    )
-                )
+@dataclass(frozen=True)
+class _Category:
+    """Everything the ingestion needs to load one category into its table."""
 
-            sess.add_all(rows)
-            total += len(rows)
-            print(f"  pass_rush {season} {position}: {len(rows)} rows")
+    model: type
+    folder: str
+    positions: tuple[str, ...]
+    data_columns: tuple
+    snap_header: str  # The column a row must have a value in (dropna subset).
+    skip_label: str  # Wording of the "file not found" line.
+    row_label: str  # Wording of the per-position "N rows" line.
 
-    sess.commit()
-    return total
 
+_PASS_RUSH = _Category(
+    model=NFLPassRushStat,
+    folder="front_7_pass_rush",
+    positions=FRONT_7_POSITIONS,
+    data_columns=PASS_RUSH_DATA_COLUMNS,
+    snap_header="PR Opp",
+    skip_label="pass rush",
+    row_label="pass_rush",
+)
 
-def ingest_pass_block(sess: Session, data_dir: Path, seasons: list[int]) -> int:
+_PASS_BLOCK = _Category(
+    model=NFLPassBlockStat,
+    folder="ol_pass_block",
+    positions=OL_POSITIONS,
+    data_columns=PASS_BLOCK_DATA_COLUMNS,
+    snap_header="Non Spike PB Snaps",
+    skip_label="pass block",
+    row_label="pass_block",
+)
+
+
+def ingest_category(
+    sess: Session, data_dir: Path, seasons: list[int], category: _Category
+) -> int:
+    """Load one category's seasons into its table, replacing each slice first.
+
+    The body is shared by the two categories (and later the two leagues); the
+    column reads come from the declaration in `database/columns.py`.
+    """
+    required = required_excel_headers(category.data_columns)
     total = 0
+
     for season in seasons:
-        path = data_dir / NFL_SUBDIR / "ol_pass_block" / f"{season}.xlsx"
+        path = data_dir / NFL_SUBDIR / category.folder / f"{season}.xlsx"
+
         if not path.exists():
-            print(f"Skip pass block {season}: {path} not found")
+            print(f"Skip {category.skip_label} {season}: {path} not found")
             continue
 
         sheets = pd.read_excel(path, sheet_name=None)
-        for position in OL_POSITIONS:
+        for position in category.positions:
             position_df = sheets.get(position)
             if position_df is None or position_df.empty:
                 continue
 
-            _require_columns(position_df, PASS_BLOCK_COLUMNS, path, position)
-            position_df.dropna(subset=["Non Spike PB Snaps"], inplace=True)
+            _require_columns(position_df, required, path, position)
+            position_df.dropna(subset=[category.snap_header], inplace=True)
 
+            # Wipe slice before re-insertion.
             sess.execute(
-                delete(NFLPassBlockStat).where(
-                    NFLPassBlockStat.season == season,
-                    NFLPassBlockStat.position == position,
+                delete(category.model).where(
+                    category.model.season == season,
+                    category.model.position == position,
                 )
             )
 
             rows = []
             for _, row in position_df.iterrows():
-                rows.append(
-                    NFLPassBlockStat(
-                        season=season,
-                        position=position,
-                        team_code=row["Team"],
-                        player_id=_player_id(row, path, position),
-                        player=row["Player"],
-                        abbr_name=row["Abbr Name"],
-                        games=_safe_int(row.get("Games")),
-                        non_spike_pb_snaps=_safe_int(row.get("Non Spike PB Snaps")),
-                        tps_non_spike_pb_snaps=_safe_int(
-                            row.get("TPS Non Spike PB Snaps")
-                        ),
-                        allowed_pressure_pct=_safe_float(row.get("Allowed Pressure %")),
-                        tps_allowed_pressure_pct=_safe_float(
-                            row.get("TPS Allowed Pressure %")
-                        ),
-                        allowed_havoc_pct=_safe_float(row.get("Allowed Havoc %")),
-                        tps_allowed_havoc_pct=_safe_float(
-                            row.get("TPS Allowed Havoc %")
-                        ),
-                        pbe=_safe_float(row.get("PBE")),
-                        tps_pbe=_safe_float(row.get("TPS PBE")),
+                values = {
+                    "season": season,
+                    "position": position,
+                    "team_code": row["Team"],
+                    "player_id": _player_id(row, path, position),
+                    "player": row["Player"],
+                    "abbr_name": row["Abbr Name"],
+                }
+                for column in category.data_columns:
+                    values[column.attr] = _read_value(
+                        column.kind, row.get(column.excel_header)
                     )
-                )
+
+                rows.append(category.model(**values))
 
             sess.add_all(rows)
             total += len(rows)
-            print(f"  pass_block {season} {position}: {len(rows)} rows")
+            print(f"  {category.row_label} {season} {position}: {len(rows)} rows")
 
     sess.commit()
     return total
 
 
+def ingest_pass_rush(sess: Session, data_dir: Path, seasons: list[int]) -> int:
+    return ingest_category(sess, data_dir, seasons, _PASS_RUSH)
+
+
+def ingest_pass_block(sess: Session, data_dir: Path, seasons: list[int]) -> int:
+    return ingest_category(sess, data_dir, seasons, _PASS_BLOCK)
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])