"""
Web tools module for OllamaTools.
This module provides functions for making web requests and processing responses,
including weather information fetching and web search functionality.
"""
import requests
from typing import Dict, Any, Optional, Union, List, Set
import logging
import json
from urllib.parse import urlparse, quote
from bs4 import BeautifulSoup
# Import the URL validation module
from ollamatools.core.utils.validators.url import (
validate_url,
URLValidationError,
configure_domain_lists,
create_url_validator
)
# Get logger
logger = logging.getLogger(__name__)
def fetch_url(
url: str,
method: str = 'GET',
params: Optional[Dict[str, str]] = None,
headers: Optional[Dict[str, str]] = None,
data: Optional[Dict[str, Any]] = None,
timeout: int = 30,
permission_manager = None,
allowed_schemes: Optional[Set[str]] = None,
domain_allowlist: Optional[Set[str]] = None,
domain_blocklist: Optional[Set[str]] = None
) -> Dict[str, Any]:
"""
Fetch a URL with specified parameters.
Args:
url: URL to fetch
method: HTTP method (GET, POST, PUT, DELETE, etc.)
params: URL parameters
headers: HTTP headers
data: Request data
timeout: Request timeout in seconds
permission_manager: Optional permission manager to request permission
allowed_schemes: Optional set of allowed URL schemes
domain_allowlist: Optional set of allowed domains
domain_blocklist: Optional set of blocked domains
Returns:
dict: Response with status, headers, and content
Raises:
ValueError: If URL or method is invalid
requests.RequestException: If the request fails
PermissionError: If permission is denied
"""
# Request permission if permission manager is provided
if permission_manager:
args = {
"url": url,
"method": method,
"params": params,
"headers": headers,
"data": data
}
if not permission_manager.request_permission("fetch_url", args):
logger.warning(f"Permission denied for web request to {url}")
raise PermissionError(f"Permission denied for web request to {url}")
# Validate and normalize the URL
try:
validated_url = validate_url(
url,
allowed_schemes=allowed_schemes,
domain_allowlist=domain_allowlist,
domain_blocklist=domain_blocklist
)
logger.debug(f"URL validated and normalized: {validated_url}")
except URLValidationError as e:
logger.error(f"URL validation error: {str(e)}")
raise ValueError(f"URL validation failed: {str(e)}")
# Validate method
valid_methods = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS']
method = method.upper()
if method not in valid_methods:
raise ValueError(f"Invalid HTTP method: {method}. Must be one of {valid_methods}")
# Set default headers if none provided
if headers is None:
headers = {}
# Add default User-Agent if not present
if 'User-Agent' not in headers:
headers['User-Agent'] = 'OllamaTools/0.1.0 (https://github.com/yourusername/ollamatools)'
logger.info(f"Making {method} request to {validated_url}")
try:
# Create a session to manage cookies and connection pooling
with requests.Session() as session:
# Configure session with reasonable security defaults
session.verify = True # Verify SSL certificates
# Make the request
response = session.request(
method=method,
url=validated_url,
params=params,
headers=headers,
json=data if method in ['POST', 'PUT', 'PATCH'] else None,
timeout=timeout
)
# Try to parse response as JSON, fall back to text
content_type = response.headers.get('Content-Type', '')
if 'application/json' in content_type:
try:
content = response.json()
except ValueError:
content = response.text
else:
content = response.text
result = {
"status": response.status_code,
"headers": dict(response.headers),
"content": content,
"url": response.url
}
logger.debug(f"Request to {validated_url} completed with status {response.status_code}")
return result
except requests.RequestException as e:
logger.error(f"Request to {validated_url} failed: {str(e)}")
raise
def get_weather(
location: str,
days: int = 3,
units: str = 'metric',
api_key: Optional[str] = None,
permission_manager = None,
domain_allowlist: Optional[Set[str]] = None,
domain_blocklist: Optional[Set[str]] = None
) -> Dict[str, Any]:
"""
Get weather information for a location.
Args:
location: Location name or coordinates
days: Number of forecast days (1-7)
units: Temperature units ('metric' or 'imperial')
api_key: OpenWeatherMap API key (if None, uses config)
permission_manager: Optional permission manager to request permission
domain_allowlist: Optional set of allowed domains
domain_blocklist: Optional set of blocked domains
Returns:
dict: Weather information including current conditions and forecast
Raises:
ValueError: If parameters are invalid
requests.RequestException: If the API request fails
PermissionError: If permission is denied
"""
# Validate parameters
if not location:
raise ValueError("Location cannot be empty")
if days < 1 or days > 7:
raise ValueError("Days must be between 1 and 7")
valid_units = ['metric', 'imperial']
if units not in valid_units:
raise ValueError(f"Units must be one of {valid_units}")
# Request permission if permission manager is provided
if permission_manager:
args = {
"location": location,
"days": days,
"units": units
}
if not permission_manager.request_permission("get_weather", args):
logger.warning(f"Permission denied for weather request for {location}")
raise PermissionError(f"Permission denied for weather request for {location}")
# Check if API key is provided
if api_key is None:
# In a real implementation, this would get the API key from the config
# For now, we'll use a placeholder message
logger.error("No API key provided for weather API")
raise ValueError("Weather API key is required. Please provide an API key or set it in the configuration.")
logger.info(f"Fetching weather for {location}")
try:
# Ensure allowlist includes the weather API domain
weather_domain_allowlist = {"api.openweathermap.org"}
if domain_allowlist:
weather_domain_allowlist.update(domain_allowlist)
# Base URL for OpenWeatherMap API
base_url = "https://api.openweathermap.org/data/2.5"
# Get current weather
current_url = f"{base_url}/weather"
params = {
"q": location,
"appid": api_key,
"units": units
}
current_response = fetch_url(
current_url,
params=params,
timeout=10,
permission_manager=permission_manager,
domain_allowlist=weather_domain_allowlist,
domain_blocklist=domain_blocklist
)
current_data = current_response["content"]
# Get forecast
forecast_url = f"{base_url}/forecast"
params = {
"q": location,
"appid": api_key,
"units": units,
"cnt": days * 8 # 8 forecasts per day (every 3 hours)
}
forecast_response = fetch_url(
forecast_url,
params=params,
timeout=10,
permission_manager=permission_manager,
domain_allowlist=weather_domain_allowlist,
domain_blocklist=domain_blocklist
)
forecast_data = forecast_response["content"]
# Format the response
weather_info = {
"location": {
"name": current_data.get("name", ""),
"country": current_data.get("sys", {}).get("country", ""),
"coordinates": current_data.get("coord", {})
},
"current": {
"temperature": current_data.get("main", {}).get("temp"),
"feels_like": current_data.get("main", {}).get("feels_like"),
"humidity": current_data.get("main", {}).get("humidity"),
"pressure": current_data.get("main", {}).get("pressure"),
"wind_speed": current_data.get("wind", {}).get("speed"),
"wind_direction": current_data.get("wind", {}).get("deg"),
"description": current_data.get("weather", [{}])[0].get("description", ""),
"icon": current_data.get("weather", [{}])[0].get("icon", "")
},
"forecast": []
}
# Process forecast data
for item in forecast_data.get("list", [])[:days * 8]:
forecast_item = {
"timestamp": item.get("dt"),
"temperature": item.get("main", {}).get("temp"),
"feels_like": item.get("main", {}).get("feels_like"),
"humidity": item.get("main", {}).get("humidity"),
"pressure": item.get("main", {}).get("pressure"),
"wind_speed": item.get("wind", {}).get("speed"),
"wind_direction": item.get("wind", {}).get("deg"),
"description": item.get("weather", [{}])[0].get("description", ""),
"icon": item.get("weather", [{}])[0].get("icon", "")
}
weather_info["forecast"].append(forecast_item)
return weather_info
except Exception as e:
logger.error(f"Weather API request failed: {str(e)}")
raise
def search_web(
query: str,
num_results: int = 5,
api_key: Optional[str] = None,
permission_manager = None,
domain_allowlist: Optional[Set[str]] = None,
domain_blocklist: Optional[Set[str]] = None
) -> List[Dict[str, str]]:
"""
Search the web for a query.
Args:
query: Search query
num_results: Number of results to return (1-10)
api_key: Search API key (if None, uses config)
permission_manager: Optional permission manager to request permission
domain_allowlist: Optional set of allowed domains
domain_blocklist: Optional set of blocked domains
Returns:
List[dict]: Search results with title, URL, and snippet
Raises:
ValueError: If parameters are invalid
requests.RequestException: If the API request fails
PermissionError: If permission is denied
"""
# Validate parameters
if not query:
raise ValueError("Query cannot be empty")
if num_results < 1 or num_results > 10:
raise ValueError("Number of results must be between 1 and 10")
# Request permission if permission manager is provided
if permission_manager:
args = {
"query": query,
"num_results": num_results
}
if not permission_manager.request_permission("search_web", args):
logger.warning(f"Permission denied for web search: {query}")
raise PermissionError(f"Permission denied for web search: {query}")
# Check if API key is provided
if api_key is None:
# In a real implementation, this would get the API key from the config
# For now, we'll use a fallback to a simple search method that doesn't require API key
logger.warning("No API key provided for search API, using fallback method")
return _fallback_search(
query,
num_results,
permission_manager=permission_manager,
domain_allowlist=domain_allowlist,
domain_blocklist=domain_blocklist
)
logger.info(f"Searching the web for: {query}")
try:
# Ensure allowlist includes the search API domain
search_domain_allowlist = {"serpapi.com"}
if domain_allowlist:
search_domain_allowlist.update(domain_allowlist)
# Example using the Serpapi (Google Search API)
# In a real implementation, you would use a proper search API
url = "https://serpapi.com/search"
params = {
"q": query,
"api_key": api_key,
"num": num_results
}
response = fetch_url(
url,
params=params,
timeout=15,
permission_manager=permission_manager,
domain_allowlist=search_domain_allowlist,
domain_blocklist=domain_blocklist
)
data = response["content"]
results = []
# Process search results
for item in data.get("organic_results", [])[:num_results]:
result = {
"title": item.get("title", ""),
"url": item.get("link", ""),
"snippet": item.get("snippet", "")
}
results.append(result)
return results
except Exception as e:
logger.error(f"Search API request failed: {str(e)}")
# Fall back to the simple search method if API request fails
logger.info("Falling back to simple search method")
return _fallback_search(
query,
num_results,
permission_manager=permission_manager,
domain_allowlist=domain_allowlist,
domain_blocklist=domain_blocklist
)
def _fallback_search(
query: str,
num_results: int = 5,
permission_manager = None,
domain_allowlist: Optional[Set[str]] = None,
domain_blocklist: Optional[Set[str]] = None
) -> List[Dict[str, str]]:
"""
Fallback search method using DuckDuckGo HTML scraping when API is not available.
Note: This is a fallback method and not recommended for production use.
It may break if the search engine changes its HTML structure.
Args:
query: Search query
num_results: Number of results to return
permission_manager: Optional permission manager to request permission
domain_allowlist: Optional set of allowed domains
domain_blocklist: Optional set of blocked domains
Returns:
List[dict]: Search results with title, URL, and snippet
"""
results = []
encoded_query = quote(query)
url = f"https://html.duckduckgo.com/html/?q={encoded_query}"
try:
# Ensure allowlist includes the search domain
search_domain_allowlist = {"html.duckduckgo.com"}
if domain_allowlist:
search_domain_allowlist.update(domain_allowlist)
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
}
response = fetch_url(
url,
headers=headers,
timeout=15,
permission_manager=permission_manager,
domain_allowlist=search_domain_allowlist,
domain_blocklist=domain_blocklist
)
soup = BeautifulSoup(response["content"], 'html.parser')
# Find search results
result_elements = soup.select('.result')
for element in result_elements[:num_results]:
title_element = element.select_one('.result__title')
link_element = element.select_one('.result__url')
snippet_element = element.select_one('.result__snippet')
if title_element and link_element:
title = title_element.get_text(strip=True)
# The actual URL is in the href attribute of the <a> tag
link = title_element.find('a').get('href', '')
snippet = snippet_element.get_text(strip=True) if snippet_element else ""
# Validate URL
try:
validate_url(link)
results.append({
"title": title,
"url": link,
"snippet": snippet
})
except URLValidationError:
logger.warning(f"Skipping result with invalid URL: {link}")
return results
except Exception as e:
logger.error(f"Fallback search failed: {str(e)}")
# Return empty results if even the fallback fails
return []