aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/utils
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--cogs/utils/file_search.py7
-rw-r--r--cogs/utils/settings.py108
-rw-r--r--cogs/utils/temp.py18
3 files changed, 89 insertions, 44 deletions
diff --git a/cogs/utils/file_search.py b/cogs/utils/file_search.py
index 2108074..8111216 100644
--- a/cogs/utils/file_search.py
+++ b/cogs/utils/file_search.py
@@ -8,11 +8,8 @@ import logging
8import aiohttp 8import aiohttp
9import os 9import os
10from typing import List 10from typing import List
11from dotenv import load_dotenv
12from cogs.utils.settings import get_baseurl 11from cogs.utils.settings import get_baseurl
13 12
14load_dotenv()
15
16logger = logging.getLogger(__name__) 13logger = logging.getLogger(__name__)
17 14
18 15
@@ -20,7 +17,7 @@ class UniversalSearcher:
20 """Handles searching for files via copyparty API.""" 17 """Handles searching for files via copyparty API."""
21 18
22 @staticmethod 19 @staticmethod
23 async def search_copyparty(query: str, max_results: int = 25, file_extensions: List[str] = None): 20 async def search_copyparty(query: str, max_results: int = 25, file_extensions: List[str] | None = None):
24 """Search for files using copyparty API (jPOST method). 21 """Search for files using copyparty API (jPOST method).
25 22
26 Args: 23 Args:
@@ -36,7 +33,7 @@ class UniversalSearcher:
36 33
37 query = query.strip() 34 query = query.strip()
38 logger.info(f"Starting copyparty search for query: '{query}'") 35 logger.info(f"Starting copyparty search for query: '{query}'")
39 36
40 # Build the query with optional extension filtering 37 # Build the query with optional extension filtering
41 name_query = f'name like "*{query}*"' 38 name_query = f'name like "*{query}*"'
42 if file_extensions: 39 if file_extensions:
diff --git a/cogs/utils/settings.py b/cogs/utils/settings.py
index 0a1716f..984ec16 100644
--- a/cogs/utils/settings.py
+++ b/cogs/utils/settings.py
@@ -7,56 +7,86 @@ import json
7import logging 7import logging
8import aiofiles 8import aiofiles
9import os 9import os
10from dataclasses import dataclass, field
10 11
11logger = logging.getLogger(__name__) 12logger = logging.getLogger(__name__)
12 13
13_settings = None
14settings_file = "data/settings.json" 14settings_file = "data/settings.json"
15 15
16async def load_settings(): 16
17 global _settings 17@dataclass
18 global settings_file 18class Settings:
19 19 volume: float = 1.0
20 default_settings = { 20 baseurl: str = "http://localhost:8080"
21 'volume': 1.0, 21 _data: dict = field(default_factory=dict, init=False)
22 'baseurl': 'http://localhost:8080' 22
23 } 23 async def load(self) -> "Settings":
24 24 default_settings = {
25 try: 25 "volume": 1.0,
26 if os.path.exists(settings_file): 26 "baseurl": "http://localhost:8080",
27 async with aiofiles.open(settings_file, 'r') as f: 27 }
28 content = await f.read() 28
29 if content.strip(): 29 try:
30 _settings = json.loads(content) 30 if os.path.exists(settings_file):
31 logger.info("Settings loaded into memory") 31 async with aiofiles.open(settings_file, "r") as f:
32 else: 32 content = await f.read()
33 _settings = default_settings 33 if content.strip():
34 await save_settings() 34 self._data = json.loads(content)
35 logger.warning("Empty settings.json, reset to defaults") 35 self._sync_from_data()
36 else: 36 logger.info("Settings loaded into memory")
37 _settings = default_settings 37 else:
38 await save_settings() 38 self._data = default_settings.copy()
39 logger.info("Created default settings.json") 39 self._sync_from_data()
40 except (json.JSONDecodeError, Exception) as e: 40 await self.save()
41 logger.error(f"Error loading settings: {e}") 41 logger.warning("Empty settings.json, reset to defaults")
42 _settings = default_settings 42 else:
43 43 self._data = default_settings.copy()
44 return _settings 44 self._sync_from_data()
45 await self.save()
46 logger.info("Created default settings.json")
47 except (json.JSONDecodeError, Exception) as e:
48 logger.error(f"Error loading settings: {e}")
49 self._data = default_settings.copy()
50 self._sync_from_data()
51
52 return self
53
54 def _sync_from_data(self):
55 self.volume = self._data.get("volume", 1.0)
56 self.baseurl = self._data.get("baseurl", "http://localhost:8080")
57
58 def _sync_to_data(self):
59 self._data["volume"] = self.volume
60 self._data["baseurl"] = self.baseurl
61
62 async def save(self):
63 self._sync_to_data()
64 async with aiofiles.open(settings_file, "w") as f:
65 await f.write(json.dumps(self._data, indent=2))
66
67
68_settings_instance: Settings | None = None
69
70
71async def load_settings() -> Settings:
72 global _settings_instance
73 _settings_instance = Settings()
74 await _settings_instance.load()
75 return _settings_instance
45 76
46 77
47async def save_settings(): 78async def save_settings():
48 if _settings: 79 if _settings_instance:
49 async with aiofiles.open(settings_file, 'w') as f: 80 await _settings_instance.save()
50 await f.write(json.dumps(_settings, indent=2))
51 81
52 82
53def get_settings(): 83def get_settings() -> Settings | None:
54 return _settings 84 return _settings_instance
55 85
56 86
57def get_volume(): 87def get_volume() -> float:
58 return _settings.get('volume', 1.0) if _settings else 1.0 88 return _settings_instance.volume if _settings_instance else 1.0
59 89
60 90
61def get_baseurl(): 91def get_baseurl() -> str:
62 return _settings.get('baseurl', 'http://localhost:8080').rstrip('/') if _settings else 'http://localhost:8080' \ No newline at end of file 92 return _settings_instance.baseurl.rstrip("/") if _settings_instance else "http://localhost:8080"
diff --git a/cogs/utils/temp.py b/cogs/utils/temp.py
new file mode 100644
index 0000000..528d6a3
--- /dev/null
+++ b/cogs/utils/temp.py
@@ -0,0 +1,18 @@
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 os
7import tempfile
8
9def get_temp_dir(subdir: str = "senator_sperma") -> str:
10 """Get or create a temp subdirectory.
11
12 Uses system temp directory (/tmp on Linux, %TEMP% on Windows).
13 OS handles cleanup automatically.
14 """
15 base_temp = tempfile.gettempdir()
16 temp_path = os.path.join(base_temp, subdir)
17 os.makedirs(temp_path, exist_ok=True)
18 return temp_path \ No newline at end of file