import os import uuid import logging from fastapi import FastAPI, BackgroundTasks, Request, Form from fastapi.responses import HTMLResponse, JSONResponse from pydantic import BaseModel import requests from curl_cffi import requests as cf_requests from tasks import run_download_task # Configure Logger logging.basicConfig(level=logging.INFO) logger = logging.getLogger("musicseerr-main") app = FastAPI(title="MusicSeerr Ingestion Pipeline", version="1.0.0") # In-memory jobs database JOBS = {} # Read configurations from Environment Variables PROXY_URL = os.getenv("GLUETUN_PROXY_URL", "http://gluetun-svc:8888") TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") class SubmitRequest(BaseModel): query: str def send_telegram_message(chat_id: int, text: str): """ Sends a message back to Telegram. Outgoing traffic uses the normal cluster network. """ if not TELEGRAM_BOT_TOKEN or not chat_id: return url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" payload = {"chat_id": chat_id, "text": text} try: requests.post(url, json=payload, timeout=10) except Exception as e: logger.error(f"Failed to send Telegram message to chat {chat_id}: {e}") @app.get("/health") def health_check(): return {"status": "healthy", "proxy": PROXY_URL} @app.get("/jobs") def get_jobs(): """ Returns the list of all jobs, sorted by timestamp. """ return JSONResponse(content=list(JOBS.values())) @app.get("/status/{task_id}") def get_job_status(task_id: str): """ Returns the status of a specific job. """ if task_id not in JOBS: return JSONResponse(status_code=404, content={"error": "Job not found"}) return JSONResponse(content=JOBS[task_id]) @app.post("/submit") def submit_job(req: SubmitRequest, background_tasks: BackgroundTasks): """ API endpoint to submit download jobs via JSON. """ task_id = str(uuid.uuid4()) query = req.query.strip() if not query: return JSONResponse(status_code=400, content={"error": "Query cannot be empty"}) JOBS[task_id] = { "id": task_id, "query": query, "status": "pending", "priority_used": None, "files": [], "error": None } background_tasks.add_task(run_download_task, task_id, query, PROXY_URL, JOBS) return {"task_id": task_id, "status": "pending"} @app.post("/telegram-webhook") async def telegram_webhook(request: Request, background_tasks: BackgroundTasks): """ Endpoint for Telegram bot webhooks. """ try: payload = await request.json() logger.info(f"Received Telegram Webhook Payload: {payload}") message = payload.get("message", {}) chat = message.get("chat", {}) chat_id = chat.get("id") text = message.get("text", "").strip() if not text or not chat_id: return {"status": "ignored"} # Ignore commands like /start unless they have arguments if text.startswith("/start"): parts = text.split(" ", 1) if len(parts) > 1: text = parts[1].strip() else: send_telegram_message( chat_id, "Welcome to MusicSeerr! 🎵\nSend me a song name, Spotify URL, YouTube URL, or Tubidy link, and I'll ingest it into the Navidrome library!" ) return {"status": "welcomed"} task_id = str(uuid.uuid4()) JOBS[task_id] = { "id": task_id, "query": text, "status": "pending", "priority_used": None, "files": [], "error": None } # 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) send_telegram_message(chat_id, f"📥 Queued: '{text}'\nTask ID: {task_id}") except Exception as e: logger.error(f"Error handling Telegram webhook: {e}") return {"status": "processed"} @app.get("/", response_class=HTMLResponse) def index_page(): """ Serves a highly-styled, responsive dark-mode portal. """ html_content = """ MusicSeerr - Music Ingestion Portal

MusicSeerr 🎵

Sovereign Music Ingestion Portal for Navidrome

Active Ingestion Jobs

No active downloads. Submit a link above!

