Medios-Macina/ProviderCore/base.py

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple


@dataclass
class SearchResult:
    """Unified search result format across all search providers."""

    table: str  # Provider name: "libgen", "soulseek", "bandcamp", "youtube", etc.
    title: str  # Display title/filename
    path: str  # Download target (URL, path, magnet, identifier)

    detail: str = ""  # Additional description
    annotations: List[str] = field(default_factory=list)  # Tags: ["120MB", "flac", "ready"]
    media_kind: str = "other"  # Type: "book", "audio", "video", "game", "magnet"
    size_bytes: Optional[int] = None
    tag: set[str] = field(default_factory=set)  # Searchable tag values
    columns: List[Tuple[str, str]] = field(default_factory=list)  # Display columns
    full_metadata: Dict[str, Any] = field(default_factory=dict)  # Extra metadata

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for pipeline processing."""

        return {
            "table": self.table,
            "title": self.title,
            "path": self.path,
            "detail": self.detail,
            "annotations": self.annotations,
            "media_kind": self.media_kind,
            "size_bytes": self.size_bytes,
            "tag": list(self.tag),
            "columns": list(self.columns),
            "full_metadata": self.full_metadata,
        }


class Provider(ABC):
    """Unified provider base class.

    This replaces the older split between "search providers" and "file providers".
    Concrete providers may implement any subset of:
    - search(query, ...)
    - download(result, output_dir)
    - upload(file_path, ...)
    - login(...)
    - validate()
    """

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.name = self.__class__.__name__.lower()

    # Standard lifecycle/auth hook.
    def login(self, **_kwargs: Any) -> bool:
        return True

    def search(
        self,
        query: str,
        limit: int = 50,
        filters: Optional[Dict[str, Any]] = None,
        **kwargs: Any,
    ) -> List[SearchResult]:
        """Search for items matching the query."""
        raise NotImplementedError(f"Provider '{self.name}' does not support search")

    def download(self, result: SearchResult, output_dir: Path) -> Optional[Path]:
        """Download an item from a search result."""

        return None

    def upload(self, file_path: str, **kwargs: Any) -> str:
        """Upload a file and return a URL or identifier."""
        raise NotImplementedError(f"Provider '{self.name}' does not support upload")

    def validate(self) -> bool:
        """Check if provider is available and properly configured."""

        return True

    def selector(self, selected_items: List[Any], *, ctx: Any, stage_is_last: bool = True, **_kwargs: Any) -> bool:
        """Optional hook for handling `@N` selection semantics.

        The CLI can delegate selection behavior to a provider/store instead of
        applying the default selection filtering.

        Return True if the selection was handled and default behavior should be skipped.
        """

        _ = selected_items
        _ = ctx
        _ = stage_is_last
        return False


class SearchProvider(Provider):
    """Compatibility alias for older code.

    Prefer inheriting from Provider directly.
    """


class FileProvider(Provider):
    """Compatibility alias for older code.

    Prefer inheriting from Provider directly.
    """