147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
"""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()
|