1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# 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}'")
# Build the query with optional extension filtering
name_query = f'name like "*{query}*"'
if file_extensions:
# Create extension filters using copyparty's .ext$ syntax
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)
|
