197 lines
6.6 KiB
Python
197 lines
6.6 KiB
Python
"""Match scanned discs against the catalog: hash -> serial -> fuzzy name.
|
|
|
|
When hashes are unavailable (quick scans), serial matches are disambiguated by
|
|
the disc's exact size, and fuzzy matches are narrowed the same way.
|
|
"""
|
|
import json
|
|
import re
|
|
from difflib import SequenceMatcher
|
|
|
|
from . import config
|
|
|
|
|
|
_ARTICLE_RE = re.compile(r"^(.+),\s*(the|a|an)(?:\s+-\s+(.+))?$")
|
|
|
|
|
|
def _normalize(name):
|
|
name = name.lower()
|
|
name = re.sub(r"\.(iso|cso|pbp|img|bin)$", "", name) # strip disc extension
|
|
name = re.sub(r"^\s*\d+\s*-\s*", "", name) # legacy No-Intro "#### - Name" catalog number
|
|
name = re.sub(r"\([^)]*\)", " ", name) # parenthetical tags
|
|
name = " ".join(name.split())
|
|
m = _ARTICLE_RE.match(name)
|
|
if m:
|
|
name = f"{m.group(2)} {m.group(1)}"
|
|
if m.group(3):
|
|
name += f" {m.group(3)}"
|
|
name = re.sub(r"[^a-z0-9]+", " ", name)
|
|
return " ".join(name.split())
|
|
|
|
|
|
_supplement_cache = None
|
|
_releases_cache = {} # id(conn) -> [(normalized_name, candidate)] for the fuzzy step
|
|
|
|
|
|
def clear_cache():
|
|
"""Drop cached release lists (call after re-ingesting a DAT)."""
|
|
_releases_cache.clear()
|
|
|
|
|
|
def _all_release_candidates(conn):
|
|
"""Catalog releases + supplement entries with pre-normalized names."""
|
|
key = id(conn)
|
|
cands = _releases_cache.get(key)
|
|
if cands is not None:
|
|
return cands
|
|
cands = [
|
|
{
|
|
"id": row["id"], "name": row["name"], "region": row["region"],
|
|
"serial": row["serial"], "system": row["system"],
|
|
}
|
|
for row in conn.execute(
|
|
"SELECT r.id, r.name, r.region, r.serial, d.system FROM releases r JOIN dats d ON d.id = r.dat_id"
|
|
)
|
|
]
|
|
cands.extend(_supplement_release(e) for e in _load_supplement())
|
|
cands = [(_normalize(c["name"]), c) for c in cands]
|
|
_releases_cache[key] = cands
|
|
return cands
|
|
|
|
|
|
def _load_supplement():
|
|
global _supplement_cache
|
|
if _supplement_cache is None:
|
|
path = config.DATA_DIR / "supplement.json"
|
|
if path.exists():
|
|
_supplement_cache = json.loads(path.read_text(encoding="utf-8"))
|
|
else:
|
|
_supplement_cache = []
|
|
return _supplement_cache
|
|
|
|
|
|
def _supplement_release(entry):
|
|
return {
|
|
"id": None,
|
|
"name": entry.get("name"),
|
|
"region": entry.get("region"),
|
|
"serial": entry.get("serial"),
|
|
"system": entry.get("system"),
|
|
"size": entry.get("size"),
|
|
}
|
|
|
|
|
|
def _narrow(conn, candidates, size):
|
|
"""Prefer candidates whose known disc size matches the on-disk file size."""
|
|
if size is None or len(candidates) <= 1:
|
|
return candidates
|
|
matched = []
|
|
for cand in candidates:
|
|
if cand.get("id") is not None:
|
|
row = conn.execute(
|
|
"SELECT 1 FROM roms WHERE release_id = ? AND size = ? LIMIT 1",
|
|
(cand["id"], size),
|
|
).fetchone()
|
|
if row:
|
|
matched.append(cand)
|
|
elif cand.get("size") == size:
|
|
matched.append(cand)
|
|
return matched or candidates
|
|
|
|
|
|
def normalize_name(name):
|
|
"""Public wrapper around the fuzzy normalizer (used by the Store)."""
|
|
return _normalize(name)
|
|
|
|
|
|
def serial_candidates(conn, record):
|
|
"""Return all catalog releases matching the disc's serial (size-narrowed)."""
|
|
serial = record.get("serial")
|
|
if not serial:
|
|
return []
|
|
rows = conn.execute(
|
|
"SELECT r.id, r.name, r.region, r.serial, d.system FROM releases r "
|
|
"JOIN dats d ON d.id = r.dat_id WHERE upper(r.serial) = upper(?)",
|
|
(serial,),
|
|
).fetchall()
|
|
if not rows:
|
|
rows = conn.execute(
|
|
"SELECT r.id, r.name, r.region, r.serial, d.system FROM roms ro "
|
|
"JOIN releases r ON r.id = ro.release_id "
|
|
"JOIN dats d ON d.id = r.dat_id WHERE upper(ro.serial) = upper(?)",
|
|
(serial,),
|
|
).fetchall()
|
|
return _narrow(conn, [dict(r) for r in rows], record.get("size"))
|
|
|
|
|
|
def match(conn, record):
|
|
size = record.get("size")
|
|
|
|
# 1. Hash is the most precise key; serials are shared across revisions.
|
|
for field in ("crc32", "sha1", "md5"):
|
|
value = record.get(field)
|
|
if not value:
|
|
continue
|
|
row = conn.execute(
|
|
f"SELECT r.id, r.name, r.region, r.serial, d.system FROM roms ro "
|
|
f"JOIN releases r ON r.id = ro.release_id "
|
|
f"JOIN dats d ON d.id = r.dat_id "
|
|
f"WHERE upper(ro.{field}) = ? LIMIT 1",
|
|
(value.upper(),),
|
|
).fetchone()
|
|
if row is not None:
|
|
return f"hash:{field}", dict(row)
|
|
|
|
# 2. Local supplement catalog (dumps not present in the DATs).
|
|
for entry in _load_supplement():
|
|
for field in ("crc32", "sha1", "md5"):
|
|
value = entry.get(field)
|
|
if value and record.get(field) and value.upper() == record[field].upper():
|
|
return f"supplement:{field}", _supplement_release(entry)
|
|
|
|
# 3. Serial (size-disambiguated for revisions that share a serial).
|
|
serial = record.get("serial")
|
|
if serial:
|
|
candidates = serial_candidates(conn, record)
|
|
if len(candidates) == 1:
|
|
return "serial", candidates[0]
|
|
if len(candidates) > 1:
|
|
saved = conn.execute(
|
|
"SELECT release_id FROM choices WHERE serial = ? AND size = ?",
|
|
(serial.upper(), size),
|
|
).fetchone()
|
|
if saved is not None:
|
|
for cand in candidates:
|
|
if cand["id"] == saved["release_id"]:
|
|
return "choice", cand
|
|
return "unknown", None
|
|
|
|
# 4. Fuzzy fallback (filename + PARAM.SFO title), size-disambiguated.
|
|
disk_norm = _normalize(record["name"])
|
|
title_norm = _normalize(record.get("title") or "")
|
|
|
|
scored = []
|
|
for cand_norm, cand in _all_release_candidates(conn):
|
|
ratio = max(
|
|
SequenceMatcher(None, disk_norm, cand_norm).ratio(),
|
|
SequenceMatcher(None, title_norm, cand_norm).ratio(),
|
|
)
|
|
scored.append((ratio, cand))
|
|
best_ratio = max((r for r, _ in scored), default=0.0)
|
|
if best_ratio < 0.9:
|
|
return "unknown", None
|
|
top = [cand for ratio, cand in scored if ratio >= best_ratio - 0.001]
|
|
top = _narrow(conn, top, size)
|
|
# The same disc can exist in multiple catalogs (e.g. No-Intro + Redump);
|
|
# collapse identical entries so they don't read as an ambiguous match.
|
|
seen = set()
|
|
deduped = []
|
|
for cand in top:
|
|
key = (cand.get("name"), size)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
deduped.append(cand)
|
|
top = deduped
|
|
if len(top) == 1:
|
|
return f"fuzzy:{best_ratio:.2f}", top[0]
|
|
return "unknown", None
|