diff --git a/apps/musicseerr/main.py b/apps/musicseerr/main.py index 85c245d..be1b1cb 100644 --- a/apps/musicseerr/main.py +++ b/apps/musicseerr/main.py @@ -15,8 +15,30 @@ logger = logging.getLogger("musicseerr-main") app = FastAPI(title="MusicSeerr Ingestion Pipeline", version="1.0.0") -# In-memory jobs database -JOBS = {} +import json + +JOBS_FILE = "/remote-music/.musicseerr_jobs.json" + +def save_jobs_db(jobs_dict): + try: + temp_file = JOBS_FILE + ".tmp" + with open(temp_file, "w") as f: + json.dump(jobs_dict, f, indent=2) + os.replace(temp_file, JOBS_FILE) + except Exception as e: + logger.error(f"Failed to save jobs database: {e}") + +def load_jobs_db(): + if os.path.exists(JOBS_FILE): + try: + with open(JOBS_FILE, "r") as f: + return json.load(f) + except Exception as e: + logger.error(f"Failed to load jobs database: {e}") + return {} + +# Load jobs database +JOBS = load_jobs_db() # Read configurations from Environment Variables PROXY_URL = os.getenv("GLUETUN_PROXY_URL", "http://gluetun-svc:8888") @@ -79,8 +101,10 @@ def submit_job(req: SubmitRequest, background_tasks: BackgroundTasks): "status": "pending", "priority_used": None, "files": [], - "error": None + "error": None, + "chat_id": None } + save_jobs_db(JOBS) background_tasks.add_task(run_download_task, task_id, query, PROXY_URL, JOBS) @@ -123,8 +147,10 @@ async def telegram_webhook(request: Request, background_tasks: BackgroundTasks): "status": "pending", "priority_used": None, "files": [], - "error": None + "error": None, + "chat_id": chat_id } + save_jobs_db(JOBS) # Pass chat_id to the background task (we will update run_download_task to support this) background_tasks.add_task(run_download_task, task_id, text, PROXY_URL, JOBS, chat_id) @@ -546,8 +572,10 @@ def run_lastfm_auto_download(username: str): "status": "pending", "priority_used": None, "files": [], - "error": None + "error": None, + "chat_id": None } + save_jobs_db(JOBS) try: run_download_task(task_id, track_query, PROXY_URL, JOBS) except Exception as e: @@ -582,6 +610,45 @@ async def lastfm_scheduler(): await asyncio.sleep(interval_hours * 3600) +async def recover_jobs(): + # Wait for remote music directory mount to be fully ready + logger.info("Starting recovery checks for interrupted/pending ingestion tasks...") + await asyncio.sleep(10) + + global JOBS + JOBS = load_jobs_db() + + recovered_count = 0 + for task_id, job in list(JOBS.items()): + if job.get("status") in ("pending", "downloading"): + logger.info(f"Recovering interrupted job {task_id}: '{job.get('query')}'") + job["status"] = "pending" + job["error"] = "Interrupted by server restart, resuming..." + + # Re-queue the task in a background thread executor using the event loop + try: + loop = asyncio.get_event_loop() + loop.run_in_executor( + None, + run_download_task, + task_id, + job.get("query"), + PROXY_URL, + JOBS, + job.get("chat_id") + ) + recovered_count += 1 + except Exception as e: + logger.error(f"Failed to recover job {task_id}: {e}") + + if recovered_count > 0: + save_jobs_db(JOBS) + logger.info(f"Successfully recovered and resumed {recovered_count} tasks.") + else: + logger.info("No pending or interrupted ingestion tasks found to recover.") + + @app.on_event("startup") async def startup_event(): asyncio.create_task(lastfm_scheduler()) + asyncio.create_task(recover_jobs()) diff --git a/apps/musicseerr/tasks.py b/apps/musicseerr/tasks.py index d19419a..1e5f1b6 100644 --- a/apps/musicseerr/tasks.py +++ b/apps/musicseerr/tasks.py @@ -18,6 +18,17 @@ DOWNLOAD_SEMAPHORE = threading.Semaphore(5) logging.basicConfig(level=logging.INFO) logger = logging.getLogger("musicseerr-tasks") +JOBS_FILE = "/remote-music/.musicseerr_jobs.json" + +def save_jobs(jobs_dict): + try: + temp_file = JOBS_FILE + ".tmp" + with open(temp_file, "w") as f: + json.dump(jobs_dict, f, indent=2) + os.replace(temp_file, JOBS_FILE) + except Exception as e: + logger.error(f"Failed to save jobs database: {e}") + def move_new_files_to_dest(temp_dir: str, dest_dir: str) -> list: moved = [] if not os.path.exists(temp_dir): @@ -495,6 +506,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: Core download manager task that runs in the background. """ jobs_dict[task_id]["status"] = "downloading" + save_jobs(jobs_dict) temp_dir = f"/tmp/downloads/{task_id}" os.makedirs(temp_dir, exist_ok=True) @@ -539,6 +551,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: if is_playlist: jobs_dict[task_id]["priority_used"] = "playlist-scraper" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Spotify playlist detected: '{query}'") track_queries = scrape_spotify_playlist(query, resolved_proxy_url) if not track_queries: @@ -619,6 +632,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: if "files" not in jobs_dict[task_id] or not jobs_dict[task_id]["files"]: jobs_dict[task_id]["files"] = [] jobs_dict[task_id]["files"].extend(new_moved) + save_jobs(jobs_dict) logger.info(f"[{task_id}] Successfully downloaded and moved {len(new_moved)} files for track '{track_query}'") if download_count > 0: @@ -635,6 +649,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: # Priority 1: spotDL (text search or Spotify URL) if not is_tubidy and not is_youtube: jobs_dict[task_id]["priority_used"] = "spotDL" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Attempting Priority 1: spotDL for '{query}'") env = os.environ.copy() @@ -660,6 +675,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: # Priority 2: yt-dlp (direct YouTube or fallback for failed spotDL) if not is_tubidy and not is_spotify: jobs_dict[task_id]["priority_used"] = "yt-dlp" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Attempting Priority 2: yt-dlp for '{query}'") env = os.environ.copy() @@ -697,6 +713,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: # Priority 3: Custom Tubidy Link Scraper (if URL is directly passed) if is_tubidy: jobs_dict[task_id]["priority_used"] = "Tubidy" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Attempting Priority 3: Tubidy Scraper for '{query}'") # Scrape MP3 URL @@ -725,6 +742,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: # Priority 4: Tubidy Search Fallback (if spotDL and yt-dlp failed and no files are present) if not download_successful and not is_tubidy: jobs_dict[task_id]["priority_used"] = "Tubidy-Search" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Attempting Fallback Priority 4: Tubidy Search Scraper for '{query}'") search_query = query @@ -780,11 +798,13 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: logger.info(f"[{task_id}] All tracks in this query were already downloaded (skipped duplicates).") jobs_dict[task_id]["files"] = [] jobs_dict[task_id]["status"] = "completed" + save_jobs(jobs_dict) if chat_id: send_telegram_notification(chat_id, f"✅ Successfully ingested request: '{query}'\n\n(All tracks already existed - skipped duplicates)") return jobs_dict[task_id]["status"] = "completed" + save_jobs(jobs_dict) logger.info(f"[{task_id}] Music Ingestion task fully completed.") if chat_id: @@ -795,6 +815,7 @@ def _run_download_task_impl(task_id: str, query: str, proxy_url: str, jobs_dict: logger.error(f"[{task_id}] Ingestion failed: {e}") jobs_dict[task_id]["status"] = "failed" jobs_dict[task_id]["error"] = str(e) + save_jobs(jobs_dict) if chat_id: send_telegram_notification(chat_id, f"❌ Failed to ingest request: '{query}'\nError: {e}")