enne2
/
qbit-agent
1
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

first commit

master
Matteo Benedetto 8 months ago
commit
88c8bb39f8
7 changed files with 442 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 12
      .env-example
  2. 39
      .gitignore
  3. 21
      LICENSE.md
  4. 53
      README.md
  5. 101
      main.py
  6. 6
      requirements.txt
  7. 210
      tools.py

12
.env-example
Unescape View File

@ -0,0 +1,12 @@
# qBittorrent API Configuration
QBIT_HOST=http://localhost:8080
QBIT_USERNAME=admin
QBIT_PASSWORD=password
# OpenAI API Key (required for the LangChain agent)
OPENAI_API_KEY=sk-proj-Rs93xxxxxxxxxxxxxxxxxxxxxUnStmeSHj_gUiEfbGzaFeZf0rgdaQzllQmvcMy6o-SywA
# DuckDuckGo Search Configuration
DUCKDUCKGO_ENABLED=true
DUCKDUCKGO_MAX_RESULTS=5
OMDB_API_KEY=3b6bc268

39
.gitignore vendored
Unescape View File

@ -0,0 +1,39 @@
# Environment variables
.env
# Python bytecode
__pycache__/
*.py[cod]
*$py.class
# Distribution / packaging
dist/
build/
*.egg-info/
# Virtual environments
venv/
env/
ENV/
.venv/
# IDE files
.idea/
.vscode/
*.swp
*.swo
# OS files
.DS_Store
Thumbs.db
# Logs
*.log
# Testing
.pytest_cache/
.coverage
htmlcov/
# Jupyter Notebooks
.ipynb_checkpoints

21
LICENSE.md
Unescape View File

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2023 qBittorrent AI Agent
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

53
README.md
Unescape View File

@ -0,0 +1,53 @@
# qBittorrent AI Agent
An AI-powered assistant for qBittorrent that allows natural language interaction with your torrent client.
## Features
- **Natural Language Interface**: Interact with qBittorrent using natural language commands
- **Search Torrents**: Search for torrents directly through the AI interface
- **Download Management**: View active downloads and add new torrents
- **Web Interface**: Built with Gradio for easy access through your browser
- **Command Line Interface**: Optional CLI mode for terminal-based interactions
## Requirements
- Python 3.8+
- qBittorrent with WebUI enabled
- OpenAI API key
## Installation
1. Clone this repository
2. Install dependencies:
```
pip install -r requirements.txt
```
3. Create a `.env` file with your configuration:
```
OPENAI_API_KEY=your_openai_api_key
QBIT_HOST=http://localhost:8080
QBIT_USERNAME=admin
QBIT_PASSWORD=adminadmin
```
## Usage
Run the web interface:
```
python main.py
```
Or use the CLI interface by uncommenting the `cli_main()` line in `main.py`.
## Tools
The agent includes several tools:
- `get_downloads_list`: Get information about current downloads
- `qbittorrent_search`: Search for torrents using qBittorrent's search functionality
- `download_torrent`: Add a torrent to the download queue
- `ForcedDuckDuckGoSearch`: Search for information about media content
## License
MIT

101
main.py
Unescape View File

@ -0,0 +1,101 @@
import os
from langchain.agents import Tool, initialize_agent, AgentType
from dotenv import load_dotenv
from tools import DownloadListTool, QBitSearchTool, DownloadTorrentTool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import init_chat_model
import gradio as gr
# Load environment variables
load_dotenv()
def create_agent():
# Initialize the language model
llm = init_chat_model("gpt-4o-mini", model_provider="openai")
# Initialize memory
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
# Initialize search tool
search_tool = DuckDuckGoSearchRun()
# Function to force DuckDuckGo for specific search types
def forced_duckduckgo_search(query: str) -> str:
"""Use DuckDuckGo to search for specific information."""
return search_tool.run(query)
# Initialize tools
tools = [
DownloadListTool(),
QBitSearchTool(),
DownloadTorrentTool(),
Tool(
name="ForcedDuckDuckGoSearch",
func=forced_duckduckgo_search,
description="Use this tool when you need to find specific information about movies, TV shows. Input should be a search query including the keyword 'imdb'.",
)
]
# Initialize the agent with memory
agent = initialize_agent(
tools,
llm,
agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
verbose=True,
memory=memory
)
return agent
def process_query(message, history):
try:
# Create agent if it doesn't exist
if not hasattr(process_query, "agent"):
process_query.agent = create_agent()
# Run the agent with the user's message
response = process_query.agent.run(message)
return response
except Exception as e:
return f"Error: {str(e)}"
def main():
print("Starting qBittorrent AI Agent...")
# Create Gradio interface
with gr.Blocks(title="qBittorrent AI Agent") as interface:
gr.Markdown("# qBittorrent AI Agent")
gr.Markdown("Ask questions about downloads, search for content, or get recommendations.")
chatbot = gr.ChatInterface(
process_query,
examples=["Find me the latest sci-fi movies",
"What are the top TV shows from 2023?",
"Download Interstellar in 1080p"],
title="qBittorrent Assistant"
)
# Launch the interface
interface.launch(share=False)
def cli_main():
print("Starting qBittorrent AI Agent in CLI mode...")
agent = create_agent()
while True:
user_input = input("\nEnter your question (or 'quit' to exit): ")
if user_input.lower() in ['quit', 'exit']:
break
try:
response = agent.run(user_input)
print(response)
except Exception as e:
print(f"Error: {str(e)}")
if __name__ == "__main__":
# Use main() for Gradio interface or cli_main() for command-line interface
main()
# Uncomment the line below to use CLI instead
# cli_main()

6
requirements.txt
Unescape View File

@ -0,0 +1,6 @@
langchain>=0.0.267
openai>=0.27.8
requests>=2.28.2
python-dotenv>=1.0.0
gradio>=3.0.0
langchain_community>=0.0.1

210
tools.py
Unescape View File

@ -0,0 +1,210 @@
import os
from langchain.tools.base import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
import requests
from typing import Optional
class DownloadListTool(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 with their details including magnet links.
'''
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"}
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 = 10 # 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": 10} # Limiting to top 10 results
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)
# Format the response
if not results:
return f"No results found for '{query}'."
response = f"Search results for '{query}':\n\n"
for i, result in enumerate(results, 1):
name = result.get("fileName", "Unknown")
size = result.get("fileSize", "Unknown")
seeds = result.get("seeders", 0)
leech = result.get("leechers", 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 DownloadTorrentTool(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)}"
Write Preview
Loading…
Cancel
Save