aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/utils/file_search.py
diff options
context:
space:
mode:
Diffstat (limited to 'cogs/utils/file_search.py')
-rw-r--r--cogs/utils/file_search.py167
1 files changed, 0 insertions, 167 deletions
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 @@
1# This file is part of Senator Sperma
2# Licensed under the GPL-3.0 License
3# Copyright (C) 2026 Salmonik @ salmonik@tuta.io
4# https://codeberg.org/salmonik/senator-sperma
5
6import json
7import logging
8import aiohttp
9import os
10from typing import List
11from cogs.utils.settings import get_baseurl
12
13logger = logging.getLogger(__name__)
14
15
16class UniversalSearcher:
17 """Handles searching for files via copyparty API."""
18
19 @staticmethod
20 async def search_copyparty(query: str, max_results: int = 25, file_extensions: List[str] | None = None):
21 """Search for files using copyparty API (jPOST method).
22
23 Args:
24 query: Search query string
25 max_results: Maximum number of results to return
26 file_extensions: Optional list of file extensions to filter by (e.g., ['mp3', 'wav'])
27
28 Returns:
29 List of matching file URLs
30 """
31 if not query:
32 return []
33
34 query = query.strip()
35 logger.info(f"Starting copyparty search for query: '{query}'")
36
37 tokens = query.split()
38 if not tokens:
39 return []
40
41 token_queries = []
42 for token in tokens:
43 path_query = f'path like "*{token}*"'
44 name_query = f'name like "*{token}*"'
45 token_queries.append(f"({path_query} or {name_query})")
46
47 if len(token_queries) == 1:
48 name_query = token_queries[0]
49 else:
50 name_query = " and ".join(token_queries)
51
52 if file_extensions:
53 ext_filters = [f"name like \"*.{ext.lower()}\"" for ext in file_extensions]
54 ext_query = " or ".join(ext_filters)
55 query = f"{name_query} and ({ext_query})"
56 else:
57 query = name_query
58
59 try:
60 baseurl = get_baseurl()
61 password = os.getenv("COPYPARTY_PASSWORD")
62
63 async with aiohttp.ClientSession() as session:
64 headers = {"Content-Type": "application/json"}
65 if password:
66 headers["PW"] = password
67 async with session.post(f"{baseurl}/?srch", headers=headers, data=json.dumps({"n": max_results, "q": query})) as response:
68 if response.status == 200:
69 data = await response.json()
70 files = UniversalSearcher._extract_files_from_response(data)
71 results = UniversalSearcher._convert_files_to_urls(files, baseurl, max_results)
72
73 logger.info(f"Copyparty search complete. Found {len(results)} matches.")
74 return results
75 else:
76 logger.error(f"Copyparty API returned status {response.status}")
77 return []
78
79 except aiohttp.ClientError as e:
80 logger.error(f"HTTP error during copyparty search: {e}")
81 return []
82 except Exception as e:
83 logger.error(f"Error during copyparty search: {str(e)}", exc_info=True)
84 return []
85
86 @staticmethod
87 def _extract_files_from_response(data):
88 """Extract files list from copyparty API response."""
89 if isinstance(data, dict):
90 if 'hits' in data:
91 return data['hits']
92 elif 'files' in data:
93 return data['files']
94 elif 'results' in data:
95 return data['results']
96 else:
97 return data if isinstance(data, list) else []
98 elif isinstance(data, list):
99 return data
100 else:
101 return []
102
103 @staticmethod
104 def _convert_files_to_urls(files, baseurl, max_results):
105 """Convert file info objects to URLs."""
106 results = []
107
108 for file_info in files[:max_results]:
109 if isinstance(file_info, dict):
110 file_url = UniversalSearcher._extract_url_from_dict(file_info, baseurl)
111 if file_url:
112 results.append(file_url)
113 elif isinstance(file_info, str):
114 if file_info.startswith('http'):
115 results.append(file_info)
116 else:
117 results.append(f"{baseurl}/{file_info.lstrip('/')}")
118
119 return results
120
121 @staticmethod
122 def _extract_url_from_dict(file_info, baseurl):
123 """Extract URL from file info dictionary."""
124 if 'url' in file_info:
125 return file_info['url']
126 elif 'rp' in file_info:
127 return f"{baseurl}/{file_info['rp']}"
128 elif 'path' in file_info:
129 return f"{baseurl}/{file_info['path'].lstrip('/')}"
130 elif 'name' in file_info:
131 return f"{baseurl}/{file_info['name'].lstrip('/')}"
132 else:
133 return None
134
135 @staticmethod
136 async def search_copyparty_autocomplete(query: str, max_results: int = 10):
137 """Wrapper for autocomplete searches using copyparty (general search - no file type filtering).
138
139 Args:
140 query: Search query string
141 max_results: Maximum number of results to return
142
143 Returns:
144 List of matching file URLs
145 """
146 if not query:
147 return []
148
149 return await UniversalSearcher.search_copyparty(query, max_results)
150
151 @staticmethod
152 async def search_copyparty_autocomplete_audio(query: str, max_results: int = 10):
153 """Wrapper for autocomplete searches using copyparty (audio files only).
154
155 Args:
156 query: Search query string
157 max_results: Maximum number of results to return
158
159 Returns:
160 List of matching file URLs (audio files only)
161 """
162 if not query:
163 return []
164
165 # Audio file extensions for music commands
166 audio_extensions = ['mp3', 'wav', 'flac', 'ogg', 'm4a', 'aac', 'mp4', "mkv"]
167 return await UniversalSearcher.search_copyparty(query, max_results, audio_extensions)
generated by cgit v1.2.3 (git 2.46.0) at 2026-11-08 18:49:44 +0000