From ec9a2d4d3692acf086bcab9441dc675921d6c65b Mon Sep 17 00:00:00 2001 From: Francwa Date: Sun, 30 Nov 2025 02:17:19 +0100 Subject: [PATCH] Knaben API --- agent/api/__init__.py | 21 +++- agent/api/knaben.py | 230 ++++++++++++++++++++++++++++++++++++++++++ agent/parameters.py | 2 +- agent/registry.py | 23 ++++- agent/tools/api.py | 96 ++++++++++++++++-- 5 files changed, 357 insertions(+), 15 deletions(-) create mode 100644 agent/api/knaben.py diff --git a/agent/api/__init__.py b/agent/api/__init__.py index a6f7e16..585f7f3 100644 --- a/agent/api/__init__.py +++ b/agent/api/__init__.py @@ -9,12 +9,31 @@ from .themoviedb import ( MediaResult ) +from .knaben import ( + KnabenClient, + knaben_client, + KnabenError, + KnabenConfigurationError, + KnabenAPIError, + KnabenNotFoundError, + TorrentResult +) + __all__ = [ + # TMDB 'TMDBClient', 'tmdb_client', 'TMDBError', 'TMDBConfigurationError', 'TMDBAPIError', 'TMDBNotFoundError', - 'MediaResult' + 'MediaResult', + # Knaben + 'KnabenClient', + 'knaben_client', + 'KnabenError', + 'KnabenConfigurationError', + 'KnabenAPIError', + 'KnabenNotFoundError', + 'TorrentResult' ] diff --git a/agent/api/knaben.py b/agent/api/knaben.py new file mode 100644 index 0000000..eaa05b4 --- /dev/null +++ b/agent/api/knaben.py @@ -0,0 +1,230 @@ +"""Knaben torrent search API client.""" +from typing import Dict, Any, Optional, List +from dataclasses import dataclass +import logging +import requests +from requests.exceptions import RequestException, Timeout, HTTPError + +from ..config import Settings, settings + +logger = logging.getLogger(__name__) + + +class KnabenError(Exception): + """Base exception for Knaben-related errors.""" + pass + + +class KnabenConfigurationError(KnabenError): + """Raised when Knaben API is not properly configured.""" + pass + + +class KnabenAPIError(KnabenError): + """Raised when Knaben API returns an error.""" + pass + + +class KnabenNotFoundError(KnabenError): + """Raised when no torrents are found.""" + pass + + +@dataclass +class TorrentResult: + """Represents a torrent search result from Knaben.""" + title: str + size: str + seeders: int + leechers: int + magnet: str + info_hash: Optional[str] = None + tracker: Optional[str] = None + upload_date: Optional[str] = None + category: Optional[str] = None + + +class KnabenClient: + """ + Client for interacting with Knaben torrent search API. + + Knaben is a torrent search engine that aggregates results from multiple trackers. + + Example: + >>> client = KnabenClient() + >>> results = client.search("Inception 1080p") + >>> for torrent in results[:5]: + ... print(f"{torrent.name} - Seeders: {torrent.seeders}") + """ + + def __init__( + self, + base_url: Optional[str] = None, + timeout: Optional[int] = None, + config: Optional[Settings] = None + ): + """ + Initialize Knaben client. + + Args: + base_url: Knaben API base URL (defaults to https://api.knaben.org/v1) + timeout: Request timeout in seconds (defaults to settings) + config: Optional Settings instance (for testing) + + Note: + Knaben API doesn't require an API key + """ + cfg = config or settings + + self.base_url = base_url or "https://api.knaben.org/v1" + self.timeout = timeout or cfg.request_timeout + + logger.info("Knaben client initialized") + + def _make_request( + self, + params: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: + """ + Make a request to Knaben API. + + Args: + endpoint: API endpoint (e.g., '/search') + params: Query parameters + + Returns: + JSON response as dict + + Raises: + KnabenAPIError: If request fails + """ + try: + logger.debug(f"Knaben request with params: {params}") + response = requests.post(self.base_url, json=params, timeout=self.timeout) + response.raise_for_status() + return response.json() + + except Timeout as e: + logger.error(f"Knaben API timeout: {e}") + raise KnabenAPIError(f"Request timeout after {self.timeout} seconds") from e + + except HTTPError as e: + logger.error(f"Knaben API HTTP error: {e}") + if e.response is not None: + status_code = e.response.status_code + if status_code == 404: + raise KnabenNotFoundError("Resource not found") from e + elif status_code == 429: + raise KnabenAPIError("Rate limit exceeded") from e + else: + raise KnabenAPIError(f"HTTP {status_code}: {e}") from e + raise KnabenAPIError(f"HTTP error: {e}") from e + + except RequestException as e: + logger.error(f"Knaben API request failed: {e}") + raise KnabenAPIError(f"Failed to connect to Knaben API: {e}") from e + + def search( + self, + query: str, + limit: int = 10 + ) -> List[TorrentResult]: + """ + Search for torrents. + + Args: + query: Search query (e.g., "Inception 1080p") + limit: Maximum number of results (default: 50) + + Returns: + List of TorrentResult objects + + Raises: + KnabenAPIError: If request fails + KnabenNotFoundError: If no results found + ValueError: If query is invalid + """ + if not query or not isinstance(query, str): + raise ValueError("Query must be a non-empty string") + + if len(query) > 500: + raise ValueError("Query is too long (max 500 characters)") + + # Build params + params = { + "query": query, + "search_field": "title", + "order_by": "peers", + "order_direction": "desc", + "from": 0, + "size": limit, + "hide_unsafe": True, + "hide_xxx": True, + } + + try: + data = self._make_request(params) + except KnabenNotFoundError as e: + # No results found + logger.info(f"No torrents found for '{query}'") + return [] + except Exception as e: + logger.error(f"Unexpected error in search: {e}", exc_info=True) + raise + + # Parse results + results = [] + torrents = data.get('hits', []) + + if not torrents: + logger.info(f"No torrents found for '{query}'") + return [] + + for torrent in torrents: + try: + result = self._parse_torrent(torrent) + results.append(result) + except Exception as e: + logger.warning(f"Failed to parse torrent: {e}") + continue + + logger.info(f"Found {len(results)} torrents for '{query}'") + return results + + def _parse_torrent(self, torrent: Dict[str, Any]) -> TorrentResult: + """ + Parse a torrent result into a TorrentResult object. + + Args: + torrent: Raw torrent dict from API + + Returns: + TorrentResult object + """ + # Extract required fields (API uses camelCase) + title = torrent.get('title', 'Unknown') + size = torrent.get('size', 'Unknown') + seeders = int(torrent.get('seeders', 0) or 0) + leechers = int(torrent.get('leechers', 0) or 0) + magnet = torrent.get('magnetUrl', '') + + # Extract optional fields + info_hash = torrent.get('hash') + tracker = torrent.get('tracker') + upload_date = torrent.get('date') + category = torrent.get('category') + + return TorrentResult( + title=title, + size=size, + seeders=seeders, + leechers=leechers, + magnet=magnet, + info_hash=info_hash, + tracker=tracker, + upload_date=upload_date, + category=category + ) + +# Global Knaben client instance (singleton) +knaben_client = KnabenClient() diff --git a/agent/parameters.py b/agent/parameters.py index 08d26f9..eee9d87 100644 --- a/agent/parameters.py +++ b/agent/parameters.py @@ -26,7 +26,7 @@ REQUIRED_PARAMETERS = [ "- download_folder: Where downloaded files arrive before being organized\n" "- tvshow_folder: Where TV show files are organized and stored\n" "- movie_folder: Where movie files are organized and stored\n" - "- torrent_folder: Where .torrent files are saved for the torrent client" + "- torrent_folder: Where torrent structures are saved for the torrent client" ), type="object", validator=lambda x: isinstance(x, dict), diff --git a/agent/registry.py b/agent/registry.py index 8667b59..8210646 100644 --- a/agent/registry.py +++ b/agent/registry.py @@ -5,7 +5,7 @@ from functools import partial from .memory import Memory from .tools.filesystem import set_path_for_folder, list_folder -from .tools.api import find_media_imdb_id +from .tools.api import find_media_imdb_id, find_torrent @dataclass @@ -20,17 +20,17 @@ class Tool: def make_tools(memory: Memory) -> Dict[str, Tool]: """ Create all available tools with memory bound to them. - + Args: memory: Memory instance to be used by the tools - + Returns: Dictionary mapping tool names to Tool instances """ # Create partial functions with memory pre-bound for filesystem tools set_path_func = partial(set_path_for_folder, memory) list_folder_func = partial(list_folder, memory) - + tools = [ Tool( name="set_path_for_folder", @@ -88,6 +88,21 @@ def make_tools(memory: Memory) -> Dict[str, Tool]: "required": ["media_title"] } ), + Tool( + name="find_torrents", + description="Finds torrents for a given media title using Knaben API.", + func=find_torrent, + parameters={ + "type": "object", + "properties": { + "media_title": { + "type": "string", + "description": "Title of the media to find torrents for" + }, + }, + "required": ["media_title"] + } + ), ] return {t.name: t for t in tools} diff --git a/agent/tools/api.py b/agent/tools/api.py index e08473e..4051d73 100644 --- a/agent/tools/api.py +++ b/agent/tools/api.py @@ -3,6 +3,7 @@ from typing import Dict, Any import logging from ..api import tmdb_client, TMDBError, TMDBNotFoundError, TMDBAPIError, TMDBConfigurationError +from ..api.knaben import knaben_client, KnabenError, KnabenNotFoundError, KnabenAPIError logger = logging.getLogger(__name__) @@ -10,18 +11,18 @@ logger = logging.getLogger(__name__) def find_media_imdb_id(media_title: str) -> Dict[str, Any]: """ Find the IMDb ID for a given media title using TMDB API. - + This is a wrapper around the TMDB client that returns a standardized dict format for compatibility with the agent's tool system. Args: media_title: Title of the media to search for - + Returns: Dict with IMDb ID or error information: - Success: {"status": "ok", "imdb_id": str, "title": str, ...} - Error: {"error": str, "message": str} - + Example: >>> result = find_media_imdb_id("Inception") >>> print(result) @@ -30,7 +31,7 @@ def find_media_imdb_id(media_title: str) -> Dict[str, Any]: try: # Use the TMDB client to search for media result = tmdb_client.search_media(media_title) - + # Check if IMDb ID was found if result.imdb_id: logger.info(f"IMDb ID found for '{media_title}': {result.imdb_id}") @@ -53,35 +54,112 @@ def find_media_imdb_id(media_title: str) -> Dict[str, Any]: "media_type": result.media_type, "tmdb_id": result.tmdb_id } - + except TMDBNotFoundError as e: logger.info(f"Media not found: {e}") return { "error": "not_found", "message": str(e) } - + except TMDBConfigurationError as e: logger.error(f"TMDB configuration error: {e}") return { "error": "configuration_error", "message": str(e) } - + except TMDBAPIError as e: logger.error(f"TMDB API error: {e}") return { "error": "api_error", "message": str(e) } - + except ValueError as e: logger.error(f"Validation error: {e}") return { "error": "validation_failed", "message": str(e) } - + + except Exception as e: + logger.error(f"Unexpected error: {e}", exc_info=True) + return { + "error": "internal_error", + "message": "An unexpected error occurred" + } + +def find_torrent(media_title: str) -> Dict[str, Any]: + """ + Find torrents for a given media title using Knaben API. + + This is a wrapper around the Knaben client that returns a standardized + dict format for compatibility with the agent's tool system. + + Args: + media_title: Title of the media to search for + + Returns: + Dict with torrent information or error details: + - Success: {"status": "ok", "torrents": List[Dict[str, Any]]} + - Error: {"error": str, "message": str} + """ + try: + # Search for torrents + results = knaben_client.search(media_title, limit=10) + + if not results: + logger.info(f"No torrents found for '{media_title}'") + return { + "error": "not_found", + "message": f"No torrents found for '{media_title}'" + } + + # Convert to dict format + torrents = [] + for torrent in results: + torrents.append({ + "name": torrent.title, + "size": torrent.size, + "seeders": torrent.seeders, + "leechers": torrent.leechers, + "magnet": torrent.magnet, + "info_hash": torrent.info_hash, + "tracker": torrent.tracker, + "upload_date": torrent.upload_date, + "category": torrent.category + }) + + logger.info(f"Found {len(torrents)} torrents for '{media_title}'") + + return { + "status": "ok", + "torrents": torrents, + "count": len(torrents) + } + + except KnabenNotFoundError as e: + logger.info(f"Torrents not found: {e}") + return { + "error": "not_found", + "message": str(e) + } + + except KnabenAPIError as e: + logger.error(f"Knaben API error: {e}") + return { + "error": "api_error", + "message": str(e) + } + + except ValueError as e: + logger.error(f"Validation error: {e}") + return { + "error": "validation_failed", + "message": str(e) + } + except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) return {