SonicForgeStudio/app/core/soundfont_scanner.py

import os
import json
import time
import logging
import threading
from app.config import settings

logger = logging.getLogger(__name__)

TRACK_FILE = os.path.join(settings.STORAGE_DIR, "sf_scan_state.json")
SYSTEM_SF_DIR = settings.SOUNDFONT_DIR
UPLOAD_SF_DIR = os.path.join(settings.STORAGE_DIR, "soundfonts")


def _file_sig(path: str) -> tuple:
    s = os.path.getsize(path)
    m = os.path.getmtime(path)
    return (s, m)


class SoundFontAutoScanner:
    def __init__(self, system_sf_dir=SYSTEM_SF_DIR, upload_sf_dir=UPLOAD_SF_DIR,
                 system_sf_dirs=None):
        # system_sf_dirs (list) — nhiều thư mục user khai báo trong Plugins
        # Manager. Fallback system_sf_dir (env/.env) nếu list rỗng.
        self.system_sf_dirs = [d for d in (system_sf_dirs or []) if d] or [system_sf_dir]
        self.upload_sf_dir = upload_sf_dir
        self._catalog = {}
        self._lock = threading.Lock()
        self._state = self._load_state()

    def _load_state(self) -> dict:
        if not os.path.exists(TRACK_FILE):
            return {}
        try:
            with open(TRACK_FILE) as f:
                return json.load(f)
        except Exception as e:
            logger.warning("scan_state load failed: %s", e)
            return {}

    def _save_state(self):
        os.makedirs(os.path.dirname(TRACK_FILE), exist_ok=True)
        with open(TRACK_FILE, "w") as f:
            json.dump(self._state, f, indent=2)

    def _sf_files(self, directory: str) -> list:
        if not os.path.isdir(directory):
            return []
        out = []
        for fname in os.listdir(directory):
            if fname.lower().endswith((".sf2", ".sf3")):
                out.append((fname, os.path.join(directory, fname)))
        return out

    def _all_sf_files(self) -> list:
        """Gộp file .sf2/.sf3 từ TẤT CẢ thư mục hiệu lực (user dirs + upload)."""
        out = []
        for d in self.system_sf_dirs:
            out.extend(self._sf_files(d))
        if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir):
            out.extend(self._sf_files(self.upload_sf_dir))
        return out

    def _inspect_single(self, fname: str, full: str, inspector) -> dict:
        if fname.lower().endswith(".sf2"):
            sf_info = inspector.inspect_sf2_file(full) or {}
        else:
            sf_info = {}
        if not sf_info.get("soundfont_id"):
            sf_id = os.path.splitext(fname)[0].lower()
            sf_info = {
                "soundfont_id": sf_id,
                "filename": fname,
                "total_instruments": 0,
                "instruments": [],
                "_sf3": True
            }
        return sf_info

    def scan_once(self) -> bool:
        from app.core.soundfont_inspector import SoundFontInspector
        inspector = SoundFontInspector(self.system_sf_dirs[0], self.upload_sf_dir)
        found_new = False
        dirs = [(d, "system") for d in self.system_sf_dirs]
        if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir):
            dirs.append((self.upload_sf_dir, "upload"))

        for directory, source in dirs:
            for fname, full in self._sf_files(directory):
                sf_id = os.path.splitext(fname)[0].lower()
                key = f"{source}:{sf_id}"
                sig = _file_sig(full)
                prev = self._state.get(key)
                unchanged = prev and prev["size"] == sig[0] and prev["mtime"] == sig[1]
                if unchanged and sf_id in self._catalog:
                    continue
                if unchanged:
                    with self._lock:
                        if sf_id in self._catalog:
                            continue
                        sf_info = self._inspect_single(fname, full, inspector)
                        self._catalog[sf_id] = sf_info
                    continue
                found_new = True
                logger.info("New/changed SF detected: %s", fname)
                sf_info = self._inspect_single(fname, full, inspector)
                with self._lock:
                    self._catalog[sf_id] = sf_info
                    self._state[key] = {"size": sig[0], "mtime": sig[1], "file": fname}

        if found_new:
            self._save_state()
            inspector.invalidate_catalog_cache()
        return found_new

    def scan_loop(self, interval: int = 30, stop_event: threading.Event = None):
        logger.info("SF auto-scanner started (interval=%ds)", interval)
        self.scan_once()
        while not (stop_event and stop_event.is_set()):
            time.sleep(interval)
            try:
                self.scan_once()
            except Exception as e:
                logger.error("scan cycle error: %s", e)

    def start_background(self, interval: int = 30) -> threading.Event:
        ev = threading.Event()
        t = threading.Thread(target=self.scan_loop, args=(interval, ev), daemon=True)
        t.start()
        return ev

    def get_catalog(self) -> dict:
        with self._lock:
            return dict(self._catalog)

    def get_instruments(self, sf_id: str) -> list:
        entry = self._catalog.get(sf_id)
        if entry:
            return entry.get("instruments", [])
        return []