voice doesn't work due to DAVE, but we got copyparty search

1 files changed, 246 insertions, 33 deletions

diff --git a/cogs/music/search_utils.py b/cogs/utils/file_search.py
index bc9e335..4a2bdc4 100644
--- a/cogs/music/search_utils.py
+++ b/cogs/utils/file_search.py
@@ -3,12 +3,57 @@
 # Copyright (C) 2026 Salmonik @ salmonik@tuta.io
 # https://codeberg.org/salmonik/senator-sperma
 
+"""
+Universal File Searcher
+
+This module provides a universal file searcher that can search for different file types:
+- Audio files (mp3, wav, mp4, opus, m4a, flac, aac, ogg)
+- Video files (mp4, avi, mkv, mov, wmv, flv, webm, m4v, 3gp)
+- Image files (jpg, jpeg, png, gif, bmp, tiff, webp, svg, ico)
+
+Usage Examples:
+    # Search all file types
+    results = await UniversalSearcher.search_local("query")
+    
+    # Search only audio files
+    results = await UniversalSearcher.search_local("query", filetype="Audio")
+    
+    # Search only video files
+    results = await UniversalSearcher.search_local("query", filetype="Video")
+    
+    # Search only image files
+    results = await UniversalSearcher.search_local("query", filetype="Image")
+    
+    # Autocomplete search
+    results = await UniversalSearcher.search_local_autocomplete("partial_query", filetype="Video")
+
+Settings.json structure:
+{
+    "file_paths": [
+        "/path/to/your/media/files",
+        "/another/path/with/mixed/content",
+        "relative/path/to/files"
+    ]
+}
+
+Note: file_paths should contain directories that may contain any mix of audio, video, 
+and image files. The searcher will filter by file type automatically based on the 
+filetype parameter.
+"""
+
 import json
 import logging
 import re
+import urllib.parse
 from pathlib import Path
 import aiofiles
 from rapidfuzz import fuzz
+import aiohttp
+import asyncio
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
 
 logger = logging.getLogger(__name__)
 
@@ -173,14 +218,34 @@ class SearchUtils:
         
         return score
 
-class AudioSearcher:
-    """Handles searching for audio files in configured paths."""
+class UniversalSearcher:
+    """Handles searching for files in configured paths with support for multiple file types."""
+    
+    FILE_TYPE_EXTENSIONS = {
+        'Audio': {'.mp4', '.mp3', '.wav', '.opus', '.m4a', '.flac', '.aac', '.ogg'},
+        'Video': {'.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.webm', '.m4v', '.3gp', '.gif'},
+        'Image': {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp', '.svg', '.ico'}
+    }
     
     @staticmethod
-    def _is_valid_audio_file(file_path: Path) -> bool:
-        """Check if file is a valid audio file."""
-        valid_extensions = {'.mp4', '.mp3', '.wav', '.opus', '.m4a'}
-        return file_path.is_file() and file_path.suffix.lower() in valid_extensions
+    def _is_valid_file(file_path: Path, filetype: str = None) -> bool:
+        """Check if file is valid for the specified filetype."""
+        if not file_path.is_file():
+            return False
+            
+        if filetype is None:
+            # Search all file types
+            all_extensions = set()
+            for extensions in UniversalSearcher.FILE_TYPE_EXTENSIONS.values():
+                all_extensions.update(extensions)
+            return file_path.suffix.lower() in all_extensions
+        
+        filetype = filetype.capitalize()
+        if filetype not in UniversalSearcher.FILE_TYPE_EXTENSIONS:
+            return False
+            
+        valid_extensions = UniversalSearcher.FILE_TYPE_EXTENSIONS[filetype]
+        return file_path.suffix.lower() in valid_extensions
     
     @staticmethod
     def _calculate_match_score(query: str, file_name: str, autocomplete: bool) -> tuple[int, str]:
@@ -201,18 +266,18 @@ class AudioSearcher:
         return -1, file_name
     
     @staticmethod
-    async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool) -> list[tuple[int, str]]:
-        """Search for audio files in a single path."""
+    async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool, filetype: str = None) -> list[tuple[int, str]]:
+        """Search for files in a single path."""
         matches = []
         
-        logger.debug(f"Searching in directory: {path}")
+        logger.debug(f"Searching in directory: {path} (filetype: {filetype or 'All'})")
         
         for file_path in path.rglob('*'):
-            if not AudioSearcher._is_valid_audio_file(file_path):
+            if not UniversalSearcher._is_valid_file(file_path, filetype):
                 continue
             
             file_name = file_path.name.lower()
-            score, _ = AudioSearcher._calculate_match_score(query, file_name, autocomplete)
+            score, _ = UniversalSearcher._calculate_match_score(query, file_name, autocomplete)
             
             if score >= 0:
                 matches.append((score, str(file_path.absolute())))
