Add spell data helper and organize screenshots
This commit is contained in:
Executable
+8
@@ -0,0 +1,8 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Command-line entry point for the local EQL spell-data index."""
|
||||
|
||||
from eql_spells import main
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,479 @@
|
||||
"""Build and query a local EverQuest Legends spell-data index.
|
||||
|
||||
The client files are the source of truth. This module only creates a local,
|
||||
rebuildable SQLite cache and deliberately does not assign meaning to the
|
||||
undocumented positional fields in spells_us.txt.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
|
||||
SCHEMA_VERSION = 2
|
||||
DEFAULT_GAME_PATH = Path("/home/pleb/Games/EverQuestLegends")
|
||||
SOURCE_FILENAMES = ("spells_us.txt", "spells_us_str.txt", "dbstr_us.txt")
|
||||
MESSAGE_COLUMNS = (
|
||||
"caster_me",
|
||||
"caster_other",
|
||||
"casted_me",
|
||||
"casted_other",
|
||||
"spell_gone",
|
||||
)
|
||||
|
||||
|
||||
class EqlSpellsError(Exception):
|
||||
"""An expected user-facing command error."""
|
||||
|
||||
|
||||
class UsageError(EqlSpellsError):
|
||||
"""The command line is invalid."""
|
||||
|
||||
|
||||
class DataError(EqlSpellsError):
|
||||
"""The EQL source files or index are invalid."""
|
||||
|
||||
|
||||
class JSONArgumentParser(argparse.ArgumentParser):
|
||||
def error(self, message: str) -> None:
|
||||
raise UsageError(message)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SourceInfo:
|
||||
game_path: Path
|
||||
fingerprint: dict[str, Any]
|
||||
|
||||
|
||||
def default_cache_path() -> Path:
|
||||
cache_home = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache"))
|
||||
return cache_home / "eqclient-conf" / "eql-spells.sqlite"
|
||||
|
||||
|
||||
def resolve_game_path(value: str | None) -> Path:
|
||||
if value:
|
||||
return Path(value).expanduser().resolve()
|
||||
if os.environ.get("EQL_GAME_PATH"):
|
||||
return Path(os.environ["EQL_GAME_PATH"]).expanduser().resolve()
|
||||
return DEFAULT_GAME_PATH
|
||||
|
||||
|
||||
def source_info(game_path: Path) -> SourceInfo:
|
||||
game_path = game_path.resolve()
|
||||
if not game_path.is_dir():
|
||||
raise DataError(
|
||||
f"EverQuest Legends directory not found: {game_path}. "
|
||||
"Set --game-path or EQL_GAME_PATH."
|
||||
)
|
||||
|
||||
sources: list[dict[str, Any]] = []
|
||||
for filename in SOURCE_FILENAMES:
|
||||
path = game_path / filename
|
||||
if not path.is_file():
|
||||
raise DataError(f"Required spell data file not found: {path}")
|
||||
try:
|
||||
stat = path.stat()
|
||||
crc_path = Path(f"{path}:crc")
|
||||
crc = crc_path.read_text(encoding="ascii").strip() if crc_path.is_file() else None
|
||||
except OSError as error:
|
||||
raise DataError(f"Unable to inspect {path}: {error}") from error
|
||||
sources.append(
|
||||
{
|
||||
"name": filename,
|
||||
"path": str(path),
|
||||
"size": stat.st_size,
|
||||
"mtime_ns": stat.st_mtime_ns,
|
||||
"crc": crc,
|
||||
}
|
||||
)
|
||||
fingerprint = {"schema_version": SCHEMA_VERSION, "sources": sources}
|
||||
return SourceInfo(game_path=game_path, fingerprint=fingerprint)
|
||||
|
||||
|
||||
def connect(path: Path) -> sqlite3.Connection:
|
||||
connection = sqlite3.connect(path)
|
||||
connection.row_factory = sqlite3.Row
|
||||
return connection
|
||||
|
||||
|
||||
def create_schema(connection: sqlite3.Connection) -> None:
|
||||
connection.executescript(
|
||||
"""
|
||||
PRAGMA foreign_keys = ON;
|
||||
CREATE TABLE metadata (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE spells (
|
||||
spell_id INTEGER PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
name_fold TEXT NOT NULL,
|
||||
raw_line TEXT NOT NULL,
|
||||
fields_json TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE spell_messages (
|
||||
spell_id INTEGER PRIMARY KEY REFERENCES spells(spell_id),
|
||||
caster_me TEXT NOT NULL DEFAULT '',
|
||||
caster_other TEXT NOT NULL DEFAULT '',
|
||||
casted_me TEXT NOT NULL DEFAULT '',
|
||||
casted_other TEXT NOT NULL DEFAULT '',
|
||||
spell_gone TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
CREATE TABLE descriptions (
|
||||
spell_id INTEGER PRIMARY KEY REFERENCES spells(spell_id),
|
||||
description TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX spells_name_fold_idx ON spells(name_fold);
|
||||
CREATE VIRTUAL TABLE spell_fts USING fts5(name, description, tokenize = 'porter unicode61');
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def source_lines(path: Path) -> Iterable[tuple[int, str]]:
|
||||
try:
|
||||
with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
|
||||
for line_number, line in enumerate(handle, start=1):
|
||||
yield line_number, line.rstrip("\r\n")
|
||||
except OSError as error:
|
||||
raise DataError(f"Unable to read {path}: {error}") from error
|
||||
|
||||
|
||||
def parse_spell_rows(path: Path) -> list[tuple[int, str, str, str, str]]:
|
||||
rows: list[tuple[int, str, str, str, str]] = []
|
||||
seen: set[int] = set()
|
||||
for line_number, line in source_lines(path):
|
||||
if not line:
|
||||
continue
|
||||
fields = line.split("^")
|
||||
if len(fields) < 2 or not fields[1]:
|
||||
raise DataError(f"Malformed spell row in {path}:{line_number}")
|
||||
try:
|
||||
spell_id = int(fields[0])
|
||||
except ValueError as error:
|
||||
raise DataError(f"Malformed spell ID in {path}:{line_number}") from error
|
||||
if spell_id in seen:
|
||||
raise DataError(f"Duplicate spell ID {spell_id} in {path}:{line_number}")
|
||||
seen.add(spell_id)
|
||||
rows.append((spell_id, fields[1], fields[1].casefold(), line, json.dumps(fields)))
|
||||
if not rows:
|
||||
raise DataError(f"No spell rows found in {path}")
|
||||
return rows
|
||||
|
||||
|
||||
def parse_messages(path: Path) -> list[tuple[int, str, str, str, str, str]]:
|
||||
rows: list[tuple[int, str, str, str, str, str]] = []
|
||||
seen: set[int] = set()
|
||||
for line_number, line in source_lines(path):
|
||||
if not line or line.startswith("#SPELLINDEX^"):
|
||||
continue
|
||||
fields = line.split("^")
|
||||
if len(fields) < 6:
|
||||
raise DataError(f"Malformed message row in {path}:{line_number}")
|
||||
try:
|
||||
spell_id = int(fields[0])
|
||||
except ValueError as error:
|
||||
raise DataError(f"Malformed message spell ID in {path}:{line_number}") from error
|
||||
if spell_id in seen:
|
||||
raise DataError(f"Duplicate message spell ID {spell_id} in {path}:{line_number}")
|
||||
seen.add(spell_id)
|
||||
rows.append((spell_id, *fields[1:6]))
|
||||
return rows
|
||||
|
||||
|
||||
def parse_descriptions(path: Path, spell_ids: set[int]) -> list[tuple[int, str]]:
|
||||
rows: list[tuple[int, str]] = []
|
||||
seen: set[int] = set()
|
||||
for line_number, line in source_lines(path):
|
||||
if not line:
|
||||
continue
|
||||
fields = line.split("^")
|
||||
if len(fields) < 3 or fields[1] != "6":
|
||||
continue
|
||||
try:
|
||||
spell_id = int(fields[0])
|
||||
except ValueError as error:
|
||||
raise DataError(f"Malformed description spell ID in {path}:{line_number}") from error
|
||||
# dbstr_us.txt is a shared string table. Type-6 entries outside the
|
||||
# client spell table belong to other UI data and are not spell rows.
|
||||
if spell_id not in spell_ids:
|
||||
continue
|
||||
if spell_id in seen:
|
||||
raise DataError(f"Duplicate spell description for ID {spell_id} in {path}:{line_number}")
|
||||
seen.add(spell_id)
|
||||
rows.append((spell_id, fields[2]))
|
||||
return rows
|
||||
|
||||
|
||||
def validate_joins(spell_ids: set[int], messages: list[tuple[int, str, str, str, str, str]]) -> None:
|
||||
orphan_messages = [row[0] for row in messages if row[0] not in spell_ids]
|
||||
if orphan_messages:
|
||||
raise DataError(f"Message row references unknown spell ID {orphan_messages[0]}")
|
||||
|
||||
|
||||
def build_index(db_path: Path, info: SourceInfo) -> dict[str, int]:
|
||||
spells_path, messages_path, descriptions_path = (
|
||||
info.game_path / filename for filename in SOURCE_FILENAMES
|
||||
)
|
||||
spells = parse_spell_rows(spells_path)
|
||||
messages = parse_messages(messages_path)
|
||||
spell_ids = {row[0] for row in spells}
|
||||
descriptions = parse_descriptions(descriptions_path, spell_ids)
|
||||
validate_joins(spell_ids, messages)
|
||||
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
descriptor, temporary_name = tempfile.mkstemp(
|
||||
prefix=f".{db_path.name}.", suffix=".tmp", dir=db_path.parent
|
||||
)
|
||||
os.close(descriptor)
|
||||
temporary_path = Path(temporary_name)
|
||||
try:
|
||||
connection = connect(temporary_path)
|
||||
try:
|
||||
create_schema(connection)
|
||||
with connection:
|
||||
connection.executemany(
|
||||
"INSERT INTO spells (spell_id, name, name_fold, raw_line, fields_json) VALUES (?, ?, ?, ?, ?)",
|
||||
spells,
|
||||
)
|
||||
connection.executemany(
|
||||
"""INSERT INTO spell_messages
|
||||
(spell_id, caster_me, caster_other, casted_me, casted_other, spell_gone)
|
||||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||||
messages,
|
||||
)
|
||||
connection.executemany(
|
||||
"INSERT INTO descriptions (spell_id, description) VALUES (?, ?)", descriptions,
|
||||
)
|
||||
connection.execute(
|
||||
"""INSERT INTO spell_fts (rowid, name, description)
|
||||
SELECT spells.spell_id, spells.name, COALESCE(descriptions.description, '')
|
||||
FROM spells LEFT JOIN descriptions USING (spell_id)"""
|
||||
)
|
||||
connection.execute(
|
||||
"INSERT INTO metadata (key, value) VALUES (?, ?)",
|
||||
("source", json.dumps({"game_path": str(info.game_path), "fingerprint": info.fingerprint}, sort_keys=True)),
|
||||
)
|
||||
count = connection.execute("SELECT count(*) FROM spells").fetchone()[0]
|
||||
fts_count = connection.execute("SELECT count(*) FROM spell_fts").fetchone()[0]
|
||||
if count != len(spells) or fts_count != len(spells):
|
||||
raise DataError("Built index failed row-count validation")
|
||||
finally:
|
||||
connection.close()
|
||||
os.replace(temporary_path, db_path)
|
||||
except Exception:
|
||||
temporary_path.unlink(missing_ok=True)
|
||||
raise
|
||||
return {"spells": len(spells), "messages": len(messages), "descriptions": len(descriptions)}
|
||||
|
||||
|
||||
def stored_source(db_path: Path) -> dict[str, Any] | None:
|
||||
if not db_path.is_file():
|
||||
return None
|
||||
try:
|
||||
connection = connect(db_path)
|
||||
try:
|
||||
row = connection.execute("SELECT value FROM metadata WHERE key = 'source'").fetchone()
|
||||
finally:
|
||||
connection.close()
|
||||
return json.loads(row[0]) if row else None
|
||||
except (OSError, sqlite3.Error, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def ensure_index(db_path: Path, info: SourceInfo) -> tuple[bool, dict[str, int] | None]:
|
||||
expected = {"game_path": str(info.game_path), "fingerprint": info.fingerprint}
|
||||
if stored_source(db_path) == expected:
|
||||
return False, None
|
||||
return True, build_index(db_path, info)
|
||||
|
||||
|
||||
def source_metadata(connection: sqlite3.Connection) -> dict[str, Any]:
|
||||
row = connection.execute("SELECT value FROM metadata WHERE key = 'source'").fetchone()
|
||||
if row is None:
|
||||
raise DataError("Index metadata is missing")
|
||||
return json.loads(row[0])
|
||||
|
||||
|
||||
def spell_from_row(row: sqlite3.Row, include_raw: bool = False) -> dict[str, Any]:
|
||||
messages = {column: row[column] for column in MESSAGE_COLUMNS}
|
||||
result: dict[str, Any] = {
|
||||
"id": row["spell_id"],
|
||||
"name": row["name"],
|
||||
"description": row["description"],
|
||||
"messages": messages,
|
||||
}
|
||||
if include_raw:
|
||||
result["raw"] = {"line": row["raw_line"], "fields": json.loads(row["fields_json"])}
|
||||
return result
|
||||
|
||||
|
||||
SELECT_SPELL = """
|
||||
SELECT spells.spell_id, spells.name, spells.raw_line, spells.fields_json,
|
||||
descriptions.description,
|
||||
COALESCE(spell_messages.caster_me, '') AS caster_me,
|
||||
COALESCE(spell_messages.caster_other, '') AS caster_other,
|
||||
COALESCE(spell_messages.casted_me, '') AS casted_me,
|
||||
COALESCE(spell_messages.casted_other, '') AS casted_other,
|
||||
COALESCE(spell_messages.spell_gone, '') AS spell_gone
|
||||
FROM spells
|
||||
LEFT JOIN descriptions USING (spell_id)
|
||||
LEFT JOIN spell_messages USING (spell_id)
|
||||
"""
|
||||
|
||||
|
||||
def get_spell(connection: sqlite3.Connection, value: str, include_raw: bool) -> dict[str, Any]:
|
||||
if value.isdecimal():
|
||||
row = connection.execute(SELECT_SPELL + " WHERE spells.spell_id = ?", (int(value),)).fetchone()
|
||||
if row is None:
|
||||
raise DataError(f"No spell found with ID {value}")
|
||||
return spell_from_row(row, include_raw)
|
||||
|
||||
rows = connection.execute(
|
||||
SELECT_SPELL + " WHERE spells.name_fold = ? ORDER BY spells.spell_id", (value.casefold(),)
|
||||
).fetchall()
|
||||
if not rows:
|
||||
raise DataError(f"No spell found with exact name {value!r}")
|
||||
if len(rows) > 1:
|
||||
ids = [row["spell_id"] for row in rows]
|
||||
raise DataError(f"Exact name {value!r} is ambiguous; matching IDs: {ids}")
|
||||
return spell_from_row(rows[0], include_raw)
|
||||
|
||||
|
||||
def find_spells(connection: sqlite3.Connection, text: str, limit: int) -> list[dict[str, Any]]:
|
||||
escaped = text.casefold().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
||||
rows = connection.execute(
|
||||
"""SELECT spells.spell_id, spells.name, descriptions.description
|
||||
FROM spells LEFT JOIN descriptions USING (spell_id)
|
||||
WHERE spells.name_fold LIKE ? ESCAPE '\\'
|
||||
ORDER BY spells.name_fold, spells.spell_id LIMIT ?""",
|
||||
(f"%{escaped}%", limit),
|
||||
).fetchall()
|
||||
return [{"id": row["spell_id"], "name": row["name"], "description": row["description"]} for row in rows]
|
||||
|
||||
|
||||
def fts_query(text: str) -> str:
|
||||
terms = re.findall(r"[^\W_]+", text, flags=re.UNICODE)
|
||||
if not terms:
|
||||
raise UsageError("search text must contain at least one letter or number")
|
||||
return " AND ".join(f'"{term}"' for term in terms)
|
||||
|
||||
|
||||
def search_spells(connection: sqlite3.Connection, text: str, limit: int) -> list[dict[str, Any]]:
|
||||
rows = connection.execute(
|
||||
"""SELECT spells.spell_id, spells.name, descriptions.description, bm25(spell_fts) AS score
|
||||
FROM spell_fts
|
||||
JOIN spells ON spells.spell_id = spell_fts.rowid
|
||||
LEFT JOIN descriptions USING (spell_id)
|
||||
WHERE spell_fts MATCH ?
|
||||
ORDER BY score, spells.name_fold, spells.spell_id LIMIT ?""",
|
||||
(fts_query(text), limit),
|
||||
).fetchall()
|
||||
return [
|
||||
{"id": row["spell_id"], "name": row["name"], "description": row["description"]}
|
||||
for row in rows
|
||||
]
|
||||
|
||||
|
||||
def parse_limit(value: str) -> int:
|
||||
try:
|
||||
limit = int(value)
|
||||
except ValueError as error:
|
||||
raise argparse.ArgumentTypeError("limit must be an integer") from error
|
||||
if not 1 <= limit <= 100:
|
||||
raise argparse.ArgumentTypeError("limit must be between 1 and 100")
|
||||
return limit
|
||||
|
||||
|
||||
def parser() -> argparse.ArgumentParser:
|
||||
argument_parser = JSONArgumentParser(prog="eql-spells", description=__doc__)
|
||||
argument_parser.add_argument("--game-path", help="EverQuest Legends install directory")
|
||||
argument_parser.add_argument("--db", type=Path, help="SQLite index path (default: user cache)")
|
||||
subparsers = argument_parser.add_subparsers(dest="command", required=True)
|
||||
subparsers.add_parser("build", help="force a complete index rebuild")
|
||||
|
||||
get_parser = subparsers.add_parser("get", help="look up a spell by ID or exact name")
|
||||
get_parser.add_argument("value", help="numeric spell ID or exact spell name")
|
||||
get_parser.add_argument("--raw", action="store_true", help="include original client row and fields")
|
||||
get_parser.add_argument("--text", action="store_true", help="render compact text instead of JSON")
|
||||
|
||||
for command, help_text in (("find", "find spell names"), ("search", "search spell names and descriptions")):
|
||||
query_parser = subparsers.add_parser(command, help=help_text)
|
||||
query_parser.add_argument("text")
|
||||
query_parser.add_argument("--limit", type=parse_limit, default=20)
|
||||
query_parser.add_argument("--text", dest="as_text", action="store_true", help="render compact text instead of JSON")
|
||||
return argument_parser
|
||||
|
||||
|
||||
def text_result(command: str, payload: dict[str, Any]) -> str:
|
||||
if command == "get":
|
||||
spell = payload["spell"]
|
||||
lines = [f"{spell['id']}\t{spell['name']}"]
|
||||
if spell["description"]:
|
||||
lines.append(spell["description"])
|
||||
for label, value in spell["messages"].items():
|
||||
if value:
|
||||
lines.append(f"{label}: {value}")
|
||||
return "\n".join(lines)
|
||||
return "\n".join(
|
||||
f"{result['id']}\t{result['name']}" + (f"\t{result['description']}" if result["description"] else "")
|
||||
for result in payload["results"]
|
||||
)
|
||||
|
||||
|
||||
def emit_json(value: dict[str, Any], stream: Any = sys.stdout) -> None:
|
||||
print(json.dumps(value, ensure_ascii=False, sort_keys=True), file=stream)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
try:
|
||||
arguments = parser().parse_args(argv)
|
||||
game_path = resolve_game_path(arguments.game_path)
|
||||
db_path = (arguments.db.expanduser() if arguments.db else default_cache_path()).resolve()
|
||||
info = source_info(game_path)
|
||||
|
||||
if arguments.command == "build":
|
||||
counts = build_index(db_path, info)
|
||||
emit_json({"built": counts, "db": str(db_path), "source": {"game_path": str(info.game_path), "fingerprint": info.fingerprint}})
|
||||
return 0
|
||||
|
||||
rebuilt, _ = ensure_index(db_path, info)
|
||||
connection = connect(db_path)
|
||||
try:
|
||||
source = source_metadata(connection)
|
||||
if arguments.command == "get":
|
||||
payload = {"source": source, "spell": get_spell(connection, arguments.value, arguments.raw)}
|
||||
as_text = arguments.text
|
||||
elif arguments.command == "find":
|
||||
payload = {"source": source, "results": find_spells(connection, arguments.text, arguments.limit)}
|
||||
as_text = arguments.as_text
|
||||
else:
|
||||
payload = {"source": source, "results": search_spells(connection, arguments.text, arguments.limit)}
|
||||
as_text = arguments.as_text
|
||||
if rebuilt:
|
||||
payload["index_rebuilt"] = True
|
||||
if as_text:
|
||||
print(text_result(arguments.command, payload))
|
||||
else:
|
||||
emit_json(payload)
|
||||
finally:
|
||||
connection.close()
|
||||
return 0
|
||||
except UsageError as error:
|
||||
emit_json({"error": {"code": "usage", "message": str(error)}}, sys.stderr)
|
||||
return 2
|
||||
except (DataError, OSError, sqlite3.Error) as error:
|
||||
emit_json({"error": {"code": "data", "message": str(error)}}, sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user