scanner.py - dev - vaskaz/flexin_flow

scanner.py
100 строк3 KB
vaskaz
newborn
10 июн 2026, 00:16
10 июн 2026, 00:16dc9da65
"""Filesystem scanner for game ROM files."""

import os
import time
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

from database import RomFile, fast_hash, insert_rom, get_all_roms, remove_stale
from metadata import detect_platform, clean_title, format_size, is_rom_file, get_all_extensions


def scan_directory(root_dir: str, progress_callback=None) -> dict:
    """
    Scan a directory recursively for ROM files.
    Returns stats: {found, added, updated, skipped, errors, total_size, total_size_str}
    """
    root = Path(root_dir).expanduser().resolve()
    if not root.is_dir():
        return {"error": f"Directory does not exist: {root}"}

    stats = {"found": 0, "added": 0, "skipped": 0, "errors": 0,
             "total_size": 0, "total_size_str": "", "platforms_found": {}}

    # Walk the directory
    rom_files = []
    extensions = get_all_extensions()

    for entry in sorted(root.rglob("*")):
        if not entry.is_file():
            continue
        if entry.suffix.lower() in extensions or is_rom_file(entry.name):
            rom_files.append(entry)

    stats["found"] = len(rom_files)

    if not rom_files:
        stats["total_size_str"] = "0 B"
        return stats

    # Process files in parallel
    def process_file(filepath: Path) -> tuple[bool, int, str | None]:
        try:
            fpath = str(filepath)
            fsize = filepath.stat().st_size
            fhash = fast_hash(fpath)
            plat_id, plat_name, emulator = detect_platform(filepath.name)
            title = clean_title(filepath.name)

            rom = RomFile(
                path=fpath,
                filename=filepath.name,
                size_bytes=fsize,
                platform=plat_id,
                platform_name=plat_name,
                file_hash=fhash,
                title=title,
                emulator=emulator,
            )
            inserted = insert_rom(rom)
            return True, fsize, plat_id
        except Exception:
            return False, 0, None

    with ThreadPoolExecutor(max_workers=8) as pool:
        futures = {pool.submit(process_file, p): p for p in rom_files}
        done = 0
        total = len(rom_files)
        for future in as_completed(futures):
            ok, size, plat_id = future.result()
            done += 1
            if ok:
                stats["added"] += 1
                stats["total_size"] += size
                if plat_id:
                    stats["platforms_found"][plat_id] = stats["platforms_found"].get(plat_id, 0) + 1
            else:
                stats["errors"] += 1
            if progress_callback and done % max(1, total // 20) == 0:
                progress_callback(done, total)

    stats["total_size_str"] = format_size(stats["total_size"])
    stats["skipped"] = stats["found"] - stats["added"] - stats["errors"]

    # Remove stale DB entries
    paths_in_use = {str(p) for p in rom_files}
    remove_stale(paths_in_use)

    return stats


def scan_and_update(root_dir: str) -> dict:
    """Wrapper with human-readable output."""
    start = time.time()
    stats = scan_directory(root_dir)
    elapsed = time.time() - start
    stats["elapsed_sec"] = round(elapsed, 1)
    return stats


__all__ = ["scan_directory", "scan_and_update"]