feat(sync): change ‘both’ mode to one video download +audio extraction; update DB and events accordingly

README.md

@@ -22,6 +22,12 @@ Local-first YouTube playlist synchronization client.
 - `yt-dlp` (pip)
 - `ffmpeg` (for audio extraction)
 
+Install:
+
+```bash
+pip install -U yt-dlp
+```
+
 ## Configure
 
 Create/edit `config/yt-playlist-config.json`:

src/app/cli.py

@@ -9,6 +9,7 @@ from .core.database.db import Database
 from .core.sync.service import SyncService
 from .core.sync.executor import ActionExecutor
 from .core.events.event_bus import EventBus
+import re
 from .core.utils.yt import extract_playlist_id
 
 
@@ -17,6 +18,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--apply", action="store_true", help="Apply actions (otherwise compute-only)")
     parser.add_argument("--db", type=Path, default=Path("app/data/app.db"), help="Path to SQLite database")
     parser.add_argument("--playlist", type=int, default=None, help="Only run for a specific playlist index (0-based)")
+    parser.add_argument("--verbose", action="store_true", help="Print detailed events (rename/recycle/start)")
     args = parser.parse_args(argv)
 
     settings = Settings()
@@ -25,16 +27,45 @@ def main(argv: list[str] | None = None) -> int:
     bus = EventBus()
     executor = ActionExecutor(db, event_bus=bus)
 
-    async def log_event(payload):
-        # Placeholder subscriber for future GUI/log integration
-        print(f"EVENT: {payload}")
+    seen_errors: set[str] = set()
+
+    ansi = re.compile(r"\x1b\[[0-9;]*m")
+
+    async def on_started(payload):
+        if args.verbose:
+            vid = payload.get("video_id")
+            target = payload.get("target")
+            print(f"START: {vid} → {target}")
+
+    async def on_completed(payload):
+        pid = payload.get("playlist_id")
+        vid = payload.get("video_id")
+        target = payload.get("target")
+        print(f"OK: {vid} → {target}")
+
+    async def on_failed(payload):
+        raw = str(payload.get("error", "failed"))
+        msg = ansi.sub("", raw)
+        # Print only once per unique message
+        if msg not in seen_errors:
+            seen_errors.add(msg)
+            # Friendly hint for missing ffmpeg
+            if "ffprobe and ffmpeg not found" in msg.lower():
+                print("ERROR: ffmpeg not found. Install ffmpeg or set 'ffmpeg_path' in config.")
+            else:
+                print(f"ERROR: {msg}")
 
     # Subscribe to key events
-    bus.subscribe("DownloadStarted", log_event)
-    bus.subscribe("DownloadCompleted", log_event)
-    bus.subscribe("DownloadFailed", log_event)
-    bus.subscribe("RenameApplied", log_event)
-    bus.subscribe("FileRecycled", log_event)
+    bus.subscribe("DownloadStarted", on_started)
+    bus.subscribe("DownloadCompleted", on_completed)
+    bus.subscribe("DownloadFailed", on_failed)
+    if args.verbose:
+        async def on_rename(payload):
+            print(f"RENAME: {payload.get('video_id')} → {payload.get('to')}")
+        async def on_recycle(payload):
+            print(f"RECYCLE: {payload.get('video_id')} ← {payload.get('name')}")
+        bus.subscribe("RenameApplied", on_rename)
+        bus.subscribe("FileRecycled", on_recycle)
 
     playlists = settings.playlists
     if args.playlist is not None:

src/app/core/download/downloader.py

@@ -19,6 +19,9 @@ class Downloader:
         try:
             job.state = JobState.DOWNLOADING
             await self._download(job)
+            # Optional local audio extraction when requested
+            if job.mode == "video" and job.audio_output_path is not None:
+                await self._extract_audio(job)
             job.state = JobState.COMPLETED
         except Exception as exc:  # pragma: no cover - environment dependent
             job.state = JobState.FAILED
@@ -31,6 +34,17 @@ class Downloader:
         def run():
             import yt_dlp  # type: ignore
 
+            class _QuietLogger:
+                def debug(self, msg):
+                    pass
+                def warning(self, msg):
+                    pass
+                def error(self, msg):
+                    # swallow inner repeats; errors are surfaced via exceptions
+                    pass
+                def info(self, msg):
+                    pass
+
             outtmpl = str(job.output_path)
             if job.mode == "audio":
                 ydl_opts = {
@@ -46,6 +60,7 @@ class Downloader:
                     "noplaylist": True,
                     "quiet": True,
                     "no_warnings": True,
+                    "logger": _QuietLogger(),
                 }
             else:  # video
                 ydl_opts = {
@@ -55,12 +70,46 @@ class Downloader:
                     "noplaylist": True,
                     "quiet": True,
                     "no_warnings": True,
+                    "logger": _QuietLogger(),
                 }
 
-            if self.ffmpeg_path:
+            # Prefer job-provided path first
+            if job.ffmpeg_path:
+                ydl_opts["ffmpeg_location"] = job.ffmpeg_path
+            elif self.ffmpeg_path:
                 ydl_opts["ffmpeg_location"] = self.ffmpeg_path
 
             with yt_dlp.YoutubeDL(ydl_opts) as ydl:  # type: ignore[attr-defined]
                 ydl.download([job.url])
 
         await asyncio.to_thread(run)
