Initial commit

scripts/populate_spells_from_scrolls.py

@@ -0,0 +1,146 @@
+"""Populate the spells table using scroll information from usable_items.
+
+Assumptions
+-----------
+1. `usable_items` table has (at least) the following columns:
+   - item_name (text)
+   - description (text)
+   - type_description (text) where scrolls have the value `SCROLL`
+2. The spell name can be derived from the item name by stripping common prefixes like
+   "Scroll of ", "Scroll: ", etc. This heuristic can be adjusted if necessary.
+3. Job / level information is embedded in the description in patterns like
+   "RDM Lv. 1", "WHM Lv.75", etc. Multiple jobs may appear in one description.
+4. The database URL is provided via the `DATABASE_URL` environment variable, e.g.
+   postgresql+psycopg2://user:password@host/dbname
+
+Usage
+-----
+    $ export DATABASE_URL=postgresql+psycopg2://...
+    $ python scripts/populate_spells_from_scrolls.py
+
+The script will insert new rows or update existing rows in the `spells` table.
+"""
+from __future__ import annotations
+
+import os
+import re
+from typing import Dict, List
+
+from sqlalchemy import MetaData, Table, select, create_engine, update, insert
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session
+
+# Job abbreviations to column mapping (identity mapping here but could differ)
+JOBS: List[str] = [
+    "run", "whm", "blm", "rdm", "pld", "drk", "brd", "nin", "smn", "cor", "sch", "geo",
+]
+
+# Regex to capture patterns like "RDM Lv. 1" or "RDM Lv.1" (space optional)
+JOB_LV_PATTERN = re.compile(r"([A-Z]{3})\s*Lv\.?\s*(\d+)")
+
+def _derive_spell_name(scroll_name: str) -> str:
+    """Convert a scroll item name to a spell name.
+
+    Examples
+    --------
+    >>> _derive_spell_name("Scroll of Fire")
+    'Fire'
+    >>> _derive_spell_name("Scroll: Cure IV")
+    'Cure IV'
+    """
+    # Remove common prefixes
+    prefixes = ["Scroll of ", "Scroll: ", "Scroll "]
+    for p in prefixes:
+        if scroll_name.startswith(p):
+            return scroll_name[len(p):].strip()
+    return scroll_name.strip()
+
+
+def _parse_job_levels(description: str) -> Dict[str, int]:
+    """Extract job-level mappings from a description string."""
+    mapping: Dict[str, int] = {}
+    for job, lvl in JOB_LV_PATTERN.findall(description):
+        job_l = job.lower()
+        if job_l in JOBS:
+            mapping[job_l] = int(lvl)
+    return mapping
+
+
+from pathlib import Path
+
+def _get_engine() -> Engine:
+    """Return SQLAlchemy engine using DATABASE_URL or db.conf."""
+    url = os.getenv("DATABASE_URL")
+    if not url:
+        # Attempt to build from db.conf at project root
+        conf_path = Path(__file__).resolve().parents[1] / "db.conf"
+        if not conf_path.exists():
+            raise RuntimeError("DATABASE_URL env var not set and db.conf not found")
+        cfg: Dict[str, str] = {}
+        with conf_path.open() as fh:
+            for line in fh:
+                line = line.strip()
+                if not line or line.startswith("#"):
+                    continue
+                if "=" not in line:
+                    continue
+                k, v = line.split("=", 1)
+                cfg[k.strip()] = v.strip().strip("'\"")  # remove quotes if any
+        try:
+            url = (
+                f"postgresql+psycopg2://{cfg['PSQL_USER']}:{cfg['PSQL_PASSWORD']}@"
+                f"{cfg['PSQL_HOST']}:{cfg.get('PSQL_PORT', '5432')}/{cfg['PSQL_DBNAME']}"
+            )
+        except KeyError as e:
+            raise RuntimeError(f"Missing key in db.conf: {e}")
+    return create_engine(url)
+
+
+def main() -> None:
+    engine = _get_engine()
+
+    meta = MetaData()
+    usable_items = Table("usable_items", meta, autoload_with=engine)
+    spells = Table("spells", meta, autoload_with=engine)
+
+    with Session(engine) as session:
+        # Fetch scroll items
+        scroll_rows = session.execute(
+            select(
+                usable_items.c.name,
+                usable_items.c.description
+            ).where(usable_items.c.type_description == "SCROLL")
+        ).all()
+
+        for name, description in scroll_rows:
+            spell_name = _derive_spell_name(name)
+            job_levels = _parse_job_levels(description or "")
+
+            # Build values dict w/ None default
+            values = {job: None for job in JOBS}
+            values.update(job_levels)
+            values["name"] = spell_name
+
+            # Upsert logic
+            existing = session.execute(
+                select(spells.c.name).where(spells.c.name == spell_name)
+            ).first()
+
+            if existing:
+                # Update existing row
+                stmt = (
+                    update(spells)
+                    .where(spells.c.name == spell_name)
+                    .values(**job_levels)
+                )
+                session.execute(stmt)
+            else:
+                stmt = insert(spells).values(**values)
+                session.execute(stmt)
+
+        session.commit()
+        print(f"Processed {len(scroll_rows)} scrolls. Spells table updated.")
+
+
+if __name__ == "__main__":
+    main()