v0.30.0: Cassette delivery — requested songs arrive in your library

- Sending: a friend's requests ∩ what I still offer is copied into my
  outbox (via a .part name Syncthing never sees); whatever they no longer
  request is deleted, so the outbox cleans itself up.
- Receiving: a finished file I cassetted is imported exactly like Add to
  Library — organized into Artist/Album, tags read, date added now, none
  of their listening history — and a song wanted only for a followed
  playlist lands in the cache. Then requests.json is rewritten without
  what arrived and without what they stopped offering.
- Safe to repeat: Syncthing temp files ignored, Syncthing asked whether a
  file is whole, and an exact match already in my library is never
  imported twice.
- Folders are watched (re-armed after each rename), with a slow fallback.
- file_importer split into stage_file (off the GUI thread) and
  track_from_file (date added = now).
- pytest --syncthing now runs a real invite → share → request → deliver →
  cleanup between two Syncthing instances.

lintunes/__init__.py

@@ -1,3 +1,3 @@
 """LinTunes — iTunes-style music library manager and player for Linux."""
 
-__version__ = "0.29.0"
+__version__ = "0.30.0"

lintunes/cassette/delivery.py

@@ -0,0 +1,305 @@
+"""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
+
+
+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))
+
+    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 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

lintunes/cassette/service.py

@@ -254,6 +254,42 @@ class CassetteService(QObject):
             req.write_requests(out_dir(self.root, token), requested)
         self.changed.emit()
 
+    def apply_delivery(self, token: str, cassetted, requested):
+        """After a batch arrived: what's still wanted, and the requests file
+        that says so (rewritten only if it changed)."""
+        from lintunes.cassette import requests as req
+        with self._lock:
+            friend = self._state.friends.get(token)
+            if friend is None:
+                return
+            cassetted = sorted(int(t) for t in cassetted)
+            if cassetted != friend.cassetted:
+                friend.cassetted = cassetted
+                self._save()
+            req.write_requests(out_dir(self.root, token), requested)
+
+    def is_complete(self, token: str, path) -> bool:
+        """Has Syncthing finished writing this file of the friend's? It only
+        renames a download into place once it's whole, so a finished name is
+        normally enough; this double-checks with Syncthing when it can."""
+        with self._lock:
+            friend = self._state.friends.get(token)
+        if friend is None:
+            return False
+        try:
+            client = self.client()
+            relative = path.relative_to(in_dir(self.root, token)).as_posix()
+            info = client.file_info(friend.in_folder, relative)
+        except (st.SyncthingError, ValueError):
+            return True
+        if not info:
+            return True
+        local, global_ = info.get("local") or {}, info.get("global") or {}
+        if local.get("deleted") or global_.get("deleted"):
+            return False
+        return (local.get("version") == global_.get("version")
+                and local.get("size") == global_.get("size"))
+
     # ---- user actions ----
 
     def create_invite(self) -> str:

lintunes/gui/cassette_delivery.py

@@ -0,0 +1,143 @@
+"""Runs the delivery loop (``cassette/delivery.py``) inside the app.
+
+Watches every friend's folder (their requests, their outbox) with a
+QFileSystemWatcher — re-armed after each event, since Syncthing's
+rename-into-place drops the watch, the ``sync_watcher`` lesson — and a slow
+fallback timer for whatever a watcher misses. Each trigger is debounced into
+one pass: plan on the GUI thread, copy on the worker's thread, add tracks to
+the library back on the GUI thread, then rewrite requests.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from PyQt6.QtCore import QFileSystemWatcher, QObject, QTimer, pyqtSignal
+
+from lintunes.cassette import delivery, matching
+from lintunes.cassette.state import in_dir
+
+DEBOUNCE_MS = 2000
+FALLBACK_MS = 10 * 60 * 1000
+
+
+class DeliveryCoordinator(QObject):
+    notice = pyqtSignal(str)
+    step_failed = pyqtSignal(str, str)
+    library_grew = pyqtSignal(list)   # new track ids
+
+    def __init__(self, service, manager, parent=None):
+        super().__init__(parent)
+        self._service = service
+        self._manager = manager
+        self._worker = delivery.DeliveryWorker(self)
+        self._worker.finished.connect(self._on_finished)
+        self._again = False
+        self._watcher = QFileSystemWatcher(self)
+        self._watcher.directoryChanged.connect(self._on_change)
+        self._watcher.fileChanged.connect(self._on_change)
+        self._debounce = QTimer(self)
+        self._debounce.setSingleShot(True)
+        self._debounce.setInterval(DEBOUNCE_MS)
+        self._debounce.timeout.connect(self.run)
+        self._fallback = QTimer(self)
+        self._fallback.setInterval(FALLBACK_MS)
+        self._fallback.timeout.connect(self.run)
+
+    def start(self):
+        self._arm()
+        self._fallback.start()
+        self._debounce.start()
+
+    def stop(self):
+        self._fallback.stop()
+        self._debounce.stop()
+        self._worker.cancel()
+
+    def busy(self) -> bool:
+        return self._worker.busy()
+
+    def _arm(self):
+        """(Re)watch every friend's folder and outbox that exists."""
+        watched = set(self._watcher.directories()) | set(self._watcher.files())
+        wanted = []
+        for token in self._service.state().friends:
+            folder = in_dir(self._service.root, token)
+            for path in (folder, folder / delivery.OUTBOX,
+                         folder / "requests.json"):
+                if path.exists() and str(path) not in watched:
+                    wanted.append(str(path))
+        if wanted:
+            self._watcher.addPaths(wanted)
+
+    def _on_change(self, _path: str):
+        self._arm()
+        self._debounce.start()
+
+    def schedule(self):
+        self._debounce.start()
+
+    def run(self):
+        """Plan every friend in both directions and start the file work."""
+        if self._worker.busy():
+            self._again = True
+            return
+        self._arm()
+        state = self._service.state()
+        if not state.friends:
+            return
+        library = self._manager.library
+        index = matching.LibraryIndex(library.tracks.values())
+        root = self._service.root
+        sends, receives = [], []
+        for token, friend in state.friends.items():
+            sends.append(delivery.plan_send(token, root, library,
+                                            state.selection_for(friend)))
+            receives.append(delivery.plan_receive(
+                friend, root, index,
+                is_complete=lambda path, t=token: self._service.is_complete(t, path)))
+        if not any(p.copies or p.deletes for p in sends) and \
+                not any(p.arrivals for p in receives):
+            # Nothing to move, but prunes and exact-match skips still count.
+            self._finish_receives(receives, set())
+            return
+        self._worker.start(sends, receives, self._manager.organize_root())
+
+    def _on_finished(self, result, receives):
+        added = []
+        for staged in result.staged:
+            from lintunes.importers.file_importer import track_from_file
+            track = track_from_file(staged.dest, staged.fields,
+                                    staged.fallback_name)
+            self._manager.add_track(track)
+            added.append(track.track_id)
+        brought = {(s.token, s.track_id) for s in result.staged}
+        brought |= set(result.cached)
+        self._finish_receives(receives, brought)
+        if added:
+            self.library_grew.emit(added)
+        names = {t: f.name for t, f in self._service.state().friends.items()}
+        per_friend = {}
+        for token, _tid in brought:
+            per_friend[token] = per_friend.get(token, 0) + 1
+        for token, count in per_friend.items():
+            noun = "song" if count == 1 else "songs"
+            self.notice.emit(f"{count} {noun} arrived from "
+                             f"{names.get(token, 'a friend')}.")
+        for what, error in result.errors[:3]:
+            self.step_failed.emit(what, error)
+        if self._again:
+            self._again = False
+            self._debounce.start()
+
+    def _finish_receives(self, receives, brought: set):
+        """Rewrite each friend's requests without what arrived (only an
+        arrival that really landed counts — a failed copy is asked for
+        again)."""
+        for plan in receives:
+            failed = {a.track_id for a in plan.arrivals
+                      if (plan.token, a.track_id) not in brought}
+            self._service.apply_delivery(
+                plan.token, plan.cassetted | {t for t in failed
+                                              if any(a.track_id == t and a.to_library
+                                                     for a in plan.arrivals)},
+                plan.requests_after | failed)

