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