trav
35281b9864
- A followed playlist (the UFO in friend mode) appears in my sidebar with a purple UFO: mixed in with my playlists alphabetically, or inside a folder named after the friend (per-friend, in Sync Settings). It's never a Playlist of mine, so it stays out of the library and its merges. - It plays from the cache. Player's queue accepts friend-track keys, resolved through Cassette: they scrobble to Last.fm but don't count in my play journal. Songs still on their way are grey and skipped. - The cassette column there is hidden by default (right-click the header); pressing one takes the song into my library — straight from the cache if it's there, otherwise by asking for it. - Sync Settings per friend: keep followed songs separate (default) or add them to my library. Unfollowing (right-click) frees their cache. - The bottom bar says when the friend's folder last synced.
324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""The request / delivery loop — both directions, rebuilt from the folders.
|
|
|
|
**Sending** (my LinTunes serving a friend): their ``requests.json`` (in their
|
|
folder to me) ∩ what I *currently* offer them = what belongs in my
|
|
``outbox/``. Missing files are copied in (to a ``.part`` name the
|
|
``.stignore`` keeps out of sync, then renamed); files no longer requested are
|
|
deleted — a request disappears when the song has arrived, so the outbox
|
|
cleans itself up with no bookkeeping of its own.
|
|
|
|
**Receiving** (my LinTunes asking a friend): each complete file in their
|
|
outbox that I want is brought in — a cassetted song is imported exactly like
|
|
Add to Library (copied into the organized tree, tags read, date added now,
|
|
none of their listening history), a song wanted only for a followed playlist
|
|
is copied into ``cache/``. Then my ``requests.json`` is rewritten once per
|
|
batch without what arrived and without anything they no longer offer.
|
|
|
|
Each machine trusts only its own view, and every step is safe to repeat: a
|
|
song already in my library (exact match) is not imported twice, a file
|
|
Syncthing is still writing (``.syncthing.*.tmp``) is never touched, and
|
|
whatever a crash interrupted is simply planned again next time.
|
|
|
|
Planning is pure and runs on the GUI thread (it reads the library); the file
|
|
work runs on ``DeliveryWorker``'s thread; adding tracks to the library happens
|
|
back on the GUI thread.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
from PyQt6.QtCore import QObject, pyqtSignal
|
|
|
|
from lintunes.cassette import matching, publish, share
|
|
from lintunes.cassette import requests as req
|
|
from lintunes.cassette.friend_library import load_friend_library
|
|
from lintunes.cassette.state import FOLLOW_LIBRARY, cache_dir, in_dir, out_dir
|
|
from lintunes.importers.file_importer import sanitize_component
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
OUTBOX = "outbox"
|
|
PART_SUFFIX = ".part"
|
|
_NAME_RE = re.compile(r"^(\d+) ~ (.+)$")
|
|
|
|
|
|
def outbox_name(track_id: int, location: str) -> str:
|
|
return f"{track_id} ~ {sanitize_component(Path(location).name)}"
|
|
|
|
|
|
def parse_outbox_name(name: str) -> tuple[int, str] | None:
|
|
"""(track id, original filename) for one of ours, else None."""
|
|
match = _NAME_RE.match(name)
|
|
return (int(match.group(1)), match.group(2)) if match else None
|
|
|
|
|
|
def is_syncthing_temp(name: str) -> bool:
|
|
return (name.startswith(".syncthing.") or name.startswith("~syncthing~")
|
|
or name.endswith(".tmp") or name.endswith(PART_SUFFIX))
|
|
|
|
|
|
def list_outbox(folder: Path) -> dict[int, Path]:
|
|
"""{track id: path} for the finished files in an outbox."""
|
|
found = {}
|
|
try:
|
|
entries = list(folder.iterdir())
|
|
except OSError:
|
|
return found
|
|
for path in entries:
|
|
if is_syncthing_temp(path.name) or not path.is_file():
|
|
continue
|
|
parsed = parse_outbox_name(path.name)
|
|
if parsed is not None:
|
|
found[parsed[0]] = path
|
|
return found
|
|
|
|
|
|
# ---------------------------------------------------------------- sending --
|
|
|
|
@dataclass
|
|
class SendPlan:
|
|
token: str
|
|
outbox: Path
|
|
copies: list = field(default_factory=list) # (source Path, dest name)
|
|
deletes: list = field(default_factory=list) # Paths
|
|
missing: list = field(default_factory=list) # track ids with no file
|
|
|
|
|
|
def plan_send(token: str, root: Path, library, selection) -> SendPlan:
|
|
outbox = out_dir(root, token) / OUTBOX
|
|
their_requests = req.read_requests(in_dir(root, token))
|
|
offered = share.offered_ids(library, selection)
|
|
want = their_requests & offered
|
|
present = list_outbox(outbox)
|
|
plan = SendPlan(token=token, outbox=outbox)
|
|
for tid in sorted(want):
|
|
if tid in present:
|
|
continue
|
|
track = library.tracks[tid]
|
|
source = Path(track.location) if track.location else None
|
|
if source is None or not source.is_file():
|
|
plan.missing.append(tid)
|
|
continue
|
|
plan.copies.append((source, outbox_name(tid, track.location)))
|
|
for tid, path in present.items():
|
|
if tid not in want:
|
|
plan.deletes.append(path)
|
|
return plan
|
|
|
|
|
|
# -------------------------------------------------------------- receiving --
|
|
|
|
@dataclass
|
|
class Arrival:
|
|
track_id: int
|
|
path: Path
|
|
to_library: bool
|
|
to_cache: bool
|
|
|
|
|
|
@dataclass
|
|
class ReceivePlan:
|
|
token: str
|
|
arrivals: list = field(default_factory=list) # Arrival
|
|
cache: Path = None
|
|
# The requests file as it should read once the arrivals are in.
|
|
cassetted: set = field(default_factory=set)
|
|
requests_after: set = field(default_factory=set)
|
|
requests_path: Path = None
|
|
already_have: list = field(default_factory=list) # ids skipped: exact match
|
|
# Cached songs no followed playlist uses any more (unfollowed, or the
|
|
# friend took them out): deleted to give the space back.
|
|
cache_prune: list = field(default_factory=list) # Paths
|
|
|
|
|
|
def followed_needed(friend_library, friend, cache: Path) -> set[int]:
|
|
"""Songs of followed playlists that aren't in the cache yet."""
|
|
cached = {int(p.stem) for p in _cache_files(cache)}
|
|
needed = set()
|
|
for pid in friend.followed:
|
|
playlist = friend_library.playlist(pid)
|
|
if playlist is not None:
|
|
needed.update(t for t in playlist.track_ids if t not in cached)
|
|
return needed
|
|
|
|
|
|
def _cache_files(cache: Path):
|
|
try:
|
|
return [p for p in cache.iterdir()
|
|
if p.is_file() and p.stem.isdigit() and not is_syncthing_temp(p.name)]
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def plan_receive(friend, root: Path, my_index: matching.LibraryIndex,
|
|
is_complete=None) -> ReceivePlan:
|
|
token = friend.token
|
|
source = in_dir(root, token)
|
|
library_file = source / share.LIBRARY_FILE
|
|
their = load_friend_library(source, friend.name)
|
|
cache = cache_dir(root, token)
|
|
plan = ReceivePlan(token=token, cache=cache,
|
|
requests_path=out_dir(root, token))
|
|
|
|
cassetted = set(friend.cassetted)
|
|
# Prune what they stopped offering — but only against a library file
|
|
# that's actually here; a not-yet-synced one isn't "offers nothing".
|
|
if library_file.exists() and publish.read_json(library_file) is not None:
|
|
cassetted &= set(their.tracks)
|
|
to_library = set(cassetted)
|
|
needed = followed_needed(their, friend, cache)
|
|
if friend.followed_mode == FOLLOW_LIBRARY:
|
|
to_library |= needed
|
|
needed_cache = set(needed)
|
|
else:
|
|
needed_cache = set(needed)
|
|
|
|
for tid, path in sorted(list_outbox(source / OUTBOX).items()):
|
|
wanted_lib = tid in to_library
|
|
wanted_cache = tid in needed_cache
|
|
if not (wanted_lib or wanted_cache):
|
|
continue
|
|
if is_complete is not None and not is_complete(path):
|
|
continue
|
|
if wanted_lib and tid in their.tracks and \
|
|
my_index.classify(their.tracks[tid]) == matching.EXACT:
|
|
plan.already_have.append(tid)
|
|
wanted_lib = False
|
|
if not wanted_cache:
|
|
cassetted.discard(tid)
|
|
continue
|
|
plan.arrivals.append(Arrival(tid, path, wanted_lib, wanted_cache))
|
|
|
|
playlists_file = source / share.PLAYLISTS_FILE
|
|
if playlists_file.exists() and publish.read_json(playlists_file) is not None:
|
|
keep = set()
|
|
for pid in friend.followed:
|
|
playlist = their.playlist(pid)
|
|
if playlist is not None:
|
|
keep.update(playlist.track_ids)
|
|
plan.cache_prune = [p for p in _cache_files(cache)
|
|
if int(p.stem) not in keep]
|
|
|
|
arrived_lib = {a.track_id for a in plan.arrivals if a.to_library}
|
|
arrived_cache = {a.track_id for a in plan.arrivals if a.to_cache}
|
|
plan.cassetted = cassetted - arrived_lib
|
|
still_needed = needed - arrived_cache
|
|
if friend.followed_mode == FOLLOW_LIBRARY:
|
|
still_needed -= arrived_lib
|
|
plan.requests_after = plan.cassetted | still_needed
|
|
return plan
|
|
|
|
|
|
# ------------------------------------------------------------------ work --
|
|
|
|
@dataclass
|
|
class Staged:
|
|
token: str
|
|
track_id: int
|
|
dest: Path
|
|
fields: dict
|
|
fallback_name: str
|
|
|
|
|
|
@dataclass
|
|
class DeliveryResult:
|
|
staged: list = field(default_factory=list) # Staged, to add to library
|
|
cached: list = field(default_factory=list) # (token, track id)
|
|
sent: int = 0
|
|
removed: int = 0
|
|
errors: list = field(default_factory=list) # (what, error text)
|
|
|
|
|
|
def run_plans(send_plans, receive_plans, music_dir: Path | None,
|
|
cancel: threading.Event | None = None) -> DeliveryResult:
|
|
"""All the file work. Safe off the GUI thread: touches no library."""
|
|
from lintunes.importers.file_importer import stage_file
|
|
result = DeliveryResult()
|
|
for plan in send_plans:
|
|
for path in plan.deletes:
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
result.removed += 1
|
|
except OSError as exc:
|
|
result.errors.append((f"Removing {path.name}", str(exc)))
|
|
if plan.copies:
|
|
plan.outbox.mkdir(parents=True, exist_ok=True)
|
|
for source, name in plan.copies:
|
|
if cancel is not None and cancel.is_set():
|
|
return result
|
|
part = plan.outbox / ("." + name + PART_SUFFIX)
|
|
try:
|
|
shutil.copyfile(source, part)
|
|
part.replace(plan.outbox / name)
|
|
result.sent += 1
|
|
except OSError as exc:
|
|
part.unlink(missing_ok=True)
|
|
result.errors.append((f"Sending {source.name}", str(exc)))
|
|
for plan in receive_plans:
|
|
for path in plan.cache_prune:
|
|
try:
|
|
path.unlink(missing_ok=True)
|
|
except OSError as exc:
|
|
result.errors.append((f"Tidying the cache ({path.name})", str(exc)))
|
|
for arrival in plan.arrivals:
|
|
if cancel is not None and cancel.is_set():
|
|
return result
|
|
parsed = parse_outbox_name(arrival.path.name)
|
|
original = parsed[1] if parsed else arrival.path.name
|
|
try:
|
|
if arrival.to_cache:
|
|
plan.cache.mkdir(parents=True, exist_ok=True)
|
|
dest = plan.cache / f"{arrival.track_id}{arrival.path.suffix.lower()}"
|
|
tmp = dest.with_name("." + dest.name + PART_SUFFIX)
|
|
shutil.copyfile(arrival.path, tmp)
|
|
tmp.replace(dest)
|
|
result.cached.append((plan.token, arrival.track_id))
|
|
if arrival.to_library and music_dir is not None:
|
|
dest, fields = stage_file(arrival.path, music_dir, original)
|
|
result.staged.append(Staged(plan.token, arrival.track_id,
|
|
dest, fields,
|
|
Path(original).stem))
|
|
except OSError as exc:
|
|
result.errors.append((f"Bringing in {original}", str(exc)))
|
|
return result
|
|
|
|
|
|
class DeliveryWorker(QObject):
|
|
"""``run_plans`` on a daemon thread (the ExportWorker shape)."""
|
|
|
|
finished = pyqtSignal(object, object) # DeliveryResult, receive plans
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._busy = False
|
|
self._cancel = threading.Event()
|
|
|
|
def busy(self) -> bool:
|
|
return self._busy
|
|
|
|
def cancel(self):
|
|
self._cancel.set()
|
|
|
|
def start(self, send_plans, receive_plans, music_dir):
|
|
if self._busy:
|
|
return False
|
|
self._busy = True
|
|
self._cancel.clear()
|
|
|
|
def run():
|
|
try:
|
|
result = run_plans(send_plans, receive_plans, music_dir,
|
|
self._cancel)
|
|
except Exception as exc: # never die silently
|
|
log.exception("cassette delivery")
|
|
result = DeliveryResult(errors=[("Delivering songs", str(exc))])
|
|
self._busy = False
|
|
self.finished.emit(result, receive_plans)
|
|
threading.Thread(target=run, daemon=True, name="cassette-delivery").start()
|
|
return True
|