123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
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.
|
|
"""
|