psp-preservation-tool/app/scanner.py

"""Scan directories for PSP .iso discs and read their identity + hashes.

A persistent scan cache (in the library DB) keyed by file size + mtime means
re-scanning only re-hashes files that actually changed; unchanged discs are
served from cache instantly.  Hashing of new/changed files runs in parallel.
"""
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from . import config, hasher, iso_reader

log = logging.getLogger("psp.scan")


def iter_discs(roots):
    seen = set()
    for root in roots:
        root = Path(root)
        if not root.exists():
            continue
        for path in sorted(root.rglob("*")):
            if not path.is_file():
                continue
            suffix = path.suffix.lower()
            if suffix in config.IGNORED_EXTENSIONS:
                continue
            if suffix not in config.DISC_EXTENSIONS:
                continue
            key = str(path.resolve())
            if key in seen:
                continue
            seen.add(key)
            yield path


def _process_disc(path, hash_files=True, cancel=None):
    stat = path.stat()
    try:
        identity = iso_reader.read_identity(path)
        serial = identity.get("serial")
        title = identity.get("title")
        error = None
    except Exception as exc:  # noqa: BLE001 - keep scanning on bad discs
        serial = None
        title = None
        error = f"{type(exc).__name__}: {exc}"
    if hash_files:
        hashes = hasher.hash_file(path, cancel=cancel)
    else:
        hashes = {"crc32": None, "md5": None, "sha1": None}
    return {
        "path": str(path),
        "name": path.name,
        "size": stat.st_size,
        "mtime_ns": stat.st_mtime_ns,
        "serial": serial,
        "title": title,
        "error": error,
        **hashes,
    }


def _cached_record(conn, path, stat, need_hashes=False):
    if conn is None:
        return None
    row = conn.execute(
        "SELECT size, mtime_ns, serial, title, error, crc32, md5, sha1 FROM files WHERE path = ?",
        (str(path),),
    ).fetchone()
    if row is None or row["size"] != stat.st_size or row["mtime_ns"] != stat.st_mtime_ns:
        return None
    if need_hashes and not row["crc32"]:
        return None
    return {
        "path": str(path),
        "name": path.name,
        "size": stat.st_size,
        "mtime_ns": stat.st_mtime_ns,
        "serial": row["serial"],
        "title": row["title"],
        "error": row["error"],
        "crc32": row["crc32"],
        "md5": row["md5"],
        "sha1": row["sha1"],
    }


def _upsert(conn, rec):
    conn.execute(
        """
        INSERT INTO files (path, name, size, mtime_ns, crc32, md5, sha1, serial, title, error, scanned_at)
        VALUES (?,?,?,?,?,?,?,?,?,?,datetime('now'))
        ON CONFLICT(path) DO UPDATE SET
            name=excluded.name, size=excluded.size, mtime_ns=excluded.mtime_ns,
            crc32=excluded.crc32, md5=excluded.md5, sha1=excluded.sha1,
            serial=excluded.serial, title=excluded.title, error=excluded.error,
            scanned_at=excluded.scanned_at
        """,
        (rec["path"], rec["name"], rec["size"], rec["mtime_ns"], rec["crc32"], rec["md5"], rec["sha1"], rec["serial"], rec["title"], rec["error"]),
    )


def rename_and_recache(conn, src, dst):
    """Rename a disc and carry its cache entry over so it isn't re-hashed."""
    src.rename(dst)
    if conn is not None:
        conn.execute(
            "UPDATE files SET path = ?, name = ? WHERE path = ?",
            (str(dst), dst.name, str(src)),
        )


def scan(roots, conn=None, workers=1, hash_files=True, cancel=None):
    pending = []
    cached_count = 0
    for path in iter_discs(roots):
        if cancel is not None and cancel.is_set():
            return
        stat = path.stat()
        cached = _cached_record(conn, path, stat, need_hashes=hash_files)
        if cached is not None:
            cached_count += 1
            log.info("cached  %s", path.name)
            yield cached
            continue
        pending.append(path)

    if cached_count:
        log.info("%d disc(s) served from cache", cached_count)
    if not pending:
        return

    log.info("processing %d disc(s) with %d worker(s)...", len(pending), workers)

    try:
        if workers and workers > 1:
            with ThreadPoolExecutor(max_workers=workers) as pool:
                futures = {pool.submit(_process_disc, p, hash_files, cancel): p for p in pending}
                for future in as_completed(futures):
                    if cancel is not None and cancel.is_set():
                        return
                    rec = future.result()
                    log.info("scanned %s", rec["name"])
                    if conn is not None:
                        _upsert(conn, rec)
                    yield rec
        else:
            for path in pending:
                if cancel is not None and cancel.is_set():
                    return
                rec = _process_disc(path, hash_files, cancel=cancel)
                log.info("scanned %s", rec["name"])
                if conn is not None:
                    _upsert(conn, rec)
                yield rec
    except hasher.HashCancelled:
        return