aboutsummaryrefslogtreecommitdiffstats
path: root/cogs/utils
diff options
context:
space:
mode:
authorsalmonik <raisersoft@proton.me>2026-07-08 02:13:35 +0200
committersalmonik <raisersoft@proton.me>2026-07-08 02:13:35 +0200
commitcd8ed8bc5793985df2c1692072a932ceee517b95 (patch)
tree42d18a4140259b5c68b4dd05b7fcb398f78b8d7f /cogs/utils
parentMerge pull request 'Voice related fixes, alongside simplifying assistant' (#5... (diff)
downloadsenator-sperma-cd8ed8bc5793985df2c1692072a932ceee517b95.tar.gz
senator-sperma-cd8ed8bc5793985df2c1692072a932ceee517b95.zip
OPEN THE RUST GATES
Signed-off-by: salmonik <raisersoft@proton.me>
Diffstat (limited to '')
-rw-r--r--cogs/utils/database.py126
-rw-r--r--cogs/utils/file_search.py167
-rw-r--r--cogs/utils/settings.py114
-rw-r--r--cogs/utils/temp.py18
-rw-r--r--cogs/utils/url_helper.py21
5 files changed, 0 insertions, 446 deletions
diff --git a/cogs/utils/database.py b/cogs/utils/database.py
deleted file mode 100644
index ea7d356..0000000
--- a/cogs/utils/database.py
+++ /dev/null
@@ -1,126 +0,0 @@
1import os
2import sqlite3
3import logging
4from contextlib import contextmanager
5
6logger = logging.getLogger(__name__)
7
8DB_PATH = "data/senator.db"
9
10_initialized = False
11
12
13def get_connection():
14 conn = sqlite3.connect(get_db_path())
15 return _ConnectionContextManager(conn)
16
17
18def get_db_path():
19 return os.getenv("SENATOR_DB_PATH", DB_PATH)
20
21
22def set_db_path(path):
23 global DB_PATH
24 DB_PATH = path
25
26
27class _ConnectionContextManager:
28 """Context manager for sqlite3 connections that auto-closes."""
29 def __init__(self, conn):
30 self.conn = conn
31
32 def __enter__(self):
33 return self.conn
34
35 def __exit__(self, exc_type, exc_val, exc_tb):
36 self.conn.close()
37 return False
38
39
40def init_database():
41 global _initialized
42 if _initialized:
43 return
44
45 _initialized = True
46
47 db_path = get_db_path()
48 db_dir = os.path.dirname(db_path)
49 if db_dir and not os.path.exists(db_dir):
50 os.makedirs(db_dir)
51
52 conn = sqlite3.connect(db_path)
53 try:
54 c = conn.cursor()
55
56 c.execute("PRAGMA foreign_keys = ON")
57
58 c.execute("""
59 CREATE TABLE IF NOT EXISTS settings (
60 key TEXT PRIMARY KEY,
61 value TEXT
62 )""")
63
64 c.execute("""
65 CREATE TABLE IF NOT EXISTS cron_jobs (
66 id INTEGER PRIMARY KEY AUTOINCREMENT,
67 cron_expression TEXT NOT NULL,
68 type TEXT NOT NULL,
69 channel_id INTEGER NOT NULL,
70 active INTEGER DEFAULT 1,
71 audio_filename TEXT,
72 message TEXT,
73 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
74 )""")
75
76 c.execute("""
77 CREATE TABLE IF NOT EXISTS voice_activity (
78 user_id TEXT PRIMARY KEY,
79 join_time REAL,
80 total_active REAL DEFAULT 0,
81 total_afk REAL DEFAULT 0,
82 is_afk INTEGER DEFAULT 0,
83 afk_start REAL,
84 active_start REAL,
85 channel_name TEXT,
86 last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
87 )""")
88
89 c.execute("CREATE INDEX IF NOT EXISTS idx_voice_activity_user_id ON voice_activity(user_id)")
90 c.execute("CREATE INDEX IF NOT EXISTS idx_cron_jobs_active ON cron_jobs(active)")
91
92 c.execute("""
93 CREATE TABLE IF NOT EXISTS triggers (
94 id INTEGER PRIMARY KEY AUTOINCREMENT,
95 name TEXT NOT NULL UNIQUE,
96 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
97 last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
98 )""")
99
100 c.execute("""
101 CREATE TABLE IF NOT EXISTS trigger_keywords (
102 id INTEGER PRIMARY KEY AUTOINCREMENT,
103 trigger_id INTEGER NOT NULL,
104 keyword TEXT NOT NULL,
105 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
106 FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE,
107 UNIQUE(trigger_id, keyword)
108 )""")
109
110 c.execute("""
111 CREATE TABLE IF NOT EXISTS trigger_responses (
112 id INTEGER PRIMARY KEY AUTOINCREMENT,
113 trigger_id INTEGER NOT NULL,
114 response_text TEXT NOT NULL,
115 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
116 FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE
117 )""")
118
119 c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_keywords_trigger_id ON trigger_keywords(trigger_id)")
120 c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_keywords_keyword ON trigger_keywords(keyword)")
121 c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_responses_trigger_id ON trigger_responses(trigger_id)")
122
123 conn.commit()
124 logger.info("Database initialized successfully")
125 finally:
126 conn.close()
diff --git a/cogs/utils/file_search.py b/cogs/utils/file_search.py
deleted file mode 100644
index 788f50c..0000000
--- a/cogs/utils/file_search.py
+++ /dev/null
@@ -1,167 +0,0 @@
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 json
7import logging
8import aiohttp
9import os
10from typing import List
11from cogs.utils.settings import get_baseurl
12
13logger = logging.getLogger(__name__)
14
15
16class UniversalSearcher:
17 """Handles searching for files via copyparty API."""
18
19 @staticmethod
20 async def search_copyparty(query: str, max_results: int = 25, file_extensions: List[str] | None = None):
21 """Search for files using copyparty API (jPOST method).
22
23 Args:
24 query: Search query string
25 max_results: Maximum number of results to return
26 file_extensions: Optional list of file extensions to filter by (e.g., ['mp3', 'wav'])
27
28 Returns:
29 List of matching file URLs
30 """
31 if not query:
32 return []
33
34 query = query.strip()
35 logger.info(f"Starting copyparty search for query: '{query}'")
36
37 tokens = query.split()
38 if not tokens:
39 return []
40
41 token_queries = []
42 for token in tokens:
43 path_query = f'path like "*{token}*"'
44 name_query = f'name like "*{token}*"'
45 token_queries.append(f"({path_query} or {name_query})")
46
47 if len(token_queries) == 1:
48 name_query = token_queries[0]
49 else:
50 name_query = " and ".join(token_queries)
51
52 if file_extensions:
53 ext_filters = [f"name like \"*.{ext.lower()}\"" for ext in file_extensions]
54 ext_query = " or ".join(ext_filters)
55 query = f"{name_query} and ({ext_query})"
56 else:
57 query = name_query
58
59 try:
60 baseurl = get_baseurl()
61 password = os.getenv("COPYPARTY_PASSWORD")
62
63 async with aiohttp.ClientSession() as session:
64 headers = {"Content-Type": "application/json"}
65 if password:
66 headers["PW"] = password
67 async with session.post(f"{baseurl}/?srch", headers=headers, data=json.dumps({"n": max_results, "q": query})) as response:
68 if response.status == 200:
69 data = await response.json()
70 files = UniversalSearcher._extract_files_from_response(data)
71 results = UniversalSearcher._convert_files_to_urls(files, baseurl, max_results)
72
73 logger.info(f"Copyparty search complete. Found {len(results)} matches.")
74 return results
75 else:
76 logger.error(f"Copyparty API returned status {response.status}")
77 return []
78
79 except aiohttp.ClientError as e:
80 logger.error(f"HTTP error during copyparty search: {e}")
81 return []
82 except Exception as e:
83 logger.error(f"Error during copyparty search: {str(e)}", exc_info=True)
84 return []
85
86 @staticmethod
87 def _extract_files_from_response(data):
88 """Extract files list from copyparty API response."""
89 if isinstance(data, dict):
90 if 'hits' in data:
91 return data['hits']
92 elif 'files' in data:
93 return data['files']
94 elif 'results' in data:
95 return data['results']
96 else:
97 return data if isinstance(data, list) else []
98 elif isinstance(data, list):
99 return data
100 else:
101 return []
102
103 @staticmethod
104 def _convert_files_to_urls(files, baseurl, max_results):
105 """Convert file info objects to URLs."""
106 results = []
107
108 for file_info in files[:max_results]:
109 if isinstance(file_info, dict):
110 file_url = UniversalSearcher._extract_url_from_dict(file_info, baseurl)
111 if file_url:
112 results.append(file_url)
113 elif isinstance(file_info, str):
114 if file_info.startswith('http'):
115 results.append(file_info)
116 else:
117 results.append(f"{baseurl}/{file_info.lstrip('/')}")
118
119 return results
120
121 @staticmethod
122 def _extract_url_from_dict(file_info, baseurl):
123 """Extract URL from file info dictionary."""
124 if 'url' in file_info:
125 return file_info['url']
126 elif 'rp' in file_info:
127 return f"{baseurl}/{file_info['rp']}"
128 elif 'path' in file_info:
129 return f"{baseurl}/{file_info['path'].lstrip('/')}"
130 elif 'name' in file_info:
131 return f"{baseurl}/{file_info['name'].lstrip('/')}"
132 else:
133 return None
134
135 @staticmethod
136 async def search_copyparty_autocomplete(query: str, max_results: int = 10):
137 """Wrapper for autocomplete searches using copyparty (general search - no file type filtering).
138
139 Args:
140 query: Search query string
141 max_results: Maximum number of results to return
142
143 Returns:
144 List of matching file URLs
145 """
146 if not query:
147 return []
148
149 return await UniversalSearcher.search_copyparty(query, max_results)
150
151 @staticmethod
152 async def search_copyparty_autocomplete_audio(query: str, max_results: int = 10):
153 """Wrapper for autocomplete searches using copyparty (audio files only).
154
155 Args:
156 query: Search query string
157 max_results: Maximum number of results to return
158
159 Returns:
160 List of matching file URLs (audio files only)
161 """
162 if not query:
163 return []
164
165 # Audio file extensions for music commands
166 audio_extensions = ['mp3', 'wav', 'flac', 'ogg', 'm4a', 'aac', 'mp4', "mkv"]
167 return await UniversalSearcher.search_copyparty(query, max_results, audio_extensions)
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 @@
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 logging
7import sqlite3
8import os
9from dataclasses import dataclass, field
10
11from cogs.utils.database import get_connection, init_database
12
13logger = logging.getLogger(__name__)
14
15DEFAULT_BASEURL = "http://localhost:3939"
16
17init_database()
18
19
20@dataclass
21class Settings:
22 volume: float = 1.0
23 baseurl: str = DEFAULT_BASEURL
24 _data: dict = field(default_factory=dict, init=False)
25
26 async def load(self) -> "Settings":
27 default_settings = {
28 "volume": 1.0,
29 "baseurl": DEFAULT_BASEURL,
30 }
31
32 try:
33 with get_connection() as conn:
34 c = conn.cursor()
35 c.execute("SELECT key, value FROM settings")
36 rows = c.fetchall()
37
38 if rows:
39 self._data = {row[0]: row[1] for row in rows}
40 self._sync_from_data()
41 logger.info("Settings loaded from database")
42 else:
43 self._data = default_settings.copy()
44 self._sync_from_data()
45 await self.save()
46 logger.info("Created default settings in database")
47 except 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 try:
56 self.volume = float(self._data.get("volume", 1.0))
57 except (ValueError, TypeError):
58 self.volume = 1.0
59 self.baseurl = self._data.get("baseurl", DEFAULT_BASEURL)
60
61 def _sync_to_data(self):
62 self._data["volume"] = str(self.volume)
63 self._data["baseurl"] = self.baseurl
64
65 async def save(self):
66 self._sync_to_data()
67 with get_connection() as conn:
68 c = conn.cursor()
69 c.execute("PRAGMA foreign_keys = ON")
70 for key, value in self._data.items():
71 c.execute(
72 "INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)",
73 (key, value),
74 )
75 conn.commit()
76
77
78_settings_instance: Settings | None = None
79
80
81async def load_settings() -> Settings:
82 global _settings_instance
83 _settings_instance = Settings()
84 await _settings_instance.load()
85 return _settings_instance
86
87
88async def save_settings():
89 if _settings_instance:
90 await _settings_instance.save()
91
92
93def get_settings() -> Settings | None:
94 return _settings_instance
95
96
97def get_volume() -> float:
98 return _settings_instance.volume if _settings_instance else 1.0
99
100
101def get_baseurl() -> str:
102 url = os.getenv("COPYPARTY_URL")
103 if not url:
104 logger.warning("COPYPARTY_URL not set! Using default http://localhost:3939")
105 return DEFAULT_BASEURL
106 return url.rstrip("/")
107
108
109def get_uploadurl() -> str:
110 url = os.getenv("COPYPARTY_URL_UPLOAD")
111 if not url:
112 logger.warning("COPYPARTY_URL_UPLOAD not set! Using default http://localhost:3939")
113 return DEFAULT_BASEURL
114 return url.rstrip("/")
diff --git a/cogs/utils/temp.py b/cogs/utils/temp.py
deleted file mode 100644
index 0b2c8d7..0000000
--- a/cogs/utils/temp.py
+++ /dev/null
@@ -1,18 +0,0 @@
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
diff --git a/cogs/utils/url_helper.py b/cogs/utils/url_helper.py
deleted file mode 100644
index eac06f1..0000000
--- a/cogs/utils/url_helper.py
+++ /dev/null
@@ -1,21 +0,0 @@
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
6
7def clean_url_for_ytdlp(url: str) -> str:
8 """Clean URL by removing everything after the first '&'.
9
10 This prevents yt-dlp from interpreting additional query parameters
11 (like &list=...) as playlist URLs.
12
13 Example:
14 Input: https://www.youtube.com/watch?v=lPGipwoJiOM&list=RDlPGipwoJiOM&
15 Output: https://www.youtube.com/watch?v=lPGipwoJiOM
16 """
17 # Find the first '&' and truncate
18 first_ampersand = url.find('&')
19 if first_ampersand != -1:
20 return url[:first_ampersand]
21 return url
generated by cgit v1.2.3 (git 2.46.0) at 2026-11-08 09:13:19 +0000