domotic-agent/tools/qbit.py

import os
from langchain.tools.base import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
import requests
from typing import Optional
from langchain_community.tools import DuckDuckGoSearchRun

class QbitDownloadListTool(BaseTool):
    name: str = "get_downloads_list"
    description: str = '''Useful for getting a list of current downloads from the qBittorrent API and
    information about them. The response will include the name, size, and status of each download.
    '''
    
    def _run(self, query: str = "", run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        """Get the list of downloads from qBittorrent API."""
        try:
            # Configuration for qBittorrent API
            QBIT_HOST = os.environ.get("QBIT_HOST", "http://localhost:8080")
            QBIT_USERNAME = os.environ.get("QBIT_USERNAME", "admin")
            QBIT_PASSWORD = os.environ.get("QBIT_PASSWORD", "adminadmin")
            
            # First login to get the auth cookie
            login_url = f"{QBIT_HOST}/api/v2/auth/login"
            login_data = {"username": QBIT_USERNAME, "password": QBIT_PASSWORD}
            
            session = requests.Session()
            login_response = session.post(login_url, data=login_data)
            
            if login_response.status_code != 200:
                return f"Failed to login to qBittorrent API: {login_response.text}"
            
            # Get the list of torrents
            torrents_url = f"{QBIT_HOST}/api/v2/torrents/info"
            response = session.get(torrents_url)
            
            if response.status_code != 200:
                return f"Failed to fetch downloads: {response.text}"
            
            torrents = response.json()
            
            # Format the response
            if not torrents:
                return "No active downloads found."
            
            result = "Current downloads:\n"
            for i, torrent in enumerate(torrents, 1):
                result += f"{i}:"
                for key, value in torrent.items():
                    if key in ["name", "progress", "size", "state", "eta"]:
                        result += f" {key}: {value},"
                result += "\n"
            result += "\n"
            result += "Total downloads: " + str(len(torrents))
                
            return result
            
        except Exception as e:
            return f"Error getting downloads list: {str(e)}"

class QbitSearchTool(BaseTool):
    name: str = "qbittorrent_search"
    description: str = '''Useful for searching torrents using qBittorrent's search functionality. 
    Input should be a search query for content the user wants to find. 
    The tool will return a list of matching torrents ordered by the number of seeders (highest first).
    '''
    
    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        """Search for torrents using qBittorrent's search functionality."""
        try:
            # Configuration for qBittorrent API
            QBIT_HOST = os.environ.get("QBIT_HOST", "http://localhost:8080")
            QBIT_USERNAME = os.environ.get("QBIT_USERNAME", "admin")
            QBIT_PASSWORD = os.environ.get("QBIT_PASSWORD", "adminadmin")
            
            # First login to get the auth cookie
            login_url = f"{QBIT_HOST}/api/v2/auth/login"
            login_data = {"username": QBIT_USERNAME, "password": QBIT_PASSWORD}
            
            session = requests.Session()
            login_response = session.post(login_url, data=login_data)
            
            if login_response.status_code != 200:
                return f"Failed to login to qBittorrent API: {login_response.text}"
            
            # Start a search
            start_search_url = f"{QBIT_HOST}/api/v2/search/start"
            search_data = {"pattern": query, "plugins": "all", "category": "all", "limit": 5, "sort": "seeders", "order": "desc"}
            
            search_response = session.post(start_search_url, data=search_data)
            
            if search_response.status_code != 200:
                return f"Failed to start search: {search_response.text}"
            
            search_id = search_response.json().get("id")
            
            if not search_id:
                return "Failed to get search ID"
            
            # Wait for results (simple implementation, can be improved)
            import time
            max_wait = 5  # seconds
            wait_time = 0
            step = 1
            
            while wait_time < max_wait:
                time.sleep(step)
                wait_time += step
                
                # Get search status
                status_url = f"{QBIT_HOST}/api/v2/search/status"
                status_params = {"id": search_id}
                status_response = session.get(status_url, params=status_params)
                
                if status_response.status_code != 200:
                    return f"Failed to get search status: {status_response.text}"
                
                status_data = status_response.json()
                if status_data[0].get("status") == "Stopped":
                    break
            
            # Get search results
            results_url = f"{QBIT_HOST}/api/v2/search/results"
            results_params = {"id": search_id, "limit": 5}  # Increased limit to find more seeders
            
            results_response = session.get(results_url, params=results_params)
            
            if results_response.status_code != 200:
                return f"Failed to get search results: {results_response.text}"
            
            results_data = results_response.json()
            results = results_data.get("results", [])
            
            # Stop the search
            stop_url = f"{QBIT_HOST}/api/v2/search/stop"
            stop_params = {"id": search_id}
            session.post(stop_url, params=stop_params)
        
            print(results)
            # Limit to top 10 results after sorting
            results = results[:10]
            
            # Format the response
            if not results:
                return f"No results found for '{query}'."
            
            response = f"Search results for '{query}' (sorted by seeders):\n\n"
            
            for i, result in enumerate(results, 1):
                name = result.get("fileName", "Unknown")
                size = result.get("fileSize", "Unknown")
                seeds = result.get("nbSeeders", 0)
                leech = result.get("nbLeechers", 0)
                magnet = result.get("fileUrl", "")
                
                # Convert size to human-readable format
                if isinstance(size, (int, float)):
                    units = ["B", "KB", "MB", "GB", "TB"]
                    size_index = 0
                    while size >= 1024 and size_index < len(units) - 1:
                        size /= 1024
                        size_index += 1
                    size = f"{size:.2f} {units[size_index]}"
                
                response += f"{i}. {name}\n"
                response += f"   Size: {size}, Seeds: {seeds}, Leechers: {leech}\n"
                if magnet:
                    response += f"   Magnet: {magnet}\n"
            
            return response
            
        except Exception as e:
            return f"Error searching torrents: {str(e)}"

class QbitDownloadTorrentTool(BaseTool):
    name: str = "download_torrent"
    description: str = '''Useful for starting a new torrent download in qBittorrent. 
    Input should be a magnet link or a torrent URL that the user wants to download.
    The tool will add the torrent to qBittorrent's download queue and return status.
    '''
    
    def _run(self, torrent_url: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        """Start downloading a torrent by adding it to qBittorrent."""
        try:
            # Check if the input is a valid torrent URL or magnet link
            if not (torrent_url.startswith("http") or torrent_url.startswith("magnet:")):
                return "Error: Please provide a valid torrent URL or magnet link."
            
            # Configuration for qBittorrent API
            QBIT_HOST = os.environ.get("QBIT_HOST", "http://localhost:8080")
            QBIT_USERNAME = os.environ.get("QBIT_USERNAME", "admin")
            QBIT_PASSWORD = os.environ.get("QBIT_PASSWORD", "adminadmin")
            
            # First login to get the auth cookie
            login_url = f"{QBIT_HOST}/api/v2/auth/login"
            login_data = {"username": QBIT_USERNAME, "password": QBIT_PASSWORD}
            
            session = requests.Session()
            login_response = session.post(login_url, data=login_data)
            
            if login_response.status_code != 200:
                return f"Failed to login to qBittorrent API: {login_response.text}"
            
            # Add torrent to download queue
            add_url = f"{QBIT_HOST}/api/v2/torrents/add"
            add_data = {"urls": torrent_url}
            
            add_response = session.post(add_url, data=add_data)
            
            if add_response.status_code != 200:
                return f"Failed to add torrent: {add_response.text}"
            
            return f"Torrent has been added to download queue successfully. Check downloads list for status."
            
        except Exception as e:
            return f"Error adding torrent: {str(e)}"