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/database.py | 126 ---------------------------------- cogs/utils/file_search.py | 167 ---------------------------------------------- cogs/utils/settings.py | 114 ------------------------------- cogs/utils/temp.py | 18 ----- cogs/utils/url_helper.py | 21 ------ 5 files changed, 446 deletions(-) delete mode 100644 cogs/utils/database.py delete mode 100644 cogs/utils/file_search.py delete mode 100644 cogs/utils/settings.py delete mode 100644 cogs/utils/temp.py delete mode 100644 cogs/utils/url_helper.py (limited to 'cogs/utils') 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 @@ -import os -import sqlite3 -import logging -from contextlib import contextmanager - -logger = logging.getLogger(__name__) - -DB_PATH = "data/senator.db" - -_initialized = False - - -def get_connection(): - conn = sqlite3.connect(get_db_path()) - return _ConnectionContextManager(conn) - - -def get_db_path(): - return os.getenv("SENATOR_DB_PATH", DB_PATH) - - -def set_db_path(path): - global DB_PATH - DB_PATH = path - - -class _ConnectionContextManager: - """Context manager for sqlite3 connections that auto-closes.""" - def __init__(self, conn): - self.conn = conn - - def __enter__(self): - return self.conn - - def __exit__(self, exc_type, exc_val, exc_tb): - self.conn.close() - return False - - -def init_database(): - global _initialized - if _initialized: - return - - _initialized = True - - db_path = get_db_path() - db_dir = os.path.dirname(db_path) - if db_dir and not os.path.exists(db_dir): - os.makedirs(db_dir) - - conn = sqlite3.connect(db_path) - try: - c = conn.cursor() - - c.execute("PRAGMA foreign_keys = ON") - - c.execute(""" - CREATE TABLE IF NOT EXISTS settings ( - key TEXT PRIMARY KEY, - value TEXT - )""") - - c.execute(""" - CREATE TABLE IF NOT EXISTS cron_jobs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - cron_expression TEXT NOT NULL, - type TEXT NOT NULL, - channel_id INTEGER NOT NULL, - active INTEGER DEFAULT 1, - audio_filename TEXT, - message TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - )""") - - c.execute(""" - CREATE TABLE IF NOT EXISTS voice_activity ( - user_id TEXT PRIMARY KEY, - join_time REAL, - total_active REAL DEFAULT 0, - total_afk REAL DEFAULT 0, - is_afk INTEGER DEFAULT 0, - afk_start REAL, - active_start REAL, - channel_name TEXT, - last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP - )""") - - c.execute("CREATE INDEX IF NOT EXISTS idx_voice_activity_user_id ON voice_activity(user_id)") - c.execute("CREATE INDEX IF NOT EXISTS idx_cron_jobs_active ON cron_jobs(active)") - - c.execute(""" - CREATE TABLE IF NOT EXISTS triggers ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL UNIQUE, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP - )""") - - c.execute(""" - CREATE TABLE IF NOT EXISTS trigger_keywords ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - trigger_id INTEGER NOT NULL, - keyword TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE, - UNIQUE(trigger_id, keyword) - )""") - - c.execute(""" - CREATE TABLE IF NOT EXISTS trigger_responses ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - trigger_id INTEGER NOT NULL, - response_text TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE - )""") - - c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_keywords_trigger_id ON trigger_keywords(trigger_id)") - c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_keywords_keyword ON trigger_keywords(keyword)") - c.execute("CREATE INDEX IF NOT EXISTS idx_trigger_responses_trigger_id ON trigger_responses(trigger_id)") - - conn.commit() - logger.info("Database initialized successfully") - finally: - 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 @@ -# 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 json -import logging -import aiohttp -import os -from typing import List -from cogs.utils.settings import get_baseurl - -logger = logging.getLogger(__name__) - - -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 = None): - """Search for files using copyparty API (jPOST method). - - Args: - query: Search query string - max_results: Maximum number of results to return - file_extensions: Optional list of file extensions to filter by (e.g., ['mp3', 'wav']) - - Returns: - List of matching file URLs - """ - if not query: - return [] - - query = query.strip() - logger.info(f"Starting copyparty search for query: '{query}'") - - tokens = query.split() - if not tokens: - return [] - - token_queries = [] - for token in tokens: - path_query = f'path like "*{token}*"' - name_query = f'name like "*{token}*"' - token_queries.append(f"({path_query} or {name_query})") - - if len(token_queries) == 1: - name_query = token_queries[0] - else: - name_query = " and ".join(token_queries) - - if file_extensions: - ext_filters = [f"name like \"*.{ext.lower()}\"" for ext in file_extensions] - ext_query = " or ".join(ext_filters) - query = f"{name_query} and ({ext_query})" - else: - query = name_query - - try: - baseurl = get_baseurl() - password = os.getenv("COPYPARTY_PASSWORD") - - async with aiohttp.ClientSession() as session: - headers = {"Content-Type": "application/json"} - if password: - headers["PW"] = password - async with session.post(f"{baseurl}/?srch", headers=headers, data=json.dumps({"n": max_results, "q": query})) as response: - if response.status == 200: - data = await response.json() - files = UniversalSearcher._extract_files_from_response(data) - results = UniversalSearcher._convert_files_to_urls(files, baseurl, max_results) - - logger.info(f"Copyparty search complete. Found {len(results)} matches.") - return results - else: - logger.error(f"Copyparty API returned status {response.status}") - return [] - - except aiohttp.ClientError as e: - logger.error(f"HTTP error during copyparty search: {e}") - return [] - except Exception as e: - logger.error(f"Error during copyparty search: {str(e)}", exc_info=True) - return [] - - @staticmethod - def _extract_files_from_response(data): - """Extract files list from copyparty API response.""" - if isinstance(data, dict): - if 'hits' in data: - return data['hits'] - elif 'files' in data: - return data['files'] - elif 'results' in data: - return data['results'] - else: - return data if isinstance(data, list) else [] - elif isinstance(data, list): - return data - else: - return [] - - @staticmethod - def _convert_files_to_urls(files, baseurl, max_results): - """Convert file info objects to URLs.""" - results = [] - - for file_info in files[:max_results]: - if isinstance(file_info, dict): - file_url = UniversalSearcher._extract_url_from_dict(file_info, baseurl) - if file_url: - results.append(file_url) - elif isinstance(file_info, str): - if file_info.startswith('http'): - results.append(file_info) - else: - results.append(f"{baseurl}/{file_info.lstrip('/')}") - - return results - - @staticmethod - def _extract_url_from_dict(file_info, baseurl): - """Extract URL from file info dictionary.""" - if 'url' in file_info: - return file_info['url'] - elif 'rp' in file_info: - return f"{baseurl}/{file_info['rp']}" - elif 'path' in file_info: - return f"{baseurl}/{file_info['path'].lstrip('/')}" - elif 'name' in file_info: - return f"{baseurl}/{file_info['name'].lstrip('/')}" - else: - return None - - @staticmethod - async def search_copyparty_autocomplete(query: str, max_results: int = 10): - """Wrapper for autocomplete searches using copyparty (general search - no file type filtering). - - Args: - query: Search query string - max_results: Maximum number of results to return - - Returns: - List of matching file URLs - """ - if not query: - return [] - - return await UniversalSearcher.search_copyparty(query, max_results) - - @staticmethod - async def search_copyparty_autocomplete_audio(query: str, max_results: int = 10): - """Wrapper for autocomplete searches using copyparty (audio files only). - - Args: - query: Search query string - max_results: Maximum number of results to return - - Returns: - List of matching file URLs (audio files only) - """ - if not query: - return [] - - # Audio file extensions for music commands - audio_extensions = ['mp3', 'wav', 'flac', 'ogg', 'm4a', 'aac', 'mp4', "mkv"] - 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 @@ -# 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("/") 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 @@ -# 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 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 @@ -# 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 - - -def clean_url_for_ytdlp(url: str) -> str: - """Clean URL by removing everything after the first '&'. - - This prevents yt-dlp from interpreting additional query parameters - (like &list=...) as playlist URLs. - - Example: - Input: https://www.youtube.com/watch?v=lPGipwoJiOM&list=RDlPGipwoJiOM& - Output: https://www.youtube.com/watch?v=lPGipwoJiOM - """ - # Find the first '&' and truncate - first_ampersand = url.find('&') - if first_ampersand != -1: - return url[:first_ampersand] - return url -- cgit v1.2.3