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 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)