416 lines
16 KiB
Python
416 lines
16 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 (a web mix gets no m3u: it is a page to upload, not a
|
|
music folder to open in a player). An interrupted export therefore never
|
|
leaves behind a page or 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 # "" for a web mix, which writes no m3u
|
|
title: str = ""
|
|
description: str = ""
|
|
images: list = field(default_factory=list) # [] = 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, folder: str | None = None) -> 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
|
|
(or `folder`, when given — an upload names it by its URL slug).
|
|
"""
|
|
folder = folder or 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="" if kind == WEB else 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_names) -> 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.
|
|
|
|
`image_names` are the copied images, shown one after another with a
|
|
blank line between each (a single name still works).
|
|
"""
|
|
if isinstance(image_names, str):
|
|
image_names = [image_names]
|
|
image_tag = "\n <br><br>\n".join(
|
|
f' <img src="{html.escape(name, quote=True)}" width="400"><br>'
|
|
for name in image_names or [])
|
|
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 page for a web mix, the .m3u for a folder. Written last.
|
|
|
|
A web mix gets no m3u. It used to ship one on the theory that it cost
|
|
nothing and let the folder double as a plain music folder — but the
|
|
folder is a page you upload, and a stray playlist file next to
|
|
index.html is one more thing to explain to whoever receives it.
|
|
"""
|
|
plan = self._plan
|
|
|
|
if plan.kind == WEB:
|
|
for name in _WEB_ASSETS:
|
|
shutil.copyfile(templates() / name, plan.dest_dir / name)
|
|
image_names = []
|
|
for image in plan.images:
|
|
image = Path(image)
|
|
if not image.is_file():
|
|
continue
|
|
name = sanitize_name(image.stem) + image.suffix.lower()
|
|
# Two pictures called cover.jpg from different folders must
|
|
# not overwrite each other.
|
|
stem, ext = os.path.splitext(name)
|
|
n = 2
|
|
while name in image_names:
|
|
name = f"{stem} {n}{ext}"
|
|
n += 1
|
|
shutil.copyfile(image, plan.dest_dir / name)
|
|
image_names.append(name)
|
|
if not image_names:
|
|
image_names = ["placeholder.png"]
|
|
shutil.copyfile(templates() / "placeholder.png",
|
|
plan.dest_dir / "placeholder.png")
|
|
(plan.dest_dir / "index.html").write_text(
|
|
build_index_html(plan, image_names), encoding="utf-8")
|
|
return
|
|
|
|
entries = [(item.dest_name, item.secs, item.display)
|
|
for item in completed]
|
|
(plan.dest_dir / plan.m3u_name).write_text(
|
|
build_m3u(entries), encoding="utf-8")
|