Skip to content

Instantly share code, notes, and snippets.

@gitgotgitgotit
Last active September 8, 2026 10:32
Show Gist options
  • Select an option

  • Save gitgotgitgotit/bfe0f85c312c06a111556a872e31568b to your computer and use it in GitHub Desktop.

Select an option

Save gitgotgitgotit/bfe0f85c312c06a111556a872e31568b to your computer and use it in GitHub Desktop.
Quod Libet plugin LOG checker (EAC)
# -*- 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