106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Dict, List, Optional, Type
|
|
import requests
|
|
import sys
|
|
|
|
from helper.logger import log, debug
|
|
|
|
|
|
class MetadataProvider(ABC):
|
|
"""Base class for metadata providers (music, movies, books, etc.)."""
|
|
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
|
|
self.config = config or {}
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.__class__.__name__.replace("Provider", "").lower()
|
|
|
|
@abstractmethod
|
|
def search(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""Return a list of candidate metadata records."""
|
|
|
|
def to_tags(self, item: Dict[str, Any]) -> List[str]:
|
|
"""Convert a result item into a list of tags."""
|
|
tags: List[str] = []
|
|
title = item.get("title")
|
|
artist = item.get("artist")
|
|
album = item.get("album")
|
|
year = item.get("year")
|
|
|
|
if title:
|
|
tags.append(f"title:{title}")
|
|
if artist:
|
|
tags.append(f"artist:{artist}")
|
|
if album:
|
|
tags.append(f"album:{album}")
|
|
if year:
|
|
tags.append(f"year:{year}")
|
|
|
|
tags.append(f"source:{self.name}")
|
|
return tags
|
|
|
|
|
|
class ITunesProvider(MetadataProvider):
|
|
"""Metadata provider using the iTunes Search API."""
|
|
|
|
def search(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
params = {"term": query, "media": "music", "entity": "song", "limit": limit}
|
|
try:
|
|
resp = requests.get("https://itunes.apple.com/search", params=params, timeout=10)
|
|
resp.raise_for_status()
|
|
results = resp.json().get("results", [])
|
|
except Exception as exc:
|
|
log(f"iTunes search failed: {exc}", file=sys.stderr)
|
|
return []
|
|
|
|
items: List[Dict[str, Any]] = []
|
|
for r in results:
|
|
item = {
|
|
"title": r.get("trackName"),
|
|
"artist": r.get("artistName"),
|
|
"album": r.get("collectionName"),
|
|
"year": str(r.get("releaseDate", ""))[:4],
|
|
"provider": self.name,
|
|
"raw": r,
|
|
}
|
|
items.append(item)
|
|
debug(f"iTunes returned {len(items)} items for '{query}'")
|
|
return items
|
|
|
|
|
|
# Registry ---------------------------------------------------------------
|
|
|
|
_METADATA_PROVIDERS: Dict[str, Type[MetadataProvider]] = {
|
|
"itunes": ITunesProvider,
|
|
}
|
|
|
|
|
|
def register_provider(name: str, provider_cls: Type[MetadataProvider]) -> None:
|
|
_METADATA_PROVIDERS[name.lower()] = provider_cls
|
|
|
|
|
|
def list_metadata_providers(config: Optional[Dict[str, Any]] = None) -> Dict[str, bool]:
|
|
availability: Dict[str, bool] = {}
|
|
for name, cls in _METADATA_PROVIDERS.items():
|
|
try:
|
|
provider = cls(config)
|
|
# Basic availability check: perform lightweight validation if defined
|
|
availability[name] = True
|
|
except Exception:
|
|
availability[name] = False
|
|
return availability
|
|
|
|
|
|
def get_metadata_provider(name: str, config: Optional[Dict[str, Any]] = None) -> Optional[MetadataProvider]:
|
|
cls = _METADATA_PROVIDERS.get(name.lower())
|
|
if not cls:
|
|
return None
|
|
try:
|
|
return cls(config)
|
|
except Exception as exc:
|
|
log(f"Provider init failed for '{name}': {exc}", file=sys.stderr)
|
|
return None
|