commit 34d69f373cd2ef1f4359f9829ddb78f9c26d8385
parent afd761fe4a4074143508b09470c87429d0e17228
Author: Amit Dutta <mail@amit.is-a.dev>
Date: Wed, 16 Sep 2026 22:30:01 +0530
Update main.py
Diffstat:
| M | app/main.py | | | 103 | +++++++++++++++++++++++++++++++++++++++++++++++-------------------------------- |
1 file changed, 61 insertions(+), 42 deletions(-)
diff --git a/app/main.py b/app/main.py
@@ -10,6 +10,7 @@ import tempfile
import asyncio
import socket
import ipaddress
+import random
from urllib.parse import urlparse, unquote
import httpx
from collections import deque
@@ -313,17 +314,23 @@ async def url_ytmusic_page(request: Request):
class YtRequest(BaseModel):
url: str
+COBALT_FALLBACK_URLS = [
+ "https://api.cobalt.tools/",
+ "https://co.wukko.me/"
+]
+
@app.post("/api/yt/download")
async def yt_download_cobalt(request: Request, body: YtRequest):
url = body.url.strip()
if not url:
raise HTTPException(status_code=400, detail="Missing YouTube URL.")
- # The official api.cobalt.tools now enforces Cloudflare Turnstile/JWT which fails
- # for automated backend requests. Defaulting to a free community instance.
- cobalt_api = os.getenv("COBALT_API_URL", "https://api.cobalt.canine.tools/")
- if not cobalt_api.endswith("/"):
- cobalt_api += "/"
+ # Determine which instance URL to use
+ user_defined_api = os.getenv("COBALT_API_URL")
+
+ # If the user defines an API URL, strictly use it. Otherwise, attempt fallbacks
+ # in case community instances go offline (DNS/NXDOMAIN errors).
+ instances_to_try = [user_defined_api] if user_defined_api else COBALT_FALLBACK_URLS
headers = {
"Accept": "application/json",
@@ -331,16 +338,13 @@ async def yt_download_cobalt(request: Request, body: YtRequest):
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
}
- # Optionally bind an authorization header if your own instance token is set
cobalt_api_token = os.getenv("COBALT_API_TOKEN")
if cobalt_api_token:
- # Standardize Api-Key auth if not strictly provided as Bearer
if " " not in cobalt_api_token:
headers["Authorization"] = f"Api-Key {cobalt_api_token}"
else:
headers["Authorization"] = cobalt_api_token
- # Cobalt v11 API Payload
payload = {
"url": url,
"downloadMode": "audio",
@@ -348,41 +352,55 @@ async def yt_download_cobalt(request: Request, body: YtRequest):
"filenameStyle": "basic"
}
- async with httpx.AsyncClient(timeout=60.0) as client:
- try:
- res = await client.post(cobalt_api, json=payload, headers=headers)
-
- # Non-200 could be bot protection, ratelimits, or API validation errors.
- if res.status_code != 200:
- err_text = res.text
- try:
- data = res.json()
- if "error" in data and "code" in data["error"]:
- err_text = data["error"]["code"]
- else:
- err_text = data.get("text", err_text)
- except Exception:
- # Truncate raw HTML (like Turnstile challenge pages) so it doesn't flood logs
- if len(err_text) > 200:
- err_text = err_text[:200] + "..."
- raise HTTPException(status_code=400, detail=f"Cobalt API failed: {err_text}")
-
- data = res.json()
- status = data.get("status")
-
- if status == "error":
- err_code = data.get("error", {}).get("code", "Unknown Cobalt error")
- raise HTTPException(status_code=400, detail=f"Cobalt error: {err_code}")
-
- # The API might tunnel the file directly or redirect to a download link
- dl_url = data.get("url")
- if not dl_url:
- raise HTTPException(status_code=500, detail=f"Cobalt succeeded but returned no download link. (Status: {status})")
-
- return {"url": dl_url, "title": "Audio"}
+ last_error = "Could not reach any Cobalt API instances."
+
+ for instance in instances_to_try:
+ if not instance.endswith("/"):
+ instance += "/"
+
+ async with httpx.AsyncClient(timeout=30.0) as client:
+ try:
+ res = await client.post(instance, json=payload, headers=headers)
+
+ # Non-200 could be bot protection, ratelimits, or API validation errors.
+ if res.status_code != 200:
+ err_text = res.text
+ try:
+ data = res.json()
+ if "error" in data and "code" in data["error"]:
+ err_text = data["error"]["code"]
+ else:
+ err_text = data.get("text", err_text)
+ except Exception:
+ if len(err_text) > 200:
+ err_text = err_text[:200] + "..."
+
+ last_error = f"API error ({res.status_code}): {err_text}"
+ # Do not fallback if the instance actually responded with an auth/format error
+ if res.status_code in (400, 401, 403):
+ raise HTTPException(status_code=400, detail=f"Cobalt returned an error: {err_text}")
+ continue
+
+ data = res.json()
+ status = data.get("status")
+
+ if status == "error":
+ err_code = data.get("error", {}).get("code", "Unknown Cobalt error")
+ raise HTTPException(status_code=400, detail=f"Cobalt error: {err_code}")
+
+ dl_url = data.get("url")
+ if not dl_url:
+ raise HTTPException(status_code=500, detail=f"Cobalt succeeded but returned no download link. (Status: {status})")
+
+ return {"url": dl_url, "title": "Audio"}
+
+ except (httpx.RequestError, socket.gaierror) as exc:
+ last_error = f"Failed to connect to {instance}: {exc}"
+ continue # Try the next instance in the list
+
+ # If the loop finishes without returning, all instances failed
+ raise HTTPException(status_code=502, detail=last_error)
- except httpx.RequestError as exc:
- raise HTTPException(status_code=502, detail=f"Error connecting to Cobalt: {exc}")
# ---------------------------------------------------------------------------------
@app.get("/shorten")
@@ -714,3 +732,4 @@ async def serve(request: Request, path: str):
ctx = await render_context(extra_ctx)
return templates.TemplateResponse(request, "index.html", ctx)
+```eof
