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
|
# 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 dotenv import load_dotenv
from cogs.utils.settings import get_baseurl
load_dotenv()
logger = logging.getLogger(__name__)
class UniversalSearcher:
"""Handles searching for files via copyparty API."""
@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 = 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()
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 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)
|
