Last active
September 8, 2026 10:32
-
-
Save gitgotgitgotit/bfe0f85c312c06a111556a872e31568b to your computer and use it in GitHub Desktop.
Quod Libet plugin LOG checker (EAC)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # -*- coding: utf-8 -*- | |
| # eac_log_checker.py — Quod Libet plugin | |
| # | |
| # Scores CD rip logs (EAC / XLD / whipper / CUERipper) next to the | |
| # selected track(s) using a Cambia server (https://github.com/arg274/cambia), | |
| # the same backend used by the foobar2000/MusicBee "loggers" integration | |
| # (https://gitlab.com/SuperSaltyGamer/loggers). | |
| # | |
| # By default it talks to the public instance at logs.musichoarders.xyz. | |
| # For privacy, or to work offline, point it at your own `cambia --server` | |
| # instance in the plugin preferences. | |
| # | |
| # Install: copy this file to ~/.quodlibet/plugins/songsmenu/ | |
| # (create the folder if it doesn't exist), then enable it in | |
| # Music -> Plugins. | |
| # | |
| # License: MIT (same spirit as Cambia itself) | |
| import json | |
| import os | |
| import threading | |
| import urllib.error | |
| import urllib.request | |
| from gi.repository import Gtk, GLib, Pango | |
| from quodlibet import _ | |
| from quodlibet.plugins.songsmenu import SongsMenuPlugin | |
| from quodlibet.plugins import PluginConfigMixin | |
| try: | |
| from quodlibet.qltk.msg import ErrorMessage | |
| except ImportError: # pragma: no cover - older/newer QL versions | |
| ErrorMessage = None | |
| DEFAULT_SERVER = "https://logs.musichoarders.xyz" | |
| UPLOAD_PATH = "/api/v1/upload" | |
| USER_AGENT = "quodlibet-eac-log-checker/1.0 (+https://github.com/arg274/cambia)" | |
| CLASS_COLORS = { | |
| "Critical": "#e01b24", | |
| "Bad": "#e5a50a", | |
| "Neutral": "#9a9996", | |
| "Good": "#33d17a", | |
| "Perfect": "#26a269", | |
| } | |
| FIELD_LABELS = { | |
| "Encoding": "Encoding", | |
| "RipperVersion": "Ripper version", | |
| "Drive": "Drive", | |
| "Ripper": "Ripper", | |
| "Offset": "Read offset", | |
| "Cache": "Audio cache", | |
| "TestAndCopy": "Test & copy", | |
| "Encoder": "Encoder", | |
| "Checksum": "Checksum", | |
| "MediaType": "Media type", | |
| "ReadMode": "Read mode", | |
| "MaxRetryCount": "Max retry count", | |
| "AccurateStream": "Accurate stream", | |
| "C2": "C2 pointers", | |
| "SilentSamples": "Silent samples", | |
| "NullSamples": "Null samples", | |
| "Gap": "Gap handling", | |
| "Tag": "Tag", | |
| "Gain": "Gain", | |
| "RangeSplit": "Range split", | |
| "Samples": "Samples", | |
| "SilentBlocks": "Silent blocks", | |
| "Normalization": "Normalization", | |
| "Filename": "Filename", | |
| "ReadError": "Read error", | |
| "SkipError": "Skip error", | |
| "JitterGenericError": "Jitter error", | |
| "JitterEdgeError": "Jitter (edge) error", | |
| "JitterAtomError": "Jitter (atom) error", | |
| "DriftError": "Drift error", | |
| "DroppedError": "Dropped bytes error", | |
| "DuplicatedError": "Duplicated bytes error", | |
| "InconsistentErrorSectors": "Inconsistent error sectors", | |
| "DamagedSector": "Damaged sector", | |
| "Abort": "Aborted", | |
| } | |
| def _score_color(score_str): | |
| try: | |
| score = float(score_str) | |
| except (TypeError, ValueError): | |
| return "#9a9996" | |
| if score < 0: | |
| return "#9a9996" | |
| if score >= 95: | |
| return "#26a269" | |
| if score >= 80: | |
| return "#33d17a" | |
| if score >= 50: | |
| return "#e5a50a" | |
| return "#e01b24" | |
| def find_log_file(folder): | |
| """Return the most likely EAC/XLD/whipper log file in a folder, or None.""" | |
| try: | |
| entries = os.listdir(folder) | |
| except OSError: | |
| return None | |
| candidates = [e for e in entries if e.lower().endswith(".log")] | |
| if not candidates: | |
| return None | |
| # Prefer filenames that look like rip logs over any stray .log file | |
| keywords = ("eac", "exactaudiocopy", "xld", "whipper", "cueripper", "log") | |
| def sort_key(name): | |
| lname = name.lower() | |
| hit = next((i for i, kw in enumerate(keywords) if kw in lname), len(keywords)) | |
| return (hit, lname) | |
| candidates.sort(key=sort_key) | |
| return os.path.join(folder, candidates[0]) | |
| def upload_log(server, data): | |
| """POST raw log bytes to a Cambia server and return the parsed JSON.""" | |
| url = server.rstrip("/") + UPLOAD_PATH | |
| req = urllib.request.Request( | |
| url, | |
| data=data, | |
| method="POST", | |
| headers={ | |
| "Content-Type": "application/octet-stream", | |
| "User-Agent": USER_AGENT, | |
| }, | |
| ) | |
| try: | |
| with urllib.request.urlopen(req, timeout=25) as resp: | |
| body = resp.read() | |
| except urllib.error.HTTPError as e: | |
| raw = e.read() | |
| try: | |
| msg = raw.decode("utf-8", "replace") | |
| except Exception: | |
| msg = str(e) | |
| raise RuntimeError( | |
| _("Cambia rejected the log (HTTP %d): %s") % (e.code, msg.strip()) | |
| ) | |
| except urllib.error.URLError as e: | |
| raise RuntimeError(_("Could not reach Cambia server at %s: %s") % (url, e.reason)) | |
| return json.loads(body) | |
| class ResultWindow(Gtk.Window): | |
| def __init__(self, jobs, server): | |
| super().__init__(title=_("Rip Log Results")) | |
| self.set_default_size(680, 520) | |
| self.set_border_width(6) | |
| self._notebook = Gtk.Notebook() | |
| self._notebook.set_scrollable(True) | |
| self.add(self._notebook) | |
| spinner_page = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) | |
| spinner_page.set_halign(Gtk.Align.CENTER) | |
| spinner_page.set_valign(Gtk.Align.CENTER) | |
| spinner = Gtk.Spinner() | |
| spinner.start() | |
| spinner_page.pack_start(spinner, False, False, 0) | |
| spinner_page.pack_start(Gtk.Label(label=_("Checking logs…")), False, False, 0) | |
| self._notebook.append_page(spinner_page, Gtk.Label(label=_("Working"))) | |
| threading.Thread( | |
| target=self._worker, args=(jobs, server), daemon=True | |
| ).start() | |
| def _worker(self, jobs, server): | |
| results = [] | |
| for log_path, folder in jobs: | |
| try: | |
| with open(log_path, "rb") as fh: | |
| data = fh.read() | |
| response = upload_log(server, data) | |
| results.append((log_path, response, None)) | |
| except Exception as e: # noqa: BLE001 - surface any failure to the UI | |
| results.append((log_path, None, str(e))) | |
| GLib.idle_add(self._show_results, results) | |
| def _show_results(self, results): | |
| while self._notebook.get_n_pages(): | |
| self._notebook.remove_page(0) | |
| for log_path, response, error in results: | |
| title = os.path.basename(log_path) | |
| tab_label = Gtk.Label(label=title) | |
| tab_label.set_ellipsize(Pango.EllipsizeMode.MIDDLE) | |
| tab_label.set_max_width_chars(24) | |
| if error: | |
| page = self._build_error_page(log_path, error) | |
| else: | |
| page = self._build_result_page(log_path, response) | |
| self._notebook.append_page(page, tab_label) | |
| self._notebook.show_all() | |
| return False | |
| @staticmethod | |
| def _build_error_page(log_path, error): | |
| box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) | |
| box.set_border_width(12) | |
| label = Gtk.Label() | |
| label.set_line_wrap(True) | |
| label.set_xalign(0) | |
| label.set_markup( | |
| "<b>%s</b>\n\n%s" | |
| % (GLib.markup_escape_text(os.path.basename(log_path)), | |
| GLib.markup_escape_text(error)) | |
| ) | |
| box.pack_start(label, False, False, 0) | |
| return box | |
| def _build_result_page(self, log_path, response): | |
| scrolled = Gtk.ScrolledWindow() | |
| scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) | |
| outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10) | |
| outer.set_border_width(12) | |
| parsed_logs = response.get("parsed", {}).get("parsed_logs", []) | |
| if parsed_logs: | |
| meta = parsed_logs[0] | |
| release = meta.get("release_info", {}) | |
| artist = release.get("artist", "") | |
| album = release.get("title", "") | |
| header = " – ".join(p for p in (artist, album) if p) or os.path.basename(log_path) | |
| header_label = Gtk.Label() | |
| header_label.set_xalign(0) | |
| header_label.set_markup("<span size='large' weight='bold'>%s</span>" % | |
| GLib.markup_escape_text(header)) | |
| outer.pack_start(header_label, False, False, 0) | |
| details = [] | |
| if meta.get("ripper"): | |
| details.append("%s %s" % (meta.get("ripper", ""), meta.get("ripper_version", ""))) | |
| if meta.get("drive"): | |
| details.append(_("Drive: %s") % meta["drive"]) | |
| if details: | |
| detail_label = Gtk.Label(label=" • ".join(details)) | |
| detail_label.set_xalign(0) | |
| detail_label.get_style_context().add_class("dim-label") | |
| outer.pack_start(detail_label, False, False, 0) | |
| # Score summary, one row per evaluator (Cambia / RED / OPS) | |
| score_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=18) | |
| for ev in response.get("evaluation_combined", []): | |
| score_box.pack_start(self._score_widget(ev), False, False, 0) | |
| outer.pack_start(score_box, False, False, 0) | |
| outer.pack_start(Gtk.Separator(), False, False, 4) | |
| # Flagged issues, grouped by evaluator | |
| for ev in response.get("evaluation_combined", []): | |
| issues = self._collect_issues(ev) | |
| if not issues: | |
| continue | |
| group_label = Gtk.Label() | |
| group_label.set_xalign(0) | |
| group_label.set_markup("<b>%s %s</b>" % ( | |
| GLib.markup_escape_text(ev.get("evaluator", "")), _("notes"))) | |
| outer.pack_start(group_label, False, False, 0) | |
| listbox = Gtk.ListBox() | |
| listbox.set_selection_mode(Gtk.SelectionMode.NONE) | |
| for scope_text, field, message, klass in issues: | |
| listbox.add(self._issue_row(scope_text, field, message, klass)) | |
| outer.pack_start(listbox, False, False, 0) | |
| scrolled.add(outer) | |
| return scrolled | |
| @staticmethod | |
| def _score_widget(evaluation_combined): | |
| evaluator = evaluation_combined.get("evaluator", "?") | |
| score = evaluation_combined.get("combined_score", "N/A") | |
| color = _score_color(score) | |
| box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) | |
| score_label = Gtk.Label() | |
| score_label.set_markup( | |
| "<span size='xx-large' weight='bold' foreground='%s'>%s</span>" % (color, score) | |
| ) | |
| name_label = Gtk.Label(label=evaluator) | |
| name_label.get_style_context().add_class("dim-label") | |
| box.pack_start(score_label, False, False, 0) | |
| box.pack_start(name_label, False, False, 0) | |
| return box | |
| @staticmethod | |
| def _collect_issues(evaluation_combined): | |
| """Flatten evaluation units, skipping the boring 'Perfect' ones.""" | |
| issues = [] | |
| for evaluation in evaluation_combined.get("evaluations", []): | |
| for unit in evaluation.get("evaluation_units", []): | |
| data = unit.get("data", {}) | |
| klass = data.get("class", "Neutral") | |
| if klass in ("Perfect", "Good"): | |
| continue | |
| scope = data.get("scope", "Release") | |
| if isinstance(scope, dict) and "Track" in scope: | |
| track_num = scope["Track"] | |
| scope_text = _("Track %s") % track_num if track_num is not None else _("Track") | |
| else: | |
| scope_text = _("Release") | |
| field = FIELD_LABELS.get(data.get("field", ""), data.get("field", "")) | |
| message = data.get("message", "") | |
| issues.append((scope_text, field, message, klass)) | |
| # Worst issues first | |
| order = {"Critical": 0, "Bad": 1, "Neutral": 2} | |
| issues.sort(key=lambda i: order.get(i[3], 3)) | |
| return issues | |
| @staticmethod | |
| def _issue_row(scope_text, field, message, klass): | |
| row = Gtk.ListBoxRow() | |
| row.set_selectable(False) | |
| hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) | |
| hbox.set_border_width(4) | |
| color = CLASS_COLORS.get(klass, "#9a9996") | |
| dot = Gtk.Label() | |
| dot.set_markup("<span foreground='%s'>●</span>" % color) | |
| hbox.pack_start(dot, False, False, 0) | |
| text = Gtk.Label() | |
| text.set_xalign(0) | |
| text.set_line_wrap(True) | |
| prefix = "%s — %s: " % (scope_text, field) if field else "%s: " % scope_text | |
| text.set_markup( | |
| "<b>%s</b>%s" | |
| % (GLib.markup_escape_text(prefix), GLib.markup_escape_text(message)) | |
| ) | |
| hbox.pack_start(text, True, True, 0) | |
| row.add(hbox) | |
| return row | |
| class EacLogChecker(SongsMenuPlugin, PluginConfigMixin): | |
| PLUGIN_ID = "EAC Log Checker" | |
| PLUGIN_NAME = _("Rip Log Checker") | |
| PLUGIN_DESC = _( | |
| "Scores EAC/XLD/whipper/CUERipper rip logs next to the selected " | |
| "tracks using a Cambia server, and shows the result." | |
| ) | |
| PLUGIN_ICON = "text-x-generic-symbolic" | |
| CONFIG_SECTION = "eac_log_checker" | |
| @classmethod | |
| def PluginPreferences(cls, window): | |
| vb = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) | |
| vb.set_border_width(6) | |
| hb = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) | |
| hb.pack_start(Gtk.Label(label=_("Cambia server URL:")), False, False, 0) | |
| entry = Gtk.Entry() | |
| entry.set_text(cls.config_get("server", DEFAULT_SERVER)) | |
| entry.set_width_chars(38) | |
| entry.connect( | |
| "changed", lambda e: cls.config_set("server", e.get_text().strip()) | |
| ) | |
| hb.pack_start(entry, True, True, 0) | |
| vb.pack_start(hb, False, False, 0) | |
| note = Gtk.Label() | |
| note.set_markup( | |
| "<small>%s</small>" | |
| % GLib.markup_escape_text( | |
| _( | |
| "Logs are uploaded to this server for scoring. The default " | |
| "is the public logs.musichoarders.xyz instance; for " | |
| "privacy or offline use, run `cambia --server` yourself " | |
| "and point this at e.g. http://localhost:3030." | |
| ) | |
| ) | |
| ) | |
| note.set_line_wrap(True) | |
| note.set_xalign(0) | |
| vb.pack_start(note, False, False, 0) | |
| return vb | |
| def plugin_songs(self, songs): | |
| folders = {} | |
| for song in songs: | |
| path = song("~filename") | |
| folder = os.path.dirname(path) | |
| folders.setdefault(folder, None) | |
| jobs = [] | |
| seen = set() | |
| for folder in folders: | |
| log_path = find_log_file(folder) | |
| if log_path and log_path not in seen: | |
| seen.add(log_path) | |
| jobs.append((log_path, folder)) | |
| if not jobs: | |
| message = _("Could not find a .log file next to the selected track(s).") | |
| if ErrorMessage is not None: | |
| ErrorMessage(None, _("No log file found"), message).run() | |
| else: | |
| print("Rip Log Checker: %s" % message) | |
| return | |
| server = self.config_get("server", DEFAULT_SERVER) | |
| window = ResultWindow(jobs, server) | |
| window.show_all() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment