File size: 3,701 Bytes
3b8b8aa ff041f0 49ca13a 3b8b8aa 875ff40 5d5bfd7 49ca13a fcfeeda 3b8b8aa fcfeeda 3b8b8aa fcfeeda 3b8b8aa afbaf1d fcfeeda 3b8b8aa fcfeeda 3b8b8aa fcfeeda 3b8b8aa afbaf1d 3b8b8aa afbaf1d 49ca13a afbaf1d bd5b698 db08f36 afbaf1d 3b8b8aa afbaf1d 0ff620e afbaf1d 49ca13a 5d5bfd7 afbaf1d 49ca13a ff041f0 e8421fb 49ca13a afbaf1d 49ca13a afbaf1d 49ca13a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 |
import os
import logging
import sys
from datetime import datetime
import tidalapi
from tidalapi import Track, Quality
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
app = FastAPI()
logger = logging.getLogger("tidalapi_app")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
def load_tokens_from_secret(session):
secret = os.getenv("TIDAL_OAUTH_TOKENS")
if not secret:
logger.error("Environment variable TIDAL_OAUTH_TOKENS not set")
return False
lines = secret.strip().splitlines()
if len(lines) != 4:
logger.error("TIDAL_OAUTH_TOKENS secret malformed: expected 4 lines")
return False
try:
expiry_time = datetime.fromisoformat(lines[3])
except Exception as e:
logger.error(f"Failed to parse expiry_time: {e}")
return False
session.load_oauth_session(lines[0], lines[1], lines[2], expiry_time)
logger.info("OAuth tokens loaded successfully from secret")
return True
# Initialize session globally
session = tidalapi.Session()
if not load_tokens_from_secret(session):
raise RuntimeError("Failed to load OAuth tokens from secret")
if not session.check_login():
raise RuntimeError("Failed to login with saved tokens")
# Map client quality strings to tidalapi Quality enums
QUALITY_MAP = {
"LOW": Quality.low_96k,
"HIGH": Quality.low_320k,
"LOSSLESS": Quality.high_lossless,
"HI_RES": Quality.hi_res_lossless,
}
@app.get("/")
async def root():
logger.info("GET /")
return {
"message": "Internal Server Error",
"author": "made by Cody from chrunos.com"
}
@app.get("/track/")
def get_track_download_url(
id: int = Query(..., description="TIDAL Track ID"),
quality: str = Query("HI_RES", description="Audio quality", regex="^(LOW|HIGH|LOSSLESS|HI_RES)$"),
):
logger.info(f"Request received for track_id: {id} with quality: {quality}")
if quality not in QUALITY_MAP:
logger.error(f"Invalid quality requested: {quality}")
raise HTTPException(status_code=400, detail=f"Invalid quality. Available: {list(QUALITY_MAP.keys())}")
session.audio_quality = QUALITY_MAP[quality]
try:
track = Track(session, id)
logger.info(f"Track found: {track.name} by {track.artist.name}")
except Exception as e:
logger.error(f"Track lookup failed: {e}")
raise HTTPException(status_code=404, detail="Track not found")
stream = track.get_stream()
manifest = stream.get_stream_manifest()
result = {
"track_id": id,
"track_name": track.name,
"artist_name": track.artist.name,
"audio_quality": str(stream.audio_quality),
"stream_type": None,
"download_urls": [],
}
if stream.is_bts:
urls = manifest.get_urls()
if not urls:
logger.error("No direct URLs found in manifest")
raise HTTPException(status_code=500, detail="No downloadable URLs found")
result["stream_type"] = "bts"
result["download_urls"] = urls
logger.info(f"Returning {len(urls)} direct download URL(s)")
elif stream.is_mpd:
mpd_manifest = stream.get_manifest_data()
result["stream_type"] = "mpd"
result["mpd_manifest"] = mpd_manifest
logger.info("Returning MPEG-DASH (MPD) manifest data")
else:
logger.error("Unsupported stream type")
raise HTTPException(status_code=500, detail="Unsupported stream type")
return JSONResponse(content=result)
|