From a2b5b10625126b7e98643af9fe40569c9e004047 Mon Sep 17 00:00:00 2001 From: salmonik Date: Sat, 18 Apr 2026 00:24:50 +0200 Subject: refactor + preparing bot for public use Signed-off-by: salmonik --- cogs/utils/file_search.py | 7 +-- cogs/utils/settings.py | 108 +++++++++++++++++++++++++++++----------------- cogs/utils/temp.py | 18 ++++++++ 3 files changed, 89 insertions(+), 44 deletions(-) create mode 100644 cogs/utils/temp.py (limited to 'cogs/utils') 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 import aiohttp import os from typing import List -from dotenv import load_dotenv from cogs.utils.settings import get_baseurl -load_dotenv() - logger = logging.getLogger(__name__) @@ -20,7 +17,7 @@ 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): + 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: @@ -36,7 +33,7 @@ class UniversalSearcher: 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: 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 import logging import aiofiles import os +from dataclasses import dataclass, field logger = logging.getLogger(__name__) -_settings = None settings_file = "data/settings.json" -async def load_settings(): - global _settings - global settings_file - - default_settings = { - 'volume': 1.0, - 'baseurl': 'http://localhost:8080' - } - - try: - if os.path.exists(settings_file): - async with aiofiles.open(settings_file, 'r') as f: - content = await f.read() - if content.strip(): - _settings = json.loads(content) - logger.info("Settings loaded into memory") - else: - _settings = default_settings - await save_settings() - logger.warning("Empty settings.json, reset to defaults") - else: - _settings = default_settings - await save_settings() - logger.info("Created default settings.json") - except (json.JSONDecodeError, Exception) as e: - logger.error(f"Error loading settings: {e}") - _settings = default_settings - - return _settings + +@dataclass +class Settings: + volume: float = 1.0 + baseurl: str = "http://localhost:8080" + _data: dict = field(default_factory=dict, init=False) + + async def load(self) -> "Settings": + default_settings = { + "volume": 1.0, + "baseurl": "http://localhost:8080", + } + + try: + if os.path.exists(settings_file): + async with aiofiles.open(settings_file, "r") as f: + content = await f.read() + if content.strip(): + self._data = json.loads(content) + self._sync_from_data() + logger.info("Settings loaded into memory") + else: + self._data = default_settings.copy() + self._sync_from_data() + await self.save() + logger.warning("Empty settings.json, reset to defaults") + else: + self._data = default_settings.copy() + self._sync_from_data() + await self.save() + logger.info("Created default settings.json") + except (json.JSONDecodeError, Exception) as e: + logger.error(f"Error loading settings: {e}") + self._data = default_settings.copy() + self._sync_from_data() + + return self + + def _sync_from_data(self): + self.volume = self._data.get("volume", 1.0) + self.baseurl = self._data.get("baseurl", "http://localhost:8080") + + def _sync_to_data(self): + self._data["volume"] = self.volume + self._data["baseurl"] = self.baseurl + + async def save(self): + self._sync_to_data() + async with aiofiles.open(settings_file, "w") as f: + await f.write(json.dumps(self._data, indent=2)) + + +_settings_instance: Settings | None = None + + +async def load_settings() -> Settings: + global _settings_instance + _settings_instance = Settings() + await _settings_instance.load() + return _settings_instance async def save_settings(): - if _settings: - async with aiofiles.open(settings_file, 'w') as f: - await f.write(json.dumps(_settings, indent=2)) + if _settings_instance: + await _settings_instance.save() -def get_settings(): - return _settings +def get_settings() -> Settings | None: + return _settings_instance -def get_volume(): - return _settings.get('volume', 1.0) if _settings else 1.0 +def get_volume() -> float: + return _settings_instance.volume if _settings_instance else 1.0 -def get_baseurl(): - return _settings.get('baseurl', 'http://localhost:8080').rstrip('/') if _settings else 'http://localhost:8080' \ No newline at end of file +def get_baseurl() -> str: + 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 @@ +# 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 os +import tempfile + +def get_temp_dir(subdir: str = "senator_sperma") -> str: + """Get or create a temp subdirectory. + + Uses system temp directory (/tmp on Linux, %TEMP% on Windows). + OS handles cleanup automatically. + """ + base_temp = tempfile.gettempdir() + temp_path = os.path.join(base_temp, subdir) + os.makedirs(temp_path, exist_ok=True) + return temp_path \ No newline at end of file -- cgit v1.2.3