app.py

import os
import logging
import sys
from datetime import datetime
from typing import List, Optional

import tidalapi
from tidalapi import Track, Quality
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse

app = FastAPI()

# --- Logging Setup ---
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)

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

# --- Constants ---
QUALITY_MAP = {
    "LOW": Quality.low_96k,
    "HIGH": Quality.low_320k,
    "LOSSLESS": Quality.high_lossless,
    "HI_RES": Quality.hi_res_lossless,
}

# --- Helper Functions ---
def _extract_stream_info(track_id: int):
    """
    Internal helper to fetch stream info for a specific track ID.
    Returns a dictionary or raises an exception/returns None on failure.
    """
    try:
        track = Track(session, track_id)
        # Verify track exists/can be fetched
        _ = track.name 
    except Exception as e:
        logger.error(f"Track lookup failed for ID {track_id}: {e}")
        return None

    try:
        stream = track.get_stream()
        manifest = stream.get_stream_manifest()
    except Exception as e:
        logger.error(f"Stream fetching failed for {track.name} ({track_id}): {e}")
        return None

    result = {
        "track_id": track_id,
        "track_name": track.name,
        "artist_name": track.artist.name,
        "album_name": track.album.name if track.album else "Unknown",
        "audio_quality": str(stream.audio_quality),
        "stream_type": None,
        "download_urls": [],
        "mpd_manifest": None
    }

    if stream.is_bts:
        urls = manifest.get_urls()
        if urls:
            result["stream_type"] = "bts"
            result["download_urls"] = urls
    elif stream.is_mpd:
        mpd_manifest = stream.get_manifest_data()
        result["stream_type"] = "mpd"
        result["mpd_manifest"] = mpd_manifest
    else:
        logger.warning(f"Unsupported stream type for track {track_id}")
        return None

    return result

# --- Endpoints ---

@app.get("/")
async def root():
    logger.info("GET /")
    return {
        "message": "Internal Server Error",
        "author": "made by Cody from chrunos.com"
    }

@app.get("/search")
def search_tracks(
    query: str = Query(..., description="Search query (e.g., song title)"),
    limit: int = Query(10, description="Max results to return", ge=1, le=50),
    offset: int = Query(0, description="Pagination offset", ge=0)
):
    """
    Search for tracks on Tidal.
    """
    logger.info(f"Searching for: {query} (limit={limit}, offset={offset})")
    
    try:
        # models=[Track] ensures we only get tracks back
        search_results = session.search(query, models=[Track], limit=limit, offset=offset)
        tracks = search_results.get('tracks', [])
        
        results = []
        for t in tracks:
            results.append({
                "id": t.id,
                "title": t.name,
                "artist": t.artist.name,
                "album": t.album.name if t.album else None,
                "duration": t.duration,
                "explicit": t.explicit
            })
            
        return JSONResponse(content={"query": query, "count": len(results), "results": results})
        
    except Exception as e:
        logger.error(f"Search failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/playlist")
def get_playlist(
    id: str = Query(..., description="Tidal Playlist UUID"),
    quality: str = Query("HI_RES", description="Audio quality", pattern="^(LOW|HIGH|LOSSLESS|HI_RES)$"),
    limit: int = Query(50, description="Max tracks to process (prevent timeouts)", ge=1, le=100),
    offset: int = Query(0, description="Track offset", ge=0)
):
    """
    Fetch a playlist and return download URLs for its tracks.
    """
    logger.info(f"Request received for playlist: {id} with quality: {quality}")

    if quality not in QUALITY_MAP:
        raise HTTPException(status_code=400, detail=f"Invalid quality. Available: {list(QUALITY_MAP.keys())}")

    session.audio_quality = QUALITY_MAP[quality]

    try:
        playlist = session.playlist(id)
        # Fetch tracks from the playlist
        tracks = playlist.tracks(limit=limit, offset=offset)
        logger.info(f"Found {len(tracks)} tracks in playlist {playlist.name}")
    except Exception as e:
        logger.error(f"Playlist lookup failed: {e}")
        raise HTTPException(status_code=404, detail="Playlist not found or API error")

    processed_tracks = []
    failed_tracks = []

    for t in tracks:
        # Reuse the helper to get stream info
        stream_info = _extract_stream_info(t.id)
        if stream_info:
            processed_tracks.append(stream_info)
        else:
            failed_tracks.append({"id": t.id, "name": t.name})

    return JSONResponse(content={
        "playlist_id": id,
        "playlist_name": playlist.name,
        "processed_count": len(processed_tracks),
        "failed_count": len(failed_tracks),
        "tracks": processed_tracks,
        "failed": failed_tracks
    })

@app.get("/track/")
def get_track_download_url(
    id: int = Query(..., description="TIDAL Track ID"),
    quality: str = Query("HI_RES", description="Audio quality", pattern="^(LOW|HIGH|LOSSLESS|HI_RES)$"),
):
    logger.info(f"Request received for track_id: {id} with quality: {quality}")

    if quality not in QUALITY_MAP:
        raise HTTPException(status_code=400, detail=f"Invalid quality. Available: {list(QUALITY_MAP.keys())}")

    session.audio_quality = QUALITY_MAP[quality]

    result = _extract_stream_info(id)
    
    if not result:
        raise HTTPException(status_code=404, detail="Track not found or no stream available")

    return JSONResponse(content=result)