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
|
# 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("/")
|
