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)