#!/usr/bin/env python3
"""
Test script to directly verify OpenSearchNG functionality without importing from ollamatools.
"""
import logging
import json
import sys
import requests
from typing import Dict, Any, Optional, Set
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Default OpenSearchNG configuration
DEFAULT_OPENSEARCH_CONFIG = {
"base_url": "http://localhost:8081", # Default local OpenSearchNG URL
"search_path": "/search", # Path for search endpoint
"preferences_path": "/preferences", # Path for preferences endpoint
"categories": ["general"], # Default search categories
"engines": ["google", "bing", "duckduckgo"], # Default search engines
"language": "en-US", # Default language
"format": "json", # Response format (html, json, csv, rss)
"timeout": 10, # Default timeout in seconds
"num_results": 10, # Default number of results
"safe_search": 0 # 0=off, 1=moderate, 2=strict
}
def opensearch_query(
query: str,
config: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Perform a search query using a local OpenSearchNG instance.
Args:
query: Search query string
config: OpenSearchNG configuration (defaults used if None)
Returns:
dict: Search results with metadata
"""
# Validate input
if not query or not isinstance(query, str):
raise ValueError("Query must be a non-empty string")
# Use default config if none provided
search_config = DEFAULT_OPENSEARCH_CONFIG.copy()
if config:
search_config.update(config)
# Construct the search URL
base_url = search_config["base_url"]
search_path = search_config["search_path"]
search_url = f"{base_url}{search_path}"
# Prepare query parameters
params = {
"q": query,
"format": search_config["format"],
"categories": ",".join(search_config["categories"]),
"engines": ",".join(search_config["engines"]),
"language": search_config["language"],
"safesearch": search_config["safe_search"],
"time_range": "", # Optional time range
"pageno": 1, # First page of results
}
# Create session with appropriate headers
with requests.Session() as session:
headers = {
"User-Agent": "OllamaTools/0.2.0",
"Accept": "application/json"
}
session.headers.update(headers)
try:
logger.info(f"Making OpenSearchNG query: {query}")
response = session.get(
search_url,
params=params,
timeout=search_config["timeout"]
)
# Check for successful response
if response.status_code != 200:
logger.error(f"OpenSearchNG query failed with status code: {response.status_code}")
return {"error": f"Status code: {response.status_code}", "results": []}
# Parse the response based on format
if search_config["format"] == "json":
try:
result_data = response.json()
except json.JSONDecodeError:
logger.error("Failed to parse JSON response from OpenSearchNG")
return {"error": "JSON decode error", "results": []}
else:
# For other formats, return the raw content
result_data = {"raw_content": response.text}
# Process the results
processed_results = process_opensearch_results(result_data, query)
return processed_results
except requests.RequestException as e:
logger.error(f"OpenSearchNG request failed: {str(e)}")
return {"error": str(e), "results": []}
def process_opensearch_results(
result_data: Dict[str, Any],
query: str
) -> Dict[str, Any]:
"""
Process and format the OpenSearchNG results.
Args:
result_data: Raw result data from OpenSearchNG
query: Original search query
Returns:
dict: Processed search results with metadata
"""
processed_results = {
"query": query,
"results": [],
"suggestion": None,
"answer": None,
"infobox": None,
"engines": [],
"timing": {}
}
# Extract search results
if "results" in result_data:
for result in result_data["results"]:
processed_result = {
"title": result.get("title", ""),
"url": result.get("url", ""),
"content": result.get("content", ""),
"engine": result.get("engine", ""),
"score": result.get("score", 0),
"category": result.get("category", ""),
"pretty_url": result.get("pretty_url", "")
}
processed_results["results"].append(processed_result)
# Extract suggestions
if "suggestions" in result_data and result_data["suggestions"]:
try:
suggestion = result_data["suggestions"][0]
# Handle case where suggestion is a dict instead of a string
if isinstance(suggestion, dict):
processed_results["suggestion"] = suggestion.get("value", str(suggestion))
else:
processed_results["suggestion"] = suggestion
except (IndexError, TypeError) as e:
logger.warning(f"Error processing suggestions: {str(e)}")
processed_results["suggestion"] = None
# Extract answer if available
if "answers" in result_data and result_data["answers"]:
try:
answer = result_data["answers"][0]
# Handle case where answer is a dict instead of a string
if isinstance(answer, dict):
processed_results["answer"] = answer.get("content", str(answer))
else:
processed_results["answer"] = answer
except (IndexError, TypeError) as e:
logger.warning(f"Error processing answers: {str(e)}")
processed_results["answer"] = None
# Extract infobox if available
if "infoboxes" in result_data and result_data["infoboxes"]:
try:
infobox = result_data["infoboxes"][0]
# Handle case where infobox is a dict (which is expected)
processed_results["infobox"] = infobox
except (IndexError, TypeError) as e:
logger.warning(f"Error processing infoboxes: {str(e)}")
processed_results["infobox"] = None
# Extract engines used
if "engines" in result_data:
processed_results["engines"] = result_data["engines"]
# Extract timing information
if "timing" in result_data:
processed_results["timing"] = result_data["timing"]
return processed_results
def get_opensearch_info(
config: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Get information about the OpenSearchNG instance.
Args:
config: OpenSearchNG configuration (defaults used if None)
Returns:
dict: Information about the OpenSearchNG instance
"""
# Use default config if none provided
search_config = DEFAULT_OPENSEARCH_CONFIG.copy()
if config:
search_config.update(config)
# Construct the URL
base_url = search_config["base_url"]
config_url = f"{base_url}/config"
try:
logger.info("Getting OpenSearchNG instance information")
response = requests.get(
config_url,
timeout=search_config["timeout"]
)
# Check for successful response
if response.status_code != 200:
logger.error(f"OpenSearchNG info request failed with status code: {response.status_code}")
return {"error": f"Status code: {response.status_code}"}
# Parse the response
try:
config_data = response.json()
return config_data
except json.JSONDecodeError:
logger.error("Failed to parse JSON response from OpenSearchNG")
return {"error": "JSON decode error"}
except requests.RequestException as e:
logger.error(f"OpenSearchNG info request failed: {str(e)}")
return {"error": str(e)}
def test_opensearch_functionality():
"""Test the OpenSearchNG functionality."""
try:
# Test getting OpenSearchNG info
logger.info("Testing OpenSearchNG info...")
info_result = get_opensearch_info()
if "error" in info_result:
logger.error(f"Error getting OpenSearchNG info: {info_result['error']}")
return False
else:
logger.info(f"OpenSearchNG info: {json.dumps(info_result, indent=2)[:200]}...")
# Test OpenSearchNG query
logger.info("Testing OpenSearchNG query...")
query_result = opensearch_query("python programming")
if "error" in query_result:
logger.error(f"Error executing query: {query_result['error']}")
return False
num_results = len(query_result.get("results", []))
logger.info(f"Got {num_results} results for 'python programming'")
# Print the first result
if num_results > 0:
first_result = query_result["results"][0]
logger.info(f"First result: {json.dumps(first_result, indent=2)}")
return True
except Exception as e:
logger.error(f"Error: {e}")
return False
if __name__ == "__main__":
success = test_opensearch_functionality()
if success:
logger.info("OpenSearchNG functionality test successful!")
sys.exit(0)
else:
logger.error("OpenSearchNG functionality test failed!")
sys.exit(1)