lintunes/gui/cassette_ui.py

@@ -37,6 +37,7 @@ class CassetteUi(QObject):
         self._service_factory = service_factory
         self.service = None
         self.friend_mode = None   # FriendMode while browsing a friend
+        self.delivery = None      # DeliveryCoordinator (host machine only)
 
     # ---- the service (host machine only) ----
 
@@ -70,6 +71,15 @@ class CassetteUi(QObject):
             self.service.start()
             # Whatever changed while LinTunes was closed.
             self._republish_timer.start()
+            if self._manager is not None:
+                from lintunes.gui.cassette_delivery import DeliveryCoordinator
+                self.delivery = DeliveryCoordinator(self.service, self._manager, self)
+                self.delivery.notice.connect(self._on_notice)
+                self.delivery.step_failed.connect(self._on_step_failed)
+                self.delivery.library_grew.connect(self._on_library_grew)
+                self.service.friends_added.connect(self.delivery.schedule)
+                self.delivery.start()
+            self._refresh_selector()
         return self.service
 
     def _library(self):
@@ -134,6 +144,9 @@ class CassetteUi(QObject):
         self.friend_mode = FriendMode(
             self._window, self.service, self._manager.library,
             music_dir or self.service.root)
+        if self.delivery is not None:
+            # A save changed what I ask for: act on it now, not in ten minutes.
+            self.friend_mode.left.connect(self.delivery.schedule)
         self.friend_mode.enter(token)
 
     def leave_friend_mode(self):