@@ -222,14 +287,41 @@ class AudioSearcher:
         
         return matches
     
+    @staticmethod
+    def _get_search_paths(data: dict, filetype: str = None) -> list:
+        """Get the search paths from settings."""
+        return data.get('file_paths', [])
+    
     @classmethod
-    async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False):
-        """Search for audio files with fuzzy matching and optional autocomplete support.
+    async def _search_paths(cls, search_paths: list, query: str, max_results: int, autocomplete: bool, filetype: str = None) -> tuple[list, int]:
+        """Search through the provided paths and return matches and file count."""
+        matches = []
+        file_count = 0
+        
+        for path_str in search_paths:
+            path = Path(path_str)
+            if not path.exists():
+                logger.warning(f"Path does not exist: {path}")
+                continue
+            
+            path_matches = await cls._search_single_path(path, query, max_results, autocomplete, filetype)
+            matches.extend(path_matches)
+            file_count += len(path_matches)
+            
+            if len(matches) >= max_results * 5:
+                break
+        
+        return matches, file_count
+    
+    @classmethod
+    async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False, filetype: str = None):
+        """Search for files with fuzzy matching and optional autocomplete support.
         
         Args:
             query: Search query string (can be None for autocomplete)
             max_results: Maximum number of results to return
             autocomplete: If True, returns results even with empty query
+            filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all)
             
         Returns:
             List of matching file paths with scores
@@ -238,30 +330,17 @@ class AudioSearcher:
             return []
             
         query = (query or '').lower().strip()
-        logger.info(f"Starting local search for query: '{query}'")
+        logger.info(f"Starting local search for query: '{query}' (filetype: {filetype or 'All'})")
         
         try:
             async with aiofiles.open('settings.json') as f:
                 content = await f.read()
                 data = json.loads(content)
-                search_paths = data.get('audio_paths', [])
-                logger.info(f"Searching in paths: {search_paths}")
                 
-                matches = []
-                file_count = 0
+                search_paths = cls._get_search_paths(data, filetype)
+                logger.info(f"Searching in paths: {search_paths}")
                 
-                for path_str in search_paths:
-                    path = Path(path_str)
-                    if not path.exists():
-                        logger.warning(f"Path does not exist: {path}")
-                        continue
-                    
-                    path_matches = await cls._search_single_path(path, query, max_results, autocomplete)
-                    matches.extend(path_matches)
-                    file_count += len(path_matches)
-                    
-                    if len(matches) >= max_results * 5:
-                        break
+                matches, file_count = await cls._search_paths(search_paths, query, max_results, autocomplete, filetype)
                 
                 logger.info(f"Scanned {file_count} files total")
                 
@@ -278,10 +357,144 @@ class AudioSearcher:
             return []
     
     @classmethod
-    async def search_local_autocomplete(cls, query: str = None, max_results: int = 10):
-        """Wrapper for autocomplete searches that returns more results and allows empty query."""
+    async def search_local_autocomplete(cls, query: str = None, max_results: int = 10, filetype: str = None):
+        """Wrapper for autocomplete searches that returns more results and allows empty query.
+        
+        Args:
+            query: Search query string (can be None for autocomplete)
+            max_results: Maximum number of results to return
+            filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all)
+        """
         # For autocomplete, we want to show more options and allow empty queries
         if not query:
             return []
         
-        return await cls.search_local(query, max_results=max_results, autocomplete=True)
+        return await cls.search_local(query, max_results=max_results, autocomplete=True, filetype=filetype)
+    
+    @staticmethod
+    async def _load_baseurl():
+        """Load baseurl from settings.json"""
+        try:
+            async with aiofiles.open('settings.json') as f:
+                content = await f.read()
+                data = json.loads(content)
+                return data.get('baseurl', 'http://localhost:8080').rstrip('/')
+        except Exception as e:
+            logger.error(f"Error loading baseurl: {e}")
+            return 'http://localhost:8080'
+    
+    @staticmethod
+    async def search_copyparty(query: str, max_results: int = 25):
+        """Search for files using copyparty API (jPOST method).
+        
+        Args:
+            query: Search query string
+            max_results: Maximum number of results to return
+            
+        Returns:
+            List of matching file URLs
+        """
+
+
+        if not query:
+            return []
+            
+        query = query.strip()
+        logger.info(f"Starting copyparty search for query: '{query}'")
+        query = f'name like "*{query}*"'
+        
+        try:
+            baseurl = await UniversalSearcher._load_baseurl()
+            password = os.getenv("COPYPARTY_R_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()
+                        logger.info(f"Copyparty response: {data}")
+                        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 it's just a string path/URL
+                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.
+        
+        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)
+
+