psp-preservation-tool/app/matcher.py

"""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|cue)$", "", 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.

    Each entry is ``(normalized_name, significant_word_set, candidate)`` so the
    fuzzy step can cheaply skip candidates that share no significant word.
    """
    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 = [
        (norm, frozenset(w for w in norm.split() if len(w) >= 3), c)
        for c in cands
        if (norm := _normalize(c["name"] or ""))
    ]
    _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)


_EUROPEAN_REGIONS = {
    "europe", "germany", "france", "spain", "italy", "russia", "netherlands",
    "portugal", "poland", "united kingdom", "greece", "norway", "sweden",
    "denmark", "finland", "austria", "belgium", "switzerland",
}


def region_matches(region, prefer):
    """True if ``region`` satisfies the user's preferred-region hint."""
    if not prefer or not region:
        return False
    tokens = {t.strip().lower() for t in str(region).split(",") if t.strip()}
    if not tokens:
        return False
    prefer = prefer.strip().lower()
    if prefer in tokens:
        return True
    if prefer == "europe":
        return bool(tokens & _EUROPEAN_REGIONS)
    return False


def apply_region_preference(candidates, prefer):
    """Narrow ``candidates`` to the preferred region (fall back to all)."""
    if not prefer:
        return candidates
    filtered = [c for c in candidates if region_matches(c.get("region"), prefer)]
    return filtered or candidates


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, prefer_region=None):
    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
            if prefer_region:
                filtered = [c for c in candidates if region_matches(c.get("region"), prefer_region)]
                if len(filtered) == 1:
                    return "region", filtered[0]
            return "unknown", None

    # 4. Fuzzy fallback (filename + PARAM.SFO title), size-disambiguated.
    disk_norm = _normalize(record["name"])
    title_norm = _normalize(record.get("title") or "")

    # Cheap pre-filter: only run the expensive difflib comparison against
    # candidates sharing at least one significant (>=3 char) word with the
    # disc's filename or PARAM.SFO title.
    target_words = frozenset(w for w in disk_norm.split() if len(w) >= 3) | \
        frozenset(w for w in title_norm.split() if len(w) >= 3)
    if not target_words:
        return "unknown", None

    scored = []
    for cand_norm, cand_words, cand in _all_release_candidates(conn):
        if not (target_words & cand_words):
            continue
        ratio = SequenceMatcher(None, disk_norm, cand_norm).ratio()
        if title_norm:
            ratio = max(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