168 lines
4.8 KiB
Python
168 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Load <Craft>_v2.csv into PostgreSQL.
|
|
|
|
Usage:
|
|
python load_recipes_v2_to_db.py <CRAFT>
|
|
|
|
The script drop-creates table `recipes_<craft>` (lowercased) with the generic
|
|
v2 schema, then bulk-loads from the CSV produced by recipes_to_csv_v2.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import csv
|
|
import json
|
|
import pathlib
|
|
import re
|
|
from typing import Dict
|
|
|
|
import asyncpg
|
|
|
|
PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
CONF_PATH = PROJECT_ROOT / "db.conf"
|
|
DATASETS_DIR = PROJECT_ROOT / "datasets"
|
|
|
|
RE_KEY = re.compile(r"^([A-Z0-9_]+)=(.*)$")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Category mapping
|
|
# ---------------------------------------------------------------------------
|
|
CATEGORY_RANGES = [
|
|
("Amateur", 1, 10),
|
|
("Recruit", 8, 20),
|
|
("Initiate", 18, 30),
|
|
("Novice", 28, 40),
|
|
("Apprentice", 38, 50),
|
|
("Journeyman", 48, 60),
|
|
("Craftsman", 58, 70),
|
|
("Artisan", 68, 80),
|
|
("Adept", 78, 90),
|
|
("Veteran", 88, 100),
|
|
("Expert", 98, 110),
|
|
("Authority", 111, 120),
|
|
]
|
|
|
|
def category_for_level(level: int) -> str:
|
|
for name, lo, hi in CATEGORY_RANGES:
|
|
if lo <= level <= hi:
|
|
return name
|
|
return "Unknown"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def parse_db_conf(path: pathlib.Path) -> Dict[str, str]:
|
|
conf: Dict[str, str] = {}
|
|
for line in path.read_text().splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith('#'):
|
|
continue
|
|
if (m := RE_KEY.match(line)):
|
|
k, v = m.group(1), m.group(2).strip().strip("'\"")
|
|
conf[k] = v
|
|
return conf
|
|
|
|
|
|
async def recreate_table(conn: asyncpg.Connection, craft: str):
|
|
table = f"recipes_{craft.lower()}"
|
|
await conn.execute(f"DROP TABLE IF EXISTS {table};")
|
|
await conn.execute(
|
|
f"""
|
|
CREATE TABLE {table} (
|
|
id SERIAL PRIMARY KEY,
|
|
category TEXT NOT NULL,
|
|
level INT NOT NULL,
|
|
subcrafts JSONB,
|
|
name TEXT NOT NULL,
|
|
crystal TEXT NOT NULL,
|
|
key_item TEXT,
|
|
ingredients JSONB,
|
|
hq_yields JSONB
|
|
);
|
|
"""
|
|
)
|
|
|
|
|
|
async def insert_csv(conn: asyncpg.Connection, craft: str, csv_path: pathlib.Path):
|
|
table = f"recipes_{craft.lower()}"
|
|
with csv_path.open(encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
records = []
|
|
for row in reader:
|
|
records.append(
|
|
(
|
|
category_for_level(int(row["level"])),
|
|
int(row["level"]),
|
|
json.dumps(json.loads(row["subcrafts"] or "[]")),
|
|
row["name"],
|
|
row["crystal"],
|
|
row["key_item"] or None,
|
|
json.dumps(json.loads(row["ingredients"] or "[]")),
|
|
json.dumps(json.loads(row["hq_yields"] or "[]")),
|
|
)
|
|
)
|
|
await conn.copy_records_to_table(
|
|
table,
|
|
records=records,
|
|
columns=[
|
|
"category",
|
|
"level",
|
|
"subcrafts",
|
|
"name",
|
|
"crystal",
|
|
"key_item",
|
|
"ingredients",
|
|
"hq_yields",
|
|
],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def process_craft(conn: asyncpg.Connection, craft: str):
|
|
csv_path = DATASETS_DIR / f"{craft}_v2.csv"
|
|
if not csv_path.exists():
|
|
print(f"CSV not found for {craft}, skipping.")
|
|
return
|
|
await recreate_table(conn, craft)
|
|
await insert_csv(conn, craft, csv_path)
|
|
print(f"Loaded {craft} -> recipes_{craft.lower()} table.")
|
|
|
|
|
|
async def main_async(craft: str | None):
|
|
conf = parse_db_conf(CONF_PATH)
|
|
conn = await asyncpg.connect(
|
|
host=conf["PSQL_HOST"],
|
|
port=int(conf["PSQL_PORT"]),
|
|
user=conf["PSQL_USER"],
|
|
password=conf["PSQL_PASSWORD"],
|
|
database=conf["PSQL_DBNAME"],
|
|
)
|
|
try:
|
|
if craft:
|
|
await process_craft(conn, craft)
|
|
else:
|
|
# Scan datasets dir
|
|
for p in DATASETS_DIR.glob("*_v2.csv"):
|
|
c = p.stem.replace("_v2", "")
|
|
await process_craft(conn, c)
|
|
finally:
|
|
await conn.close()
|
|
|
|
|
|
def main() -> None:
|
|
p = argparse.ArgumentParser(description="Load <Craft>_v2.csv into DB")
|
|
p.add_argument("craft", nargs="?", help="Craft name; if omitted, load all *_v2.csv files")
|
|
args = p.parse_args()
|
|
craft_arg = args.craft.strip() if args.craft else None
|
|
asyncio.run(main_async(craft_arg))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|