480 lines
18 KiB
Python
480 lines
18 KiB
Python
"""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())
|