"""PSP save-data inventory and two-way sync (PSP / PPSSPP / cloud folder). PSP saves live in ``PSP/SAVEDATA//`` folders (``PARAM.SFO`` + data). PPSSPP uses the exact same layout under its own "memory stick" directory, so a save folder can be copied verbatim between the PSP, PPSSPP, and a cloud folder. """ import hashlib import sqlite3 from pathlib import Path from . import config, iso_reader from .db import get_connection SAVEDATA_NAME = "SAVEDATA" def default_ppsspp_saves(): """Best-guess PPSSPP SAVEDATA directory (Linux-first, falls back to a default).""" home = Path.home() candidates = [ home / ".config" / "ppsspp" / "PSP" / SAVEDATA_NAME, home / ".local" / "share" / "ppsspp" / "PSP" / SAVEDATA_NAME, home / ".config" / "ppsspp" / SAVEDATA_NAME, home / ".var" / "app" / "org.ppsspp.PPSSPP" / "data" / "ppsspp" / "PSP" / SAVEDATA_NAME, home / "Documents" / "PPSSPP" / "PSP" / SAVEDATA_NAME, ] for cand in candidates: if cand.is_dir(): return cand return candidates[0] def read_save_meta(folder): """Return {title, detail, size, mtime, encrypted} for one save folder.""" folder = Path(folder) title, detail = "", "" sfo_path = folder / "PARAM.SFO" if sfo_path.is_file(): try: sfo = iso_reader.parse_sfo(sfo_path.read_bytes()) except Exception: # noqa: BLE001 sfo = {} title = sfo.get("SAVEDATA_TITLE") or sfo.get("TITLE") or "" detail = sfo.get("SAVEDATA_DETAIL") or "" size = 0 mtime = 0.0 mtime_ns = 0 encrypted = False for p in folder.rglob("*"): if not p.is_file(): continue if p.name.upper() == "SECURE.BIN": encrypted = True try: st = p.stat() except OSError: continue size += st.st_size mtime = max(mtime, st.st_mtime) mtime_ns = max(mtime_ns, st.st_mtime_ns) return {"title": title or folder.name, "detail": detail, "size": size, "mtime": mtime, "mtime_ns": mtime_ns, "encrypted": encrypted} def list_saves(root): """Return [{id, path, title, detail, size, mtime, encrypted}] for a SAVEDATA dir.""" root = Path(root) if root else None if root is None or not root.is_dir(): return [] out = [] for folder in sorted(root.iterdir()): if not folder.is_dir(): continue meta = read_save_meta(folder) meta["id"] = folder.name meta["path"] = str(folder) out.append(meta) return out _sig_cache = {} # (save_id, size, mtime_ns) -> hexdigest _cache_loaded = False def _load_cache(): """Populate the signature cache from the library DB (once per process).""" global _cache_loaded if _cache_loaded: return _cache_loaded = True try: conn = get_connection(config.DB_PATH) conn.execute( "CREATE TABLE IF NOT EXISTS save_sigs (" " save_id TEXT NOT NULL, size INTEGER NOT NULL, mtime_ns INTEGER NOT NULL," " sig TEXT NOT NULL, PRIMARY KEY (save_id, size, mtime_ns))" ) for r in conn.execute("SELECT save_id, size, mtime_ns, sig FROM save_sigs"): _sig_cache[(r["save_id"], r["size"], r["mtime_ns"])] = r["sig"] conn.close() except sqlite3.Error: pass def save_signature(folder, size=None, mtime_ns=None): """Content fingerprint of a save folder (paths + bytes), mtime-independent. Two saves with identical files count as the same "version" even when their modification times differ (after a copy, or across filesystems). Results are cached by ``(save_id, size, mtime_ns)`` in memory and persisted to the library DB, so an unchanged save is never re-hashed — even after a restart. """ folder = Path(folder) save_id = folder.name if size is None or mtime_ns is None: meta = read_save_meta(folder) size = meta["size"] mtime_ns = meta["mtime_ns"] _load_cache() key = (save_id, size, mtime_ns) cached = _sig_cache.get(key) if cached is not None: return cached h = hashlib.sha256() for p in sorted(folder.rglob("*")): if not p.is_file(): continue h.update(p.relative_to(folder).as_posix().encode("utf-8", "replace")) h.update(b"\0") try: with open(p, "rb") as f: for chunk in iter(lambda: f.read(1 << 16), b""): h.update(chunk) except OSError: # noqa: BLE001 pass h.update(b"\0") sig = h.hexdigest() _sig_cache[key] = sig try: conn = get_connection(config.DB_PATH) conn.execute( "INSERT OR REPLACE INTO save_sigs (save_id, size, mtime_ns, sig) VALUES (?,?,?,?)", (save_id, size, mtime_ns, sig), ) conn.commit() conn.close() except sqlite3.Error: pass return sig def sync_plan(locations): """Converge every location to the newest *version* of each save. ``locations`` is ``{name: root_path}`` (empty/missing roots are ignored). Saves are fingerprinted by content, so a location whose copy is already the same version is never re-synced (mtime differences are ignored). Returns a list of ``{"src": path, "dst_root": path, "dst_name": save_id}``. """ per_loc = {name: {s["id"]: s for s in list_saves(root)} for name, root in locations.items() if root} active = [loc for loc, root in locations.items() if root] all_ids = set() for saves_map in per_loc.values(): all_ids.update(saves_map) plan = [] for sid in sorted(all_ids): entries = {} for loc in active: save = per_loc[loc].get(sid) if save is not None: save["sig"] = save_signature(save["path"], save["size"], save["mtime_ns"]) entries[loc] = save # Group existing copies into distinct versions by content signature. versions = {} for save in entries.values(): versions.setdefault(save["sig"], []).append(save) # The newest version wins (best-effort tiebreak by mtime). newest = max(versions.values(), key=lambda group: max(s["mtime"] for s in group))[0] # Replace any copy that is a different version. for loc, save in entries.items(): if save["sig"] != newest["sig"]: plan.append({"src": newest["path"], "dst_root": locations[loc], "dst_name": sid}) # Fill locations that don't have this save at all. for loc in active: if sid not in per_loc[loc]: plan.append({"src": newest["path"], "dst_root": locations[loc], "dst_name": sid}) return plan