+
+    async def _extract_audio(self, job: DownloadJob):
+        import asyncio
+        from shutil import which
+
+        src = job.output_path
+        dst = job.audio_output_path
+        if not src or not dst:
+            return
+
+        def run():
+            ffmpeg_exe = job.ffmpeg_path or self.ffmpeg_path or which("ffmpeg") or "ffmpeg"
+            import subprocess
+            cmd = [
+                str(ffmpeg_exe),
+                "-y",
+                "-i",
+                str(src),
+                "-vn",
+                "-codec:a",
+                "libmp3lame",
+                "-q:a",
+                "0",
+                str(dst),
+            ]
+            subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+        # Mark converting state only for local clarity; not published
+        job.state = JobState.CONVERTING
+        await asyncio.to_thread(run)

src/app/core/download/queue_manager.py

@@ -27,6 +27,8 @@ class DownloadJob:
     mode: str = "audio"  # audio|video
     state: JobState = JobState.QUEUED
     error: Optional[str] = None
+    ffmpeg_path: Optional[str] = None
+    audio_output_path: Optional[Path] = None  # when mode=video and we also want mp3
 
 
 class QueueManager:

src/app/core/download/workers.py

@@ -6,7 +6,7 @@ from .queue_manager import DownloadJob, JobState
 
 
 async def default_worker(job: DownloadJob, *, max_retries: int = 2, delay_seconds: float = 1.5):
-    dl = Downloader()
+    dl = Downloader(ffmpeg_path=job.ffmpeg_path)
     attempt = 0
     while attempt <= max_retries:
         await dl.handle_job(job)

src/app/core/sync/executor.py

@@ -116,15 +116,62 @@ class ActionExecutor:
         await queue.start(worker)
         try:
             jobs: List[DownloadJob] = []
+
+            # Collapse 'both' into single video download + local audio extraction
+            # Build per-video desired outputs
+            by_vid: dict[str, dict[str, str]] = {}
             for a in actions:
                 if a.type != SyncActionType.DOWNLOAD or not a.item or not a.to_name:
                     continue
+                d = by_vid.setdefault(a.item.video_id, {})
+                if a.to_name.endswith(".mp3"):
+                    d["audio"] = a.to_name
+                elif a.to_name.endswith(".mp4"):
+                    d["video"] = a.to_name
+
+            ffmpeg_cfg = str(playlist_cfg.get("ffmpeg_path", "ffmpeg")) if playlist_cfg.get("ffmpeg_path") is not None else None
+
+            for a in actions:
+                if a.type != SyncActionType.DOWNLOAD or not a.item or not a.to_name:
+                    continue
+                vid = a.item.video_id
+                targets = by_vid.get(vid, {})
+
+                # If both audio and video requested for this video id, enqueue only video job with audio_output_path
+                if targets.get("audio") and targets.get("video"):
+                    # only create job once, when encountering the video target
+                    if a.to_name.endswith(".mp4"):
+                        video_path = video_root / targets["video"]
+                        audio_path = audio_root / targets["audio"]
+                        for p in (video_path.parent, audio_path.parent):
+                            p.mkdir(parents=True, exist_ok=True)
+                        url = f"https://www.youtube.com/watch?v={vid}"
+                        job = DownloadJob(
+                            item=a.item,
+                            output_path=video_path,
+                            url=url,
+                            mode="video",
+                            ffmpeg_path=ffmpeg_cfg,
+                            audio_output_path=audio_path,
+                        )
+                        jobs.append(job)
+                        await queue.enqueue(job)
+                    # skip creating a separate audio job
+                    continue
+
+                # Normal single-output path
                 is_audio = a.to_name.endswith(".mp3")
                 root = audio_root if is_audio else video_root
                 output_path = root / a.to_name
                 output_path.parent.mkdir(parents=True, exist_ok=True)
-                url = f"https://www.youtube.com/watch?v={a.item.video_id}"
-                job = DownloadJob(item=a.item, output_path=output_path, url=url, mode=("audio" if is_audio else "video"))
+                url = f"https://www.youtube.com/watch?v={vid}"
+                job = DownloadJob(
+                    item=a.item,
+                    output_path=output_path,
+                    url=url,
+                    mode=("audio" if is_audio else "video"),
+                    ffmpeg_path=ffmpeg_cfg,
+                )
                 jobs.append(job)
                 await queue.enqueue(job)
         finally:
@@ -136,10 +183,12 @@ class ActionExecutor:
             if job.item and job.output_path:
                 try:
                     if job.state.name == "COMPLETED":
-                        self.db.update_local_filename(playlist_id, job.item.video_id, job.output_path.name)
+                        # Prefer audio filename if produced
+                        final_name = job.audio_output_path.name if job.audio_output_path is not None else job.output_path.name
+                        self.db.update_local_filename(playlist_id, job.item.video_id, final_name)
                         self.db.mark_downloaded(playlist_id, job.item.video_id, True)
                         if self.bus:
-                            await self.bus.publish("DownloadCompleted", {"playlist_id": playlist_id, "video_id": job.item.video_id, "target": str(job.output_path)})
+                            await self.bus.publish("DownloadCompleted", {"playlist_id": playlist_id, "video_id": job.item.video_id, "target": str(job.audio_output_path or job.output_path)})
                     else:
                         # Ensure not marked as downloaded if failed
                         self.db.mark_downloaded(playlist_id, job.item.video_id, False)