@@ -145,7 +158,14 @@ class CassetteUi(QObject):
             return True
         return self.friend_mode.confirm_leave()
 
+    def _on_library_grew(self, track_ids):
+        refresh = getattr(self._window, "refresh_library_view", None)
+        if refresh is not None:
+            refresh(track_ids)
+
     def shutdown(self):
+        if self.delivery is not None:
+            self.delivery.stop()
         if self.service is not None:
             self.service.stop()
 
@@ -263,3 +283,5 @@ class CassetteUi(QObject):
         except st.SyncthingError as exc:
             QMessageBox.warning(self._window, "Sync Settings", failure_text(exc))
         service.kick()
+        if self.delivery is not None:
+            self.delivery.schedule()

lintunes/gui/main_window.py

@@ -403,6 +403,11 @@ class MainWindow(QMainWindow):
     def sidebar(self):
         return self._sidebar
 
+    def refresh_library_view(self, _track_ids=None):
+        """Songs arrived from somewhere other than an import dialog (a friend's
+        delivery): show them."""
+        self._library_view.reload()
+
     def show_my_library(self):
         self._sidebar.reflect_library()
         self._show_library()

lintunes/importers/file_importer.py

@@ -79,9 +79,36 @@ def import_file(source: Path, music_dir: Path, manager,
         dest.parent.mkdir(parents=True, exist_ok=True)
         shutil.copy2(source, dest)
 
+    track = track_from_file(dest, fields, source.stem)
+    manager.add_track(track)
+    by_location[Path(track.location)] = track
+    return track
+
+
+def stage_file(source: Path, music_dir: Path,
+               filename: str | None = None) -> tuple[Path, dict]:
+    """The file-system half of an import, safe off the GUI thread: read the
+    tags and copy the file to its organized home. Returns (dest, fields);
+    ``track_from_file`` + ``manager.add_track`` finish it on the GUI thread.
+    ``filename`` overrides the name the copy gets (a Cassette delivery
+    arrives as ``<id> ~ <name>``)."""
+    source = Path(source)
+    fields = tagging.read_tags(source)
+    dest = unique_path(organized_destination(music_dir, fields,
+                                             filename or source.name))
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(source, dest)
+    return dest, fields
+
+
+def track_from_file(dest: Path, fields: dict, fallback_name: str) -> Track:
+    """A new Track for a file just imported. Date added is *now* — every
+    individual import is new to this library, whatever the file carries;
+    only the iTunes migration keeps old dates, and it doesn't come here.
+    Only the file's own tags come across: nobody's listening history."""
     now = datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
     track = Track(
-        name=fields.get("name") or source.stem,
+        name=fields.get("name") or fallback_name,
         location=str(dest),
         date_added=now,
         date_modified=now,
@@ -91,8 +118,6 @@ def import_file(source: Path, music_dir: Path, manager,
                 "comments", "total_time", "bit_rate", "sample_rate", "size", "kind"):
         if key in fields:
             setattr(track, key, fields[key])
-    manager.add_track(track)
-    by_location[Path(track.location)] = track
     return track