"""API tools for interacting with external services.""" 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 from ..api.qbittorrent import qbittorrent_client, QBittorrentError, QBittorrentAuthError, QBittorrentAPIError 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) {'status': 'ok', 'imdb_id': 'tt1375666', 'title': 'Inception', ...} """ 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}") return { "status": "ok", "imdb_id": result.imdb_id, "title": result.title, "media_type": result.media_type, "tmdb_id": result.tmdb_id, "overview": result.overview, "release_date": result.release_date, "vote_average": result.vote_average } else: logger.warning(f"No IMDb ID available for '{media_title}'") return { "error": "no_imdb_id", "message": f"No IMDb ID available for '{result.title}'", "title": result.title, "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 { "error": "internal_error", "message": "An unexpected error occurred" } def add_torrent_to_qbittorrent(magnet_link: str) -> Dict[str, Any]: """ Add a torrent to qBittorrent using a magnet link. This is a wrapper around the qBittorrent client that returns a standardized dict format for compatibility with the agent's tool system. Args: magnet_link: Magnet link of the torrent to add Returns: Dict with success or error information: - Success: {"status": "ok", "message": str} - Error: {"error": str, "message": str} Example: >>> result = add_torrent_to_qbittorrent("magnet:?xt=urn:btih:...") >>> print(result) {'status': 'ok', 'message': 'Torrent added successfully'} """ try: # Validate magnet link if not magnet_link or not isinstance(magnet_link, str): raise ValueError("Magnet link must be a non-empty string") if not magnet_link.startswith("magnet:"): raise ValueError("Invalid magnet link format") logger.info("Adding torrent to qBittorrent") # Add torrent to qBittorrent success = qbittorrent_client.add_torrent(magnet_link) if success: logger.info("Torrent added successfully to qBittorrent") return { "status": "ok", "message": "Torrent added successfully to qBittorrent" } else: logger.warning("Failed to add torrent to qBittorrent") return { "error": "add_failed", "message": "Failed to add torrent to qBittorrent" } except QBittorrentAuthError as e: logger.error(f"qBittorrent authentication error: {e}") return { "error": "authentication_failed", "message": "Failed to authenticate with qBittorrent" } except QBittorrentAPIError as e: logger.error(f"qBittorrent 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" }