From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- cogs/utils/file_search.py | 167 ---------------------------------------------- 1 file changed, 167 deletions(-) delete mode 100644 cogs/utils/file_search.py (limited to 'cogs/utils/file_search.py') diff --git a/cogs/utils/file_search.py b/cogs/utils/file_search.py deleted file mode 100644 index 788f50c..0000000 --- a/cogs/utils/file_search.py +++ /dev/null @@ -1,167 +0,0 @@ -# This file is part of Senator Sperma -# Licensed under the GPL-3.0 License -# Copyright (C) 2026 Salmonik @ salmonik@tuta.io -# https://codeberg.org/salmonik/senator-sperma - -import json -import logging -import aiohttp -import os -from typing import List -from cogs.utils.settings import get_baseurl - -logger = logging.getLogger(__name__) - - -class UniversalSearcher: - """Handles searching for files via copyparty API.""" - - @staticmethod - async def search_copyparty(query: str, max_results: int = 25, file_extensions: List[str] | None = None): - """Search for files using copyparty API (jPOST method). - - Args: - query: Search query string - max_results: Maximum number of results to return - file_extensions: Optional list of file extensions to filter by (e.g., ['mp3', 'wav']) - - Returns: - List of matching file URLs - """ - if not query: - return [] - - query = query.strip() - logger.info(f"Starting copyparty search for query: '{query}'") - - tokens = query.split() - if not tokens: - return [] - - token_queries = [] - for token in tokens: - path_query = f'path like "*{token}*"' - name_query = f'name like "*{token}*"' - token_queries.append(f"({path_query} or {name_query})") - - if len(token_queries) == 1: - name_query = token_queries[0] - else: - name_query = " and ".join(token_queries) - - if file_extensions: - ext_filters = [f"name like \"*.{ext.lower()}\"" for ext in file_extensions] - ext_query = " or ".join(ext_filters) - query = f"{name_query} and ({ext_query})" - else: - query = name_query - - try: - baseurl = get_baseurl() - password = os.getenv("COPYPARTY_PASSWORD") - - async with aiohttp.ClientSession() as session: - headers = {"Content-Type": "application/json"} - if password: - headers["PW"] = password - async with session.post(f"{baseurl}/?srch", headers=headers, data=json.dumps({"n": max_results, "q": query})) as response: - if response.status == 200: - data = await response.json() - files = UniversalSearcher._extract_files_from_response(data) - results = UniversalSearcher._convert_files_to_urls(files, baseurl, max_results) - - logger.info(f"Copyparty search complete. Found {len(results)} matches.") - return results - else: - logger.error(f"Copyparty API returned status {response.status}") - return [] - - except aiohttp.ClientError as e: - logger.error(f"HTTP error during copyparty search: {e}") - return [] - except Exception as e: - logger.error(f"Error during copyparty search: {str(e)}", exc_info=True) - return [] - - @staticmethod - def _extract_files_from_response(data): - """Extract files list from copyparty API response.""" - if isinstance(data, dict): - if 'hits' in data: - return data['hits'] - elif 'files' in data: - return data['files'] - elif 'results' in data: - return data['results'] - else: - return data if isinstance(data, list) else [] - elif isinstance(data, list): - return data - else: - return [] - - @staticmethod - def _convert_files_to_urls(files, baseurl, max_results): - """Convert file info objects to URLs.""" - results = [] - - for file_info in files[:max_results]: - if isinstance(file_info, dict): - file_url = UniversalSearcher._extract_url_from_dict(file_info, baseurl) - if file_url: - results.append(file_url) - elif isinstance(file_info, str): - if file_info.startswith('http'): - results.append(file_info) - else: - results.append(f"{baseurl}/{file_info.lstrip('/')}") - - return results - - @staticmethod - def _extract_url_from_dict(file_info, baseurl): - """Extract URL from file info dictionary.""" - if 'url' in file_info: - return file_info['url'] - elif 'rp' in file_info: - return f"{baseurl}/{file_info['rp']}" - elif 'path' in file_info: - return f"{baseurl}/{file_info['path'].lstrip('/')}" - elif 'name' in file_info: - return f"{baseurl}/{file_info['name'].lstrip('/')}" - else: - return None - - @staticmethod - async def search_copyparty_autocomplete(query: str, max_results: int = 10): - """Wrapper for autocomplete searches using copyparty (general search - no file type filtering). - - Args: - query: Search query string - max_results: Maximum number of results to return - - Returns: - List of matching file URLs - """ - if not query: - return [] - - return await UniversalSearcher.search_copyparty(query, max_results) - - @staticmethod - async def search_copyparty_autocomplete_audio(query: str, max_results: int = 10): - """Wrapper for autocomplete searches using copyparty (audio files only). - - Args: - query: Search query string - max_results: Maximum number of results to return - - Returns: - List of matching file URLs (audio files only) - """ - if not query: - return [] - - # Audio file extensions for music commands - audio_extensions = ['mp3', 'wav', 'flac', 'ogg', 'm4a', 'aac', 'mp4', "mkv"] - return await UniversalSearcher.search_copyparty(query, max_results, audio_extensions) -- cgit v1.2.3