From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- cogs/utils/settings.py | 114 ------------------------------------------------- 1 file changed, 114 deletions(-) delete mode 100644 cogs/utils/settings.py (limited to 'cogs/utils/settings.py') diff --git a/cogs/utils/settings.py b/cogs/utils/settings.py deleted file mode 100644 index b1a585c..0000000 --- a/cogs/utils/settings.py +++ /dev/null @@ -1,114 +0,0 @@ -# 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 logging -import sqlite3 -import os -from dataclasses import dataclass, field - -from cogs.utils.database import get_connection, init_database - -logger = logging.getLogger(__name__) - -DEFAULT_BASEURL = "http://localhost:3939" - -init_database() - - -@dataclass -class Settings: - volume: float = 1.0 - baseurl: str = DEFAULT_BASEURL - _data: dict = field(default_factory=dict, init=False) - - async def load(self) -> "Settings": - default_settings = { - "volume": 1.0, - "baseurl": DEFAULT_BASEURL, - } - - try: - with get_connection() as conn: - c = conn.cursor() - c.execute("SELECT key, value FROM settings") - rows = c.fetchall() - - if rows: - self._data = {row[0]: row[1] for row in rows} - self._sync_from_data() - logger.info("Settings loaded from database") - else: - self._data = default_settings.copy() - self._sync_from_data() - await self.save() - logger.info("Created default settings in database") - except 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): - try: - self.volume = float(self._data.get("volume", 1.0)) - except (ValueError, TypeError): - self.volume = 1.0 - self.baseurl = self._data.get("baseurl", DEFAULT_BASEURL) - - def _sync_to_data(self): - self._data["volume"] = str(self.volume) - self._data["baseurl"] = self.baseurl - - async def save(self): - self._sync_to_data() - with get_connection() as conn: - c = conn.cursor() - c.execute("PRAGMA foreign_keys = ON") - for key, value in self._data.items(): - c.execute( - "INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", - (key, value), - ) - conn.commit() - - -_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_instance: - await _settings_instance.save() - - -def get_settings() -> Settings | None: - return _settings_instance - - -def get_volume() -> float: - return _settings_instance.volume if _settings_instance else 1.0 - - -def get_baseurl() -> str: - url = os.getenv("COPYPARTY_URL") - if not url: - logger.warning("COPYPARTY_URL not set! Using default http://localhost:3939") - return DEFAULT_BASEURL - return url.rstrip("/") - - -def get_uploadurl() -> str: - url = os.getenv("COPYPARTY_URL_UPLOAD") - if not url: - logger.warning("COPYPARTY_URL_UPLOAD not set! Using default http://localhost:3939") - return DEFAULT_BASEURL - return url.rstrip("/") -- cgit v1.2.3