FastAPI Backend Active
VPN Proxy Bound
""" return HTMLResponse(content=html_content) def run_lastfm_auto_download(username: str): url = f"https://lfm.xiffy.nl/{username}/recommended" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } logger.info(f"Fetching Last.fm recommendations feed from {url}") response = None try: logger.info("Fetching Last.fm feed directly (no proxy)") response = cf_requests.get(url, headers=headers, impersonate="chrome", timeout=20) response.raise_for_status() except Exception as e: logger.warning(f"Failed to fetch Last.fm feed directly: {e}. Retrying with proxy...") if PROXY_URL: proxies = {"http": PROXY_URL, "https": PROXY_URL} try: response = cf_requests.get(url, headers=headers, proxies=proxies, impersonate="chrome", timeout=20) response.raise_for_status() except Exception as e2: logger.error(f"Failed to fetch Last.fm feed with proxy: {e2}") return else: return try: import xml.etree.ElementTree as ET root = ET.fromstring(response.content) items = root.findall(".//item") except Exception as e: logger.error(f"Failed to parse Last.fm recommendations XML: {e}") return logger.info(f"Found {len(items)} items in Last.fm feed.") # Import tasks dynamically to avoid circular import from tasks import run_download_task import re def is_track_already_downloaded(track_query: str) -> bool: remote_base = "/remote-music/music" if not os.path.isdir(remote_base): return False clean_query = re.sub(r'[^\w\s-]', '', track_query).lower() parts = track_query.split(" - ", 1) if len(parts) == 2: artist, title = parts[0].strip().lower(), parts[1].strip().lower() artist_clean = re.sub(r'[^\w\s-]', '', artist) title_clean = re.sub(r'[^\w\s-]', '', title) else: artist_clean = clean_query title_clean = clean_query for r, d, files in os.walk(remote_base): for file in files: if file.lower().endswith((".mp3", ".flac", ".m4a", ".ogg", ".wav", ".opus")): file_lower = file.lower() if len(parts) == 2: if artist_clean in file_lower and title_clean in file_lower: return True else: if clean_query in file_lower: return True return False for item in items: title_el = item.find("title") if title_el is not None and title_el.text: track_query = title_el.text.strip() # Check if already queued already_queued = any(job.get("query") == f"[Last.fm Auto] {track_query}" for job in JOBS.values()) if already_queued: continue if is_track_already_downloaded(track_query): logger.info(f"Skipping Last.fm auto-download for '{track_query}' (already exists).") continue logger.info(f"Queueing Last.fm recommendation: '{track_query}'") task_id = str(uuid.uuid4()) JOBS[task_id] = { "id": task_id, "query": f"[Last.fm Auto] {track_query}", "status": "pending", "priority_used": None, "files": [], "error": None } try: run_download_task(task_id, track_query, PROXY_URL, JOBS) except Exception as e: logger.error(f"Failed to process auto-download job {task_id}: {e}") @app.post("/lastfm/trigger") def trigger_lastfm_sync(background_tasks: BackgroundTasks): username = os.getenv("LASTFM_AUTO_DOWNLOAD_USER") if not username: return JSONResponse(status_code=400, content={"error": "Last.fm auto-download is not enabled (LASTFM_AUTO_DOWNLOAD_USER not configured)"}) background_tasks.add_task(run_lastfm_auto_download, username) return {"status": "triggered", "user": username} import asyncio async def lastfm_scheduler(): # Wait a short bit after startup before the first check await asyncio.sleep(30) while True: username = os.getenv("LASTFM_AUTO_DOWNLOAD_USER") if username: try: logger.info(f"Triggering scheduled Last.fm recommendations fetch for '{username}'...") loop = asyncio.get_event_loop() await loop.run_in_executor(None, run_lastfm_auto_download, username) except Exception as e: logger.error(f"Error in Last.fm scheduler task: {e}") interval_hours = float(os.getenv("LASTFM_AUTO_DOWNLOAD_INTERVAL_HOURS", "12")) await asyncio.sleep(interval_hours * 3600) @app.on_event("startup") async def startup_event(): asyncio.create_task(lastfm_scheduler())