diff --git a/test/add_cover_art.py b/test/add_cover_art.py deleted file mode 100644 index 5eaeacf..0000000 --- a/test/add_cover_art.py +++ /dev/null @@ -1,1299 +0,0 @@ -# Standard library -import os -import sys -import re -import csv -import asyncio -import logging -import traceback - -# Third-party -import aiohttp -from pathlib import Path -from dotenv import load_dotenv -from rapidfuzz import fuzz -from music_tag import load_file # type: ignore -from rich.console import Console -from rich.table import Table -from rich.progress import ( - Progress, - BarColumn, - TextColumn, - TimeElapsedColumn, - TaskProgressColumn, -) - -# Local imports -sys.path.insert(0, "..") -from utils.sr_wrapper import SRUtil - -import musicbrainzngs # type: ignore -from discogs_client import Client # type: ignore - -# typing helper -from typing import Any, cast, Optional - -# Optional: use the popular `itunespy` PyPI package when available -try: - import itunespy # type: ignore - - HAVE_ITUNESPY = True -except Exception: - itunespy = None - HAVE_ITUNESPY = False - -# Optional: use `spotipy` when available for Spotify lookups -try: - import spotipy # type: ignore - - HAVE_SPOTIPY = True -except Exception: - spotipy = None - HAVE_SPOTIPY = False -# Reminder: If you see 'Import "music_tag" could not be resolved', run: -# uv add music-tag - -# Configurable paths and extensions -MUSIC_DIR = Path("/storage/music2/completed/FLAC/review") -AUDIO_EXTS = {".flac", ".mp3", ".m4a", ".ogg", ".wav", ".aac"} -REPORT_CSV = "cover_art_report.csv" -ALBUM_ART_CACHE: dict = {} - - -# Reminder: If you see 'Import "music_tag" could not be resolved', run: -# uv add music-tag -async def search_musicbrainz_cover( - artist, album, session: aiohttp.ClientSession, limiter: "AsyncRateLimiter" -): - # Use musicbrainzngs to search for a release-group matching artist+album - try: - # search for release-groups using a thread to avoid blocking - query = f"artist:{artist} AND release:{album}" - try: - res = await asyncio.to_thread( - musicbrainzngs.search_release_groups, query, 5 - ) - except Exception: - res = {} - if COVER_DEBUG_QUERIES: - try: - rgs_dbg = res.get("release-group-list") or [] - dbg_info = [] - for rg in rgs_dbg[:3]: - dbg_info.append( - { - "id": rg.get("id"), - "title": rg.get("title"), - "artist": artist_credit_to_name( - rg.get("artist-credit", []) - ), - } - ) - console.print( - f"[cyan][DEBUG] MusicBrainz candidates: {dbg_info}[/cyan]" - ) - except Exception: - pass - rgs = res.get("release-group-list") or [] - if COVER_DEBUG_QUERIES: - try: - dbg_info = [] - for rg in (rgs or [])[:3]: - dbg_info.append( - { - "id": rg.get("id"), - "title": rg.get("title"), - "artist": artist_credit_to_name( - rg.get("artist-credit", []) - ), - } - ) - console.print( - f"[cyan][DEBUG] MusicBrainz top candidates: {dbg_info}[/cyan]" - ) - except Exception: - pass - for rg in rgs: - # try to get cover art via Cover Art Archive for releases in the group - # check releases for a cover - releases = rg.get("release-list") or [] - for rel in releases: - relid = rel.get("id") - if relid: - caa_url = f"https://coverartarchive.org/release/{relid}/front-500" - try: - await limiter.acquire() - timeout = aiohttp.ClientTimeout(total=15) - async with session.get(caa_url, timeout=timeout) as resp: - if resp.status == 200: - return await resp.read() - except Exception: - continue - return None - except Exception as e: - console.print(f"[red]MusicBrainz search exception: {e}[/red]") - return None - - -async def search_discogs_cover( - artist, album, session: aiohttp.ClientSession, limiter: "AsyncRateLimiter" -): - # Use discogs_client to search for releases matching artist+album - try: - if not DISCOGS_TOKEN: - return None - # Use the discogs client (synchronous) to search in a thread - try: - await limiter.acquire() - if COVER_DEBUG_QUERIES: - console.print( - f"[cyan][DEBUG] Discogs query: album='{album}' artist='{artist}'" - ) - results = await asyncio.to_thread( - discogs_client.search, album, {"artist": artist, "type": "release"} - ) - except Exception: - results = [] - if COVER_DEBUG_QUERIES: - try: - dbg = [] - for rr in (results or [])[:3]: - try: - data = getattr(rr, "data", {}) or {} - dbg.append( - { - "id": data.get("id"), - "title": data.get("title") - or getattr(rr, "title", None), - "cover_image": data.get("cover_image"), - } - ) - except Exception: - continue - console.print(f"[cyan][DEBUG] Discogs candidates: {dbg}[/cyan]") - except Exception: - pass - if not results: - # conservative normalized fallback: try a combined normalized string - try: - await limiter.acquire() - combined = f"{normalize_name(artist)} {normalize_name(album)}" - if COVER_DEBUG_QUERIES: - console.print(f"[cyan][DEBUG] Discogs fallback query: {combined}") - results = await asyncio.to_thread( - discogs_client.search, combined, {"type": "release"} - ) - except Exception: - results = [] - for r in results: - # r.data may contain 'cover_image' or images - cover = None - try: - cover = r.data.get("cover_image") - except Exception: - cover = None - if not cover: - # try images list - imgs = r.data.get("images") or [] - if imgs and isinstance(imgs, list) and imgs[0].get("uri"): - cover = imgs[0].get("uri") - if cover: - # fetch image via aiohttp - try: - await limiter.acquire() - timeout = aiohttp.ClientTimeout(total=15) - async with session.get(cover, timeout=timeout) as resp: - if resp.status == 200: - return await resp.read() - except Exception: - continue - return None - except Exception as e: - console.print(f"[red]Discogs search exception: {e}[/red]") - return None - - -# Load env once -load_dotenv() - -# Console for pretty output -console = Console() - -# If set to '1'|'true', run only Spotify searches (useful for quick testing) -ONLY_SPOTIFY = os.getenv("ONLY_SPOTIFY", "").lower() in ("1", "true", "yes") -# If set, print query strings and brief response info for debugging -COVER_DEBUG_QUERIES = os.getenv("COVER_DEBUG_QUERIES", "").lower() in ( - "1", - "true", - "yes", -) -# If set, use more aggressive fuzzy thresholds and extra fallbacks -COVER_AGGRESSIVE = os.getenv("COVER_AGGRESSIVE", "").lower() in ("1", "true", "yes") - - -def _log_attempt(artist, album, title, source, result): - """Log a single, clean attempt line to console and to `search_attempts.log`. - - result should be a short string like 'Success', 'No match', 'Timeout', or an error message. - """ - try: - a = artist or "Unknown Artist" - al = album or "Unknown Album" - t = title or "Unknown Title" - line = f"SEARCH: {a} - {al} / {t} | Source: {source} | Result: {result}" - console.print(line) - try: - with open("search_attempts.log", "a", encoding="utf-8") as lf: - lf.write(line + "\n") - except Exception: - pass - except Exception: - # Never crash logging - pass - - -# Define a lightweight async rate limiter -class AsyncRateLimiter: - def __init__(self, rate_seconds: float): - self._rate = float(rate_seconds) - self._lock = asyncio.Lock() - self._last = 0.0 - - async def acquire(self) -> None: - async with self._lock: - now = asyncio.get_event_loop().time() - wait = self._rate - (now - self._last) - if wait > 0: - await asyncio.sleep(wait) - self._last = asyncio.get_event_loop().time() - - -# Initialize MusicBrainz client -musicbrainzngs.set_useragent("cover-art-script", "1.0", "your-email@example.com") - -# Initialize Discogs client -DISCOGS_TOKEN = os.getenv("DISCOGS_TOKEN") -discogs_client = Client("cover-art-script/1.0", user_token=DISCOGS_TOKEN) - - -# Define the log_api_response function at the top of the script -async def log_api_response(api_name, response): - """Log relevant parts of API responses for debugging purposes.""" - try: - data = await response.json() - if api_name == "MusicBrainz": - release_groups = data.get("release-groups", []) - relevant_info = [ - { - "id": rg.get("id"), - "title": rg.get("title"), - "artist": artist_credit_to_name(rg.get("artist-credit", [])), - } - for rg in release_groups - ] - console.print( - f"[cyan][DEBUG] {api_name} relevant response: {relevant_info}[/cyan]" - ) - elif api_name == "Discogs": - results = data.get("results", []) - relevant_info = [ - { - "id": result.get("id"), - "title": result.get("title"), - "cover_image": result.get("cover_image"), - } - for result in results - ] - console.print( - f"[cyan][DEBUG] {api_name} relevant response: {relevant_info}[/cyan]" - ) - elif api_name == "iTunes": - results = data.get("results", []) - relevant_info = [ - { - "collectionId": result.get("collectionId"), - "collectionName": result.get("collectionName"), - "artworkUrl100": result.get("artworkUrl100"), - } - for result in results - ] - console.print( - f"[cyan][DEBUG] {api_name} relevant response: {relevant_info}[/cyan]" - ) - else: - console.print(f"[cyan][DEBUG] {api_name} response: {data}[/cyan]") - except Exception as e: - console.print(f"[red][DEBUG] Failed to parse {api_name} response: {e}[/red]") - - -# Helper to strip common parenthetical tags from album names -def strip_album_tags(album): - """Remove common parenthetical tags from the end of album names.""" - pattern = r"\s*\((deluxe|remaster(ed)?|original mix|expanded|bonus|edition|version|mono|stereo|explicit|clean|anniversary|special|reissue|expanded edition|bonus track(s)?|international|digital|single|ep|live|instrumental|karaoke|radio edit|explicit version|clean version|acoustic|demo|re-recorded|remix|mix|edit|feat\.?|featuring|with .+|from .+|soundtrack|ost|score|session|vol(ume)? ?\d+|disc ?\d+|cd ?\d+|lp ?\d+|vinyl|202[0-9]|20[0-1][0-9]|19[0-9][0-9])\)$" - return re.sub(pattern, "", album, flags=re.IGNORECASE).strip() - - -# Helper to strip common trailing tags like EP, LP, Single, Album, etc. from album names -def strip_album_suffix(album): - # Remove trailing tags like ' EP', ' LP', ' Single', ' Album', ' Remix', ' Version', etc. - # Only if they appear at the end, case-insensitive, with or without punctuation - suffix_pattern = r"[\s\-_:]*(ep|lp|single|album|remix|version|edit|mix|deluxe|expanded|anniversary|reissue|instrumental|karaoke|ost|score|session|mono|stereo|explicit|clean|bonus|disc ?\d+|cd ?\d+|vinyl|202[0-9]|20[0-1][0-9]|19[0-9][0-9])$" - return re.sub(suffix_pattern, "", album, flags=re.IGNORECASE).strip() - - -# iTunes/Apple Music API fallback (async) -async def search_itunes_cover( - session: aiohttp.ClientSession, artist, album, limiter: "AsyncRateLimiter" -): - # Use only the `itunespy` library for iTunes album lookups. - if not HAVE_ITUNESPY: - console.print( - f"[yellow]iTunes: itunespy not available; skipping iTunes album search for '{artist} - {album}'[/yellow]" - ) - return None - - try: - mod = cast(Any, itunespy) - - def _search(): - try: - # try common itunespy APIs safely - if hasattr(mod, "search_album"): - return mod.search_album(f"{artist} {album}") - if hasattr(mod, "Album") and hasattr(mod.Album, "search"): - return mod.Album.search(f"{artist} {album}") - if hasattr(mod, "search"): - return mod.search(f"{artist} {album}", entity="album") - return None - except Exception: - return None - - albums = await asyncio.to_thread(_search) - if COVER_DEBUG_QUERIES and albums: - try: - dbg = [] - for a in (albums or [])[:3]: - try: - aid = getattr(a, "collectionId", None) or ( - a.get("collectionId") if isinstance(a, dict) else None - ) - except Exception: - aid = None - try: - aname = getattr(a, "collectionName", None) or ( - a.get("collectionName") if isinstance(a, dict) else None - ) - except Exception: - aname = None - dbg.append({"id": aid, "name": aname}) - console.print(f"[cyan][DEBUG] iTunes album candidates: {dbg}[/cyan]") - except Exception: - pass - if not albums: - if COVER_DEBUG_QUERIES: - console.print( - f"[cyan][DEBUG] iTunes album: no results for '{artist} - {album}', trying normalized fallback" - ) - norm_q = f"{normalize_name(artist)} {normalize_name(album)}" - - def _search_norm(): - try: - if hasattr(mod, "search_album"): - return mod.search_album(norm_q) - if hasattr(mod, "Album") and hasattr(mod.Album, "search"): - return mod.Album.search(norm_q) - if hasattr(mod, "search"): - return mod.search(norm_q, entity="album") - return None - except Exception: - return None - - albums = await asyncio.to_thread(_search_norm) - if not albums: - return None - - first = albums[0] - art_url = getattr(first, "artwork_url", None) or getattr( - first, "artworkUrl100", None - ) - if not art_url: - return None - - # Normalize to higher-res if possible - if "100x100" in art_url: - art_url = art_url.replace("100x100bb", "600x600bb") - - await limiter.acquire() - img_timeout = aiohttp.ClientTimeout(total=15) - try: - async with session.get(art_url, timeout=img_timeout) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - except Exception: - return None - except Exception as e: - console.print(f"[red][ERROR] itunespy album search exception: {e}[/red]") - return None - - -async def search_itunes_track( - session: aiohttp.ClientSession, artist, title, limiter: "AsyncRateLimiter" -): - # Use only the `itunespy` library for iTunes track lookups. - if not HAVE_ITUNESPY: - console.print( - f"[yellow]iTunes: itunespy not available; skipping iTunes track search for '{artist} - {title}'[/yellow]" - ) - return None - - try: - mod = cast(Any, itunespy) - - def _search(): - try: - if hasattr(mod, "search_track"): - return mod.search_track(f"{artist} {title}") - if hasattr(mod, "Track") and hasattr(mod.Track, "search"): - return mod.Track.search(f"{artist} {title}") - if hasattr(mod, "search"): - return mod.search(f"{artist} {title}", entity="song") - return None - except Exception: - return None - - tracks = await asyncio.to_thread(_search) - if not tracks: - if COVER_DEBUG_QUERIES: - console.print( - f"[cyan][DEBUG] iTunes track: no results for '{artist} - {title}', trying normalized fallback" - ) - norm_q = f"{normalize_name(artist)} {normalize_name(title)}" - - def _search_norm_track(): - try: - if hasattr(mod, "search_track"): - return mod.search_track(norm_q) - if hasattr(mod, "Track") and hasattr(mod.Track, "search"): - return mod.Track.search(norm_q) - if hasattr(mod, "search"): - return mod.search(norm_q, entity="song") - return None - except Exception: - return None - - tracks = await asyncio.to_thread(_search_norm_track) - if not tracks: - return None - - first = tracks[0] - art_url = getattr(first, "artwork_url", None) or getattr( - first, "artworkUrl100", None - ) - if not art_url: - return None - if "100x100" in art_url: - art_url = art_url.replace("100x100bb", "600x600bb") - - await limiter.acquire() - img_timeout = aiohttp.ClientTimeout(total=15) - try: - async with session.get(art_url, timeout=img_timeout) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - except Exception: - return None - except Exception as e: - console.print(f"[red][ERROR] itunespy track search exception: {e}[/red]") - return None - - -async def search_deezer_cover( - session: aiohttp.ClientSession, artist, album, limiter: "AsyncRateLimiter" -): - """Search Deezer for an album cover. Uses Deezer public API (no auth).""" - try: - # build simple query - from urllib.parse import quote - - query = f"{artist} {album}" - if COVER_DEBUG_QUERIES: - console.print(f"[cyan][DEBUG] Deezer query: {query}") - url = f"https://api.deezer.com/search/album?q={quote(query)}&limit=1" - await limiter.acquire() - timeout = aiohttp.ClientTimeout(total=10) - async with session.get(url, timeout=timeout) as resp: - if resp.status != 200: - return None - data = await resp.json() - items = data.get("data") or [] - if COVER_DEBUG_QUERIES: - try: - dbg = [] - for it in (items or [])[:3]: - dbg.append( - { - "id": it.get("id"), - "title": it.get("title"), - "cover_xl": it.get("cover_xl"), - "cover_big": it.get("cover_big"), - } - ) - console.print(f"[cyan][DEBUG] Deezer candidates: {dbg}[/cyan]") - except Exception: - pass - if not items: - # try a conservative normalized fallback - norm_q = f"{normalize_name(artist)} {normalize_name(album)}" - if COVER_DEBUG_QUERIES: - console.print(f"[cyan][DEBUG] Deezer fallback query: {norm_q}") - url2 = f"https://api.deezer.com/search/album?q={quote(norm_q)}&limit=1" - async with session.get(url2, timeout=timeout) as resp2: - if resp2.status != 200: - return None - data2 = await resp2.json() - items = data2.get("data") or [] - if not items: - return None - first = items[0] - # prefer XL or big covers - art_url = ( - first.get("cover_xl") or first.get("cover_big") or first.get("cover") - ) - if not art_url: - return None - await limiter.acquire() - img_timeout = aiohttp.ClientTimeout(total=15) - async with session.get(art_url, timeout=img_timeout) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - except Exception: - return None - return None - - -async def search_lastfm_cover( - session: aiohttp.ClientSession, artist, album, limiter: "AsyncRateLimiter" -): - """Search Last.fm for album cover using album.getInfo. Requires LASTFM_API_KEY in env.""" - LASTFM_API_KEY = os.getenv("LASTFM_API_KEY") - if not LASTFM_API_KEY: - console.print( - f"[yellow]LastFM: LASTFM_API_KEY not configured; skipping LastFM search for '{artist} - {album}'[/yellow]" - ) - return None - try: - params = { - "method": "album.getinfo", - "api_key": LASTFM_API_KEY, - "artist": artist, - "album": album, - "format": "json", - } - from urllib.parse import quote - - qs = "&".join(f"{quote(str(k))}={quote(str(v))}" for k, v in params.items()) - url = f"http://ws.audioscrobbler.com/2.0/?{qs}" - await limiter.acquire() - timeout = aiohttp.ClientTimeout(total=10) - async with session.get(url, timeout=timeout) as resp: - if resp.status != 200: - return None - data = await resp.json() - album_data = data.get("album") or {} - images = album_data.get("image") or [] - # images is a list of dicts with '#text' and 'size' - art_url = None - # prefer 'extralarge' or 'mega' - for size_name in ("mega", "extralarge", "large", "medium"): - for img in images: - if img.get("size") == size_name and img.get("#text"): - art_url = img.get("#text") - break - if art_url: - break - if not art_url: - return None - await limiter.acquire() - img_timeout = aiohttp.ClientTimeout(total=15) - async with session.get(art_url, timeout=img_timeout) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - except Exception: - return None - return None - - -_SPOTIFY_CLIENT = None - - -def get_spotify_client(): - """Lazily create and cache a spotipy.Spotify client using client-credentials. - - Returns None if spotipy is not installed or credentials are not configured. - """ - global _SPOTIFY_CLIENT - if _SPOTIFY_CLIENT is not None: - return _SPOTIFY_CLIENT - if not HAVE_SPOTIPY: - return None - client_id = os.getenv("SPOTIFY_CLIENT_ID") - client_secret = os.getenv("SPOTIFY_CLIENT_SECRET") - if not client_id or not client_secret: - return None - try: - import importlib - - sp_mod = importlib.import_module("spotipy") - creds_mod = importlib.import_module("spotipy.oauth2") - SpotifyClientCredentials = getattr(creds_mod, "SpotifyClientCredentials", None) - SpotifyCls = getattr(sp_mod, "Spotify", None) - if SpotifyClientCredentials is None or SpotifyCls is None: - return None - creds = SpotifyClientCredentials( - client_id=client_id, client_secret=client_secret - ) - _SPOTIFY_CLIENT = SpotifyCls(client_credentials_manager=creds) - return _SPOTIFY_CLIENT - except Exception: - return None - - -async def search_spotify_cover( - session: aiohttp.ClientSession, - artist, - album, - limiter: "AsyncRateLimiter", - isrc: Optional[str] = None, -): - """Search Spotify for album cover with multiple strategies: - - If `isrc` provided, try track search by ISRC first. - - Try quoted album+artist queries, then looser queries. - - Use fuzzy matching to validate results. - - Pick the largest available image. - """ - client = get_spotify_client() - if client is None: - console.print( - f"[yellow]Spotify: client not configured or spotipy not installed; skipping search for '{artist} - {album}'[/yellow]" - ) - return None - - def _sp_search(q, typ="album", limit=3): - try: - return client.search(q=q, type=typ, limit=limit) - except Exception: - return None - - try: - # 1) ISRC search (track -> album) - if isrc: - res = await asyncio.to_thread(_sp_search, f"isrc:{isrc}", "track", 1) - if res: - tracks = res.get("tracks", {}).get("items", []) - if tracks: - album_obj = tracks[0].get("album") or {} - images = album_obj.get("images") or [] - if images: - # pick largest - best = max(images, key=lambda x: x.get("width") or 0) - art_url = best.get("url") - if art_url: - await limiter.acquire() - async with session.get( - art_url, timeout=aiohttp.ClientTimeout(total=15) - ) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - - # Prepare normalized variants for querying - quoted_q = f'album:"{album}" artist:"{artist}"' - exact_q = f"artist:{artist} album:{album}" - norm_artist = normalize_name(artist) - norm_album = normalize_name(album) - simple_q = f"album:{norm_album} artist:{norm_artist}" - queries = [ - quoted_q, - exact_q, - simple_q, - f'album:"{album}"', - f'artist:"{artist}"', - ] - - for q in queries: - res = await asyncio.to_thread(_sp_search, q, "album", 3) - if not res: - continue - albums = res.get("albums", {}).get("items", []) - if COVER_DEBUG_QUERIES: - try: - dbg = [] - for a in (albums or [])[:3]: - dbg.append( - { - "id": a.get("id"), - "name": a.get("name"), - "artists": [ - ar.get("name") - for ar in (a.get("artists") or [])[:3] - if ar.get("name") - ], - "images": [ - img.get("url") - for img in (a.get("images") or [])[:3] - ], - } - ) - console.print( - f"[cyan][DEBUG] Spotify album candidates for query '{q}': {dbg}[/cyan]" - ) - except Exception: - pass - if not albums: - continue - # examine candidates and pick the best match via fuzzy matching - for a in albums: - found_album = a.get("name") or "" - found_artist = " ".join( - [ - ar.get("name") - for ar in (a.get("artists") or []) - if ar.get("name") - ] - ) - if is_fuzzy_match(artist, found_artist, threshold=75) and ( - not album or is_fuzzy_match(album, found_album, threshold=70) - ): - images = a.get("images") or [] - if not images: - continue - best = max(images, key=lambda x: x.get("width") or 0) - art_url = best.get("url") - if art_url: - await limiter.acquire() - try: - async with session.get( - art_url, timeout=aiohttp.ClientTimeout(total=15) - ) as img_resp: - if img_resp.status == 200: - return await img_resp.read() - except Exception: - continue - - return None - except Exception: - return None - - -# Fuzzy match helper for metadata -def is_fuzzy_match(expected, actual, threshold=80): - if not expected or not actual: - return False - return fuzz.token_set_ratio(expected.lower(), actual.lower()) >= threshold - - -# Fuzzy match for all fields -def is_metadata_match( - expected_artist, - expected_album, - expected_title, - found_artist, - found_album, - found_title, - threshold=80, -): - artist_match = is_fuzzy_match(expected_artist, found_artist, threshold) - album_match = ( - is_fuzzy_match(expected_album, found_album, threshold) - if expected_album - else True - ) - title_match = is_fuzzy_match(expected_title, found_title, threshold) - return artist_match and album_match and title_match - - -# Utility to normalize artist/song names for searching -def normalize_name(name): - # Lowercase, strip, remove extra spaces, and remove common punctuation - name = name.lower().strip() - name = re.sub(r"\([0-9]\)$", "", name) # remove (1), (2), etc. at end - name = re.sub(r"[\s_]+", " ", name) - name = re.sub(r"[\(\)\[\]\{\}\'\"\!\?\.,:;`~@#$%^&*+=|\\/<>]", "", name) - return name - - -def artist_credit_to_name(ac): - """Safely convert a MusicBrainz artist-credit array into a single artist name string.""" - parts = [] - for a in ac: - if isinstance(a, dict): - # Common formats: {'name': 'Artist Name'} or {'artist': {'name': 'Artist Name'}} - name = None - if a.get("name"): - name = a.get("name") - elif ( - a.get("artist") - and isinstance(a.get("artist"), dict) - and a.get("artist", {}).get("name") - ): - name = a.get("artist", {}).get("name") - if name: - parts.append(name) - return " ".join(parts) - - -# Suppress noisy loggers (aiohttp, urllib3, etc.) -for noisy_logger in [ - "aiohttp.client", - "aiohttp.server", - "aiohttp.access", - "urllib3", - "asyncio", - "chardet", - "requests.packages.urllib3", -]: - logging.getLogger(noisy_logger).setLevel(logging.CRITICAL) - logging.getLogger(noisy_logger).propagate = False - -# Also suppress root logger to CRITICAL for anything not our own -logging.getLogger().setLevel(logging.CRITICAL) - - -async def fetch_srutil_cover( - sr, artist, song, session: aiohttp.ClientSession, limiter: AsyncRateLimiter -): - try: - album = await sr.get_album_by_name(artist, song) - if not album or not album.get("id"): - return None - cover_url = await sr.get_cover_by_album_id(album["id"], 640) - if cover_url: - await limiter.acquire() - try: - timeout = aiohttp.ClientTimeout(total=15) - async with session.get(cover_url, timeout=timeout) as resp: - if resp.status == 200: - return await resp.read() - else: - console.print( - f"[red]SRUtil: Failed to fetch cover art from URL (status {resp.status}): {cover_url}[/red]" - ) - except Exception as e: - console.print(f"[red]SRUtil: Exception fetching cover url: {e}[/red]") - except Exception as e: - msg = str(e) - if "Cannot combine AUTHORIZATION header with AUTH argument" in msg: - console.print( - "[red]SRUtil: Skipping due to conflicting authentication method in dependency (AUTHORIZATION header + AUTH argument).[/red]" - ) - else: - console.print(f"[red]SRUtil: Exception: {e}[/red]") - return None - - -async def get_isrc(file): - try: - - def _read_isrc(): - f = load_file(file) - # music_tag may store ISRC under 'isrc' or 'ISRC' - try: - val = f["isrc"].value - except Exception: - try: - val = f["ISRC"].value - except Exception: - val = None - if isinstance(val, list): - return val[0] if val else None - return val - - return await asyncio.to_thread(_read_isrc) - except Exception as e: - console.print(f"[red]Error reading ISRC for {file}: {e}[/red]") - return None - - -async def search_musicbrainz_by_isrc(session, isrc, limiter: AsyncRateLimiter): - if not isrc: - return None - headers = {"User-Agent": "cover-art-script/1.0"} - # Use the ISRC lookup endpoint which returns recordings - url = f"https://musicbrainz.org/ws/2/isrc/{isrc}?fmt=json" - try: - await limiter.acquire() - timeout = aiohttp.ClientTimeout(total=15) - async with session.get(url, headers=headers, timeout=timeout) as resp: - if resp.status != 200: - return None - try: - data = await resp.json() - except Exception: - return None - recordings = data.get("recordings") or [] - for rec in recordings: - # try releases tied to this recording - releases = rec.get("releases") or [] - if releases: - relid = releases[0].get("id") - if relid: - caa_url = ( - f"https://coverartarchive.org/release/{relid}/front-500" - ) - async with session.get(caa_url, timeout=timeout) as caa_resp: - if caa_resp.status == 200: - console.print( - f"[green]Found cover art via ISRC {isrc}[/green]" - ) - return await caa_resp.read() - return None - except Exception as e: - console.print(f"[red]MusicBrainz ISRC lookup exception for {isrc}: {e}[/red]") - return None - - -# Concurrency limit for async processing -CONCURRENCY = 18 - - -# Helper for formatting failure reasons in a consistent way -def format_failure_reason(e, resp_status=None): - """Format a failure reason from an exception or response status""" - if isinstance(e, asyncio.TimeoutError): - return "timeout" - elif isinstance(e, aiohttp.ClientError): - return f"network error: {str(e)}" - elif resp_status: - return f"HTTP {resp_status}" - elif e: - return str(e) - return "no match found" - - -async def process_file( - file, sr, table, results, sem, session: aiohttp.ClientSession, limiters: dict -): - """Process a single audio file to find and embed cover art.""" - async with sem: - if await has_cover(file): - table.add_row(file, "Already Present", "-") - results.append([file, "Already Present", "-"]) - return - - artist, album, title = await get_artist_album_title(file) - album_key = (artist, album) - image_bytes = ALBUM_ART_CACHE.get(album_key) - source = None - status = "Failed" - - # Try ISRC-based lookup first - isrc = await get_isrc(file) - if isrc: - img = await search_musicbrainz_by_isrc( - session, isrc, limiters["musicbrainz"] - ) - if img: - image_bytes = img - source = f"MusicBrainz (ISRC:{isrc})" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, source, "Success") - else: - _log_attempt( - artist, album, title, f"MusicBrainz (ISRC:{isrc})", "No match" - ) - - # If ONLY_SPOTIFY testing mode is enabled, attempt only Spotify and return - if ONLY_SPOTIFY: - img = await search_spotify_cover( - session, artist, album, limiters["spotify"], isrc - ) - if img: - image_bytes = img - source = "Spotify" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "Spotify", "Success") - file_basename = os.path.basename(file) - ok = await embed_cover(file, image_bytes) - if ok: - console.print( - f"[green][FINAL RESULT] {file_basename} — {artist} / {album} | Success via {source}[/green]" - ) - else: - status = "Embed Failed" - console.print( - f"[red][FINAL RESULT] {file_basename} — {artist} / {album} | Embed Failed from {source}[/red]" - ) - else: - _log_attempt(artist, album, title, "Spotify", "No match") - console.print( - f"[yellow][FINAL RESULT] {os.path.basename(file)} — {artist} / {album} | No Spotify cover art found[/yellow]" - ) - - table.add_row(file, status, source if source else "-") - results.append([file, status, source if source else "-"]) - return - - # SRUtil - if not image_bytes: - img = await fetch_srutil_cover( - sr, artist, album, session, limiters["srutil"] - ) - if img: - image_bytes = img - source = "SRUtil" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "SRUtil", "Success") - else: - _log_attempt(artist, album, title, "SRUtil", "No match") - - # MusicBrainz - if not image_bytes: - img = await search_musicbrainz_cover( - artist, album, session, limiters["musicbrainz"] - ) - if img: - image_bytes = img - source = "MusicBrainz" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "MusicBrainz", "Success") - else: - _log_attempt(artist, album, title, "MusicBrainz", "No match") - - # Discogs - if not image_bytes: - img = await search_discogs_cover( - artist, album, session, limiters["discogs"] - ) - if img: - image_bytes = img - source = "Discogs" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "Discogs", "Success") - else: - _log_attempt(artist, album, title, "Discogs", "No match") - - # Deezer - if not image_bytes: - img = await search_deezer_cover(session, artist, album, limiters["deezer"]) - if img: - image_bytes = img - source = "Deezer" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "Deezer", "Success") - else: - _log_attempt(artist, album, title, "Deezer", "No match") - - # Spotify - if not image_bytes: - img = await search_spotify_cover( - session, artist, album, limiters["spotify"], isrc - ) - if img: - image_bytes = img - source = "Spotify" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "Spotify", "Success") - else: - _log_attempt(artist, album, title, "Spotify", "No match") - - # iTunes album - if not image_bytes: - img = await search_itunes_cover(session, artist, album, limiters["itunes"]) - if img: - image_bytes = img - source = "iTunes(album)" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "iTunes(album)", "Success") - else: - _log_attempt(artist, album, title, "iTunes(album)", "No match") - - # iTunes track - if not image_bytes: - img = await search_itunes_track(session, artist, title, limiters["itunes"]) - if img: - image_bytes = img - source = "iTunes(track)" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "iTunes(track)", "Success") - else: - _log_attempt(artist, album, title, "iTunes(track)", "No match") - - # Last.fm - if not image_bytes: - img = await search_lastfm_cover(session, artist, album, limiters["lastfm"]) - if img: - image_bytes = img - source = "LastFM" - status = "Success" - ALBUM_ART_CACHE[album_key] = image_bytes - _log_attempt(artist, album, title, "LastFM", "Success") - else: - _log_attempt(artist, album, title, "LastFM", "No match") - - # Embed and summary - file_basename = os.path.basename(file) - if image_bytes and source: - ok = await embed_cover(file, image_bytes) - if ok: - console.print( - f"[green][FINAL RESULT] {file_basename} — {artist} / {album} | Success via {source}[/green]" - ) - else: - status = "Embed Failed" - console.print( - f"[red][FINAL RESULT] {file_basename} — {artist} / {album} | Embed Failed from {source}[/red]" - ) - else: - console.print( - f"[yellow][FINAL RESULT] {file_basename} — {artist} / {album} | No cover art found[/yellow]" - ) - - table.add_row(file, status, source if source else "-") - results.append([file, status, source if source else "-"]) - - -async def has_cover(file): - # Check if the audio file already has embedded cover art - try: - f = load_file(file) - # music_tag stores artwork in 'artwork' which may be a list-like field - art = f["artwork"] - # If there is any artwork, consider it present - try: - return bool(art.first) - except Exception: - # fallback if .first not available - return bool(art) - except Exception: - return False - - -async def get_artist_album_title(file): - # Extract artist, album, and title from audio file tags - try: - f = load_file(file) - artist = str(f["artist"].first) if f["artist"].first else "Unknown Artist" - album = str(f["album"].first) if f["album"].first else "Unknown Album" - title = str(f["title"].first) if f["title"].first else "Unknown Title" - return artist, album, title - except Exception: - return "Unknown Artist", "Unknown Album", "Unknown Title" - - -async def embed_cover(file, image_bytes): - # Embed cover art into audio file metadata using music_tag - try: - f = load_file(file) - f["artwork"] = image_bytes - f.save() - return True - except Exception as e: - console.print(f"[red][ERROR] Failed to embed cover: {e}[/red]") - return False - - -async def main(): - try: - console.print(f"[bold blue]Scanning directory: {MUSIC_DIR}[/bold blue]") - sr = SRUtil() - results = [] - files = [] - for root, _, filenames in os.walk(MUSIC_DIR): - for fn in filenames: - if os.path.splitext(fn)[1].lower() in AUDIO_EXTS: - file_path = os.path.join(root, fn) - files.append(file_path) - - table = Table(title="Cover Art Embedding Report") - table.add_column("File", style="cyan", overflow="fold") - table.add_column("Status", style="green") - table.add_column("Source", style="magenta") - - # create rate limiters (seconds between requests) - RATE_SRUTIL = 0.1 - RATE_MUSICBRAINZ = 1.0 - RATE_ITUNES = 0.5 - RATE_DISCOGS = 1.0 - RATE_DEEZER = 0.5 - RATE_LASTFM = 1.0 - RATE_SPOTIFY = 0.5 - - limiters = { - "srutil": AsyncRateLimiter(RATE_SRUTIL), - "musicbrainz": AsyncRateLimiter(RATE_MUSICBRAINZ), - "itunes": AsyncRateLimiter(RATE_ITUNES), - "discogs": AsyncRateLimiter(RATE_DISCOGS), - "deezer": AsyncRateLimiter(RATE_DEEZER), - "lastfm": AsyncRateLimiter(RATE_LASTFM), - "spotify": AsyncRateLimiter(RATE_SPOTIFY), - } - - sem = asyncio.Semaphore(CONCURRENCY) - - def format_failure_reason(e, resp_status=None): - """Format a failure reason from an exception or response status""" - if isinstance(e, asyncio.TimeoutError): - return "timeout" - elif isinstance(e, aiohttp.ClientError): - return f"network error: {str(e)}" - elif resp_status: - return f"HTTP {resp_status}" - elif e: - return str(e) - return "no match found" - - async def worker( - file, sr, table, results, sem, progress, task_id, session, limiters - ): - await process_file(file, sr, table, results, sem, session, limiters) - progress.update(task_id, advance=1) - - async with aiohttp.ClientSession() as session: - with Progress( - TextColumn("[progress.description]{task.description}"), - BarColumn(), - TaskProgressColumn(), - TimeElapsedColumn(), - ) as progress: - task_id = progress.add_task("Processing files...", total=len(files)) - # Schedule all workers - await asyncio.gather( - *( - worker( - file, - sr, - table, - results, - sem, - progress, - task_id, - session, - limiters, - ) - for file in files - ) - ) - - # Print summary table and CSV after progress bar - console.print(table) - with open(REPORT_CSV, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["File", "Status", "Source"]) - writer.writerows(results) - console.print(f"[bold green]CSV report written to {REPORT_CSV}[/bold green]") - except Exception as e: - console.print(f"[red][ERROR] Unhandled exception: {e}[/red]") - traceback.print_exc() - sys.exit(1) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/test/liquidsoap.liq b/test/liquidsoap.liq deleted file mode 100644 index eb96797..0000000 --- a/test/liquidsoap.liq +++ /dev/null @@ -1,252 +0,0 @@ -#!/usr/bin/env liquidsoap - -set("log.file.path","/home/kyle/.lsl.txt") -set("log.stdout",true) -set("harbor.bind_addrs", ["127.0.0.1"]) - -# Buffer and timing settings -set("frame.duration",0.02) -set("root.max_latency",2.) -set("audio.converter.samplerate.libsamplerate.quality","best") -set("clock.allow_streaming_errors",false) - - -# Get next track dynamically [Each station] - -def get_next_main() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py main")) - [request.create(uri)] -end - -def get_next_rock() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py rock")) - [request.create(uri)] -end - -def get_next_electronic() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py electronic")) - [request.create(uri)] -end - -def get_next_rap() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py rap")) - [request.create(uri)] -end - -#def get_next_classical() = -# uri = list.hd(default="", process.read.lines("uv run get_next_track.py classical")) -# [request.create(uri)] -#end - -def get_next_pop() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py pop")) - [request.create(uri)] -end - - -# Set up queues [Each station] - -main_list = request.dynamic( - id="requests", - get_next_main, - retry_delay=1.0, - timeout=20.0 -) - - -rock_list = request.dynamic( - id="rock_requests", - get_next_rock, - retry_delay=1.0, - timeout=20.0 -) - -electronic_list = request.dynamic( - id="electronic_requests", - get_next_electronic, - retry_delay=1.0, - timeout=20.0 -) - -rap_list = request.dynamic( - id="rap_requests", - get_next_rap, - retry_delay=1.0, - timeout=20.0 -) - -#classical_list = request.dynamic.list( -# id="classical_requests", -# get_next_classical, -# prefetch=0 -#) - -pop_list = request.dynamic( - id="pop_requests", - get_next_pop, - retry_delay=1.0, - timeout=20.0 -) - - -# Standard - -silence = single("/home/kyle/ls/silence.ogg") - -# Queue [Each station] - -def main_queue(remaining, _) = - log("MAIN: Queueing with #{remaining} seconds remaining") - if not main_list.fetch() then - log("Fetching next query failed") - end -end - -def rock_queue(remaining, _) = - log("ROCK: Queueing with #{remaining} seconds remaining") - if not rock_list.fetch() then - log("Fetching next query failed") - end -end - -def electronic_queue(remaining, _) = - log("ELECTRONIC: Queueing with #{remaining} seconds remaining") - if not electronic_list.fetch() then - log("Fetching next query failed") - end -end - -def rap_queue(remaining, _) = - log("RAP: Queueing with #{remaining} seconds remaining") - if not rap_list.fetch() then - log("Fetching next query failed") - end -end - -#def classical_queue(remaining, _) = -# log("CLASSICAL: Queueing with #{remaining} seconds remaining") -# if not classical_list.fetch() then -# log("Fetching next query failed") -# end -#end - -def pop_queue(remaining, _) = - log("POP: Queueing with #{remaining} seconds remaining") - if not pop_list.fetch() then - log("Fetching next query failed") - end -end - - - -# Initial fetch [Each station] - -main_list.fetch() -rock_list.fetch() -electronic_list.fetch() -rap_list.fetch() -#classical_list.fetch() -pop_list.fetch() - -# Source setup [Each station] - -def create_source(s,q) = - source.dynamic(s, track_sensitive=true, {q()}) -end - -main_source = create_source(main_list, main_queue) -rock_source = create_source(rock_list, rock_queue) -electronic_source = create_source(electronic_list, electronic_queue) -rap_source = create_source(rap_list, rap_queue) -#classical_source = create_source(classical_list, classical_queue) -pop_source = create_source(pop_list, pop_queue) - -all_tracks_main = fallback(track_sensitive=false, [main_source, silence]) -all_tracks_rock = fallback(track_sensitive=false, [rock_source, silence]) -all_tracks_electronic = fallback(track_sensitive=false, [electronic_source, silence]) -all_tracks_rap = fallback(track_sensitive=false, [rap_source, silence]) -#all_tracks_classical = fallback(track_sensitive=false, [classical_source, silence]) -all_tracks_pop = fallback(track_sensitive=false, [pop_source, silence]) - -# HLS Setup [Standard] - -aac_lofi = %ffmpeg(format="mpegts", - %audio(codec="aac", - channels=2, - ar=48000, - b="128k")) - -aac_midfi = %ffmpeg(format="mpegts", - %audio(codec="aac", - channels=2, - ar=48000, - b="256k")) - -aac_hifi = %ffmpeg(format="mpegts", - %audio(codec="aac", - channels=2, - ar=48000, - b="512k")) - - -streams = - [("aac_lofi", aac_lofi), ("aac_midfi", aac_midfi), ("aac_hifi", aac_hifi)] - - -# HLS Outputs [Each station] - -def create_hls_output(~name, source) = - output.file.hls( - playlist="#{name}.m3u8", - segment_duration=0.5, - segments=10, - segments_overhead=5, - persist_at="/nvme/pub/hls/#{name}/state.config", - "/nvme/pub/hls/#{name}", - streams, - source - ) -end - -create_hls_output(name="main", mksafe(main_source)) - -create_hls_output(name="rock", mksafe(rock_source)) -create_hls_output(name="electronic", mksafe(electronic_source)) -create_hls_output(name="rap", mksafe(rap_source)) - -#output.file.hls( -# playlist="classical.m3u8", -# segment_duration=0.45, -# segments=9, -# segments_overhead=3, -# persist_at="/nvme/pub/hls/classical_state.config", -# "/nvme/pub/hls/classical", -# streams, -# mksafe(classical_source) -#) - -create_hls_output(name="pop", mksafe(pop_source)) - -# HTTP Server - -def get_next_http(~protocol,~data,~headers,uri) = - source = - if data == "main" then main_source - elsif data == "rock" then rock_source - elsif data == "electronic" then electronic_source - elsif data == "rap" then rap_source - elsif data == "pop" then pop_source - else null() end - - if source != null() then - source.skip(source) - http.response( - protocol=protocol, - code=200, - data="OK #{data}" - ) -end - -harbor.http.register(port=29000, method="POST", "/next", get_next_http) - -# EOF diff --git a/test/liquidsoap.ls b/test/liquidsoap.ls deleted file mode 100644 index b6ffbb6..0000000 --- a/test/liquidsoap.ls +++ /dev/null @@ -1,270 +0,0 @@ -#!/usr/bin/liquidsoap -set("log.file.path", "/home/kyle/.lsl.txt") -set("log.stdout", true) -set("harbor.bind_addrs", ["127.0.0.1"]) - - -# Get next track dynamically [Each station] - -def get_next_main() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py main")) - [request.create(uri)] -end - -def get_next_rock() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py rock")) - [request.create(uri)] -end - -def get_next_electronic() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py electronic")) - [request.create(uri)] -end - -def get_next_rap() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py rap")) - [request.create(uri)] -end - -#def get_next_classical() = -# uri = list.hd(default="", process.read.lines("uv run get_next_track.py classical")) -# [request.create(uri)] -#end - -def get_next_pop() = - uri = list.hd(default="", process.read.lines("uv run get_next_track.py pop")) - [request.create(uri)] -end - - -# Set up queues [Each station] - -main_list = request.dynamic.list( - id="requests", - get_next_main, - prefetch=0 -) - - -rock_list = request.dynamic.list( - id="rock_requests", - get_next_rock, - prefetch=0 -) - -electronic_list = request.dynamic.list( - id="electronic_requests", - get_next_electronic, - prefetch=0 -) - -rap_list = request.dynamic.list( - id="rap_requests", - get_next_rap, - prefetch=0 -) - -#classical_list = request.dynamic.list( -# id="classical_requests", -# get_next_classical, -# prefetch=0 -#) - -pop_list = request.dynamic.list( - id="pop_requests", - get_next_pop, - prefetch=0 -) - - -# Standard - -silence = single("/home/kyle/ls/silence.ogg") - -# Queue [Each station] - -def main_queue(remaining, _) = - log("MAIN: Queueing with #{remaining} seconds remaining") - if not main_list.fetch() then - log("Fetching next query failed") - end -end - -def rock_queue(remaining, _) = - log("ROCK: Queueing with #{remaining} seconds remaining") - if not rock_list.fetch() then - log("Fetching next query failed") - end -end - -def electronic_queue(remaining, _) = - log("ELECTRONIC: Queueing with #{remaining} seconds remaining") - if not electronic_list.fetch() then - log("Fetching next query failed") - end -end - -def rap_queue(remaining, _) = - log("RAP: Queueing with #{remaining} seconds remaining") - if not rap_list.fetch() then - log("Fetching next query failed") - end -end - -#def classical_queue(remaining, _) = -# log("CLASSICAL: Queueing with #{remaining} seconds remaining") -# if not classical_list.fetch() then -# log("Fetching next query failed") -# end -#end - -def pop_queue(remaining, _) = - log("POP: Queueing with #{remaining} seconds remaining") - if not pop_list.fetch() then - log("Fetching next query failed") - end -end - - - -# Initial fetch [Each station] - -main_list.fetch() -rock_list.fetch() -electronic_list.fetch() -rap_list.fetch() -#classical_list.fetch() -pop_list.fetch() - -# Source setup [Each station] - -main_source = source.on_end(delay=1.0, main_list, main_queue) -rock_source = source.on_end(delay=1.0, rock_list, rock_queue) -electronic_source = source.on_end(delay=1.0, electronic_list, electronic_queue) -rap_source = source.on_end(delay=1.0, rap_list, rap_queue) -#classical_source = source.on_end(delay=1.0, classical_list, classical_queue) -pop_source = source.on_end(delay=1.0, pop_list, pop_queue) - -all_tracks_main = fallback(track_sensitive=false, [main_source, silence]) -all_tracks_rock = fallback(track_sensitive=false, [rock_source, silence]) -all_tracks_electronic = fallback(track_sensitive=false, [electronic_source, silence]) -all_tracks_rap = fallback(track_sensitive=false, [rap_source, silence]) -#all_tracks_classical = fallback(track_sensitive=false, [classical_source, silence]) -all_tracks_pop = fallback(track_sensitive=false, [pop_source, silence]) - -# HLS Setup [Standard] - -aac_lofi = - %ffmpeg(format = "mpegts", %audio(codec = "aac", channels = 2, ar = 44100)) - -aac_midfi = - %ffmpeg( - format = "mpegts", - %audio(codec = "aac", channels = 2, ar = 44100, b = "96k") - ) - -aac_hifi = - %ffmpeg( - format = "mpegts", - %audio(codec = "aac", channels = 2, ar = 44100, b = "448k") - ) - - -streams = - [("aac_lofi", aac_lofi), ("aac_midfi", aac_midfi), ("aac_hifi", aac_hifi)] - - -# HLS Outputs [Each station] - -output.file.hls( - playlist="main.m3u8", - segment_duration=0.5, - segments=9, - segments_overhead=4, - persist_at="/nvme/pub/hls/state.config", - "/nvme/pub/hls/main", - streams, - mksafe(main_source) -) - -output.file.hls( - playlist="rock.m3u8", - segment_duration=0.5, - segments=9, - segments_overhead=4, - persist_at="/nvme/pub/hls/rock/state.config", - "/nvme/pub/hls/rock", - streams, - mksafe(rock_source) -) - -output.file.hls( - playlist="electronic.m3u8", - segment_duration=0.5, - segments=9, - segments_overhead=4, - persist_at="/nvme/pub/hls/electronic/state.config", - "/nvme/pub/hls/electronic", - streams, - mksafe(electronic_source) -) - -output.file.hls( - playlist="rap.m3u8", - segment_duration=0.5, - segments=9, - segments_overhead=4, - persist_at="/nvme/pub/hls/rap_state.config", - "/nvme/pub/hls/rap", - streams, - mksafe(rap_source) -) - -#output.file.hls( -# playlist="classical.m3u8", -# segment_duration=0.45, -# segments=9, -# segments_overhead=3, -# persist_at="/nvme/pub/hls/classical_state.config", -# "/nvme/pub/hls/classical", -# streams, -# mksafe(classical_source) -#) - -output.file.hls( - playlist="pop.m3u8", - segment_duration=0.5, - segments=9, - segments_overhead=4, - persist_at="/nvme/pub/hls/pop_state.config", - "/nvme/pub/hls/pop", - streams, - mksafe(pop_source) -) - -# HTTP Server - -def get_next_http(~protocol,~data,~headers,uri) = - if data == "main" then - _req = source.skip(main_source) - elsif data == "rock" then - _req = source.skip(rock_source) - elsif data == "electronic" then - _req = source.skip(electronic_source) - elsif data == "rap" then - _req = source.skip(rap_source) -#elsif data == "classical" then -# _req = source.skip(classical_source) - elsif data == "pop" then - _req = source.skip(pop_source) - end - http.response( - protocol=protocol, - code=200, - data="OK #{data}" - ) -end - -harbor.http.register(port=29000, method="POST", "/next", get_next_http) - -# EOF diff --git a/test/minimal_test.liq b/test/minimal_test.liq deleted file mode 100644 index e69de29..0000000