quoije/psp-preservation-tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
922c01d17c9b0836b4b71c18aaecb715097b2a2a
psp-preservation-tool/app/gui.py
T
quoije 83627e788b Initial commit
2026-09-06 22:30:13 -04:00

4316 lines
166 KiB
Python
Raw Blame History

"""Native Linux GUI for the PSP Preservation Tool (PySide6 / Qt).
Run with::
python3 -m app.main
"""
import datetime
import logging
import sys
import threading
from pathlib import Path
from PySide6.QtCore import QBuffer, QByteArray, QObject, QRect, QSettings, QSize, Qt, QThread, QTimer, QUrl, Signal
from PySide6.QtGui import QColor, QDesktopServices, QFont, QIcon, QImage, QPainter, QPixmap
from PySide6.QtWidgets import (
QAbstractItemView, QApplication, QCheckBox, QComboBox, QDialog, QDialogButtonBox,
QFileDialog, QFormLayout, QGridLayout, QHBoxLayout, QHeaderView, QLabel, QLineEdit,
QListWidget, QListWidgetItem, QMainWindow, QMenu, QMessageBox, QPlainTextEdit,
QProgressBar, QProgressDialog, QPushButton, QScrollArea, QSpinBox, QSplitter, QStackedWidget,
QTabWidget, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget,
)
from . import config, covers, info, iso_reader, launcher, matcher, namer, pipeline, psp, queue, saves, scanner, sources, store, transfers
from .db import get_connection, init_db
log = logging.getLogger("gui")
class ScanWorker(QObject):
record = Signal(dict)
prompt = Signal(dict, list)
done = Signal(int)
cancelled = Signal()
failed = Signal(str)
def __init__(self, roots, db_path, hash_files=True, workers=None):
super().__init__()
self.roots = roots
self.db_path = db_path
self.hash_files = hash_files
self.workers = workers
self._cancel = threading.Event()
self._reply = [None]
self._reply_event = threading.Event()
self._prompt_pending = False
def cancel(self):
self._cancel.set()
def choose(self, release):
self._reply[0] = release
self._reply_event.set()
def _ask(self, rec, candidates):
self._reply = [None]
self._reply_event.clear()
self._prompt_pending = True
self.prompt.emit(rec, candidates)
self._reply_event.wait()
self._prompt_pending = False
return self._reply[0]
def run(self):
conn = init_db(self.db_path)
count = 0
try:
for rec in scanner.scan(
self.roots, conn=conn, workers=self.workers or config.SCAN_WORKERS,
hash_files=self.hash_files, cancel=self._cancel,
):
ident = pipeline.identify(conn, rec)
if ident["method"] == "unknown" and rec.get("serial") and not self._cancel.is_set():
candidates = matcher.serial_candidates(conn, rec)
if len(candidates) > 1:
chosen = self._ask(rec, candidates)
if chosen is not None:
conn.execute(
"INSERT OR REPLACE INTO choices (serial, size, release_id) VALUES (?,?,?)",
(rec["serial"].upper(), rec["size"], chosen["id"]),
)
ident = pipeline.identify_with_release(rec, chosen, "choice")
rel = ident.get("release") or {}
conn.execute(
"UPDATE files SET release_id = ? WHERE path = ?",
(rel.get("id"), rec["path"]),
)
self.record.emit({**rec, **ident})
count += 1
conn.commit()
if self._cancel.is_set():
self.cancelled.emit()
else:
self.done.emit(count)
except Exception as exc: # noqa: BLE001
self.failed.emit(str(exc))
finally:
conn.close()
def _thread_running(thread):
try:
return thread is not None and thread.isRunning()
except RuntimeError:
return False
class CoverLoader(QObject):
loaded = Signal(int, int, str) # gen, item index, cached image path
failed = Signal(int, int, str) # gen, item index, name (no cover found)
finished = Signal()
def __init__(self, items, gen):
super().__init__()
self.items = items
self.gen = gen
self._cancel = threading.Event()
def cancel(self):
self._cancel.set()
def run(self):
from concurrent.futures import ThreadPoolExecutor, as_completed
pool = ThreadPoolExecutor(max_workers=8)
futures = {
pool.submit(load_cover, it): i
for i, it in enumerate(self.items)
}
try:
for fut in as_completed(futures):
if self._cancel.is_set():
break
i = futures[fut]
try:
path = fut.result()
except Exception: # noqa: BLE001
path = None
if path is not None:
self.loaded.emit(self.gen, i, str(path))
else:
self.failed.emit(self.gen, i, self.items[i]["name"])
finally:
pool.shutdown(wait=True, cancel_futures=True)
self.finished.emit()
_template_image = None
def _umd_template():
global _template_image
if _template_image is None:
img = QImage()
for path in (
config.COVERS_DIR / "UMD_COVER_TEMPLATE.png",
config.DATA_DIR / "covers" / "UMD_COVER_TEMPLATE.png",
):
if path.exists():
img.load(str(path))
if not img.isNull():
break
_template_image = img.convertToFormat(QImage.Format_ARGB32)
return _template_image
def _transparent_window(img):
minx, miny, maxx, maxy = img.width(), img.height(), -1, -1
for y in range(img.height()):
for x in range(img.width()):
if ((img.pixel(x, y) >> 24) & 0xFF) == 0:
minx, miny = min(minx, x), min(miny, y)
maxx, maxy = max(maxx, x), max(maxy, y)
if maxx < minx or maxy < miny:
return QRect()
return QRect(minx, miny, maxx - minx + 1, maxy - miny + 1)
def compose_umd_cover(poster_bytes):
"""Paste a movie poster into the UMD template's transparent window (PNG bytes)."""
template = _umd_template()
poster = QImage.fromData(poster_bytes)
if template.isNull() or poster.isNull():
return None
poster = poster.convertToFormat(QImage.Format_ARGB32)
out = QImage(template.size(), QImage.Format_ARGB32)
out.fill(Qt.transparent)
window = _transparent_window(template)
painter = QPainter(out)
if window.width() > 0 and window.height() > 0:
scaled = poster.scaled(window.size(), Qt.KeepAspectRatioByExpanding, Qt.SmoothTransformation)
ox = window.x() + (window.width() - scaled.width()) // 2
oy = window.y() + (window.height() - scaled.height()) // 2
painter.save()
painter.setClipRect(window)
painter.drawImage(ox, oy, scaled)
painter.restore()
painter.drawImage(0, 0, template)
painter.end()
ba = QByteArray()
buf = QBuffer(ba)
buf.open(QBuffer.WriteOnly)
out.save(buf, "PNG")
buf.close()
return bytes(ba)
def load_cover(item, timeout=8, quick=False):
name = item["name"]
rid = item["release_id"]
system = item["system"]
if system == "videos":
return _load_video_cover(name, rid, system, timeout)
return covers.fetch_cover(name, rid, system, timeout, quick=quick)
def _load_video_cover(name, rid, system, timeout):
dest = covers.cover_path(rid, system)
if dest.exists() and covers.file_is_image(dest):
return dest
data = covers.fetch_manual(name, system)
if data is not None:
logging.info("cover %r <- manual", name)
else:
poster = covers.fetch_poster(covers.movie_search_title(name), timeout)
if poster is not None:
data = compose_umd_cover(poster)
logging.info("cover %r <- poster+template", name)
else:
logging.warning("no cover found for %r", name)
if data is None:
return None
return covers.save_cover(rid, system, data)
def _placeholder_pixmap(w, h):
pm = QPixmap(w, h)
pm.fill(QColor(38, 40, 46))
painter = QPainter(pm)
painter.setPen(QColor(90, 95, 105))
painter.drawRect(1, 1, w - 2, h - 2)
font = QFont()
font.setPointSize(10)
painter.setFont(font)
painter.setPen(QColor(150, 155, 165))
painter.drawText(pm.rect(), Qt.AlignCenter, "No cover")
painter.end()
return pm
def _fit_pixmap(pm, w, h):
"""Scale a cover to fit inside ``w x h`` (letterboxed, full cover visible)."""
canvas = QPixmap(w, h)
canvas.fill(Qt.transparent)
if pm.isNull():
return canvas
scaled = pm.scaled(w, h, Qt.KeepAspectRatio, Qt.SmoothTransformation)
x = (w - scaled.width()) // 2
y = (h - scaled.height()) // 2
painter = QPainter(canvas)
painter.drawPixmap(x, y, scaled)
painter.end()
return canvas
def _cover_pixmap(rid, system, max_w, max_h):
"""Return a scaled cover pixmap for an item, or None if unavailable."""
if rid is None:
return None
path = covers.cover_path(rid, system)
if path is None or not path.exists() or not covers.file_is_image(path):
return None
return QPixmap(str(path)).scaled(max_w, max_h, Qt.KeepAspectRatio, Qt.SmoothTransformation)
class NumericItem(QTableWidgetItem):
"""Table item that sorts numerically by a stored value."""
def __init__(self, text, value):
super().__init__(text)
self._sort_value = value
def __lt__(self, other):
a = getattr(self, "_sort_value", None)
b = getattr(other, "_sort_value", None)
try:
return float(a) < float(b)
except (TypeError, ValueError):
return super().__lt__(other)
def _item_label(it):
"""Grid label: manual title override > canonical name > catalog name."""
label = it.get("title_override") or it["canonical"] or it["name"]
if label.endswith(".iso"):
label = label[:-4]
if len(label) > 44:
label = label[:43] + "…"
if it.get("on_psp"):
label = "\u25cf " + label
return label
def open_containing_folder(path):
"""Open the folder containing a disc/archive in the file manager."""
p = Path(path) if path else None
if p is not None and p.exists():
QDesktopServices.openUrl(QUrl.fromLocalFile(str(p.parent)))
else:
QMessageBox.warning(None, "Not found", f"File not found:\n{path}")
class CoverShowcase(QWidget):
selected = Signal(dict) # emitted when a cover is selected
cover_loaded = Signal(dict) # emitted when an item's cover finishes loading
copy_requested = Signal(dict) # emitted when the user wants to copy an item to the PSP
delete_requested = Signal(dict) # emitted when the user wants to delete an item
delete_psp_requested = Signal(dict) # emitted when the user wants to delete an item from the PSP
play_requested = Signal(dict) # emitted when the user wants to launch an item in PPSSPP
scrape_requested = Signal(dict) # emitted when the user wants to scrape metadata + cover
missing_changed = Signal(int) # emitted with the number of missing covers
ICON_W, ICON_H = 150, 220
GRID_W, GRID_H = 182, 262
def __init__(self, parent=None):
super().__init__(parent)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.list = QListWidget()
self.list.setViewMode(QListWidget.IconMode)
self.list.setIconSize(QSize(self.ICON_W, self.ICON_H))
self.list.setGridSize(QSize(self.GRID_W, self.GRID_H))
self.list.setResizeMode(QListWidget.Adjust)
self.list.setMovement(QListWidget.Static)
self.list.setUniformItemSizes(True)
self.list.setWordWrap(True)
self.list.setSpacing(6)
self.list.setContextMenuPolicy(Qt.CustomContextMenu)
self.list.customContextMenuRequested.connect(self._on_context_menu)
self.list.currentItemChanged.connect(self._on_current_changed)
self.list.itemDoubleClicked.connect(self._on_item_double_clicked)
layout.addWidget(self.list)
self.status = QLabel("")
layout.addWidget(self.status)
self.items = []
self.thread = None
self.loader = None
self._generation = 0
self._threads = []
self._loaders = []
self.missing = []
self._loading = False
self._closing = False
def set_items(self, items):
self._generation += 1
gen = self._generation
self.items = list(items)
self.missing = []
self._loading = True
self.status.setText("Loading covers…")
self.missing_changed.emit(0)
placeholder = self._placeholder_icon()
self.list.clear()
for it in self.items:
label = _item_label(it)
item = QListWidgetItem(placeholder, label)
item.setTextAlignment(Qt.AlignHCenter | Qt.AlignTop)
item.setToolTip(it["name"])
item.setData(Qt.UserRole, it)
self.list.addItem(item)
self._start_loader(gen)
def _start_loader(self, gen):
if self._closing or not self.items:
return
# Cancel any previous loaders so reloads/filter changes don't pile up
# overlapping network fetches.
for loader in self._loaders:
loader.cancel()
self.thread = QThread(self)
self.loader = CoverLoader(self.items, gen)
self.loader.moveToThread(self.thread)
self.thread.started.connect(self.loader.run)
self.loader.loaded.connect(self._on_loaded)
self.loader.failed.connect(self._on_failed)
self.loader.finished.connect(self._on_loading_done)
self.loader.finished.connect(self.thread.quit)
self._threads.append(self.thread)
self._loaders.append(self.loader)
self.thread.start()
def _placeholder_icon(self):
pm = QPixmap(self.ICON_W, self.ICON_H)
pm.fill(QColor(38, 40, 46))
painter = QPainter(pm)
painter.setPen(QColor(90, 95, 105))
painter.drawRect(1, 1, self.ICON_W - 2, self.ICON_H - 2)
font = QFont()
font.setPointSize(10)
painter.setFont(font)
painter.setPen(QColor(150, 155, 165))
painter.drawText(pm.rect(), Qt.AlignCenter, "No cover")
painter.end()
return QIcon(pm)
def _on_loaded(self, gen, index, path):
if gen != self._generation:
return
item = self.list.item(index)
if item is not None:
item.setIcon(self._fit_icon(QPixmap(path)))
if 0 <= index < len(self.items):
self.cover_loaded.emit(self.items[index])
def _on_current_changed(self, current, previous):
if current is not None:
data = current.data(Qt.UserRole)
if data:
self.selected.emit(data)
def _on_item_double_clicked(self, item):
if item is None:
return
data = item.data(Qt.UserRole)
if data and data.get("system") != "videos":
self.play_requested.emit(data)
def _on_failed(self, gen, index, name):
if gen != self._generation:
return
if index not in self.missing:
self.missing.append(index)
logging.warning("no cover for %r", name)
self._update_status()
def _on_loading_done(self):
sender = self.sender()
if sender is not None and getattr(sender, "gen", None) != self._generation:
return
self._loading = False
self._update_status()
def _update_status(self):
if self._loading:
self.status.setText("Loading covers…")
self.missing_changed.emit(0)
return
total = len(self.items)
missing = len(self.missing)
self.status.setText(f"{total - missing}/{total} covers · {missing} missing")
self.missing_changed.emit(missing)
def _on_context_menu(self, pos):
item = self.list.itemAt(pos)
if item is None:
return
idx = self.list.row(item)
it = self.items[idx] if 0 <= idx < len(self.items) else None
menu = QMenu(self)
if it is not None and it.get("path"):
menu.addAction("Open containing folder", lambda: open_containing_folder(it.get("path")))
if it is not None:
if it.get("system") != "videos":
menu.addAction("Play on PPSSPP", lambda: self.play_requested.emit(it))
menu.addAction("Copy to PSP", lambda: self.copy_requested.emit(it))
if it.get("on_psp"):
menu.addAction("Delete from PSP", lambda: self.delete_psp_requested.emit(it))
menu.addAction("Delete…", lambda: self.delete_requested.emit(it))
if it is not None:
menu.addAction("Scrape…", lambda: self.scrape_requested.emit(it))
menu.exec(self.list.mapToGlobal(pos))
def search_missing(self):
"""Open the scrape dialog for the first missing cover (if any)."""
if self.missing:
idx = self.missing[0]
if 0 <= idx < len(self.items):
self.scrape_requested.emit(self.items[idx])
def refresh_item(self, it):
"""Re-read an item's cover from disk and update its icon."""
rid = it.get("release_id") or covers.stable_key(it.get("name") or "")
system = it.get("system") or "games"
path = covers.cover_path(rid, system)
if path is None or not path.exists():
return
pm = QPixmap(str(path))
if pm.isNull():
return
for i in range(self.list.count()):
item = self.list.item(i)
if item.data(Qt.UserRole) is it:
item.setIcon(self._fit_icon(pm))
if i in self.missing:
self.missing.remove(i)
self._update_status()
break
def update_item_label(self, it):
"""Refresh a single item's label after a manual retag."""
for i in range(self.list.count()):
list_item = self.list.item(i)
data = list_item.data(Qt.UserRole)
if data is not None and data.get("release_id") == it.get("release_id"):
list_item.setText(_item_label(it))
break
def _fit_icon(self, pm):
canvas = QPixmap(self.ICON_W, self.ICON_H)
canvas.fill(Qt.transparent)
if pm.isNull():
return QIcon(canvas)
scaled = pm.scaled(self.ICON_W, self.ICON_H, Qt.KeepAspectRatio, Qt.SmoothTransformation)
x = (self.ICON_W - scaled.width()) // 2
y = (self.ICON_H - scaled.height()) // 2
painter = QPainter(canvas)
painter.drawPixmap(x, y, scaled)
painter.end()
return QIcon(canvas)
def shutdown(self):
self._closing = True
for loader in self._loaders:
loader.cancel()
for t in self._threads:
if _thread_running(t):
t.quit()
t.wait()
self._threads = []
self._loaders = []
self.thread = None
self.loader = None
class MetadataWorker(QObject):
result = Signal(int, dict) # gen, meta
finished = Signal()
def __init__(self, title, gen, system, release_id):
super().__init__()
self.title = title
self.gen = gen
self.system = system
self.release_id = release_id
def run(self):
meta = info.get_or_fetch(self.release_id, self.system, self.title)
self.result.emit(self.gen, meta or {})
self.finished.emit()
class GameDetailsPanel(QWidget):
edit_requested = Signal() # emitted when the user wants to hand-tag the item
link_requested = Signal() # emitted when the user wants to pick a movie/game to link
copy_requested = Signal() # emitted when the user wants to copy the current item to the PSP
delete_requested = Signal() # emitted when the user wants to delete the current item
play_requested = Signal() # emitted when the user wants to launch the item in PPSSPP
delete_psp_requested = Signal() # emitted when the user wants to delete the current item from the PSP
def __init__(self, parent=None):
super().__init__(parent)
self.setFixedWidth(300)
self._youtube_url = ""
self._meta = None
outer = QVBoxLayout(self)
outer.setContentsMargins(12, 12, 0, 12)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
outer.addWidget(scroll)
content = QWidget()
self.layout = QVBoxLayout(content)
self.layout.setContentsMargins(6, 6, 6, 6)
self.layout.setSpacing(6)
scroll.setWidget(content)
self.cover = QLabel()
self.cover.setAlignment(Qt.AlignCenter)
self.cover.setVisible(False)
self.layout.addWidget(self.cover)
self.title = QLabel("Select a game")
title_font = QFont()
title_font.setPointSize(16)
title_font.setBold(True)
self.title.setFont(title_font)
self.title.setWordWrap(True)
self.layout.addWidget(self.title)
self.subtitle = QLabel("")
self.subtitle.setStyleSheet("color: #999;")
self.layout.addWidget(self.subtitle)
self.layout.addSpacing(8)
self.meta_grid = QGridLayout()
self.meta_grid.setHorizontalSpacing(14)
self.meta_grid.setVerticalSpacing(6)
self.layout.addLayout(self.meta_grid)
self.layout.addSpacing(8)
self.synopsis_label = QLabel("<b>Synopsis</b>")
self.layout.addWidget(self.synopsis_label)
self.synopsis = QLabel("")
self.synopsis.setWordWrap(True)
self.layout.addWidget(self.synopsis)
self.layout.addSpacing(12)
self.youtube_btn = QPushButton("\u25b6 Watch trailer")
self.youtube_btn.setVisible(False)
self.youtube_btn.clicked.connect(self._open_youtube)
self.layout.addWidget(self.youtube_btn)
self.layout.addSpacing(8)
self.play_btn = QPushButton("\u25b6 Play on PPSSPP")
self.play_btn.clicked.connect(self.play_requested.emit)
self.play_btn.setEnabled(False)
self.play_btn.setToolTip("Launch this game in PPSSPP")
self.layout.addWidget(self.play_btn)
btn_grid = QGridLayout()
btn_grid.setHorizontalSpacing(6)
btn_grid.setVerticalSpacing(6)
self.link_btn = QPushButton("Scrape…")
self.link_btn.clicked.connect(self.link_requested.emit)
btn_grid.addWidget(self.link_btn, 0, 0)
self.edit_btn = QPushButton("Edit details…")
self.edit_btn.clicked.connect(self.edit_requested.emit)
btn_grid.addWidget(self.edit_btn, 0, 1)
self.copy_btn = QPushButton("Copy to PSP")
self.copy_btn.clicked.connect(self.copy_requested.emit)
self.copy_btn.setEnabled(False)
self.copy_btn.setToolTip("Copy this disc to the connected PSP")
btn_grid.addWidget(self.copy_btn, 1, 0)
self.delete_btn = QPushButton("Delete")
self.delete_btn.clicked.connect(self.delete_requested.emit)
self.delete_btn.setStyleSheet("color: #e06c6c;")
self.delete_btn.setToolTip("Permanently delete this disc from disk")
btn_grid.addWidget(self.delete_btn, 1, 1)
self.del_psp_btn = QPushButton("Delete from PSP")
self.del_psp_btn.clicked.connect(self.delete_psp_requested.emit)
self.del_psp_btn.setEnabled(False)
self.del_psp_btn.setStyleSheet("color: #e06c6c;")
self.del_psp_btn.setToolTip("Delete this disc from the connected PSP")
btn_grid.addWidget(self.del_psp_btn, 2, 0)
self.layout.addLayout(btn_grid)
self.layout.addStretch(1)
def set_item(self, item):
self.system = item.get("system")
path = item.get("path") or ""
self.play_btn.setEnabled(item.get("system") == "games" and bool(path) and Path(path).suffix.lower() == ".iso")
self.del_psp_btn.setEnabled(bool(item.get("on_psp")))
self.title.setText(item.get("title_override") or item.get("name") or "Unknown")
parts = []
if item.get("serial"):
parts.append(item["serial"])
if item.get("region"):
parts.append(item["region"])
self.subtitle.setText(" · ".join(parts))
self._clear_meta()
self.synopsis.setText("")
self.youtube_btn.setVisible(False)
self._youtube_url = ""
self._meta = None
self._set_cover(item.get("release_id"), item.get("system"))
def set_copy_available(self, available):
self.copy_btn.setEnabled(bool(available))
def set_psp_installed(self, on_psp):
self.del_psp_btn.setEnabled(bool(on_psp))
def _set_cover(self, rid, system):
pm = _cover_pixmap(rid, system, 180, 260)
if pm is not None:
self.cover.setPixmap(pm)
self.cover.setVisible(True)
else:
self.cover.clear()
self.cover.setVisible(False)
def set_metadata(self, meta):
self._meta = meta
self._clear_meta()
self.synopsis.setText("")
self.youtube_btn.setVisible(False)
self._youtube_url = ""
if not meta:
self.synopsis_label.setVisible(False)
self.synopsis.setVisible(False)
return
self.synopsis_label.setVisible(True)
self.synopsis.setVisible(True)
if getattr(self, "system", None) == "videos":
fields = (
("Released", meta.get("release_date")),
("Genre", ", ".join(meta.get("genres") or [])),
("Rating", meta.get("rating")),
)
else:
fields = (
("Released", meta.get("release_date")),
("Developer", ", ".join(meta.get("developers") or [])),
("Publisher", ", ".join(meta.get("publishers") or [])),
("Genre", ", ".join(meta.get("genres") or [])),
("Players", meta.get("players")),
("Rating", meta.get("rating")),
)
rows = []
for label, value in fields:
if value:
rows.append((label, str(value)))
for r, (label, value) in enumerate(rows):
lbl = QLabel(f"<b>{label}</b>")
val = QLabel(value)
val.setWordWrap(True)
self.meta_grid.addWidget(lbl, r, 0, Qt.AlignTop)
self.meta_grid.addWidget(val, r, 1)
overview = meta.get("overview")
if overview:
self.synopsis.setText(overview)
youtube = meta.get("youtube")
if youtube:
self._youtube_url = youtube
self.youtube_btn.setVisible(True)
def _clear_meta(self):
while self.meta_grid.count():
it = self.meta_grid.takeAt(0)
w = it.widget()
if w is not None:
w.deleteLater()
def _open_youtube(self):
if self._youtube_url:
QDesktopServices.openUrl(QUrl(self._youtube_url))
def current_meta(self):
return self._meta
class TagDialog(QDialog):
"""Hand-edit the title and metadata of a library item."""
def __init__(self, item, meta, parent=None):
super().__init__(parent)
self.setWindowTitle("Edit details")
self.resize(480, 540)
self.reset_pressed = False
meta = meta or {}
layout = QVBoxLayout(self)
grid = QGridLayout()
grid.setHorizontalSpacing(8)
grid.setVerticalSpacing(8)
self.title_edit = QLineEdit(item.get("title_override") or item.get("name") or "")
self.date_edit = QLineEdit(str(meta.get("release_date") or ""))
self.genre_edit = QLineEdit(", ".join(meta.get("genres") or []))
self.rating_edit = QLineEdit(str(meta.get("rating") or ""))
self.youtube_edit = QLineEdit(str(meta.get("youtube") or ""))
self.synopsis_edit = QPlainTextEdit(str(meta.get("overview") or ""))
for r, (label, widget) in enumerate((
("Title", self.title_edit),
("Released", self.date_edit),
("Genre", self.genre_edit),
("Rating", self.rating_edit),
("Trailer URL", self.youtube_edit),
)):
grid.addWidget(QLabel(f"<b>{label}</b>"), r, 0, Qt.AlignTop)
grid.addWidget(widget, r, 1)
layout.addLayout(grid)
layout.addWidget(QLabel("<b>Synopsis</b>"))
layout.addWidget(self.synopsis_edit, 1)
buttons = QDialogButtonBox(QDialogButtonBox.Save | QDialogButtonBox.Reset | QDialogButtonBox.Cancel)
buttons.button(QDialogButtonBox.Save).setText("Save")
buttons.button(QDialogButtonBox.Reset).setText("Reset to automatic")
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
buttons.button(QDialogButtonBox.Reset).clicked.connect(self._reset)
layout.addWidget(buttons)
def _reset(self):
self.reset_pressed = True
self.accept()
def result_meta(self):
genres = [g.strip() for g in self.genre_edit.text().split(",") if g.strip()]
return {
"title": self.title_edit.text().strip(),
"release_date": self.date_edit.text().strip(),
"genres": genres,
"rating": self.rating_edit.text().strip() or None,
"youtube": self.youtube_edit.text().strip(),
"overview": self.synopsis_edit.toPlainText().strip(),
}
class ScrapeSearchWorker(QObject):
results = Signal(list) # candidates with thumb_bytes added
finished = Signal()
def __init__(self, system, query):
super().__init__()
self.system = system
self.query = query
def run(self):
found = covers.search_metadata_candidates(self.system, self.query, limit=20, timeout=10)
for r in found:
url = r.get("thumb_url")
r["thumb_bytes"] = covers.download_image(url, timeout=8) if url else None
self.results.emit(found)
self.finished.emit()
class ScrapeFetchWorker(QObject):
result = Signal(dict) # {"meta": {...}, "cover": bytes|None}
finished = Signal()
def __init__(self, system, rid, candidate):
super().__init__()
self.system = system
self.rid = rid
self.candidate = candidate
def run(self):
meta = covers.link_metadata(self.system, self.candidate, timeout=10) or {}
cover = _scrape_cover(self.system, self.candidate, self.rid, timeout=20)
self.result.emit({"meta": meta, "cover": cover})
self.finished.emit()
def _scrape_cover(system, candidate, rid, timeout=20):
"""Download the cover for a metadata candidate; returns bytes or None."""
url = candidate.get("image_url")
if url:
data = covers.download_image(url, timeout)
if data:
if system == "videos":
return compose_umd_cover(data)
if candidate.get("source") == "igdb":
return covers.compose_game_cover(data)
return data
title = candidate.get("title") or ""
if title:
path = covers.fetch_cover(title, rid, system, timeout=timeout)
if path is not None:
try:
return path.read_bytes()
except OSError:
return None
return None
class ScrapeDialog(QDialog):
"""Scrape metadata + cover for an item in one go (Plex-style)."""
def __init__(self, item, parent=None, rid=None, system=None):
super().__init__(parent)
self.system = system or item.get("system") or "games"
self.rid = str(rid or item.get("release_id") or covers.stable_key(item.get("name") or ""))
self.setWindowTitle("Scrape metadata & cover")
self.resize(600, 580)
self.candidates = []
self.selected_meta = None
layout = QVBoxLayout(self)
top = QHBoxLayout()
top.addWidget(QLabel("Search:"))
if self.system == "videos":
hint = covers.movie_search_title(item.get("name") or "")
else:
hint = covers.plain_title(item.get("name") or "")
self.query_edit = QLineEdit(hint)
self.query_edit.returnPressed.connect(self._do_search)
top.addWidget(self.query_edit, 1)
self.search_btn = QPushButton("Search")
self.search_btn.clicked.connect(self._do_search)
top.addWidget(self.search_btn)
layout.addLayout(top)
self.list = QListWidget()
self.list.setViewMode(QListWidget.IconMode)
self.list.setIconSize(QSize(90, 120))
self.list.setGridSize(QSize(120, 170))
self.list.setResizeMode(QListWidget.Adjust)
self.list.setWordWrap(True)
self.list.setUniformItemSizes(True)
self.list.itemDoubleClicked.connect(self._scrape)
layout.addWidget(self.list)
self.status = QLabel("Searching…")
layout.addWidget(self.status)
buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
self.ok_btn = buttons.button(QDialogButtonBox.Ok)
self.ok_btn.setText("Scrape")
self.ok_btn.setEnabled(False)
buttons.accepted.connect(self._scrape)
buttons.rejected.connect(self.reject)
layout.addWidget(buttons)
self._threads = []
self._workers = []
self._do_search()
def _do_search(self):
query = self.query_edit.text().strip()
if not query:
return
self.status.setText("Searching…")
self.search_btn.setEnabled(False)
self.ok_btn.setEnabled(False)
self.list.clear()
self.candidates = []
thread = QThread(self)
worker = ScrapeSearchWorker(self.system, query)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.results.connect(self._on_results)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_results(self, results):
self.search_btn.setEnabled(True)
self.candidates = results
self.list.clear()
if not results:
self.status.setText("No results. Try another term.")
return
for r in results:
icon = QIcon()
if r.get("thumb_bytes"):
pm = QPixmap()
pm.loadFromData(r["thumb_bytes"])
icon = QIcon(pm)
year = r.get("year") or (r.get("release_date") or "")[:4]
label = f"{r['title']} ({year})" if year else r["title"]
item = QListWidgetItem(icon, label)
if r.get("overview"):
item.setToolTip(r["overview"])
item.setData(Qt.UserRole, r)
self.list.addItem(item)
self.list.setCurrentRow(0)
self.status.setText(f"{len(results)} result(s). Select one and press Scrape.")
self.ok_btn.setEnabled(True)
def _scrape(self):
item = self.list.currentItem()
if item is None:
return
candidate = item.data(Qt.UserRole)
self.ok_btn.setEnabled(False)
self.search_btn.setEnabled(False)
self.status.setText("Scraping…")
thread = QThread(self)
worker = ScrapeFetchWorker(self.system, self.rid, candidate)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_scraped)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_scraped(self, payload):
meta = payload.get("meta") or {}
cover = payload.get("cover")
if cover:
covers.save_cover(self.rid, self.system, cover)
if meta:
self.selected_meta = meta
self.accept()
elif cover:
self.accept() # a cover with no metadata is still a win
else:
self.ok_btn.setEnabled(True)
self.search_btn.setEnabled(True)
self.status.setText("Could not fetch anything for that selection.")
def closeEvent(self, event):
for t in self._threads:
if _thread_running(t):
t.quit()
t.wait()
self._workers = []
event.accept()
class SettingsPage(QWidget):
saved = Signal()
def __init__(self, games_dir, videos_dir, store_dir, ppsspp_dir, cloud_dir, ppsspp_exe,
store_server_url, store_token, store_archive_id, store_archive_sig, store_archive_user,
iso_category, parent=None):
super().__init__(parent)
outer = QVBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
self.tabs = QTabWidget()
outer.addWidget(self.tabs, 1)
s = QSettings("PSPPreserve", "PSPPreserve")
# -- General ----------------------------------------------------- #
page, form = self._new_form()
self.games_edit = self._dir_row(form, "Games directory:", games_dir)
self.videos_edit = self._dir_row(form, "Videos directory:", videos_dir)
self.auto_scan = QCheckBox("Scan directories automatically on startup")
self.auto_scan.setChecked(str(s.value("auto_scan", False)).lower() == "true")
form.addRow(self.auto_scan)
self.workers_spin = QSpinBox()
self.workers_spin.setRange(1, 16)
self.workers_spin.setValue(int(s.value("scan_workers", 1)))
self.workers_spin.setToolTip("Raise on SSDs / fast network shares; keep 1 on spinning disks.")
form.addRow("Parallel scan workers:", self.workers_spin)
self.tabs.addTab(page, "General")
# -- Store ------------------------------------------------------- #
page, form = self._new_form()
self.store_server_edit = self._line_row(form, "Server URL:", store_server_url)
self.store_token_edit = self._secret_row(form, "API token:", store_token)
self.archive_edit = self._line_row(form, "Archive.org item:", store_archive_id)
self.archive_sig_edit = self._secret_row(form, "Archive.org logged-in-sig:", store_archive_sig)
self.archive_user_edit = self._secret_row(form, "Archive.org logged-in-user:", store_archive_user)
self.store_edit = self._dir_row(form, "Local store dir:", store_dir)
self.tabs.addTab(page, "Store")
# -- Saves ------------------------------------------------------- #
page, form = self._new_form()
self.ppsspp_edit = self._dir_row(form, "PPSSPP saves dir:", ppsspp_dir)
self.cloud_edit = self._dir_row(form, "Cloud saves dir:", cloud_dir)
self.ppsspp_exe_edit = self._file_row(form, "PPSSPP executable:", ppsspp_exe)
self.auto_sync = QCheckBox("Auto-sync saves when the PSP is connected")
self.auto_sync.setChecked(str(s.value("auto_sync_saves", "false")).lower() == "true")
form.addRow(self.auto_sync)
self.tabs.addTab(page, "Saves")
# -- PSP --------------------------------------------------------- #
page, form = self._new_form()
self.iso_category_edit = QLineEdit(iso_category)
self.iso_category_edit.setPlaceholderText("e.g. Homebrew (blank = ISO root)")
form.addRow("ISO category (Category Lite):", self.iso_category_edit)
self.tabs.addTab(page, "PSP")
# -- Cover sources ------------------------------------------------ #
page, form = self._new_form()
creds = sources.load()
self.fields = {}
def add_field(label, key, password=False, placeholder="(optional)"):
edit = QLineEdit()
if password:
edit.setEchoMode(QLineEdit.Password)
edit.setText(creds.get(key, ""))
edit.setPlaceholderText(placeholder)
form.addRow(label, edit)
self.fields[key] = edit
note = QLabel(
"libretro-thumbnails is always on and needs no key — it's the default "
"game-cover source. The keys below are optional upgrades."
)
note.setWordWrap(True)
form.addRow(note)
form.addRow(QLabel("<b>Games — covers &amp; metadata</b>"))
add_field("IGDB Client ID:", "igdb_client_id")
add_field("IGDB Client Secret:", "igdb_client_secret", password=True)
add_field("ScreenScraper dev ID:", "screenscraper_devid")
add_field("ScreenScraper dev password:", "screenscraper_devpassword", password=True)
add_field("ScreenScraper user ID:", "screenscraper_ssid")
add_field("ScreenScraper user password:", "screenscraper_sspassword", password=True)
add_field("ScreenScraper software name:", "screenscraper_softname")
form.addRow(QLabel("<b>Videos — metadata &amp; posters</b>"))
add_field("TheMovieDB API key:", "tmdb_api_key", password=True)
hint = QLabel(
"Manual UMD covers: drop images into "
f"<tt>{config.DATA_DIR / 'manual_covers'}</tt> "
"(subfolders <tt>games/</tt> and <tt>videos/</tt>)."
)
hint.setWordWrap(True)
form.addRow(hint)
self.clear_covers_btn = QPushButton("Clear cover cache (re-fetch all covers)")
self.clear_covers_btn.clicked.connect(self._clear_cover_cache)
form.addRow(self.clear_covers_btn)
self.tabs.addTab(page, "Cover sources")
self.save_btn = QPushButton("Save settings")
self.save_btn.clicked.connect(self._save)
outer.addWidget(self.save_btn)
def _new_form(self):
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
page = QWidget()
form = QFormLayout(page)
form.setContentsMargins(16, 16, 16, 16)
form.setHorizontalSpacing(16)
form.setVerticalSpacing(10)
form.setFieldGrowthPolicy(QFormLayout.AllNonFixedFieldsGrow)
scroll.setWidget(page)
return scroll, form
def _dir_row(self, form, label, value):
edit = QLineEdit(value)
edit.setPlaceholderText("Select a directory…")
browse = QPushButton("…")
browse.clicked.connect(lambda _=False, e=edit: self._browse(e))
row = QWidget()
h = QHBoxLayout(row)
h.setContentsMargins(0, 0, 0, 0)
h.addWidget(edit, 1)
h.addWidget(browse)
form.addRow(label, row)
return edit
def _browse(self, edit):
path = QFileDialog.getExistingDirectory(self, "Select directory")
if path:
edit.setText(path)
def _file_row(self, form, label, value):
edit = QLineEdit(value)
edit.setPlaceholderText("Auto-detect (optional)")
browse = QPushButton("…")
browse.clicked.connect(lambda _=False, e=edit: self._browse_file(e))
row = QWidget()
h = QHBoxLayout(row)
h.setContentsMargins(0, 0, 0, 0)
h.addWidget(edit, 1)
h.addWidget(browse)
form.addRow(label, row)
return edit
def _browse_file(self, edit):
path, _ = QFileDialog.getOpenFileName(self, "Select PPSSPP executable")
if path:
edit.setText(path)
def _line_row(self, form, label, value):
edit = QLineEdit(value)
form.addRow(label, edit)
return edit
def _secret_row(self, form, label, value):
edit = QLineEdit(value)
edit.setEchoMode(QLineEdit.Password)
form.addRow(label, edit)
return edit
def games_dir(self):
return self.games_edit.text().strip()
def videos_dir(self):
return self.videos_edit.text().strip()
def store_dir(self):
return self.store_edit.text().strip()
def store_server_url(self):
return self.store_server_edit.text().strip()
def store_token(self):
return self.store_token_edit.text().strip()
def archive_item(self):
return self.archive_edit.text().strip()
def archive_sig(self):
return self.archive_sig_edit.text().strip()
def archive_user(self):
return self.archive_user_edit.text().strip()
def ppsspp_saves_dir(self):
return self.ppsspp_edit.text().strip()
def cloud_saves_dir(self):
return self.cloud_edit.text().strip()
def auto_sync_saves(self):
return self.auto_sync.isChecked()
def ppsspp_exe(self):
return self.ppsspp_exe_edit.text().strip()
def iso_category(self):
return self.iso_category_edit.text().strip()
def _save(self):
sources.save({key: edit.text() for key, edit in self.fields.items()})
covers.clear_miss_cache()
s = QSettings("PSPPreserve", "PSPPreserve")
s.setValue("auto_scan", self.auto_scan.isChecked())
s.setValue("scan_workers", self.workers_spin.value())
s.setValue("auto_sync_saves", self.auto_sync.isChecked())
self.saved.emit()
def _clear_cover_cache(self):
covers.clear_cover_cache()
QMessageBox.information(self, "Cover cache cleared",
"Cached covers deleted.\nRefresh the Library or Store to re-fetch them.")
class StoreScanWorker(QObject):
results = Signal(list)
failed = Signal(str)
finished = Signal()
def __init__(self, root):
super().__init__()
self.root = root
def run(self):
try:
items = store.scan(self.root)
except Exception as exc: # noqa: BLE001
self.failed.emit(str(exc))
self.finished.emit()
return
self.results.emit(items)
self.finished.emit()
class StoreCoverLoader(QObject):
loaded = Signal(int, int, str) # gen, grid row, cached image path
finished = Signal()
def __init__(self, tasks, gen, quick=True):
super().__init__()
self.tasks = tasks # list of (grid_row, item_dict)
self.gen = gen
self.quick = quick
self._cancel = threading.Event()
def cancel(self):
self._cancel.set()
def run(self):
from concurrent.futures import ThreadPoolExecutor, as_completed
pool = ThreadPoolExecutor(max_workers=8)
futures = {
pool.submit(load_cover, it, 8, self.quick): row
for row, it in self.tasks
}
try:
for fut in as_completed(futures):
if self._cancel.is_set():
break
row = futures[fut]
try:
path = fut.result()
except Exception: # noqa: BLE001
path = None
if path is not None:
self.loaded.emit(self.gen, row, str(path))
finally:
pool.shutdown(wait=True, cancel_futures=True)
self.finished.emit()
class StoreMetaWorker(QObject):
result = Signal(int, dict) # gen, meta
finished = Signal()
def __init__(self, title, gen, rid, system="games"):
super().__init__()
self.title = title
self.gen = gen
self.rid = rid
self.system = system
def run(self):
meta = info.get_or_fetch(self.rid, self.system, self.title)
self.result.emit(self.gen, meta or {})
self.finished.emit()
class StoreDetailsPanel(QWidget):
"""Metadata sidebar for a selected store item."""
link_requested = Signal()
download_requested = Signal()
def __init__(self, parent=None):
super().__init__(parent)
self.setFixedWidth(300)
self._youtube_url = ""
outer = QVBoxLayout(self)
outer.setContentsMargins(12, 12, 0, 12)
scroll = QScrollArea()
scroll.setWidgetResizable(True)
scroll.setFrameShape(QScrollArea.NoFrame)
outer.addWidget(scroll)
content = QWidget()
self.layout = QVBoxLayout(content)
self.layout.setContentsMargins(6, 6, 6, 6)
self.layout.setSpacing(6)
scroll.setWidget(content)
self.cover = QLabel()
self.cover.setAlignment(Qt.AlignCenter)
self.cover.setVisible(False)
self.layout.addWidget(self.cover)
self.title = QLabel("Select a game")
title_font = QFont()
title_font.setPointSize(16)
title_font.setBold(True)
self.title.setFont(title_font)
self.title.setWordWrap(True)
self.layout.addWidget(self.title)
self.subtitle = QLabel("")
self.subtitle.setStyleSheet("color: #999;")
self.layout.addWidget(self.subtitle)
self.layout.addSpacing(8)
self.meta_grid = QGridLayout()
self.meta_grid.setHorizontalSpacing(14)
self.meta_grid.setVerticalSpacing(6)
self.layout.addLayout(self.meta_grid)
self.layout.addSpacing(8)
self.synopsis_label = QLabel("<b>Synopsis</b>")
self.layout.addWidget(self.synopsis_label)
self.synopsis = QLabel("")
self.synopsis.setWordWrap(True)
self.layout.addWidget(self.synopsis)
self.layout.addSpacing(12)
self.youtube_btn = QPushButton("\u25b6 Watch trailer")
self.youtube_btn.setVisible(False)
self.youtube_btn.clicked.connect(self._open_youtube)
self.layout.addWidget(self.youtube_btn)
self.layout.addSpacing(8)
self.download_btn = QPushButton("\u2b07 Download")
self.download_btn.clicked.connect(self.download_requested.emit)
self.download_btn.setToolTip("Download and extract this game into your library")
self.layout.addWidget(self.download_btn)
self.layout.addSpacing(6)
self.link_btn = QPushButton("Scrape…")
self.link_btn.clicked.connect(self.link_requested.emit)
self.link_btn.setToolTip("Scrape metadata and cover for this archive")
self.layout.addWidget(self.link_btn)
self.layout.addStretch(1)
def set_item(self, item):
self._item = item
self.title.setText(item.get("title_override") or store.display_name(item["name"]))
parts = []
size = item.get("size")
if size:
parts.append(f"{size / (1024 * 1024):.1f} MB")
if item.get("in_library"):
parts.append("\u2713 In library")
self.subtitle.setText(" · ".join(parts))
self._clear_meta()
self.synopsis.setText("")
self.youtube_btn.setVisible(False)
self._youtube_url = ""
lc = item.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(item["name"]), "games"
self._set_cover(rid, system)
def _set_cover(self, rid, system):
pm = _cover_pixmap(rid, system, 180, 260)
if pm is not None:
self.cover.setPixmap(pm)
self.cover.setVisible(True)
else:
self.cover.clear()
self.cover.setVisible(False)
def set_metadata(self, meta):
self._clear_meta()
self.synopsis.setText("")
self.youtube_btn.setVisible(False)
self._youtube_url = ""
if not meta:
self.synopsis_label.setVisible(False)
self.synopsis.setVisible(False)
return
self.synopsis_label.setVisible(True)
self.synopsis.setVisible(True)
fields = (
("Released", meta.get("release_date")),
("Developer", ", ".join(meta.get("developers") or [])),
("Publisher", ", ".join(meta.get("publishers") or [])),
("Genre", ", ".join(meta.get("genres") or [])),
("Players", meta.get("players")),
("Rating", meta.get("rating")),
)
rows = [(label, str(value)) for label, value in fields if value]
for r, (label, value) in enumerate(rows):
lbl = QLabel(f"<b>{label}</b>")
val = QLabel(value)
val.setWordWrap(True)
self.meta_grid.addWidget(lbl, r, 0, Qt.AlignTop)
self.meta_grid.addWidget(val, r, 1)
overview = meta.get("overview")
if overview:
self.synopsis.setText(overview)
youtube = meta.get("youtube")
if youtube:
self._youtube_url = youtube
self.youtube_btn.setVisible(True)
def _clear_meta(self):
while self.meta_grid.count():
it = self.meta_grid.takeAt(0)
w = it.widget()
if w is not None:
w.deleteLater()
def _open_youtube(self):
if self._youtube_url:
QDesktopServices.openUrl(QUrl(self._youtube_url))
class StoreDateLoader(QObject):
"""Background metadata fetch: fills in release dates for store rows."""
loaded = Signal(int, int, str) # gen, row, release_date
finished = Signal()
def __init__(self, tasks, gen):
super().__init__()
self.tasks = tasks # list of (row, release_id, system, title)
self.gen = gen
self._cancel = threading.Event()
def cancel(self):
self._cancel.set()
def run(self):
for row, rid, system, title in self.tasks:
if self._cancel.is_set():
break
try:
meta = info.get_or_fetch(rid, system, title)
date = (meta or {}).get("release_date") or ""
except Exception: # noqa: BLE001
date = ""
self.loaded.emit(self.gen, row, date)
self.finished.emit()
class StoreScrapeAllWorker(QObject):
"""Bulk scrape for the whole store (release dates, details, and covers)."""
progress = Signal(int, int) # done, total
finished = Signal()
def __init__(self, tasks):
super().__init__()
self.tasks = tasks # list of (rid, system, name, title)
self._cancel = threading.Event()
def cancel(self):
self._cancel.set()
def run(self):
total = len(self.tasks)
for i, (rid, system, name, title) in enumerate(self.tasks):
if self._cancel.is_set():
break
try:
info.get_or_fetch(rid, system, title)
covers.fetch_cover(name, rid, system)
except Exception: # noqa: BLE001
pass
self.progress.emit(i + 1, total)
self.finished.emit()
class RemoteStoreScanWorker(QObject):
results = Signal(list)
failed = Signal(str)
finished = Signal()
def __init__(self, source):
super().__init__()
self.source = source
def run(self):
try:
items = self.source.scan()
except Exception as exc: # noqa: BLE001
self.failed.emit(str(exc))
self.finished.emit()
return
self.results.emit(items)
self.finished.emit()
class StorePage(QWidget):
"""Browse the Redump 7z store and queue games for download."""
download_requested = Signal(list) # list of store item dicts
ICON_W, ICON_H = 150, 220
GRID_W, GRID_H = 182, 262
def __init__(self, parent=None):
super().__init__(parent)
self._store_dir = ""
self._server_url = ""
self._token = ""
self._archive_id = ""
self._archive_sig = ""
self._archive_user = ""
self._items = []
self._filtered = []
self._lib_norm = []
self._lib_cover = {} # norm -> (release_id, system)
self._psp_norm = [] # norm of library items installed on the PSP
self._cover_attempted = set()
self._generation = 0
self._meta_gen = 0
self._closing = False
self._threads = []
self._workers = []
self._loaders = []
self._dirty = False
self._fill_timer = None
self._scraping = False
self._scrape_dialog = None
self._scrape_worker = None
self._scrape_thread = None
layout = QVBoxLayout(self)
m = layout.contentsMargins()
layout.setContentsMargins(m.left(), m.top(), 0, m.bottom())
bar = QHBoxLayout()
bar.addWidget(QLabel("Search:"))
self.search = QLineEdit()
self.search.setPlaceholderText("Filter by title…")
self.search.textChanged.connect(self._schedule_filter)
bar.addWidget(self.search, 1)
self.source_combo = QComboBox()
self.source_combo.addItems(["Server", "Archive.org", "Local folder"])
self.source_combo.currentIndexChanged.connect(self._on_source_changed)
bar.addWidget(self.source_combo)
self.view_combo = QComboBox()
self.view_combo.addItems(["Cover view", "List view"])
bar.addWidget(self.view_combo)
self.psp_filter = QComboBox()
self.psp_filter.addItems(["All PSP", "On PSP", "Not on PSP"])
self.psp_filter.setToolTip("Filter by whether the archive matches a disc installed on the connected PSP")
self.psp_filter.currentIndexChanged.connect(self._apply_filter)
bar.addWidget(self.psp_filter)
refresh_btn = QPushButton("Refresh")
refresh_btn.clicked.connect(self.refresh)
bar.addWidget(refresh_btn)
self.all_check = QCheckBox("All")
self.all_check.setChecked(False)
self.all_check.setToolTip("Show all games at once (unchecked loads more as you scroll)")
self.all_check.toggled.connect(self._apply_filter)
bar.addWidget(self.all_check)
self.download_btn = QPushButton("Download selected")
self.download_btn.setEnabled(False)
self.download_btn.clicked.connect(self._emit_download)
bar.addWidget(self.download_btn)
self.scrape_btn = QPushButton("Scrape all")
self.scrape_btn.setToolTip("Fetch metadata and covers for every game in the store")
self.scrape_btn.clicked.connect(self._on_scrape_all)
bar.addWidget(self.scrape_btn)
layout.addLayout(bar)
self._debounce = QTimer(self)
self._debounce.setSingleShot(True)
self._debounce.setInterval(250)
self._debounce.timeout.connect(self._apply_filter)
self.stack = QStackedWidget()
self.grid = QListWidget()
self.grid.setViewMode(QListWidget.IconMode)
self.grid.setIconSize(QSize(self.ICON_W, self.ICON_H))
self.grid.setGridSize(QSize(self.GRID_W, self.GRID_H))
self.grid.setResizeMode(QListWidget.Adjust)
self.grid.setWordWrap(True)
self.grid.setUniformItemSizes(True)
self.grid.setSpacing(6)
self.grid.setSelectionMode(QAbstractItemView.ExtendedSelection)
self.grid.itemSelectionChanged.connect(self._on_selection_changed)
self.grid.setContextMenuPolicy(Qt.CustomContextMenu)
self.grid.customContextMenuRequested.connect(self._on_grid_menu)
self.grid.verticalScrollBar().valueChanged.connect(self._on_scroll)
self.stack.addWidget(self.grid)
self.table = QTableWidget(0, 5)
self.table.setHorizontalHeaderLabels(["In library", "Title", "Region", "Size", "Released"])
self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
header = self.table.horizontalHeader()
header.resizeSection(0, 80)
header.resizeSection(2, 90)
header.resizeSection(3, 90)
header.resizeSection(4, 110)
self.table.verticalHeader().setVisible(False)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setSelectionMode(QAbstractItemView.ExtendedSelection)
self.table.itemSelectionChanged.connect(self._on_selection_changed)
self.table.setContextMenuPolicy(Qt.CustomContextMenu)
self.table.customContextMenuRequested.connect(self._on_table_menu)
self.table.verticalScrollBar().valueChanged.connect(self._on_scroll)
self.table.setSortingEnabled(True)
self.stack.addWidget(self.table)
self.view_combo.currentIndexChanged.connect(self.stack.setCurrentIndex)
self.details = StoreDetailsPanel()
self.details.link_requested.connect(self._link_item)
self.details.download_requested.connect(self._download_current_item)
splitter = QSplitter(Qt.Horizontal)
splitter.addWidget(self.stack)
splitter.addWidget(self.details)
splitter.setStretchFactor(0, 1)
layout.addWidget(splitter, 1)
self.status = QLabel("")
layout.addWidget(self.status)
def set_store_config(self, local_dir, server_url, token, archive_id="", archive_sig="", archive_user=""):
self._store_dir = local_dir or ""
self._server_url = server_url or ""
self._token = token or ""
self._archive_id = archive_id or ""
self._archive_sig = archive_sig or ""
self._archive_user = archive_user or ""
if self._server_url:
self.status.setText(f"Store: {self._server_url}")
elif self._archive_id:
self.status.setText(f"Store: archive.org/{self._archive_id}")
elif self._store_dir:
self.status.setText(f"Store (local): {self._store_dir}")
else:
self.status.setText("Store not configured")
def download_source(self):
if self.source_combo.currentIndex() == 0:
return transfers.RemoteStoreSource(self._server_url, self._token)
if self.source_combo.currentIndex() == 1:
return transfers.ArchiveOrgSource(self._archive_id, self._archive_sig, self._archive_user)
return transfers.LocalShareSource(self._store_dir)
def _on_source_changed(self, _idx):
self._items = []
self._filtered = []
self._dirty = False
self.refresh()
def set_library_items(self, items):
norm, seen, cover = [], set(), {}
psp_norm, psp_seen = [], set()
for it in items or []:
# The store only contains PSP game discs, so ignore UMD videos when
# matching against the library (a game archive must never match a
# same-named movie and inherit its cover/metadata).
system = it.get("system")
if system != "games":
continue
rid = it.get("release_id")
for key in (it.get("canonical"), it.get("name")):
if not key:
continue
n = matcher.normalize_name(key)
if n and n not in seen:
seen.add(n)
norm.append(n)
if rid is not None:
cover[n] = (rid, system)
if it.get("on_psp") and n and n not in psp_seen:
psp_seen.add(n)
psp_norm.append(n)
self._lib_norm = norm
self._lib_cover = cover
self._psp_norm = psp_norm
if self._items:
self._apply_filter()
def refresh(self):
if self._closing:
return
if self.source_combo.currentIndex() == 0: # Server
if not self._server_url:
self.status.setText("Set the store server URL in Settings")
return
log.info("store: fetching from server %s", self._server_url)
self.status.setText("Fetching store from server…")
worker = RemoteStoreScanWorker(transfers.RemoteStoreSource(self._server_url, self._token))
elif self.source_combo.currentIndex() == 1: # Archive.org
if not self._archive_id:
self.status.setText("Set the Archive.org item in Settings")
return
log.info("store: fetching from archive.org %s", self._archive_id)
self.status.setText("Fetching store from archive.org…")
worker = RemoteStoreScanWorker(transfers.ArchiveOrgSource(self._archive_id, self._archive_sig, self._archive_user))
else:
if not self._store_dir:
self.status.setText("Set the local store folder in Settings (Advanced)")
return
log.info("store: scanning local folder %s", self._store_dir)
self.status.setText("Scanning local store…")
worker = StoreScanWorker(self._store_dir)
thread = QThread(self)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.results.connect(self._on_scan_results)
worker.failed.connect(self._on_scan_failed)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_scan_failed(self, message):
if self._closing:
return
self._items = []
self._filtered = []
self._dirty = False
log.warning("store unavailable: %s", message)
self.status.setText(f"Store unavailable: {message}")
def _on_scan_results(self, items):
if self._closing:
return
self._items = items
self._cover_attempted.clear()
log.info("store: %d archive(s)", len(items))
self.status.setText(f"{len(items)} archive(s) in store")
self._apply_filter()
def _schedule_filter(self, _text=None):
self._debounce.start()
def _apply_filter(self):
if not hasattr(self, "_items"):
return
q = self.search.text().strip().lower()
dates = info.cached_release_dates()
psp_f = self.psp_filter.currentText() if hasattr(self, "psp_filter") else "All PSP"
filtered = []
for it in self._items:
disp = store.display_name(it["name"]).lower()
if q and q not in disp and q not in it["name"].lower():
continue
n = matcher.normalize_name(it["name"])
it["in_library"] = False
it["on_psp"] = False
it["_lib_cover"] = None
if n:
# Exact normalized match only: region/language/version tags are
# already stripped, and substring matching would false-positive
# sequels (e.g. "Open Tee" matching "Open Tee 2").
for ln in self._lib_norm:
if n == ln:
it["in_library"] = True
it["_lib_cover"] = self._lib_cover.get(ln)
break
for pn in self._psp_norm:
if n == pn:
it["on_psp"] = True
break
if psp_f == "On PSP" and not it.get("on_psp"):
continue
if psp_f == "Not on PSP" and it.get("on_psp"):
continue
lc = it.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(it["name"]), "games"
it["release_date"] = dates.get((str(rid), system)) or ""
filtered.append(it)
filtered.sort(key=self._store_sort_key)
self._filtered = filtered
if self.all_check.isChecked():
show = len(filtered)
else:
show = min(self.RECENT_LIMIT, len(filtered))
self._populate(filtered, show)
@staticmethod
def _store_sort_key(it):
d = it.get("release_date") or ""
mtime = it.get("mtime_ns") or 0
if d:
try:
date_int = int(d.replace("-", "")[:8])
except (ValueError, AttributeError):
date_int = 0
return (0, -date_int, -mtime)
return (1, 0, -mtime)
CHUNK = 150 # items filled per timer tick
COVER_BATCH = 200 # covers fetched for the initially-visible area
COVER_SCROLL_BATCH = 100
RECENT_LIMIT = 150 # default: show only the newest N store items
DATE_BATCH = 150 # release-date metadata fetched per batch
def _populate(self, items, show_count):
self._generation += 1
self._filtered = list(items)
self._show_count = show_count
self._fill_index = 0
self._cover_queue = []
self._cover_next = 0
self._date_queue = []
self._date_next = 0
self._cover_attempted.clear()
self._placeholder = QIcon(_placeholder_pixmap(self.ICON_W, self.ICON_H))
# Cancel any in-flight cover loaders (stale generation).
for loader in self._loaders:
loader.cancel()
self._loaders = []
self.grid.clear()
self.table.setUpdatesEnabled(False)
self.table.setSortingEnabled(False)
self.table.setRowCount(show_count)
if show_count <= 0:
self._finish_fill()
return
if self._fill_timer is None:
self._fill_timer = QTimer(self)
self._fill_timer.timeout.connect(self._fill_chunk)
self._fill_timer.start()
def _fill_chunk(self):
if self._closing:
return
items = self._filtered
if self._fill_index >= self._show_count:
self._finish_fill()
return
end = min(self._fill_index + self.CHUNK, self._show_count)
for i in range(self._fill_index, end):
self._add_grid_item(items[i], i)
self._add_table_row(items[i], i)
self._fill_index = end
self.status.setText(f"Loading… {self._fill_index}/{self._show_count}")
def _add_grid_item(self, it, row):
label = store.display_name(it["name"])
if len(label) > 44:
label = label[:43] + "…"
item = QListWidgetItem(self._placeholder, label)
item.setToolTip(it["name"])
item.setTextAlignment(Qt.AlignHCenter | Qt.AlignTop)
item.setData(Qt.UserRole, it)
self.grid.addItem(item)
lc = it.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(it["name"]), "games"
p = covers.cover_path(rid, system)
if p.exists() and covers.file_is_image(p):
item.setIcon(QIcon(_fit_pixmap(QPixmap(str(p)), self.ICON_W, self.ICON_H)))
elif rid not in self._cover_attempted:
self._cover_attempted.add(rid)
self._cover_queue.append((row, {
"name": covers.plain_title(it["name"]) or it["name"],
"release_id": rid,
"system": system,
}))
def _add_table_row(self, it, row):
self.table.setItem(row, 0, QTableWidgetItem("✓" if it.get("in_library") else ""))
title = QTableWidgetItem(store.display_name(it["name"]))
title.setData(Qt.UserRole, it)
self.table.setItem(row, 1, title)
self.table.setItem(row, 2, QTableWidgetItem(store.region(it["name"])))
self.table.setItem(row, 3, NumericItem(f"{it['size'] / (1024 * 1024):.1f} MB", it["size"]))
lc = it.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(it["name"]), "games"
rel = it.get("release_date") or ""
self.table.setItem(row, 4, QTableWidgetItem(rel))
if not rel and system == "games":
self._date_queue.append((row, str(rid), system,
covers.plain_title(store.display_name(it["name"])) or it["name"]))
def _finish_fill(self):
if self._fill_timer is not None:
self._fill_timer.stop()
self.table.setUpdatesEnabled(True)
self.table.setSortingEnabled(True)
total = len(self._filtered)
shown = self._show_count
if not self.all_check.isChecked() and total > shown:
self.status.setText(f"Latest {shown} of {total} — scroll for more")
else:
self.status.setText(f"{shown} archive(s) in store")
self._load_cover_batch(self.COVER_BATCH)
self._load_date_batch(self.DATE_BATCH)
def _load_cover_batch(self, n):
if self._closing or not self._cover_queue:
return
end = min(self._cover_next + n, len(self._cover_queue))
if self._cover_next >= end:
return
tasks = self._cover_queue[self._cover_next:end]
self._cover_next = end
self._start_cover_loader(self._generation, tasks)
def _active_scrollbar(self):
if self.view_combo.currentIndex() == 0:
return self.grid.verticalScrollBar()
return self.table.verticalScrollBar()
def _on_scroll(self, *_):
if self._closing:
return
bar = self._active_scrollbar()
if bar.maximum() - bar.value() >= 600:
return
self._load_cover_batch(self.COVER_SCROLL_BATCH)
self._load_date_batch(self.DATE_BATCH)
self._load_more()
def _load_more(self):
if self._closing or self.all_check.isChecked():
return
if self._show_count >= len(self._filtered):
return
self._show_count = min(self._show_count + self.RECENT_LIMIT, len(self._filtered))
self.table.setSortingEnabled(False)
self.table.setUpdatesEnabled(False)
self.table.setRowCount(self._show_count)
if self._fill_timer is not None:
self._fill_timer.start()
def _load_date_batch(self, n):
if self._closing or not self._date_queue:
return
end = min(self._date_next + n, len(self._date_queue))
if self._date_next >= end:
return
tasks = self._date_queue[self._date_next:end]
self._date_next = end
self._start_date_loader(self._generation, tasks)
def _start_date_loader(self, gen, tasks):
if self._closing or not tasks:
return
thread = QThread(self)
loader = StoreDateLoader(tasks, gen)
loader.moveToThread(thread)
thread.started.connect(loader.run)
loader.loaded.connect(self._on_date_loaded)
loader.finished.connect(thread.quit)
self._threads.append(thread)
self._loaders.append(loader)
thread.start()
def _on_date_loaded(self, gen, row, date):
if gen != self._generation or not date:
return
if 0 <= row < len(self._filtered):
self._filtered[row]["release_date"] = date
cell = self.table.item(row, 4)
if cell is not None:
cell.setText(date)
def _start_cover_loader(self, gen, tasks):
if self._closing or not tasks:
return
thread = QThread(self)
loader = StoreCoverLoader(tasks, gen, quick=True)
loader.moveToThread(thread)
thread.started.connect(loader.run)
loader.loaded.connect(self._on_cover_loaded)
loader.finished.connect(thread.quit)
self._threads.append(thread)
self._loaders.append(loader)
thread.start()
def _on_cover_loaded(self, gen, row, path):
if gen != self._generation:
return
item = self.grid.item(row)
if item is not None:
item.setIcon(QIcon(_fit_pixmap(QPixmap(path), self.ICON_W, self.ICON_H)))
it = item.data(Qt.UserRole)
shown = getattr(self.details, "_item", None)
if it is not None and shown is not None and it.get("path") == shown.get("path"):
lc = it.get("_lib_cover")
rid, system = (lc if lc is not None else (covers.stable_key(it["name"]), "games"))
self.details._set_cover(rid, system)
def _selected_items(self):
if self.view_combo.currentIndex() == 0:
return [i.data(Qt.UserRole) for i in self.grid.selectedItems() if i.data(Qt.UserRole)]
items = []
for r in sorted({i.row() for i in self.table.selectionModel().selectedRows()}):
cell = self.table.item(r, 1)
it = cell.data(Qt.UserRole) if cell is not None else None
if it is not None:
items.append(it)
return items
def _on_selection_changed(self):
items = self._selected_items()
self.download_btn.setEnabled(bool(items))
if items:
self._show_details(items[0])
def _show_details(self, it):
if self._closing:
return
self.details.set_item(it)
self._meta_gen += 1
gen = self._meta_gen
lc = it.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(it["name"]), "games"
cached = info.load(str(rid), system)
if cached is not None:
self.details.set_metadata(cached)
return
title = covers.plain_title(store.display_name(it["name"])) or it["name"]
thread = QThread(self)
worker = StoreMetaWorker(title, gen, str(rid), system)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_store_meta)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_store_meta(self, gen, meta):
if gen != self._meta_gen:
return
self.details.set_metadata(meta)
def _download_current_item(self):
item = getattr(self.details, "_item", None)
if item is not None:
self.download_requested.emit([item])
def _link_item(self):
item = getattr(self.details, "_item", None)
if item is None:
return
lc = item.get("_lib_cover")
rid, system = lc if lc is not None else (covers.stable_key(item["name"]), "games")
dlg = ScrapeDialog(item, self, rid=rid, system=system)
if dlg.exec() != QDialog.Accepted:
return
meta = dlg.selected_meta
if meta:
title = meta.get("title") or None
info.save(str(rid), system, meta, manual=True, title=title)
item["title_override"] = title or None
meta["manual"] = True
self.details.set_item(item)
self.details.set_metadata(meta)
self._refresh_item_cover(item, rid, system)
def _on_grid_menu(self, pos):
item = self.grid.itemAt(pos)
if item is None:
return
it = item.data(Qt.UserRole)
if it:
menu = QMenu(self)
menu.addAction("Download", lambda: self.download_requested.emit([it]))
if it.get("path"):
menu.addAction("Open containing folder", lambda: open_containing_folder(it.get("path")))
menu.addAction("Scrape…", lambda: self._search_cover(it))
menu.exec(self.grid.mapToGlobal(pos))
def _on_table_menu(self, pos):
item = self.table.itemAt(pos)
if item is None:
return
row = item.row()
cell = self.table.item(row, 1)
it = cell.data(Qt.UserRole) if cell is not None else None
if it is not None:
menu = QMenu(self)
menu.addAction("Download", lambda: self.download_requested.emit([it]))
if it.get("path"):
menu.addAction("Open containing folder", lambda: open_containing_folder(it.get("path")))
menu.addAction("Scrape…", lambda: self._search_cover(it))
menu.exec(self.table.viewport().mapToGlobal(pos))
def _search_cover(self, it):
rid = covers.stable_key(it["name"])
dlg = ScrapeDialog({"name": it["name"], "system": "games"}, self, rid=rid, system="games")
if dlg.exec() == QDialog.Accepted:
meta = dlg.selected_meta
if meta:
title = meta.get("title") or None
info.save(str(rid), "games", meta, manual=True, title=title)
it["title_override"] = title or None
self._refresh_item_cover(it, rid, "games")
def _refresh_item_cover(self, it, rid, system):
p = covers.cover_path(rid, system)
if not p.exists():
return
pm = _fit_pixmap(QPixmap(str(p)), self.ICON_W, self.ICON_H)
for i in range(self.grid.count()):
item = self.grid.item(i)
if item.data(Qt.UserRole) is it:
item.setIcon(QIcon(pm))
break
shown = getattr(self.details, "_item", None)
if shown is it:
self.details._set_cover(rid, system)
def _emit_download(self):
self.download_requested.emit(self._selected_items())
def _scrape_tasks(self):
tasks = []
seen = set()
for it in self._items:
lc = it.get("_lib_cover")
if lc is not None:
rid, system = lc
else:
rid, system = covers.stable_key(it["name"]), "games"
if system != "games":
continue
key = (str(rid), system)
if key in seen:
continue
seen.add(key)
tasks.append((str(rid), system, it["name"],
covers.plain_title(store.display_name(it["name"])) or it["name"]))
return tasks
def _on_scrape_all(self):
if self._closing or self._scraping:
return
tasks = self._scrape_tasks()
if not tasks:
QMessageBox.information(self, "Nothing to scrape", "The store is empty.")
return
self._scraping = True
dlg = QProgressDialog("Fetching metadata and covers for every game…", "Cancel", 0, len(tasks), self)
dlg.setWindowTitle("Scrape store")
dlg.setWindowModality(Qt.WindowModal)
dlg.setMinimumDuration(0)
dlg.setValue(0)
self._scrape_dialog = dlg
thread = QThread(self)
worker = StoreScrapeAllWorker(tasks)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.progress.connect(self._on_scrape_progress)
worker.finished.connect(thread.quit)
dlg.canceled.connect(worker.cancel)
worker.finished.connect(self._on_scrape_done)
self._scrape_thread = thread
self._scrape_worker = worker
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_scrape_progress(self, done, total):
dlg = self._scrape_dialog
if dlg is not None:
dlg.setLabelText(f"Fetching metadata + covers… {done}/{total}")
dlg.setValue(done)
def _on_scrape_done(self):
self._scraping = False
dlg = self._scrape_dialog
if dlg is not None:
dlg.close()
self._scrape_dialog = None
thread = self._scrape_thread
worker = self._scrape_worker
if thread is not None and thread in self._threads:
self._threads.remove(thread)
if worker is not None and worker in self._workers:
self._workers.remove(worker)
self._scrape_thread = None
self._scrape_worker = None
if not self._closing:
self._apply_filter()
def shutdown(self):
self._closing = True
if self._fill_timer is not None:
self._fill_timer.stop()
for loader in self._loaders:
loader.cancel()
if self._scrape_worker is not None:
self._scrape_worker.cancel()
dlg = self._scrape_dialog
if dlg is not None:
dlg.close()
for t in self._threads:
try:
if t.isRunning():
t.quit()
t.wait()
except RuntimeError:
pass
self._threads = []
self._workers = []
self._loaders = []
class QueuePage(QWidget):
"""Live view of transfer jobs with progress, cancel, and delete."""
def __init__(self, parent=None):
super().__init__(parent)
self.queue = None
self._rows = {}
layout = QVBoxLayout(self)
header = QHBoxLayout()
header.addWidget(QLabel("<b>Transfers</b>"))
header.addStretch(1)
self.clear_finished_btn = QPushButton("Clear finished")
self.clear_finished_btn.clicked.connect(self._clear_finished)
header.addWidget(self.clear_finished_btn)
layout.addLayout(header)
self.table = QTableWidget(0, 6)
self.table.setHorizontalHeaderLabels(["Item", "Status", "Progress", "Destination", "", ""])
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
self.table.horizontalHeader().setSectionResizeMode(4, QHeaderView.ResizeToContents)
self.table.horizontalHeader().setSectionResizeMode(5, QHeaderView.ResizeToContents)
self.table.verticalHeader().setVisible(False)
layout.addWidget(self.table)
def set_queue(self, q):
self.queue = q
q.job_added.connect(self._on_added)
q.job_updated.connect(self._on_updated)
q.job_finished.connect(self._on_updated)
q.job_removed.connect(self._on_removed)
def _on_added(self, job):
row = self.table.rowCount()
self.table.insertRow(row)
self.table.setItem(row, 0, QTableWidgetItem(job["title"]))
self.table.setItem(row, 1, QTableWidgetItem(self._status_text(job)))
bar = QProgressBar()
bar.setRange(0, 1000)
bar.setValue(int(job["progress"] * 1000))
self.table.setCellWidget(row, 2, bar)
self.table.setItem(row, 3, QTableWidgetItem(job["dest"]))
cancel = QPushButton("Cancel")
cancel.clicked.connect(lambda _=False, jid=job["id"]: self.queue.cancel(jid))
self.table.setCellWidget(row, 4, cancel)
delete = QPushButton("✕")
delete.setToolTip("Remove this entry from the queue history")
delete.setFixedWidth(28)
delete.clicked.connect(lambda _=False, jid=job["id"]: self.queue.remove(jid))
self.table.setCellWidget(row, 5, delete)
self._rows[job["id"]] = row
def _on_updated(self, job):
row = self._rows.get(job["id"])
if row is None:
return
status_item = self.table.item(row, 1)
if status_item is not None:
status_item.setText(self._status_text(job))
bar = self.table.cellWidget(row, 2)
if isinstance(bar, QProgressBar):
if job["progress"] < 0:
bar.setRange(0, 0)
else:
bar.setRange(0, 1000)
bar.setValue(int(job["progress"] * 1000))
cancel = self.table.cellWidget(row, 4)
if isinstance(cancel, QPushButton):
cancel.setEnabled(job["status"] in ("queued", "running"))
delete = self.table.cellWidget(row, 5)
if isinstance(delete, QPushButton):
delete.setEnabled(job["status"] not in ("running", "cancelling"))
def _on_removed(self, job_id):
row = self._rows.pop(job_id, None)
if row is None:
return
self.table.removeRow(row)
for jid, r in self._rows.items():
if r > row:
self._rows[jid] = r - 1
def _clear_finished(self):
if self.queue is not None:
self.queue.clear_finished()
@staticmethod
def _status_text(job):
status = job["status"]
if status == "running":
return f"{job['stage']}…" if job["stage"] else "Running…"
if status == "done":
return "Done"
if status == "skipped":
return "Already present"
if status == "cancelled":
return "Cancelled"
if status == "cancelling":
return "Cancelling…"
if status == "error":
return f"Error: {job['error'][:80]}"
return "Queued"
class UmdReadWorker(QObject):
"""Read PSP_GAME / UMD_VIDEO PARAM.SFO from a mounted UMD ISO."""
result = Signal(int, dict) # gen, payload
finished = Signal()
def __init__(self, gen, iso_path, mount):
super().__init__()
self.gen = gen
self.iso_path = iso_path
self.mount = mount
def run(self):
try:
identity = iso_reader.read_identity(self.iso_path)
size = Path(self.iso_path).stat().st_size
icon = None
for dir_name in ("PSP_GAME", "UMD_VIDEO"):
try:
icon = iso_reader.read_file(self.iso_path, dir_name, "ICON0.PNG")
break
except iso_reader.IsoError:
continue
self.result.emit(self.gen, {
"mount": self.mount,
"iso": self.iso_path,
"identity": identity,
"size": size,
"icon": icon,
})
except Exception as exc: # noqa: BLE001
self.result.emit(self.gen, {"error": f"{type(exc).__name__}: {exc}"})
self.finished.emit()
class PspInventoryWorker(QObject):
"""Read a PSP's inventory + free space off the GUI thread."""
result = Signal(int, dict) # gen, payload
finished = Signal()
def __init__(self, gen, root):
super().__init__()
self.gen = gen
self.root = root
def run(self):
try:
inventory = psp.list_installed(self.root)
usage = psp.free_space(self.root)
except Exception as exc: # noqa: BLE001
self.result.emit(self.gen, {"error": f"{type(exc).__name__}: {exc}"})
self.finished.emit()
return
self.result.emit(self.gen, {"inventory": inventory, "usage": usage})
self.finished.emit()
class PspSyncPage(QWidget):
"""Show what is installed on a connected PSP and read a mounted UMD."""
device_changed = Signal(str) # psp root path ("" when none)
install_requested = Signal(list) # library items to copy to the PSP
dump_requested = Signal(str, str, str) # umd iso path, system, serial
def __init__(self, parent=None):
super().__init__(parent)
self._root = ""
self._inventory = {"games": [], "videos": [], "homebrew": [], "pkg": []}
self._library_items = []
self._lib_raw = {}
self._lib_norm = {}
self._gen = 0
self._umd_gen = 0
self._umd_iso = None
self._umd_serial = None
self._umd_kind = None
self._threads = []
self._workers = []
layout = QVBoxLayout(self)
device_row = QHBoxLayout()
device_row.addWidget(QLabel("PSP device:"))
self.device_combo = QComboBox()
self.device_combo.setMinimumContentsLength(24)
self.device_combo.currentTextChanged.connect(self._on_device_selected)
device_row.addWidget(self.device_combo, 1)
browse_btn = QPushButton("Browse…")
browse_btn.clicked.connect(self._browse)
device_row.addWidget(browse_btn)
refresh_btn = QPushButton("Refresh")
refresh_btn.clicked.connect(self.refresh)
device_row.addWidget(refresh_btn)
eject_btn = QPushButton("Eject")
eject_btn.setToolTip("Unmount the PSP so it is safe to unplug")
eject_btn.clicked.connect(self._eject)
device_row.addWidget(eject_btn)
layout.addLayout(device_row)
self.status = QLabel("No PSP connected")
layout.addWidget(self.status)
self.tabs = QTabWidget()
self.games_table = self._make_table(["", "Name", "Category", "Added", "Release", "Region", "Size"], stretch_col=1)
self.videos_table = self._make_table(["", "Name", "Added", "Size"], stretch_col=1)
self.homebrew_table = self._make_table(["", "ID", "Name", "Added", "Size"], stretch_col=2)
self.pkg_table = self._make_table(["", "Name", "Added", "Size"], stretch_col=1)
self.tabs.addTab(self.games_table, "Games")
self.tabs.addTab(self.videos_table, "UMD Videos")
self.tabs.addTab(self.homebrew_table, "Homebrew / PSN")
self.tabs.addTab(self.pkg_table, "PKG")
self.tabs.addTab(self._build_umd_tab(), "UMD")
self.tabs.addTab(self._build_sync_tab(), "Sync from Library")
layout.addWidget(self.tabs, 1)
for tbl in (self.games_table, self.videos_table, self.homebrew_table, self.pkg_table):
tbl.itemSelectionChanged.connect(self._update_delete_enabled)
tbl.itemChanged.connect(self._update_delete_enabled)
tbl.setContextMenuPolicy(Qt.CustomContextMenu)
tbl.customContextMenuRequested.connect(self._on_table_menu)
self.tabs.currentChanged.connect(self._update_delete_enabled)
del_row = QHBoxLayout()
del_row.addStretch(1)
self.delete_btn = QPushButton("Delete checked from PSP")
self.delete_btn.setEnabled(False)
self.delete_btn.clicked.connect(self._delete_selected)
del_row.addWidget(self.delete_btn)
del_row.addStretch(1)
layout.addLayout(del_row)
self.refresh()
@staticmethod
def _make_table(headers, stretch_col=1):
table = QTableWidget(0, len(headers))
table.setHorizontalHeaderLabels(headers)
table.horizontalHeader().setSectionResizeMode(stretch_col, QHeaderView.Stretch)
table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Fixed)
table.horizontalHeader().resizeSection(0, 32)
table.verticalHeader().setVisible(False)
table.setEditTriggers(QAbstractItemView.NoEditTriggers)
table.setSelectionBehavior(QAbstractItemView.SelectRows)
table.setSelectionMode(QAbstractItemView.ExtendedSelection)
table.setSortingEnabled(True)
return table
def root(self):
return self._root
def inventory(self):
return self._inventory
def set_library_items(self, items):
self._library_items = list(items or [])
self._lib_raw = {}
self._lib_norm = {}
for it in self._library_items:
label = it.get("canonical") or it.get("name") or ""
if not label:
continue
stem = Path(label).stem
self._lib_raw.setdefault(stem.lower(), it)
n = matcher.normalize_name(stem)
if n:
self._lib_norm.setdefault(n, it)
self._filter_sync()
self._populate(self._inventory)
def _build_sync_tab(self):
page = QWidget()
layout = QVBoxLayout(page)
bar = QHBoxLayout()
bar.addWidget(QLabel("Search:"))
self.sync_search = QLineEdit()
self.sync_search.setPlaceholderText("Filter library…")
self.sync_search.textChanged.connect(self._filter_sync)
bar.addWidget(self.sync_search, 1)
self.sync_type = QComboBox()
self.sync_type.addItems(["All", "Games", "Videos"])
self.sync_type.currentTextChanged.connect(self._filter_sync)
bar.addWidget(self.sync_type)
self.sync_installed = QComboBox()
self.sync_installed.addItems(["All", "Not installed", "Installed"])
self.sync_installed.setToolTip("Filter by install status on the PSP")
self.sync_installed.currentTextChanged.connect(self._filter_sync)
bar.addWidget(self.sync_installed)
layout.addLayout(bar)
self.sync_table = QTableWidget(0, 4)
self.sync_table.setHorizontalHeaderLabels(["Sync", "Title", "Type", "Size"])
self.sync_table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
self.sync_table.verticalHeader().setVisible(False)
self.sync_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.sync_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.sync_table.setSortingEnabled(True)
layout.addWidget(self.sync_table, 1)
self.sync_btn = QPushButton("Sync selected to PSP")
self.sync_btn.setEnabled(False)
self.sync_btn.clicked.connect(self._emit_install)
layout.addWidget(self.sync_btn)
return page
def _build_umd_tab(self):
page = QWidget()
layout = QVBoxLayout(page)
bar = QHBoxLayout()
bar.addWidget(QLabel("UMD (USB UMD mode):"))
self.umd_refresh_btn = QPushButton("Refresh")
self.umd_refresh_btn.clicked.connect(self.refresh_umd)
bar.addWidget(self.umd_refresh_btn)
self.umd_dump_btn = QPushButton("Copy to library")
self.umd_dump_btn.setEnabled(False)
self.umd_dump_btn.setToolTip("Copy this UMD into your library, then scan and rename it")
self.umd_dump_btn.clicked.connect(self._emit_dump)
bar.addWidget(self.umd_dump_btn)
bar.addStretch(1)
layout.addLayout(bar)
self.umd_status = QLabel("No UMD mounted")
layout.addWidget(self.umd_status)
summary_row = QHBoxLayout()
self.umd_icon = QLabel()
self.umd_icon.setFixedWidth(168)
self.umd_icon.setAlignment(Qt.AlignCenter)
self.umd_icon.setVisible(False)
summary_row.addWidget(self.umd_icon)
self.umd_summary = QLabel("")
self.umd_summary.setWordWrap(True)
summary_row.addWidget(self.umd_summary, 1)
layout.addLayout(summary_row)
self.umd_table = QTableWidget(0, 2)
self.umd_table.setHorizontalHeaderLabels(["Field", "Value"])
self.umd_table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
self.umd_table.verticalHeader().setVisible(False)
self.umd_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.umd_table.setSelectionMode(QAbstractItemView.NoSelection)
layout.addWidget(self.umd_table, 1)
return page
def refresh_umd(self):
self._umd_gen += 1
gen = self._umd_gen
mounts = psp.umd_mounts()
log.info("umd: %d mount(s) detected", len(mounts))
if not mounts:
self._clear_umd()
return
mount = mounts[0]
iso = psp.find_umd_iso(mount)
if iso is None:
self._clear_umd()
return
self.umd_status.setText(f"Reading UMD… ({Path(mount).name})")
thread = QThread(self)
worker = UmdReadWorker(gen, str(iso), mount)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_umd_result)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_umd_result(self, gen, res):
if gen != self._umd_gen:
return
self.umd_table.setRowCount(0)
if "error" in res:
self._umd_iso = None
self.umd_status.setText(f"UMD read failed: {res['error']}")
self.umd_summary.setText("")
self.umd_dump_btn.setEnabled(False)
self.umd_icon.setVisible(False)
return
ident = res["identity"]
sfo = ident.get("sfo") or {}
title = str(ident.get("title") or "Unknown")
serial = str(ident.get("serial") or Path(res["mount"]).name)
kind = "Game" if ident.get("kind") == "game" else "Video"
self._umd_iso = Path(res["iso"])
self._umd_serial = serial
self._umd_kind = ident.get("kind")
self.umd_dump_btn.setEnabled(True)
self._set_umd_icon(res.get("icon"))
size_mb = res["size"] / (1024 * 1024)
self.umd_status.setText(f"UMD: {title}")
self.umd_summary.setText(
f"<b>{title}</b> · {serial} ({kind}) · {size_mb:.1f} MB · volume {Path(res['mount']).name}"
)
rows = [
("Title", title),
("Disc ID", serial),
("Type", kind),
("Size", f"{size_mb:.1f} MB"),
("Volume", Path(res["mount"]).name),
("File", Path(res["iso"]).name),
]
for key in sorted(sfo):
if key in ("TITLE", "DISC_ID"):
continue
rows.append((str(key), str(sfo[key])))
self.umd_table.setRowCount(len(rows))
for r, (k, v) in enumerate(rows):
self.umd_table.setItem(r, 0, QTableWidgetItem(k))
self.umd_table.setItem(r, 1, QTableWidgetItem(v))
def _set_umd_icon(self, data):
if data:
img = QImage.fromData(data)
if not img.isNull():
pm = QPixmap.fromImage(img).scaled(160, 120, Qt.KeepAspectRatio, Qt.SmoothTransformation)
self.umd_icon.setPixmap(pm)
self.umd_icon.setVisible(True)
return
self.umd_icon.setVisible(False)
def _clear_umd(self):
self._umd_iso = None
self._umd_serial = None
self._umd_kind = None
self.umd_status.setText("No UMD mounted")
self.umd_summary.setText("")
self.umd_table.setRowCount(0)
self.umd_dump_btn.setEnabled(False)
self.umd_icon.setVisible(False)
def _emit_dump(self):
if self._umd_iso is None or not self._umd_iso.exists():
return
system = "videos" if self._umd_kind == "video" else "games"
self.dump_requested.emit(str(self._umd_iso), system, self._umd_serial or "")
def _filter_sync(self):
if not hasattr(self, "sync_table"):
return
q = self.sync_search.text().strip().lower()
typ = self.sync_type.currentText()
inst = self.sync_installed.currentText() if hasattr(self, "sync_installed") else "All"
self.sync_table.setUpdatesEnabled(False)
self.sync_table.setSortingEnabled(False)
self.sync_table.setRowCount(0)
selectable = 0
for it in self._library_items:
if typ == "Games" and it.get("system") != "games":
continue
if typ == "Videos" and it.get("system") != "videos":
continue
on_psp = bool(it.get("on_psp"))
if inst == "Not installed" and on_psp:
continue
if inst == "Installed" and not on_psp:
continue
label = it.get("title_override") or it.get("canonical") or it.get("name") or ""
if q and q not in label.lower():
continue
row = self.sync_table.rowCount()
self.sync_table.insertRow(row)
cb = QTableWidgetItem()
if on_psp:
cb.setCheckState(Qt.Checked)
cb.setFlags(Qt.ItemIsSelectable | Qt.ItemIsEnabled)
cb.setToolTip("Already on PSP")
else:
cb.setCheckState(Qt.Unchecked)
cb.setFlags(Qt.ItemIsSelectable | Qt.ItemIsUserCheckable | Qt.ItemIsEnabled)
selectable += 1
self.sync_table.setItem(row, 0, cb)
title = QTableWidgetItem(label)
title.setData(Qt.UserRole, it)
self.sync_table.setItem(row, 1, title)
self.sync_table.setItem(row, 2, QTableWidgetItem("Game" if it.get("system") == "games" else "Video"))
size = it.get("size")
self.sync_table.setItem(row, 3, NumericItem(f"{size / (1024 * 1024):.1f} MB" if size else "", size or 0))
self.sync_table.setSortingEnabled(True)
self.sync_table.setUpdatesEnabled(True)
self.sync_btn.setEnabled(selectable > 0 and bool(self._root))
def _emit_install(self):
items = []
for row in range(self.sync_table.rowCount()):
cb = self.sync_table.item(row, 0)
if (cb is not None and cb.checkState() == Qt.Checked
and bool(cb.flags() & Qt.ItemIsUserCheckable)):
cell = self.sync_table.item(row, 1)
it = cell.data(Qt.UserRole) if cell is not None else None
if it is not None:
items.append(it)
if items:
self.install_requested.emit(items)
def refresh(self):
current = self.device_combo.currentText()
mounts = psp.detect_mounts()
umd = psp.umd_mounts()
log.info("psp sync: %d mount(s), %d umd volume(s) detected", len(mounts), len(umd))
self.device_combo.blockSignals(True)
self.device_combo.clear()
self.device_combo.addItem("")
for m in mounts:
self.device_combo.addItem(m, ("stick", m))
for u in umd:
self.device_combo.addItem(f"UMD: {Path(u).name}", ("umd", u))
if current and current in mounts:
self.device_combo.setCurrentText(current)
elif mounts:
self.device_combo.setCurrentIndex(1) # auto-select the first PSP found
self.device_combo.blockSignals(False)
self._root = self._selected_root()
self.refresh_inventory()
self.refresh_umd()
self.device_changed.emit(self._root)
def _selected_root(self):
data = self.device_combo.currentData()
if isinstance(data, tuple) and data[0] == "stick":
return data[1]
return ""
def refresh_inventory(self):
self._gen += 1
gen = self._gen
if not self._root:
self._inventory = {"games": [], "videos": [], "homebrew": [], "pkg": []}
self.status.setText("No PSP connected")
self._populate(self._inventory)
return
log.info("reading psp inventory from %s…", self._root)
self.status.setText("Reading PSP…")
thread = QThread(self)
worker = PspInventoryWorker(gen, self._root)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_inventory)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_inventory(self, gen, res):
if gen != self._gen:
return
if "error" in res:
self._inventory = {"games": [], "videos": [], "homebrew": [], "pkg": []}
self.status.setText(f"Error reading PSP: {res['error']}")
log.warning("psp inventory error: %s", res["error"])
else:
inventory = res["inventory"]
self._inventory = inventory
log.info("psp inventory: %d game(s), %d video(s), %d homebrew, %d pkg",
len(inventory["games"]), len(inventory["videos"]),
len(inventory["homebrew"]), len(inventory.get("pkg", [])))
usage = res.get("usage")
parts = [
f"{len(inventory['games'])} game(s)",
f"{len(inventory['videos'])} video(s)",
f"{len(inventory['homebrew'])} homebrew",
]
if usage is not None:
free_mb = usage["free"] / (1024 * 1024)
total_mb = usage["total"] / (1024 * 1024)
parts.append(f"free {free_mb:.0f} / {total_mb:.0f} MB")
self.status.setText(" · ".join(parts))
self._populate(self._inventory)
self._update_delete_enabled()
self.device_changed.emit(self._root)
def _populate(self, inventory):
self._fill_table(self.games_table, inventory.get("games", []), "games")
self._fill_table(self.videos_table, inventory.get("videos", []), "videos")
self._fill_table(self.homebrew_table, inventory.get("homebrew", []), "homebrew")
self._fill_table(self.pkg_table, inventory.get("pkg", []), "pkg")
@staticmethod
def _fmt_date(ts):
if not ts:
return ""
try:
return datetime.datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
except (OSError, OverflowError, ValueError):
return ""
def _match_library(self, psp_item):
stem = Path(psp_item["name"]).stem
lib = self._lib_raw.get(stem.lower())
if lib is None:
n = matcher.normalize_name(stem)
lib = self._lib_norm.get(n) if n else None
return lib or {}
def _fill_table(self, table, items, kind):
table.setUpdatesEnabled(False)
table.setSortingEnabled(False)
table.setRowCount(0)
for it in items:
row = table.rowCount()
table.insertRow(row)
cb = QTableWidgetItem()
cb.setFlags(Qt.ItemIsUserCheckable | Qt.ItemIsEnabled)
cb.setCheckState(Qt.Unchecked)
cb.setData(Qt.UserRole, it)
table.setItem(row, 0, cb)
name = QTableWidgetItem(it["name"])
name.setData(Qt.UserRole, it)
size = it.get("size") or 0
added = self._fmt_date(it.get("mtime"))
if kind == "games":
table.setItem(row, 1, name)
lib = self._match_library(it)
table.setItem(row, 2, QTableWidgetItem(it.get("category") or ""))
table.setItem(row, 3, QTableWidgetItem(added))
table.setItem(row, 4, QTableWidgetItem(lib.get("release_date") or ""))
table.setItem(row, 5, QTableWidgetItem(lib.get("region") or ""))
table.setItem(row, 6, NumericItem(f"{size / (1024 * 1024):.1f} MB" if size else "", size))
elif kind == "homebrew":
table.setItem(row, 1, QTableWidgetItem(it.get("id") or ""))
table.setItem(row, 2, name)
table.setItem(row, 3, QTableWidgetItem(added))
table.setItem(row, 4, NumericItem(f"{size / (1024 * 1024):.1f} MB" if size else "", size))
else:
table.setItem(row, 1, name)
table.setItem(row, 2, QTableWidgetItem(added))
table.setItem(row, 3, NumericItem(f"{size / (1024 * 1024):.1f} MB" if size else "", size))
table.setSortingEnabled(True)
table.setUpdatesEnabled(True)
def _current_inventory_table(self):
w = self.tabs.currentWidget()
if w in (self.games_table, self.videos_table, self.homebrew_table, self.pkg_table):
return w
return None
def _update_delete_enabled(self, *_):
table = self._current_inventory_table()
self.delete_btn.setEnabled(table is not None and bool(self._checked_items(table)))
def _checked_items(self, table):
items = []
for row in range(table.rowCount()):
cb = table.item(row, 0)
if cb is not None and cb.checkState() == Qt.Checked:
it = cb.data(Qt.UserRole)
if it is not None:
items.append(it)
return items
def _on_table_menu(self, pos):
table = self.sender()
if not isinstance(table, QTableWidget):
return
item = table.itemAt(pos)
if item is None:
return
it = None
cb = table.item(item.row(), 0)
if cb is not None:
it = cb.data(Qt.UserRole)
if it is None:
cell = table.item(item.row(), 1)
it = cell.data(Qt.UserRole) if cell is not None else None
if it is None:
return
menu = QMenu(self)
menu.addAction("Delete…", lambda: self._delete_items([it]))
menu.exec(table.viewport().mapToGlobal(pos))
def _delete_selected(self):
table = self._current_inventory_table()
if table is None:
return
items = self._checked_items(table)
if not items:
QMessageBox.information(self, "Delete from PSP",
"Check the box next to the item(s) you want to delete.")
return
self._delete_items(items)
def _delete_items(self, items):
if not items:
return
preview = ", ".join(it["name"] for it in items[:5])
if len(items) > 5:
preview += f", … (+{len(items) - 5} more)"
res = QMessageBox.question(
self, "Delete from PSP",
f"Permanently delete {len(items)} item(s) from the PSP?\n\n{preview}",
QMessageBox.Yes | QMessageBox.Cancel, QMessageBox.Cancel,
)
if res != QMessageBox.Yes:
return
failed = 0
for it in items:
target = Path(it["path"]).parent if it.get("kind") == "homebrew" else Path(it["path"])
if not transfers.remove_path(target):
failed += 1
if failed:
QMessageBox.warning(self, "Delete incomplete", f"{failed} item(s) could not be deleted.")
self.refresh_inventory()
def _on_device_selected(self, text):
self._root = self._selected_root()
self.refresh_inventory()
self.refresh_umd()
self.device_changed.emit(self._root)
def _browse(self):
path = QFileDialog.getExistingDirectory(self, "Select PSP root")
if not path:
return
if not psp.is_psp_root(path):
QMessageBox.warning(self, "Not a PSP", "That folder does not look like a PSP memory stick (no PSP/ or ISO/ folder).")
return
mounts = [self.device_combo.itemText(i) for i in range(self.device_combo.count())]
if path not in mounts:
self.device_combo.addItem(path, ("stick", path))
self.device_combo.setCurrentText(path)
def _eject(self):
if not self._root:
QMessageBox.information(self, "Eject", "No PSP selected.")
return
ok, msg = psp.eject(self._root)
log.info("eject %s -> %s", self._root, "ok" if ok else f"failed: {msg}")
self.refresh()
if ok:
self.status.setText("Ejected — safe to disconnect")
else:
QMessageBox.warning(self, "Eject failed", msg)
def shutdown(self):
for t in self._threads:
try:
if t.isRunning():
t.quit()
t.wait()
except RuntimeError:
pass
self._threads = []
self._workers = []
def build_library_items():
"""Build the library item list from the DB with a single joined query."""
items = []
seen = set()
overrides = info.override_titles()
dates = info.cached_release_dates()
conn = init_db(config.DB_PATH)
try:
rows = conn.execute(
"SELECT f.path, f.name, f.size, f.crc32, f.md5, f.sha1, f.serial, f.title, f.release_id, "
"r.id AS rid, r.name AS rname, r.region, r.serial AS rserial, d.system "
"FROM files f "
"LEFT JOIN releases r ON r.id = f.release_id "
"LEFT JOIN dats d ON d.id = r.dat_id "
"ORDER BY f.name"
).fetchall()
for row in rows:
if not Path(row["path"]).exists():
continue
key = (row["name"], row["size"])
if key in seen:
continue
seen.add(key)
if row["rid"] is not None:
release = {
"id": row["rid"], "name": row["rname"],
"region": row["region"], "serial": row["rserial"], "system": row["system"],
}
canon_serial = row["serial"] or (
iso_reader.normalize_serial(row["rserial"])
if iso_reader.looks_like_serial(row["rserial"]) else None
)
canonical = namer.canonical_name(release, canon_serial) if release.get("name") else None
else:
record = {
"path": row["path"], "name": row["name"], "size": row["size"],
"serial": row["serial"], "title": row["title"],
"crc32": row["crc32"], "md5": row["md5"], "sha1": row["sha1"],
}
ident = pipeline.identify(conn, record)
release = ident.get("release")
canonical = ident.get("canonical")
if release and release.get("id"):
conn.execute(
"UPDATE files SET release_id = ? WHERE path = ?",
(release.get("id"), row["path"]),
)
if not release or not release.get("name"):
continue
rid = release.get("id") or covers.stable_key(release["name"])
system = release.get("system")
items.append({
"name": release["name"],
"release_id": rid,
"system": system,
"canonical": canonical,
"serial": release.get("serial"),
"region": release.get("region") or namer.region_from_name(release["name"]),
"title_override": overrides.get((system, str(rid))),
"release_date": dates.get((str(rid), system)) or "",
"path": row["path"],
"size": row["size"],
})
conn.commit()
finally:
conn.close()
return items
class LibraryWorker(QObject):
result = Signal(list)
finished = Signal()
def run(self):
try:
items = build_library_items()
except Exception: # noqa: BLE001
items = []
self.result.emit(items)
self.finished.emit()
class SaveSyncPlanWorker(QObject):
"""Compute the save-sync plan off the GUI thread (content hashing is I/O-heavy)."""
plan = Signal(list, bool) # plan, switch_to_queue
finished = Signal()
def __init__(self, locations, switch_tab=False):
super().__init__()
self.locations = locations
self.switch_tab = switch_tab
def run(self):
try:
plan = saves.sync_plan(self.locations)
except Exception: # noqa: BLE001
plan = []
self.plan.emit(plan, self.switch_tab)
self.finished.emit()
class SavesPage(QWidget):
"""PSP save-data sync between PSP, PPSSPP, and a cloud folder."""
sync_requested = Signal(list, bool) # (plan, switch_to_queue)
def __init__(self, parent=None):
super().__init__(parent)
self._psp_dir = ""
self._ppsspp_dir = ""
self._cloud_dir = ""
self._syncing = False
self._closing = False
self._threads = []
self._workers = []
layout = QVBoxLayout(self)
bar = QHBoxLayout()
refresh_btn = QPushButton("Refresh")
refresh_btn.clicked.connect(self.refresh)
bar.addWidget(refresh_btn)
self.sync_btn = QPushButton("Sync now")
self.sync_btn.clicked.connect(lambda: self.sync_now(True))
bar.addWidget(self.sync_btn)
bar.addStretch(1)
layout.addLayout(bar)
self.table = QTableWidget(0, 5)
self.table.setHorizontalHeaderLabels(["Save", "ID", "PSP", "PPSSPP", "Cloud"])
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
self.table.verticalHeader().setVisible(False)
self.table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setSortingEnabled(True)
self.table.horizontalHeader().setSortIndicator(0, Qt.AscendingOrder)
layout.addWidget(self.table)
self.status = QLabel("")
layout.addWidget(self.status)
def set_locations(self, psp_dir, ppsspp_dir, cloud_dir):
self._psp_dir = psp_dir or ""
self._ppsspp_dir = ppsspp_dir or ""
self._cloud_dir = cloud_dir or ""
self.refresh()
def _locations(self):
return {"psp": self._psp_dir, "ppsspp": self._ppsspp_dir, "cloud": self._cloud_dir}
def refresh(self):
locs = self._locations()
data = {name: {s["id"]: s for s in saves.list_saves(root)} for name, root in locs.items()}
all_ids = set()
for d in data.values():
all_ids.update(d)
sort_col = self.table.horizontalHeader().sortIndicatorSection()
sort_order = self.table.horizontalHeader().sortIndicatorOrder()
self.table.setSortingEnabled(False)
self.table.setRowCount(0)
for sid in sorted(all_ids):
row = self.table.rowCount()
self.table.insertRow(row)
title = sid
enc = False
for loc in ("psp", "ppsspp", "cloud"):
s = data[loc].get(sid)
if s is None:
continue
if s.get("title"):
title = s["title"]
if s.get("encrypted"):
enc = True
if enc:
title += " \u26a0"
t = QTableWidgetItem(title)
t.setToolTip(sid)
self.table.setItem(row, 0, t)
self.table.setItem(row, 1, QTableWidgetItem(sid))
for col, loc in enumerate(("psp", "ppsspp", "cloud"), start=2):
s = data[loc].get(sid)
if s is None:
self.table.setItem(row, col, QTableWidgetItem(""))
else:
stamp = datetime.datetime.fromtimestamp(s["mtime"]).strftime("%Y-%m-%d %H:%M")
item = QTableWidgetItem(stamp)
item.setToolTip(f"{s['size'] / 1024:.1f} KB")
self.table.setItem(row, col, item)
self.table.setSortingEnabled(True)
self.table.sortByColumn(sort_col, sort_order)
self.status.setText(f"{len(all_ids)} save(s) across {sum(1 for r in locs.values() if r)} location(s)")
def sync_now(self, switch_tab=False):
if self._syncing:
return
self._syncing = True
self.sync_btn.setEnabled(False)
self.status.setText("Comparing saves…")
thread = QThread(self)
worker = SaveSyncPlanWorker(self._locations(), switch_tab)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.plan.connect(self._on_sync_plan)
worker.finished.connect(thread.quit)
self._threads.append(thread)
self._workers.append(worker)
thread.start()
def _on_sync_plan(self, plan, switch_tab):
if self._closing:
return
self._syncing = False
self.sync_btn.setEnabled(True)
if not plan:
self.status.setText("Already in sync")
return
self.status.setText(f"Queued {len(plan)} sync operation(s)…")
self.sync_requested.emit(plan, switch_tab)
def shutdown(self):
self._closing = True
for t in self._threads:
if _thread_running(t):
t.quit()
t.wait()
self._threads = []
self._workers = []
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("psp preservation tool ~ quoije")
self.resize(1240, 780)
self._rows = []
self._renames = []
self.thread = None
self.worker = None
self._games_root = None
self._videos_root = None
self._library_cache = []
self._meta_threads = []
self._meta_workers = []
self._meta_gen = 0
self._closing = False
self._psp_root = ""
self._psp_index = {}
self._psp_paths = {}
self._pending_umd_dump = None
self._auto_rename_umd = None
self._library_threads = []
self._library_workers = []
self._library_loading = False
self.settings = QSettings("PSPPreserve", "PSPPreserve")
self._games_dir = self._first_existing(config.GAMES_ROOTS) or self.settings.value("games_dir", "") or ""
self._videos_dir = self._first_existing(config.VIDEO_ROOTS) or self.settings.value("videos_dir", "") or ""
self._store_dir = str(self.settings.value("store_dir", str(config.STORE_ROOT)) or "")
self._store_server_url = str(self.settings.value("store_server_url", config.STORE_SERVER_URL) or "")
self._store_token = str(self.settings.value("store_token", config.STORE_TOKEN) or "")
self._store_archive_id = str(self.settings.value("store_archive_id", config.ARCHIVE_ITEM) or "")
self._store_archive_sig = str(self.settings.value("store_archive_sig", config.ARCHIVE_LOGGED_IN_SIG) or "")
self._store_archive_user = str(self.settings.value("store_archive_user", config.ARCHIVE_LOGGED_IN_USER) or "")
self._ppsspp_saves_dir = str(self.settings.value("ppsspp_saves_dir", str(saves.default_ppsspp_saves())) or "")
self._cloud_saves_dir = str(self.settings.value("cloud_saves_dir", "") or "")
self._auto_sync_saves = str(self.settings.value("auto_sync_saves", "false")).lower() == "true"
self._ppsspp_exe = str(self.settings.value("ppsspp_exe", "") or "")
self._iso_category = str(self.settings.value("iso_category", "") or "")
log.info("games dir: %s", self._games_dir or "(none)")
log.info("videos dir: %s", self._videos_dir or "(none)")
log.info("store: %s", self._store_server_url or self._store_dir or "(none)")
log.info("archive.org item: %s", self._store_archive_id or "(none)")
central = QWidget()
self.setCentralWidget(central)
layout = QVBoxLayout(central)
layout.setContentsMargins(8, 8, 0, 8)
self.tabs = QTabWidget()
self.tabs.setStyleSheet("QTabWidget::pane { border: 0px; padding: 0px; }")
layout.addWidget(self.tabs)
self.queue = queue.TransferQueue(self)
self.queue.job_finished.connect(self._on_queue_job_done)
self.tabs.addTab(self._build_library_tab(), "Library")
self.store_page = StorePage()
self.store_page.set_store_config(self._store_dir, self._store_server_url, self._store_token,
self._store_archive_id, self._store_archive_sig, self._store_archive_user)
self.store_page.download_requested.connect(self._on_store_download)
self.tabs.addTab(self.store_page, "Store")
self.queue_page = QueuePage()
self.queue_page.set_queue(self.queue)
self.tabs.addTab(self.queue_page, "Queue")
self.psp_page = PspSyncPage()
self.psp_page.device_changed.connect(self._on_psp_device_changed)
self.psp_page.install_requested.connect(self._copy_items_to_psp)
self.psp_page.dump_requested.connect(self._on_umd_dump)
self.tabs.addTab(self.psp_page, "PSP Sync")
self.saves_page = SavesPage()
self.saves_page.sync_requested.connect(self._on_saves_sync)
self.tabs.addTab(self.saves_page, "Saves")
self.tabs.addTab(self._build_settings_tab(), "Settings")
self.tabs.currentChanged.connect(self._on_tab_changed)
self._on_psp_device_changed(self.psp_page.root())
self._update_saves_locations()
self.statusBar().showMessage("")
self._update_dir_status()
self._current_item = None
geometry = self.settings.value("window_geometry")
if geometry:
self.restoreGeometry(geometry)
QTimer.singleShot(0, self.load_library)
QTimer.singleShot(500, self.store_page.refresh)
if (str(self.settings.value("auto_scan", False)).lower() == "true"
and (self._games_dir or self._videos_dir)):
QTimer.singleShot(0, self._start_scan)
# -- Library tab ------------------------------------------------------- #
def _build_library_tab(self):
page = QWidget()
layout = QVBoxLayout(page)
m = layout.contentsMargins()
layout.setContentsMargins(m.left(), m.top(), 0, m.bottom())
bar = QHBoxLayout()
bar.addWidget(QLabel("Search:"))
self.library_search = QLineEdit()
self.library_search.setPlaceholderText("Filter by title…")
self.library_search.textChanged.connect(self._apply_library_filter)
bar.addWidget(self.library_search, 1)
self.library_type = QComboBox()
self.library_type.addItems(["All types", "Games", "Videos"])
self.library_type.currentTextChanged.connect(self._apply_library_filter)
bar.addWidget(self.library_type)
self.library_psp = QComboBox()
self.library_psp.addItems(["All PSP", "On PSP", "Not on PSP"])
self.library_psp.setToolTip("Filter by whether the disc is installed on the connected PSP")
self.library_psp.currentTextChanged.connect(self._apply_library_filter)
bar.addWidget(self.library_psp)
self.library_view = QComboBox()
self.library_view.addItems(["Cover view", "List view"])
bar.addWidget(self.library_view)
reload_btn = QPushButton("Reload")
reload_btn.clicked.connect(self.load_library)
bar.addWidget(reload_btn)
self.search_missing_btn = QPushButton("Scrape missing…")
self.search_missing_btn.setEnabled(False)
self.search_missing_btn.setToolTip("Scrape metadata and covers for games that are missing one")
bar.addWidget(self.search_missing_btn)
self.scan_shortcut_btn = QPushButton("Scan…")
self.scan_shortcut_btn.setToolTip("Scan your folders for new discs and apply renames")
self.scan_shortcut_btn.clicked.connect(self._open_scan)
bar.addWidget(self.scan_shortcut_btn)
layout.addLayout(bar)
splitter = QSplitter(Qt.Horizontal)
self.library_subtabs = QTabWidget()
self.library_subtabs.setStyleSheet("QTabWidget::pane { border: 0px; padding: 0px; }")
self.library_subtabs.addTab(self._build_library_view(), "Library")
self.library_subtabs.addTab(self._build_scan_view(), "Scan")
splitter.addWidget(self.library_subtabs)
self.details_panel = GameDetailsPanel()
splitter.addWidget(self.details_panel)
splitter.setStretchFactor(0, 1)
layout.addWidget(splitter)
self.library_subtabs.currentChanged.connect(self._on_library_subtab_changed)
self.cover_grid.selected.connect(self._on_library_selected)
self.cover_grid.cover_loaded.connect(self._on_cover_loaded)
self.cover_grid.missing_changed.connect(lambda n: self.search_missing_btn.setEnabled(n > 0))
self.search_missing_btn.clicked.connect(self.cover_grid.search_missing)
self.cover_grid.copy_requested.connect(self._copy_to_psp)
self.cover_grid.delete_requested.connect(self._delete_library_item)
self.cover_grid.delete_psp_requested.connect(self._delete_from_psp)
self.cover_grid.play_requested.connect(self._play_item_on_ppsspp)
self.cover_grid.scrape_requested.connect(self._scrape_grid_item)
self.details_panel.edit_requested.connect(self._edit_details)
self.details_panel.link_requested.connect(self._scrape_item)
self.details_panel.copy_requested.connect(self._copy_current_to_psp)
self.details_panel.delete_requested.connect(self._delete_current)
self.details_panel.delete_psp_requested.connect(self._delete_current_from_psp)
self.details_panel.play_requested.connect(self._play_current_on_ppsspp)
return page
def _open_scan(self):
self.library_subtabs.setCurrentWidget(self.scan_view)
self._start_scan()
def _on_library_subtab_changed(self, index):
self.details_panel.setVisible(self.library_subtabs.widget(index) is not self.scan_view)
def _build_library_view(self):
self.library_stack = QStackedWidget()
self.cover_grid = CoverShowcase()
self.library_stack.addWidget(self.cover_grid)
self.library_table = QTableWidget(0, 6)
self.library_table.setHorizontalHeaderLabels(["Title", "Type", "Region", "Serial", "Size", "Released"])
self.library_table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch)
self.library_table.verticalHeader().setVisible(False)
self.library_table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.library_table.setSelectionMode(QAbstractItemView.SingleSelection)
self.library_table.itemSelectionChanged.connect(self._on_library_table_selected)
self.library_table.itemDoubleClicked.connect(self._on_library_double_clicked)
self.library_table.setContextMenuPolicy(Qt.CustomContextMenu)
self.library_table.customContextMenuRequested.connect(self._on_library_table_menu)
self.library_table.setSortingEnabled(True)
self.library_stack.addWidget(self.library_table)
self.library_view.currentIndexChanged.connect(self.library_stack.setCurrentIndex)
return self.library_stack
# -- Scan view (inside the Library tab) -------------------------------- #
def _build_scan_view(self):
self.scan_view = QWidget()
layout = QVBoxLayout(self.scan_view)
controls = QHBoxLayout()
self.scan_btn = QPushButton("Scan")
self.scan_btn.clicked.connect(self._start_scan)
controls.addWidget(self.scan_btn)
self.cancel_btn = QPushButton("Cancel")
self.cancel_btn.setEnabled(False)
self.cancel_btn.clicked.connect(self._cancel_scan)
controls.addWidget(self.cancel_btn)
self.apply_btn = QPushButton("Apply renames")
self.apply_btn.setEnabled(False)
self.apply_btn.clicked.connect(self._apply_renames)
controls.addWidget(self.apply_btn)
self.hash_check = QCheckBox("Verify integrity (full hash)")
controls.addWidget(self.hash_check)
self.progress = QProgressBar()
controls.addWidget(self.progress, 1)
layout.addLayout(controls)
self.table = QTableWidget(0, 7)
self.table.setHorizontalHeaderLabels(
["Type", "Current filename", "Path", "Status", "Serial", "Canonical name", "Rename?"]
)
self.table.horizontalHeader().setStretchLastSection(False)
header = self.table.horizontalHeader()
header.resizeSection(0, 64)
header.resizeSection(1, 280)
header.resizeSection(2, 240)
header.resizeSection(5, 300)
header.setContextMenuPolicy(Qt.CustomContextMenu)
header.customContextMenuRequested.connect(self._show_column_menu)
self.table.itemChanged.connect(self._on_item_changed)
layout.addWidget(self.table)
return self.scan_view
def _show_column_menu(self, pos):
header = self.table.horizontalHeader()
menu = QMenu(self)
for col in range(self.table.columnCount()):
header_item = self.table.horizontalHeaderItem(col)
title = header_item.text() if header_item else f"Column {col}"
action = menu.addAction(title)
action.setCheckable(True)
action.setChecked(not self.table.isColumnHidden(col))
action.toggled.connect(lambda checked, c=col: self.table.setColumnHidden(c, not checked))
menu.exec(header.mapToGlobal(pos))
# -- Settings tab ------------------------------------------------------ #
def _build_settings_tab(self):
self.settings_page = SettingsPage(self._games_dir, self._videos_dir, self._store_dir,
self._ppsspp_saves_dir, self._cloud_saves_dir, self._ppsspp_exe,
self._store_server_url, self._store_token, self._store_archive_id,
self._store_archive_sig, self._store_archive_user, self._iso_category)
self.settings_page.saved.connect(self._on_settings_saved)
return self.settings_page
def _on_settings_saved(self):
self._games_dir = self.settings_page.games_dir()
self._videos_dir = self.settings_page.videos_dir()
self._store_dir = self.settings_page.store_dir()
self._store_server_url = self.settings_page.store_server_url()
self._store_token = self.settings_page.store_token()
self._store_archive_id = self.settings_page.archive_item()
self._store_archive_sig = self.settings_page.archive_sig()
self._store_archive_user = self.settings_page.archive_user()
self.settings.setValue("games_dir", self._games_dir)
self.settings.setValue("videos_dir", self._videos_dir)
self.settings.setValue("store_dir", self._store_dir)
self.settings.setValue("store_server_url", self._store_server_url)
self.settings.setValue("store_token", self._store_token)
self.settings.setValue("store_archive_id", self._store_archive_id)
self.settings.setValue("store_archive_sig", self._store_archive_sig)
self.settings.setValue("store_archive_user", self._store_archive_user)
self.store_page.set_store_config(self._store_dir, self._store_server_url, self._store_token,
self._store_archive_id, self._store_archive_sig, self._store_archive_user)
self.store_page.refresh()
self._ppsspp_saves_dir = self.settings_page.ppsspp_saves_dir()
self._cloud_saves_dir = self.settings_page.cloud_saves_dir()
self._auto_sync_saves = self.settings_page.auto_sync_saves()
self._ppsspp_exe = self.settings_page.ppsspp_exe()
self._iso_category = self.settings_page.iso_category()
self.settings.setValue("ppsspp_saves_dir", self._ppsspp_saves_dir)
self.settings.setValue("cloud_saves_dir", self._cloud_saves_dir)
self.settings.setValue("auto_sync_saves", self._auto_sync_saves)
self.settings.setValue("ppsspp_exe", self._ppsspp_exe)
self.settings.setValue("iso_category", self._iso_category)
self._update_saves_locations()
self._update_dir_status()
def _on_tab_changed(self, index):
pass # store builds eagerly and caches its view; no per-tab work needed
def _update_saves_locations(self):
if not hasattr(self, "saves_page"):
return
psp_dir = str(Path(self._psp_root) / "PSP" / "SAVEDATA") if self._psp_root else ""
self.saves_page.set_locations(psp_dir, self._ppsspp_saves_dir, self._cloud_saves_dir)
def _maybe_auto_sync(self):
if self._auto_sync_saves and self._psp_root:
self.saves_page.sync_now(False)
def _on_saves_sync(self, plan, switch_tab=False):
for op in plan:
src = transfers.TreeCopySource(op["src"])
self.queue.enqueue(src, op["dst_root"], {"dest_name": op["dst_name"]}, op["dst_name"], kind="save-sync")
if plan and switch_tab:
self.tabs.setCurrentWidget(self.queue_page)
def _refresh_saves(self):
if hasattr(self, "saves_page"):
self.saves_page.refresh()
# -- Library data ------------------------------------------------------ #
def load_library(self):
if self._closing or self._library_loading:
return
self._library_loading = True
self.statusBar().showMessage("Loading library…")
log.info("loading library…")
thread = QThread(self)
worker = LibraryWorker()
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_library_loaded)
worker.finished.connect(thread.quit)
thread.finished.connect(lambda t=thread, w=worker: self._forget_library_thread(t, w))
self._library_threads.append(thread)
self._library_workers.append(worker)
thread.start()
def _on_library_loaded(self, items):
self._library_loading = False
self._library_cache = items or []
log.info("library loaded: %d item(s)", len(self._library_cache))
self._apply_library_filter()
if hasattr(self, "store_page"):
self.store_page.set_library_items(self._library_cache)
self.statusBar().showMessage("")
def _forget_library_thread(self, thread, worker):
if thread in self._library_threads:
self._library_threads.remove(thread)
if worker in self._library_workers:
self._library_workers.remove(worker)
def _apply_library_filter(self):
if not hasattr(self, "_library_cache"):
return
q = self.library_search.text().strip().lower()
typ = self.library_type.currentText()
psp_f = self.library_psp.currentText() if hasattr(self, "library_psp") else "All PSP"
items = []
for it in self._library_cache:
it["on_psp"] = self._on_psp(it)
if typ == "Games" and it["system"] != "games":
continue
if typ == "Videos" and it["system"] != "videos":
continue
if q and q not in (it["canonical"] or it["name"]).lower():
continue
if psp_f == "On PSP" and not it.get("on_psp"):
continue
if psp_f == "Not on PSP" and it.get("on_psp"):
continue
items.append(it)
self._library_filtered = items
self._populate_library_table(items)
self.cover_grid.set_items(items)
if hasattr(self, "psp_page"):
self.psp_page.set_library_items(self._library_cache)
def _populate_library_table(self, items):
self.library_table.setUpdatesEnabled(False)
self.library_table.setSortingEnabled(False)
try:
self.library_table.setRowCount(len(items))
for row, it in enumerate(items):
title = QTableWidgetItem(it.get("title_override") or it["canonical"] or it["name"])
title.setData(Qt.UserRole, it)
if it.get("on_psp"):
title.setIcon(self._psp_dot_icon())
title.setToolTip("Installed on PSP")
self.library_table.setItem(row, 0, title)
self.library_table.setItem(row, 1, QTableWidgetItem("Game" if it["system"] == "games" else "Video"))
self.library_table.setItem(row, 2, QTableWidgetItem(it.get("region") or ""))
self.library_table.setItem(row, 3, QTableWidgetItem(it.get("serial") or ""))
size = it.get("size")
self.library_table.setItem(row, 4, NumericItem(f"{size / (1024 * 1024):.1f} MB" if size else "", size or 0))
self.library_table.setItem(row, 5, QTableWidgetItem(it.get("release_date") or ""))
finally:
self.library_table.setSortingEnabled(True)
self.library_table.setUpdatesEnabled(True)
def _on_library_table_selected(self):
row = self.library_table.currentRow()
cell = self.library_table.item(row, 0) if row >= 0 else None
if cell is not None:
it = cell.data(Qt.UserRole)
if it:
self._on_library_selected(it)
def _on_library_double_clicked(self, item):
if item is None:
return
cell = self.library_table.item(item.row(), 0)
it = cell.data(Qt.UserRole) if cell is not None else None
if it and it.get("system") != "videos":
self._play_item_on_ppsspp(it)
def _on_library_table_menu(self, pos):
item = self.library_table.itemAt(pos)
if item is None:
return
row = item.row()
cell = self.library_table.item(row, 0)
it = cell.data(Qt.UserRole) if cell is not None else None
if not it:
return
menu = QMenu(self)
if it.get("system") != "videos":
menu.addAction("Play on PPSSPP", lambda: self._play_item_on_ppsspp(it))
menu.addAction("Copy to PSP", lambda: self._copy_to_psp(it))
if it.get("on_psp"):
menu.addAction("Delete from PSP", lambda: self._delete_from_psp(it))
menu.addAction("Open containing folder", lambda: open_containing_folder(it.get("path")))
menu.addAction("Delete…", lambda: self._delete_library_item(it))
menu.exec(self.library_table.viewport().mapToGlobal(pos))
def _on_library_selected(self, item):
self._current_item = item
self.details_panel.set_item(item)
rid = str(item["release_id"])
system = item.get("system")
cached = info.load(rid, system)
if cached is not None:
self.details_panel.set_metadata(cached)
return
self._meta_gen += 1
gen = self._meta_gen
if item.get("title_override"):
title = item["title_override"]
elif system == "videos":
title = covers.movie_search_title(item["name"])
else:
title = covers.plain_title(item["name"])
thread = QThread(self)
worker = MetadataWorker(title, gen, system, rid)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.result.connect(self._on_metadata)
worker.finished.connect(thread.quit)
self._meta_threads.append(thread)
self._meta_workers.append(worker)
thread.start()
def _on_metadata(self, gen, meta):
if gen != self._meta_gen:
return
self.details_panel.set_metadata(meta)
def _on_cover_loaded(self, item):
current = getattr(self, "_current_item", None)
if current is not None and item.get("release_id") == current.get("release_id"):
self.details_panel._set_cover(item.get("release_id"), item.get("system"))
def _edit_details(self):
item = getattr(self, "_current_item", None)
if item is None:
return
rid = str(item["release_id"])
system = item.get("system")
dlg = TagDialog(item, self.details_panel.current_meta(), self)
if dlg.exec() != QDialog.Accepted:
return
if dlg.reset_pressed:
info.reset(rid, system)
item["title_override"] = None
self.cover_grid.update_item_label(item)
self._on_library_selected(item) # re-fetch and re-display
return
result = dlg.result_meta()
title = result.get("title") or None
info.save(rid, system, result, manual=True, title=title)
item["title_override"] = title or None
self.cover_grid.update_item_label(item)
result["manual"] = True
self.details_panel.set_item(item)
self.details_panel.set_metadata(result)
def _scrape_item(self):
item = getattr(self, "_current_item", None)
if item is None:
return
rid = str(item.get("release_id") or covers.stable_key(item.get("name") or ""))
system = item.get("system") or "games"
dlg = ScrapeDialog(item, self, rid=rid, system=system)
if dlg.exec() != QDialog.Accepted:
return
meta = dlg.selected_meta
if meta:
title = meta.get("title") or None
info.save(rid, system, meta, manual=True, title=title)
item["title_override"] = title or None
meta["manual"] = True
self.details_panel.set_item(item)
self.details_panel.set_metadata(meta)
self.cover_grid.update_item_label(item)
self.cover_grid.refresh_item(item)
def _scrape_grid_item(self, it):
self._current_item = it
self.details_panel.set_item(it)
self._scrape_item()
def _on_store_download(self, items):
if not items:
return
if not self._games_dir:
QMessageBox.warning(self, "No games directory", "Set the games directory in Settings, then scan.")
return
src = self.store_page.download_source()
log.info("downloading %d item(s) from store", len(items))
for it in items:
self.queue.enqueue(src, self._games_dir, it, store.display_name(it["name"]), kind="download")
self.tabs.setCurrentWidget(self.queue_page)
# -- PSP sync ---------------------------------------------------------- #
def _on_psp_device_changed(self, root):
self._psp_root = root or ""
log.info("psp device changed: %s", self._psp_root or "(none)")
if self._psp_root:
inventory = self.psp_page.inventory() or {}
self._psp_index = psp.build_index(inventory)
self._psp_paths = {"raw": {}, "norm": {}}
for group in ("games", "videos"):
for it in inventory.get(group, []):
stem = Path(it["name"]).stem
self._psp_paths["raw"].setdefault(stem.lower(), []).append(it)
n = matcher.normalize_name(stem)
if n:
self._psp_paths["norm"].setdefault(n, []).append(it)
log.info("psp index: %d raw / %d normalized names",
len(self._psp_index.get("raw", set())),
len(self._psp_index.get("names", set())))
else:
self._psp_index = {}
self._psp_paths = {}
if hasattr(self, "details_panel"):
self.details_panel.set_copy_available(bool(self._psp_root))
current = getattr(self, "_current_item", None)
self.details_panel.set_psp_installed(bool(current and current.get("on_psp")))
if hasattr(self, "_library_cache"):
self._apply_library_filter()
if hasattr(self, "store_page"):
self.store_page.set_library_items(self._library_cache)
self._update_saves_locations()
self._maybe_auto_sync()
def _on_psp(self, item):
if not self._psp_index:
return False
stem = Path(item.get("canonical") or item.get("name") or "").stem
if stem.lower() in self._psp_index.get("raw", set()):
return True
norm = matcher.normalize_name(stem)
return bool(norm) and norm in self._psp_index.get("names", set())
def _psp_dot_icon(self):
icon = getattr(self, "_psp_icon", None)
if icon is None:
pm = QPixmap(12, 12)
pm.fill(Qt.transparent)
painter = QPainter(pm)
painter.setRenderHint(QPainter.Antialiasing)
painter.setBrush(QColor(70, 200, 120))
painter.setPen(Qt.NoPen)
painter.drawEllipse(1, 1, 10, 10)
painter.end()
icon = QIcon(pm)
self._psp_icon = icon
return icon
def _copy_current_to_psp(self):
item = getattr(self, "_current_item", None)
if item is not None:
self._copy_to_psp(item)
def _play_current_on_ppsspp(self):
self._play_item_on_ppsspp(getattr(self, "_current_item", None))
def _play_item_on_ppsspp(self, item):
if item is None:
return
if item.get("system") == "videos":
return # UMD Video discs can't be played in PPSSPP
path = item.get("path")
if not path or not Path(path).exists():
QMessageBox.warning(self, "Missing file", f"File not found:\n{path}")
return
log.info("launching ppsspp: %s", path)
ok, msg = launcher.launch_ppsspp(path, self._ppsspp_exe)
if not ok:
QMessageBox.warning(self, "PPSSPP", msg)
def _delete_current(self):
item = getattr(self, "_current_item", None)
if item is not None:
self._delete_library_item(item)
def _delete_library_item(self, item):
if item is None:
return
path = Path(item.get("path") or "")
if not path.exists():
QMessageBox.warning(self, "Missing file", f"File not found:\n{path}")
return
name = item.get("title_override") or item.get("canonical") or item.get("name") or path.name
res = QMessageBox.question(
self, "Delete disc",
f"Permanently delete this disc from disk?\n\n{name}\n{path}",
QMessageBox.Yes | QMessageBox.Cancel, QMessageBox.Cancel,
)
if res != QMessageBox.Yes:
return
if not transfers.remove_path(path):
QMessageBox.critical(self, "Delete failed", f"Could not delete:\n{path}")
return
log.info("deleted %s", path)
try:
conn = get_connection(config.DB_PATH)
conn.execute("DELETE FROM files WHERE path = ?", (str(path),))
conn.commit()
conn.close()
except Exception: # noqa: BLE001
pass
self.load_library()
def _copy_to_psp(self, item):
if item is None:
return
if not self._psp_root:
QMessageBox.warning(self, "No PSP connected", "Connect the PSP and select it in the PSP Sync tab first.")
return
if self._enqueue_psp_copy(item):
self.tabs.setCurrentWidget(self.queue_page)
def _copy_items_to_psp(self, items):
if not items:
return
if not self._psp_root:
QMessageBox.warning(self, "No PSP connected", "Connect the PSP and select it in the PSP Sync tab first.")
return
queued = 0
for item in items:
if self._enqueue_psp_copy(item):
queued += 1
if queued:
self.tabs.setCurrentWidget(self.queue_page)
def _enqueue_psp_copy(self, item):
if not item.get("path") or not Path(item["path"]).exists():
QMessageBox.warning(self, "Missing file", f"File not found:\n{item.get('path')}")
return False
system = item.get("system")
psp_root = Path(self._psp_root)
if system == "videos":
dest_dir = psp_root / "ISO" / "VIDEO"
else:
cat = "".join(c for c in (self._iso_category or "").strip() if c not in "/\\")
dest_dir = (psp_root / "ISO" / cat) if cat else (psp_root / "ISO")
dest_name = item.get("canonical") or item.get("name") or Path(item["path"]).name
if not dest_name.lower().endswith(".iso"):
dest_name += ".iso"
src = transfers.FileCopySource(item["path"])
log.info("copying to psp: %s -> %s", item["path"], dest_dir / dest_name)
self.queue.enqueue(src, dest_dir, {"dest_name": dest_name}, dest_name, kind="psp-install")
return True
def _on_umd_dump(self, iso_path, system, serial):
src = Path(iso_path)
if not src.exists():
QMessageBox.warning(self, "Missing UMD", f"UMD image not found:\n{iso_path}")
return
dest_root = self._videos_dir if system == "videos" else self._games_dir
if not dest_root:
QMessageBox.warning(self, "No target directory",
"Set the games/videos directories in Settings before dumping a UMD.")
return
serial = (serial or "").strip()
dest_name = f"UMD9660 - {serial}.iso" if serial else "UMD9660.iso"
log.info("umd dump: %s -> %s", src, Path(dest_root) / dest_name)
self._pending_umd_dump = Path(dest_root) / dest_name
self.queue.enqueue(transfers.FileCopySource(str(src)), Path(dest_root),
{"dest_name": dest_name}, f"Dump {serial or src.name}", kind="umd-dump")
self.tabs.setCurrentWidget(self.queue_page)
def _rename_umd_dump(self):
target = self._auto_rename_umd
self._auto_rename_umd = None
if target is None:
return
try:
target_resolved = Path(target).resolve()
except OSError:
return
canonical = None
for rec in self._rows:
try:
same = Path(rec["path"]).resolve() == target_resolved
except OSError:
same = False
if same and rec.get("method") != "unknown" and rec.get("canonical"):
canonical = rec["canonical"]
break
if not canonical:
return
src = Path(target)
if not src.exists() or src.name == canonical:
return
dst = src.with_name(canonical)
if dst.exists():
return
conn = init_db(config.DB_PATH)
try:
scanner.rename_and_recache(conn, src, dst)
conn.commit()
finally:
conn.close()
log.info("renamed umd dump: %s -> %s", src.name, canonical)
def _delete_current_from_psp(self):
self._delete_from_psp(getattr(self, "_current_item", None))
def _delete_from_psp(self, item):
if item is None:
return
if not self._psp_root:
QMessageBox.warning(self, "No PSP connected", "Connect the PSP and select it in the PSP Sync tab first.")
return
if not item.get("on_psp"):
QMessageBox.information(self, "Not on PSP", "That disc is not installed on the PSP.")
return
stem = Path(item.get("canonical") or item.get("name") or "").stem
matches = list(self._psp_paths.get("raw", {}).get(stem.lower(), []))
if not matches:
n = matcher.normalize_name(stem)
matches = list(self._psp_paths.get("norm", {}).get(n, []))
if not matches:
QMessageBox.warning(self, "Not found", f"Could not find the installed copy on the PSP:\n{stem}")
return
name = item.get("title_override") or item.get("canonical") or item.get("name") or stem
preview = "\n".join(m["path"] for m in matches[:5])
if len(matches) > 5:
preview += f"\n… (+{len(matches) - 5} more)"
res = QMessageBox.question(
self, "Delete from PSP",
f"Delete this disc from the PSP?\n\n{name}\n\n{preview}",
QMessageBox.Yes | QMessageBox.Cancel, QMessageBox.Cancel,
)
if res != QMessageBox.Yes:
return
failed = 0
for m in matches:
if not transfers.remove_path(Path(m["path"])):
failed += 1
log.info("deleted %d file(s) from psp: %s", len(matches) - failed, name)
if failed:
QMessageBox.warning(self, "Delete incomplete", f"{failed} file(s) could not be deleted.")
self._refresh_psp()
def _refresh_psp(self):
if hasattr(self, "psp_page"):
self.psp_page.refresh_inventory()
def _on_queue_job_done(self, job):
if self._closing:
return
status = job.get("status")
kind = job.get("kind")
if kind == "umd-dump" and status in ("done", "skipped"):
if self._pending_umd_dump is not None:
self._auto_rename_umd = self._pending_umd_dump
self._pending_umd_dump = None
QTimer.singleShot(0, self._start_scan)
if status != "done":
return
if kind == "download":
remaining = any(j.status in ("queued", "running", "cancelling") for j in self.queue.jobs)
if not remaining:
QTimer.singleShot(0, self._start_scan)
if kind == "psp-install":
QTimer.singleShot(0, self._refresh_psp)
if kind == "save-sync":
QTimer.singleShot(0, self._refresh_saves)
# -- Common ------------------------------------------------------------ #
@staticmethod
def _first_existing(roots):
for root in roots:
if Path(root).exists():
return str(root)
return ""
def _update_dir_status(self):
parts = []
if self._games_dir:
parts.append(f"Games: {self._games_dir}")
if self._videos_dir:
parts.append(f"Videos: {self._videos_dir}")
self.statusBar().showMessage(" | ".join(parts) if parts else "Set directories in Settings, then scan.")
def _start_scan(self):
if self._closing:
return
games_text = self._games_dir
videos_text = self._videos_dir
roots = [Path(t) for t in (games_text, videos_text) if t]
self._games_root = Path(games_text) if games_text else None
self._videos_root = Path(videos_text) if videos_text else None
if not roots:
QMessageBox.warning(self, "No directories", "Set at least one directory in Settings, then scan.")
return
self._rows = []
self._renames = []
self.table.setRowCount(0)
self.progress.setRange(0, 0)
log.info("scanning roots: %s (workers: %d, hash: %s)",
[str(r) for r in roots],
int(self.settings.value("scan_workers", 1) or 1),
self.hash_check.isChecked())
self.scan_btn.setEnabled(False)
self.cancel_btn.setEnabled(True)
self.apply_btn.setEnabled(False)
self.statusBar().showMessage("Scanning…")
self.thread = QThread()
workers = int(self.settings.value("scan_workers", 1) or 1)
self.worker = ScanWorker(
roots, config.DB_PATH,
hash_files=self.hash_check.isChecked(),
workers=workers,
)
self.worker.moveToThread(self.thread)
self.thread.started.connect(self.worker.run)
self.worker.record.connect(self._on_record)
self.worker.prompt.connect(self._on_prompt)
self.worker.done.connect(self._on_done)
self.worker.cancelled.connect(self._on_cancelled)
self.worker.failed.connect(self._on_failed)
self.thread.start()
def _cancel_scan(self):
if self.worker:
self.worker.cancel()
def _on_record(self, rec):
self._rows.append(rec)
rec["_type"] = self._record_type(rec)
row = self.table.rowCount()
self.table.insertRow(row)
readonly = Qt.ItemIsSelectable | Qt.ItemIsEnabled
item = QTableWidgetItem(self._type_label(rec["_type"]))
item.setFlags(readonly)
self.table.setItem(row, 0, item)
item = QTableWidgetItem(rec["name"])
item.setFlags(readonly)
self.table.setItem(row, 1, item)
item = QTableWidgetItem(rec["path"])
item.setFlags(readonly)
item.setToolTip(rec["path"])
self.table.setItem(row, 2, item)
item = QTableWidgetItem(rec["method"])
item.setFlags(readonly)
self.table.setItem(row, 3, item)
item = QTableWidgetItem(rec.get("serial") or "")
item.setFlags(readonly)
self.table.setItem(row, 4, item)
canonical = rec.get("canonical") or ""
item = QTableWidgetItem(canonical)
item.setFlags(readonly | Qt.ItemIsEditable) # editable target name
self.table.setItem(row, 5, item)
do_rename = rec["method"] != "unknown" and canonical and Path(rec["path"]).name != canonical
cb = QCheckBox()
cb.setChecked(do_rename)
cb.toggled.connect(lambda _checked=False: self._update_apply_enabled())
if do_rename:
self._renames.append((rec["path"], canonical))
self.table.setCellWidget(row, 6, cb)
def _on_item_changed(self, item):
if item.column() != 5:
return
row = item.row()
widget = self.table.cellWidget(row, 6)
if isinstance(widget, QCheckBox):
widget.setChecked(bool(item.text().strip()))
self._update_apply_enabled()
def _update_apply_enabled(self):
enabled = False
for i in range(self.table.rowCount()):
widget = self.table.cellWidget(i, 6)
cell = self.table.item(i, 5)
name_item = self.table.item(i, 1)
canonical = (cell.text().strip() if cell else "") or ""
name = name_item.text() if name_item else ""
if (isinstance(widget, QCheckBox) and widget.isChecked()
and canonical and name != canonical):
enabled = True
break
self.apply_btn.setEnabled(enabled)
def _record_type(self, rec):
release = rec.get("release")
if release and release.get("system"):
return release["system"]
try:
p = Path(rec["path"]).resolve()
except Exception: # noqa: BLE001
return "?"
if self._videos_root is not None and p.is_relative_to(self._videos_root.resolve()):
return "videos"
if self._games_root is not None and p.is_relative_to(self._games_root.resolve()):
return "games"
return "?"
@staticmethod
def _type_label(kind):
return {"games": "Game", "videos": "Video"}.get(kind, "?")
def _on_prompt(self, rec, candidates):
chosen = self._ask_user(rec, candidates)
if self.worker is not None:
self.worker.choose(chosen)
def _ask_user(self, rec, candidates):
dlg = QDialog(self)
dlg.setWindowTitle("Identify disc")
layout = QVBoxLayout(dlg)
layout.addWidget(QLabel(
f"Multiple catalog entries match:\n{rec['name']}\n\nPick the correct one:"
))
combo = QComboBox()
for cand in candidates:
combo.addItem(cand["name"])
layout.addWidget(combo)
buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
buttons.accepted.connect(dlg.accept)
buttons.rejected.connect(dlg.reject)
layout.addWidget(buttons)
if dlg.exec() == QDialog.Accepted:
idx = combo.currentIndex()
return candidates[idx] if 0 <= idx < len(candidates) else None
return None
def _on_done(self, count):
self.progress.setRange(0, 1)
self.progress.setValue(1)
self.scan_btn.setEnabled(True)
self.cancel_btn.setEnabled(False)
self._update_apply_enabled()
matched = sum(1 for r in self._rows if r["method"] != "unknown")
unknown = len(self._rows) - matched
parts = [f"Scanned {len(self._rows)} — matched {matched}, unknown {unknown}"]
try:
conn = get_connection(config.DB_PATH)
totals = {
r["system"]: r["c"]
for r in conn.execute(
"SELECT d.system, COUNT(*) AS c FROM releases r JOIN dats d ON d.id = r.dat_id GROUP BY d.system"
)
}
conn.close()
on_disk = {}
for r in self._rows:
if r.get("release"):
on_disk.setdefault(r["release"]["system"], set()).add(r["release"].get("id"))
for system in sorted(totals):
have = len(on_disk.get(system, set()))
parts.append(f"{system} {have}/{totals[system]}")
except Exception: # noqa: BLE001
pass
self.statusBar().showMessage(" | ".join(parts))
log.info("scan complete: %s", " | ".join(parts))
self._thread_cleanup()
self._rename_umd_dump()
self.load_library()
def _on_failed(self, msg):
self.progress.setRange(0, 1)
self.progress.setValue(1)
self.scan_btn.setEnabled(True)
self.cancel_btn.setEnabled(False)
self.statusBar().showMessage("Scan failed")
log.error("scan failed: %s", msg)
QMessageBox.critical(self, "Scan failed", msg)
self._thread_cleanup()
def _on_cancelled(self):
self.progress.setRange(0, 1)
self.progress.setValue(1)
self.scan_btn.setEnabled(True)
self.cancel_btn.setEnabled(False)
self._update_apply_enabled()
self.statusBar().showMessage("Scan cancelled")
log.info("scan cancelled")
self._thread_cleanup()
def _thread_cleanup(self):
if self.thread:
self.thread.quit()
self.thread.wait()
self.thread = None
self.worker = None
def _open_sources(self):
pass
def _apply_renames(self):
conn = init_db(config.DB_PATH)
renamed = 0
try:
for i, rec in enumerate(self._rows):
widget = self.table.cellWidget(i, 6)
if not (isinstance(widget, QCheckBox) and widget.isChecked()):
continue
cell = self.table.item(i, 5)
canonical = (cell.text().strip() if cell else "") or rec.get("canonical") or ""
if not canonical:
continue
src = Path(rec["path"])
dst = src.with_name(canonical)
if src.exists() and not dst.exists() and src.name != canonical:
scanner.rename_and_recache(conn, src, dst)
renamed += 1
conn.commit()
finally:
conn.close()
self._renames = []
self.apply_btn.setEnabled(False)
QMessageBox.information(self, "Done", f"Renamed {renamed} file(s).")
self.load_library()
def closeEvent(self, event):
self._closing = True
self.settings.setValue("window_geometry", self.saveGeometry())
if self.worker is not None:
if getattr(self.worker, "_prompt_pending", False):
self.worker.choose(None)
self.worker.cancel()
self._thread_cleanup()
self.cover_grid.shutdown()
for t in self._meta_threads:
if _thread_running(t):
t.quit()
t.wait()
for t in self._library_threads:
if _thread_running(t):
t.quit()
t.wait()
if hasattr(self, "queue"):
self.queue.shutdown()
if hasattr(self, "store_page"):
self.store_page.shutdown()
if hasattr(self, "psp_page"):
self.psp_page.shutdown()
if hasattr(self, "saves_page"):
self.saves_page.shutdown()
event.accept()
def _start_bundle_check(window):
"""Download any missing offline cover bundles in the background."""
server = getattr(window, "_store_server_url", "") or ""
try:
covers.ensure_cover_bundles(server)
except Exception: # noqa: BLE001 - background best-effort
pass
def main():
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)-14s %(message)s")
init_db() # migrate legacy on-share DB/covers to local storage, then schema
log.info("starting psp-preserve (db: %s)", config.DB_PATH)
app = QApplication(sys.argv)
window = MainWindow()
window.show()
log.info("main window ready")
threading.Thread(target=_start_bundle_check, args=(window,), daemon=True).start()
return app.exec()
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink