The export dialog gains an accent color — the hover background on links and
tracklist rows, and the player's progress fill — and the rest of the bar loses
its 2010 gold so that choice is the only color in it.
The accent travels as a `:root { --accent }` custom property declared in
index.html, which player.css reads as `var(--accent, #8c764a)`. That keeps the
stylesheet in the verbatim copyfile loop: index.html is still the only rendered
template. `normalize_accent` is the injection gate — the value lands raw inside
a <style> block and string.Template escapes nothing — and `contrast_text` flips
the hover text black or white, since the old page hard-coded white and a pale
accent made it unreadable.
The bar itself is now fixed light gray with black text. Its sprite glyphs are
pale lavender and yellow, drawn for the dark gold bar, so they're recolored
with `filter: brightness(0)` rather than by editing the GIF — which still ships
byte-identical, spinner and all.
Not persisted: the picker opens on #8c764a every time, so an export nobody
touches looks exactly like the mixes already online.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JBSM2bFC6UToiEg8BE4dqj
396 lines
15 KiB
Python
396 lines
15 KiB
Python
"""Export a playlist to a folder of files, or to a self-contained web mix.
|
|
|
|
A sibling of ``device_sync``: same shape (pure planning function, then a
|
|
worker on a daemon thread reporting through Qt signals), same filename
|
|
helpers, same m3u writer. The differences are that the destination is an
|
|
ordinary directory rather than an MTP mount, and that the web variant may
|
|
have to transcode a track the browser can't decode (see ``web_support``).
|
|
|
|
Two invariants worth keeping:
|
|
|
|
* **Local music files are only ever read.** Like sync, this never moves,
|
|
rewrites or deletes anything in the library.
|
|
* **The manifest is written last** — ``index.html`` for a web mix, the
|
|
``.m3u`` for a folder. An interrupted export therefore never leaves behind
|
|
a playlist promising files that aren't there.
|
|
"""
|
|
|
|
import html
|
|
import os
|
|
import re
|
|
import shutil
|
|
import string
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from importlib import resources
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtCore import QObject, pyqtSignal
|
|
|
|
from ..device_sync import (CHUNK, build_m3u, sanitize_name, track_display,
|
|
track_filename)
|
|
from . import web_support
|
|
|
|
FOLDER = "folder"
|
|
WEB = "web"
|
|
|
|
# Copied into a web mix alongside the audio. player-graphics.gif has to ship
|
|
# byte-identical: it is an animated GIF (the "loading" state is a spinner),
|
|
# not a flat sprite sheet.
|
|
_WEB_ASSETS = ("player.js", "player.css", "player-graphics.gif")
|
|
|
|
DEFAULT_DESCRIPTION = (
|
|
"Write something about this mix here — this paragraph is placeholder "
|
|
"text, and it's plain HTML, so <a href=\"https://example.com\">links</a> "
|
|
"and <br> work."
|
|
)
|
|
|
|
# The one color a web mix lets you choose: link/tracklist hover backgrounds and
|
|
# the progress fill. The default is the brown the hand-made mixes used, so an
|
|
# untouched export comes out looking exactly as it always has. Everything else
|
|
# in the player bar is deliberately fixed (light gray, black text) — see
|
|
# templates/player.css.
|
|
DEFAULT_ACCENT = "#8c764a"
|
|
|
|
_HEX_COLOR = re.compile(r"^#(?:[0-9a-fA-F]{3}|[0-9a-fA-F]{6})$")
|
|
|
|
|
|
def normalize_accent(value) -> str:
|
|
"""A `#rgb`/`#rrggbb` string, lowercased, or DEFAULT_ACCENT.
|
|
|
|
This is the injection gate, not a nicety: the result is substituted raw
|
|
into the page's <style> block, and string.Template escapes nothing.
|
|
"""
|
|
if isinstance(value, str) and _HEX_COLOR.match(value.strip()):
|
|
return value.strip().lower()
|
|
return DEFAULT_ACCENT
|
|
|
|
|
|
def contrast_text(accent: str) -> str:
|
|
"""Black or white, whichever reads on `accent`.
|
|
|
|
Same rule the app's own theme uses for highlighted text
|
|
(`theme.apply_theme`) — the old page hard-coded white, which disappears
|
|
the moment someone picks a pale accent.
|
|
"""
|
|
hex_digits = normalize_accent(accent).lstrip("#")
|
|
if len(hex_digits) == 3:
|
|
hex_digits = "".join(c * 2 for c in hex_digits)
|
|
r, g, b = (int(hex_digits[i:i + 2], 16) for i in (0, 2, 4))
|
|
luminance = (0.299 * r + 0.587 * g + 0.114 * b) / 255
|
|
return "#000" if luminance > 0.6 else "#fff"
|
|
|
|
|
|
def templates():
|
|
"""The template directory, working from a checkout or an installed wheel."""
|
|
return resources.files(__package__) / "templates"
|
|
|
|
|
|
@dataclass
|
|
class ExportItem:
|
|
track_id: int
|
|
src: Path
|
|
dest_name: str # sanitized "Artist - Title.ext"
|
|
size: int
|
|
display: str # "Artist - Title", for the m3u and progress
|
|
secs: int
|
|
convert_to: str | None = None # None = byte copy; "flac" = transcode
|
|
|
|
|
|
@dataclass
|
|
class ExportPlan:
|
|
kind: str # FOLDER | WEB
|
|
playlist_name: str
|
|
dest_dir: Path # the folder that gets created
|
|
audio_dir: Path # dest_dir, or dest_dir/"audios" for a web mix
|
|
m3u_name: str
|
|
title: str = ""
|
|
description: str = ""
|
|
image: Path | None = None # None = ship the gray placeholder
|
|
accent: str = DEFAULT_ACCENT # hover backgrounds + the progress fill
|
|
items: list = field(default_factory=list)
|
|
skipped: list = field(default_factory=list) # (display, reason)
|
|
bytes_to_copy: int = 0
|
|
|
|
@property
|
|
def conversions(self) -> list:
|
|
return [i for i in self.items if i.convert_to]
|
|
|
|
@property
|
|
def needs_ffmpeg(self) -> bool:
|
|
return bool(self.conversions)
|
|
|
|
|
|
def plan_export(playlist_name: str, tracks: list, dest_parent: Path,
|
|
kind: str = FOLDER) -> ExportPlan:
|
|
"""Work out what an export would write. Pure — creates nothing.
|
|
|
|
`tracks` is the playlist's tracks in order; `dest_parent` is the folder
|
|
the user picked, inside which a folder named after the playlist is made.
|
|
"""
|
|
folder = sanitize_name(playlist_name)
|
|
dest_dir = Path(dest_parent) / folder
|
|
plan = ExportPlan(
|
|
kind=kind,
|
|
playlist_name=playlist_name,
|
|
dest_dir=dest_dir,
|
|
audio_dir=dest_dir / "audios" if kind == WEB else dest_dir,
|
|
m3u_name=folder + ".m3u",
|
|
title=playlist_name,
|
|
description=DEFAULT_DESCRIPTION,
|
|
)
|
|
|
|
items: dict[int, ExportItem] = {}
|
|
order: list[int] = []
|
|
seen_skipped: set[int] = set()
|
|
for track in tracks:
|
|
if track.track_id in items:
|
|
order.append(track.track_id) # a repeat of a track we have
|
|
continue
|
|
if track.track_id in seen_skipped:
|
|
continue
|
|
display = track_display(track)
|
|
src = Path(track.location) if track.location else None
|
|
if src is None or not src.is_file():
|
|
seen_skipped.add(track.track_id)
|
|
plan.skipped.append((display, "its file is missing"))
|
|
continue
|
|
convert_to = None
|
|
if kind == WEB:
|
|
convert_to, reason = web_support.conversion_for(
|
|
track.location, track.kind or "")
|
|
if reason:
|
|
seen_skipped.add(track.track_id)
|
|
plan.skipped.append((display, reason))
|
|
continue
|
|
dest_name = track_filename(track)
|
|
if convert_to:
|
|
dest_name = os.path.splitext(dest_name)[0] + "." + convert_to
|
|
items[track.track_id] = ExportItem(
|
|
track_id=track.track_id, src=src, dest_name=dest_name,
|
|
size=src.stat().st_size, display=display,
|
|
secs=round(track.total_time / 1000), convert_to=convert_to)
|
|
order.append(track.track_id)
|
|
|
|
# Disambiguate name collisions with a stable [track_id] suffix on every
|
|
# member of a colliding group, so a name never depends on playlist order.
|
|
by_name: dict[str, list[ExportItem]] = {}
|
|
for item in items.values():
|
|
by_name.setdefault(item.dest_name, []).append(item)
|
|
for group in by_name.values():
|
|
if len(group) > 1:
|
|
for item in group:
|
|
stem, ext = os.path.splitext(item.dest_name)
|
|
item.dest_name = f"{stem} [{item.track_id}]{ext}"
|
|
|
|
# Playlist order, duplicates collapsed to their first appearance: a web
|
|
# page can't list the same <li> twice and an m3u repeat is just noise.
|
|
emitted: set[int] = set()
|
|
for tid in order:
|
|
if tid in emitted:
|
|
continue
|
|
emitted.add(tid)
|
|
plan.items.append(items[tid])
|
|
plan.bytes_to_copy = sum(i.size for i in plan.items)
|
|
return plan
|
|
|
|
|
|
def build_tracklist(items: list) -> str:
|
|
"""The <ol> rows for the web mix, in playlist order."""
|
|
rows = []
|
|
for item in items:
|
|
src = "./audios/" + item.dest_name
|
|
rows.append(
|
|
f' <li><a href="#" data-src="{html.escape(src, quote=True)}">'
|
|
f'{html.escape(item.display)}</a></li>')
|
|
return "\n".join(rows)
|
|
|
|
|
|
def build_index_html(plan: ExportPlan, image_name: str | None) -> str:
|
|
"""Render index.html for `plan`.
|
|
|
|
The title and track names are escaped; the description deliberately is
|
|
**not**, because the hand-made mixes lean on inline links and <br> in
|
|
exactly that spot and the dialog says so.
|
|
"""
|
|
if image_name:
|
|
image_tag = (f' <img src="{html.escape(image_name, quote=True)}"'
|
|
f' width="400"><br>')
|
|
else:
|
|
image_tag = ""
|
|
accent = normalize_accent(plan.accent)
|
|
tmpl = string.Template((templates() / "index.html.tmpl").read_text(encoding="utf-8"))
|
|
return tmpl.substitute(
|
|
title=html.escape(plan.title or plan.playlist_name),
|
|
description=plan.description,
|
|
tracklist=build_tracklist(plan.items),
|
|
image_tag=image_tag,
|
|
accent=accent,
|
|
accent_text=contrast_text(accent),
|
|
)
|
|
|
|
|
|
class _Cancelled(Exception):
|
|
"""Internal: unwinds _run when cancel() was requested."""
|
|
|
|
|
|
class ExportWorker(QObject):
|
|
"""Runs an ExportPlan on a daemon thread, reporting through signals.
|
|
|
|
Progress is emitted in KiB — a pyqtSignal(int) is a C int, and a mix of
|
|
lossless tracks overflows a byte count past 2 GiB. Threading matches the
|
|
rest of the codebase: a plain daemon thread, not a QThread.
|
|
|
|
Cancelling removes the in-flight partial file and skips the manifest
|
|
entirely, so a cancelled export leaves an obviously-incomplete folder
|
|
rather than a page that half works.
|
|
"""
|
|
|
|
progress = pyqtSignal(int, int, str) # done_kib, total_kib, label
|
|
finished = pyqtSignal(dict)
|
|
cancelled = pyqtSignal(dict)
|
|
failed = pyqtSignal(str)
|
|
|
|
def __init__(self, plan: ExportPlan, parent=None):
|
|
super().__init__(parent)
|
|
self._plan = plan
|
|
self._busy = False
|
|
self._cancel = threading.Event()
|
|
|
|
def busy(self) -> bool:
|
|
return self._busy
|
|
|
|
def cancel(self):
|
|
self._cancel.set()
|
|
|
|
def _check_cancel(self):
|
|
if self._cancel.is_set():
|
|
raise _Cancelled
|
|
|
|
def start(self):
|
|
if self._busy:
|
|
return
|
|
self._busy = True
|
|
threading.Thread(target=self._run_guarded, daemon=True).start()
|
|
|
|
def _run_guarded(self):
|
|
try:
|
|
self._run()
|
|
finally:
|
|
self._busy = False
|
|
|
|
def _run(self):
|
|
plan = self._plan
|
|
dest = None
|
|
completed: list[ExportItem] = []
|
|
vanished: set[int] = set()
|
|
converted = 0
|
|
try:
|
|
self._check_cancel()
|
|
plan.audio_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
total_kib = max(plan.bytes_to_copy // 1024, 1)
|
|
done = 0
|
|
for i, item in enumerate(plan.items, start=1):
|
|
self._check_cancel()
|
|
label = f"{i}/{len(plan.items)} {item.display}"
|
|
dest = plan.audio_dir / item.dest_name
|
|
if item.convert_to:
|
|
# ffmpeg reports no usable byte progress, so the bar jumps
|
|
# a whole track at a time here. Conversions are rare.
|
|
self.progress.emit(done // 1024, total_kib, label)
|
|
if not item.src.is_file():
|
|
vanished.add(item.track_id)
|
|
done += item.size
|
|
dest = None
|
|
continue
|
|
web_support.convert_to_flac(item.src, dest, self._cancel)
|
|
self._check_cancel()
|
|
converted += 1
|
|
done += item.size
|
|
self.progress.emit(done // 1024, total_kib, label)
|
|
else:
|
|
# A source gone since planning means Syncthing moved it
|
|
# under us — skip it, don't abort the whole export. Once
|
|
# open the fd pins the inode, so the read stays coherent.
|
|
try:
|
|
fsrc = open(item.src, "rb")
|
|
except FileNotFoundError:
|
|
vanished.add(item.track_id)
|
|
done += item.size
|
|
self.progress.emit(done // 1024, total_kib, label)
|
|
dest = None
|
|
continue
|
|
with fsrc, open(dest, "wb") as fdst:
|
|
while True:
|
|
chunk = fsrc.read(CHUNK)
|
|
if not chunk:
|
|
break
|
|
fdst.write(chunk)
|
|
done += len(chunk)
|
|
self.progress.emit(done // 1024, total_kib, label)
|
|
self._check_cancel()
|
|
try:
|
|
shutil.copystat(item.src, dest, follow_symlinks=True)
|
|
except OSError:
|
|
pass # mtime is a nicety; some filesystems refuse it
|
|
dest = None
|
|
completed.append(item)
|
|
|
|
self._check_cancel()
|
|
self._write_manifest(completed)
|
|
self.finished.emit({
|
|
"playlist": plan.playlist_name,
|
|
"kind": plan.kind,
|
|
"dest": str(plan.dest_dir),
|
|
"exported": len(completed),
|
|
"converted": converted,
|
|
"skipped": len(plan.skipped),
|
|
"vanished": len(vanished),
|
|
})
|
|
except _Cancelled:
|
|
if dest is not None:
|
|
try:
|
|
dest.unlink() # the mid-flight partial
|
|
except OSError:
|
|
pass
|
|
self.cancelled.emit({
|
|
"playlist": plan.playlist_name,
|
|
"dest": str(plan.dest_dir),
|
|
"exported": len(completed),
|
|
"total": len(plan.items),
|
|
})
|
|
except OSError as e:
|
|
if dest is not None:
|
|
try:
|
|
dest.unlink()
|
|
except OSError:
|
|
pass
|
|
self.failed.emit(f"Export failed: {e}")
|
|
|
|
def _write_manifest(self, completed: list):
|
|
"""The .m3u, plus the page and its assets for a web mix. Written last."""
|
|
plan = self._plan
|
|
entries = [(item.dest_name, item.secs, item.display) for item in completed]
|
|
|
|
if plan.kind == WEB:
|
|
for name in _WEB_ASSETS:
|
|
shutil.copyfile(templates() / name, plan.dest_dir / name)
|
|
image_name = None
|
|
if plan.image is not None and Path(plan.image).is_file():
|
|
image_name = sanitize_name(Path(plan.image).stem) \
|
|
+ Path(plan.image).suffix.lower()
|
|
shutil.copyfile(plan.image, plan.dest_dir / image_name)
|
|
else:
|
|
image_name = "placeholder.png"
|
|
shutil.copyfile(templates() / "placeholder.png",
|
|
plan.dest_dir / image_name)
|
|
# An m3u alongside the page costs nothing and lets the folder
|
|
# double as a plain music folder.
|
|
entries = [("audios/" + n, s, d) for n, s, d in entries]
|
|
(plan.dest_dir / "index.html").write_text(
|
|
build_index_html(plan, image_name), encoding="utf-8")
|
|
|
|
(plan.dest_dir / plan.m3u_name).write_text(
|
|
build_m3u(entries), encoding="utf-8")
|