From c0053ca98912901b5d387d58fbc57f9b7da91f38 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 18 Mar 2026 15:29:51 +0100 Subject: voice doesn't work due to DAVE, but we got copyparty search --- cogs/jarvis.py | 3 +- cogs/math.py | 35 ---- cogs/math_commands.py | 35 ++++ cogs/music/play_commands.py | 5 +- cogs/music/search_utils.py | 287 ------------------------- cogs/search.py | 73 +++++++ cogs/tts.py | 25 ++- cogs/utils/file_search.py | 500 ++++++++++++++++++++++++++++++++++++++++++++ cogs/vclogs.py | 55 +++-- cogs/voice.py | 140 ++++++++++--- 10 files changed, 764 insertions(+), 394 deletions(-) delete mode 100644 cogs/math.py create mode 100644 cogs/math_commands.py delete mode 100644 cogs/music/search_utils.py create mode 100644 cogs/search.py create mode 100644 cogs/utils/file_search.py (limited to 'cogs') diff --git a/cogs/jarvis.py b/cogs/jarvis.py index b199c8a..5417c5e 100644 --- a/cogs/jarvis.py +++ b/cogs/jarvis.py @@ -57,7 +57,8 @@ class Jarvis(commands.Cog): def _build_messages(message): messages = [{ "role": "system", - "content": "your task is to tell the program what to do and how to respond after receiving a message.\n" + "content": "You are an alter ego of Senator Sperma Armstrong.\n" + "your task is to tell the program what to do and how to respond after receiving a message.\n" "message field has to be filled out in polish.\n" "act like jarvis, be a personal assistant to user, do not use emojis\n" "do not ask the user if they want to do an action. the user will tell you that" diff --git a/cogs/math.py b/cogs/math.py deleted file mode 100644 index bf164b3..0000000 --- a/cogs/math.py +++ /dev/null @@ -1,35 +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 asyncio - -from discord import ApplicationContext -from discord.ext import commands - -logger = logging.getLogger(__name__) - - -class Math(commands.Cog): - def __init__(self, bot): - self.bot = bot - - @commands.slash_command( - name="math", - description="wrug szyskih matfisuf" - ) - async def math(self, ctx: ApplicationContext, equation: str): - try: - math = await asyncio.create_subprocess_exec(f"qalc -e '{equation}'") - result = await math.wait() - await ctx.respond(f"{equation} = {result.decode('utf-8').strip()}") - logger.info(f"Math command executed: {equation}") - except Exception as e: - await ctx.respond(e, ephemeral=True) - logger.error(f"Error in math command: {e}", exc_info=True) - - -def setup(bot): - bot.add_cog(Math(bot)) diff --git a/cogs/math_commands.py b/cogs/math_commands.py new file mode 100644 index 0000000..bf164b3 --- /dev/null +++ b/cogs/math_commands.py @@ -0,0 +1,35 @@ +# 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 asyncio + +from discord import ApplicationContext +from discord.ext import commands + +logger = logging.getLogger(__name__) + + +class Math(commands.Cog): + def __init__(self, bot): + self.bot = bot + + @commands.slash_command( + name="math", + description="wrug szyskih matfisuf" + ) + async def math(self, ctx: ApplicationContext, equation: str): + try: + math = await asyncio.create_subprocess_exec(f"qalc -e '{equation}'") + result = await math.wait() + await ctx.respond(f"{equation} = {result.decode('utf-8').strip()}") + logger.info(f"Math command executed: {equation}") + except Exception as e: + await ctx.respond(e, ephemeral=True) + logger.error(f"Error in math command: {e}", exc_info=True) + + +def setup(bot): + bot.add_cog(Math(bot)) diff --git a/cogs/music/play_commands.py b/cogs/music/play_commands.py index b25ce84..e3ca4c1 100644 --- a/cogs/music/play_commands.py +++ b/cogs/music/play_commands.py @@ -9,7 +9,8 @@ import mutagen from typing import List, Tuple, Callable, Awaitable from os import PathLike -from cogs.music import search_utils, audio_playback, media_processing +from cogs.music import audio_playback, media_processing +from cogs.utils import file_search as search_utils from discord.ext import commands from discord.commands import OptionChoice import logging @@ -110,7 +111,7 @@ class PlayCommands(commands.Cog): """Autocomplete handler for search queries.""" query = ctx.value.lower() try: - results = await search_utils.AudioSearcher.search_local_autocomplete(query, max_results=25) + results = await search_utils.UniversalSearcher.search_local_autocomplete(query, max_results=25, filetype="Audio") choices = [] self.path_mapping = {} # Reset the mapping for each new query diff --git a/cogs/music/search_utils.py b/cogs/music/search_utils.py deleted file mode 100644 index bc9e335..0000000 --- a/cogs/music/search_utils.py +++ /dev/null @@ -1,287 +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 re -from pathlib import Path -import aiofiles -from rapidfuzz import fuzz - -logger = logging.getLogger(__name__) - -class SearchUtils: - """Utilities for searching and scoring audio files.""" - - @staticmethod - def normalize_name(p: Path, keep_ext=False): - """Preprocess filenames: lower, replace separators, strip extensions optionally.""" - name = str(p.name if keep_ext else p.stem) - return name.lower().replace('_', ' ').replace('-', ' ') - - @staticmethod - def length_penalty(query, candidate): - """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0].""" - ql = len(query) - cl = len(candidate) - if ql == 0: - return 1.0 - ratio = cl / ql - # Prefer candidate lengths near query length; gentle falloff otherwise - return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2)) - - @staticmethod - def combined_score(query, candidate): - """Calculate a combined fuzzy matching score using multiple strategies. - - Args: - query: The search query - candidate: The candidate string to score against - - Returns: - float: Score between 0-100, where higher means better match - """ - if not query or not candidate: - return 0.0 - - query = query.lower().strip() - candidate = candidate.lower() - - # Check for direct matches first - direct_match_score = SearchUtils._check_direct_matches(query, candidate) - if direct_match_score is not None: - return direct_match_score - - # Handle short queries specially - short_query_score = SearchUtils._handle_short_queries(query, candidate) - if short_query_score is not None: - return short_query_score - - # For longer queries, use complex scoring - return SearchUtils._calculate_complex_score(query, candidate) - - @staticmethod - def _check_direct_matches(query: str, candidate: str) -> float: - """Check for direct substring matches.""" - if query in candidate: - base_score = 100 - if re.search(rf'\b{re.escape(query)}\b', candidate): - base_score = 110 - return min(100, base_score) - return 0 - - @staticmethod - def _handle_short_queries(query: str, candidate: str) -> float: - """Handle scoring for short queries (2-3 chars).""" - if len(query) <= 3: - if re.search(rf'\b{re.escape(query)}', candidate): - return 90 - if query in candidate: - return 80 - return 0 - - @staticmethod - def _calculate_complex_score(query: str, candidate: str) -> float: - """Calculate complex score for longer queries.""" - query_words = {word for word in re.split(r'\W+', query) if len(word) > 1} - if not query_words: - return 0.0 - - candidate_words = set(re.split(r'\W+', candidate)) - word_ratio = SearchUtils._calculate_word_ratio(query_words, candidate_words) - - if word_ratio == 0: - return 0.0 - - # Calculate various similarity metrics - similarity_scores = SearchUtils._get_similarity_scores(query, candidate) - word_scores = SearchUtils._get_word_scores(query, candidate, query_words, candidate_words) - - # Calculate final weighted score - final_score = SearchUtils._calculate_weighted_score(similarity_scores, word_scores, word_ratio) - - # Apply penalties and boosts - final_score = SearchUtils._apply_adjustments(final_score, query, candidate, word_ratio) - - return min(100, max(0, int(final_score))) - - @staticmethod - def _calculate_word_ratio(query_words: set, candidate_words: set) -> float: - """Calculate the ratio of matching words.""" - matching_words = sum(1 for word in query_words if word in candidate_words) - return matching_words / len(query_words) - - @staticmethod - def _get_similarity_scores(query: str, candidate: str) -> dict: - """Get various fuzzy similarity scores.""" - return { - 'ratio': fuzz.ratio(query, candidate), - 'partial': fuzz.partial_ratio(query, candidate), - 'token_set': fuzz.token_set_ratio(query, candidate) - } - - @staticmethod - def _get_word_scores(query: str, candidate: str, query_words: set, candidate_words: set) -> dict: - """Calculate word-based scores.""" - query_tokens = set(query.split()) - candidate_tokens = set(candidate.split()) - - word_intersection = len(query_tokens & candidate_tokens) - word_union = len(query_tokens | candidate_tokens) - jaccard_sim = word_intersection / word_union if word_union > 0 else 0 - - prefix_bonus = SearchUtils._calculate_prefix_bonus(query_words, candidate_words) - - return { - 'jaccard': jaccard_sim, - 'prefix': min(prefix_bonus, 100) - } - - @staticmethod - def _calculate_prefix_bonus(query_words: set, candidate_words: set) -> float: - """Calculate prefix matching bonus.""" - prefix_bonus = 0 - for q_word in query_words: - for c_word in candidate_words: - if c_word.startswith(q_word): - prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20 - return prefix_bonus - - @staticmethod - def _calculate_weighted_score(similarity_scores: dict, word_scores: dict, word_ratio: float) -> float: - """Calculate the final weighted score.""" - return ( - 0.40 * similarity_scores['token_set'] + - 0.25 * similarity_scores['ratio'] + - 0.15 * similarity_scores['partial'] + - 0.15 * (word_scores['jaccard'] * 100) + - 0.05 * word_scores['prefix'] - ) - - @staticmethod - def _apply_adjustments(score: float, query: str, candidate: str, word_ratio: float) -> float: - """Apply length penalty and word ratio boost.""" - import math - - # Apply length penalty - score *= min(1.0, 0.9 + (0.1 * SearchUtils.length_penalty(query, candidate))) - - # Apply word ratio boost - score = score * (0.7 + (0.3 * word_ratio)) - - return score - -class AudioSearcher: - """Handles searching for audio files in configured paths.""" - - @staticmethod - def _is_valid_audio_file(file_path: Path) -> bool: - """Check if file is a valid audio file.""" - valid_extensions = {'.mp4', '.mp3', '.wav', '.opus', '.m4a'} - return file_path.is_file() and file_path.suffix.lower() in valid_extensions - - @staticmethod - def _calculate_match_score(query: str, file_name: str, autocomplete: bool) -> tuple[int, str]: - """Calculate match score for a file.""" - if not query and autocomplete: - return 0, file_name - - if query and len(query) >= 2: - if query in file_name: - logger.debug(f"Found direct match: {file_name} (contains '{query}')") - - score = SearchUtils.combined_score(query, file_name) - logger.debug(f"File: {file_name} - Score: {score}") - - if score >= 30: - return score, file_name - - return -1, file_name - - @staticmethod - async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool) -> list[tuple[int, str]]: - """Search for audio files in a single path.""" - matches = [] - - logger.debug(f"Searching in directory: {path}") - - for file_path in path.rglob('*'): - if not AudioSearcher._is_valid_audio_file(file_path): - continue - - file_name = file_path.name.lower() - score, _ = AudioSearcher._calculate_match_score(query, file_name, autocomplete) - - if score >= 0: - matches.append((score, str(file_path.absolute()))) - - if len(matches) >= max_results * 5: - break - - return matches - - @classmethod - async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False): - """Search for audio files with fuzzy matching and optional autocomplete support. - - Args: - query: Search query string (can be None for autocomplete) - max_results: Maximum number of results to return - autocomplete: If True, returns results even with empty query - - Returns: - List of matching file paths with scores - """ - if not query and not autocomplete: - return [] - - query = (query or '').lower().strip() - logger.info(f"Starting local search for query: '{query}'") - - try: - async with aiofiles.open('settings.json') as f: - content = await f.read() - data = json.loads(content) - search_paths = data.get('audio_paths', []) - logger.info(f"Searching in paths: {search_paths}") - - matches = [] - file_count = 0 - - for path_str in search_paths: - path = Path(path_str) - if not path.exists(): - logger.warning(f"Path does not exist: {path}") - continue - - path_matches = await cls._search_single_path(path, query, max_results, autocomplete) - matches.extend(path_matches) - file_count += len(path_matches) - - if len(matches) >= max_results * 5: - break - - logger.info(f"Scanned {file_count} files total") - - matches.sort(reverse=True, key=lambda x: x[0]) - result = [path for score, path in matches[:max_results]] - - logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.") - if not result and query: - logger.warning(f"No matches found for query: '{query}'. Try a different search term.") - return result - - except Exception as e: - logger.error(f"Error during local search: {str(e)}", exc_info=True) - return [] - - @classmethod - async def search_local_autocomplete(cls, query: str = None, max_results: int = 10): - """Wrapper for autocomplete searches that returns more results and allows empty query.""" - # For autocomplete, we want to show more options and allow empty queries - if not query: - return [] - - return await cls.search_local(query, max_results=max_results, autocomplete=True) diff --git a/cogs/search.py b/cogs/search.py new file mode 100644 index 0000000..2f8373b --- /dev/null +++ b/cogs/search.py @@ -0,0 +1,73 @@ +# 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 urllib.parse + +import discord +from discord.ext import commands +import logging +from discord import AutocompleteContext +from cogs.utils import file_search as search_utils + +logger = logging.getLogger(__name__) + +class Search(commands.Cog): + def __init__(self, bot): + self.bot = bot + self._cached_results = {} + + async def file_autocomplete(self, ctx: AutocompleteContext): + """Autocomplete function for file search.""" + query = ctx.value.lower() if ctx.value else "" + results = await search_utils.UniversalSearcher.search_copyparty_autocomplete(query, max_results=25) + + user_id = ctx.interaction.user.id + self._cached_results[user_id] = {urllib.parse.unquote(url.split('/')[-1]): url for url in results} + + choices = [] + for url in results[:25]: + filename = urllib.parse.unquote(url.split('/')[-1]) + display_name = filename[:99] + "…" if len(filename) > 100 else filename + choices.append(discord.OptionChoice(name=display_name, value=filename)) + return choices + + @commands.slash_command( + name="search", + description="wyszukaj pliku" + ) + @discord.option( + name="filename", + description="Search for a file to send", + autocomplete=file_autocomplete + ) + async def search(self, ctx, filename: str): + """Send a file via URL from autocomplete selection.""" + if not filename: + await ctx.reply("Please select a file from the autocomplete options.") + return + + await ctx.defer() + logger.info(f"Search command called by {ctx.author} for file: {filename}") + + cached = self._cached_results.get(ctx.author.id) + if not cached: + await ctx.send_followup("Search results expired. Please search again.") + return + + matching_url = cached.get(filename) + if not matching_url: + await ctx.send_followup(f"File '{filename}' not found.") + return + + try: + await ctx.send_followup(matching_url) + logger.info(f"Successfully sent URL: {matching_url}") + del self._cached_results[ctx.author.id] + except Exception as e: + logger.error(f"Error sending URL {matching_url}: {str(e)}") + await ctx.send_followup(str(e)) + +def setup(bot): + bot.add_cog(Search(bot)) diff --git a/cogs/tts.py b/cogs/tts.py index a330b74..46ab637 100644 --- a/cogs/tts.py +++ b/cogs/tts.py @@ -40,16 +40,14 @@ class TTS(commands.Cog): logger.error(f"Error sending file: {e}") await ctx.respond("eror :c", ephemeral=True) + def _get_voice_manager(self): + from cogs.voice import get_voice_manager + return get_voice_manager(self.bot) + async def _get_or_connect_voice_client(self, channel): """Get existing voice client or connect to new channel""" - voice_client = discord.utils.get(self.bot.voice_clients, guild=channel.guild) - - if not voice_client or not voice_client.is_connected(): - voice_client = await channel.connect() - elif voice_client.channel != channel: - await voice_client.move_to(channel) - - return voice_client + voice_mgr = self._get_voice_manager() + return await voice_mgr.join_channel(channel) async def _handle_voice_output(self, ctx, temp_path, text, volume): """Handle TTS voice channel output""" @@ -65,6 +63,10 @@ class TTS(commands.Cog): try: voice_client = await self._get_or_connect_voice_client(channel) + if not voice_client: + await ctx.respond("Failed to connect to voice channel", ephemeral=True) + return + # Play the TTS audio with volume adjustment audio_source = discord.FFmpegPCMAudio( temp_path, @@ -96,9 +98,10 @@ class TTS(commands.Cog): ): await ctx.defer() - # Create temporary file path asynchronously - temp_fd, temp_path = tempfile.mkstemp(suffix=".mp3") - os.close(temp_fd) # Close the file descriptor, we'll write to it asynchronously + # Create temporary file in temp folder + temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'temp') + os.makedirs(temp_dir, exist_ok=True) + temp_path = os.path.join(temp_dir, f"tts_{os.getpid()}_{asyncio.get_event_loop().time()}.mp3") try: # Generate and save TTS to temp file diff --git a/cogs/utils/file_search.py b/cogs/utils/file_search.py new file mode 100644 index 0000000..4a2bdc4 --- /dev/null +++ b/cogs/utils/file_search.py @@ -0,0 +1,500 @@ +# 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 + +""" +Universal File Searcher + +This module provides a universal file searcher that can search for different file types: +- Audio files (mp3, wav, mp4, opus, m4a, flac, aac, ogg) +- Video files (mp4, avi, mkv, mov, wmv, flv, webm, m4v, 3gp) +- Image files (jpg, jpeg, png, gif, bmp, tiff, webp, svg, ico) + +Usage Examples: + # Search all file types + results = await UniversalSearcher.search_local("query") + + # Search only audio files + results = await UniversalSearcher.search_local("query", filetype="Audio") + + # Search only video files + results = await UniversalSearcher.search_local("query", filetype="Video") + + # Search only image files + results = await UniversalSearcher.search_local("query", filetype="Image") + + # Autocomplete search + results = await UniversalSearcher.search_local_autocomplete("partial_query", filetype="Video") + +Settings.json structure: +{ + "file_paths": [ + "/path/to/your/media/files", + "/another/path/with/mixed/content", + "relative/path/to/files" + ] +} + +Note: file_paths should contain directories that may contain any mix of audio, video, +and image files. The searcher will filter by file type automatically based on the +filetype parameter. +""" + +import json +import logging +import re +import urllib.parse +from pathlib import Path +import aiofiles +from rapidfuzz import fuzz +import aiohttp +import asyncio +import os +from dotenv import load_dotenv + +load_dotenv() + +logger = logging.getLogger(__name__) + +class SearchUtils: + """Utilities for searching and scoring audio files.""" + + @staticmethod + def normalize_name(p: Path, keep_ext=False): + """Preprocess filenames: lower, replace separators, strip extensions optionally.""" + name = str(p.name if keep_ext else p.stem) + return name.lower().replace('_', ' ').replace('-', ' ') + + @staticmethod + def length_penalty(query, candidate): + """Penalize tiny filenames when query is long; returns multiplier [0.0,1.0].""" + ql = len(query) + cl = len(candidate) + if ql == 0: + return 1.0 + ratio = cl / ql + # Prefer candidate lengths near query length; gentle falloff otherwise + return math.exp(-0.5 * ((math.log(max(ratio, 1e-6)))**2)) + + @staticmethod + def combined_score(query, candidate): + """Calculate a combined fuzzy matching score using multiple strategies. + + Args: + query: The search query + candidate: The candidate string to score against + + Returns: + float: Score between 0-100, where higher means better match + """ + if not query or not candidate: + return 0.0 + + query = query.lower().strip() + candidate = candidate.lower() + + # Check for direct matches first + direct_match_score = SearchUtils._check_direct_matches(query, candidate) + if direct_match_score is not None: + return direct_match_score + + # Handle short queries specially + short_query_score = SearchUtils._handle_short_queries(query, candidate) + if short_query_score is not None: + return short_query_score + + # For longer queries, use complex scoring + return SearchUtils._calculate_complex_score(query, candidate) + + @staticmethod + def _check_direct_matches(query: str, candidate: str) -> float: + """Check for direct substring matches.""" + if query in candidate: + base_score = 100 + if re.search(rf'\b{re.escape(query)}\b', candidate): + base_score = 110 + return min(100, base_score) + return 0 + + @staticmethod + def _handle_short_queries(query: str, candidate: str) -> float: + """Handle scoring for short queries (2-3 chars).""" + if len(query) <= 3: + if re.search(rf'\b{re.escape(query)}', candidate): + return 90 + if query in candidate: + return 80 + return 0 + + @staticmethod + def _calculate_complex_score(query: str, candidate: str) -> float: + """Calculate complex score for longer queries.""" + query_words = {word for word in re.split(r'\W+', query) if len(word) > 1} + if not query_words: + return 0.0 + + candidate_words = set(re.split(r'\W+', candidate)) + word_ratio = SearchUtils._calculate_word_ratio(query_words, candidate_words) + + if word_ratio == 0: + return 0.0 + + # Calculate various similarity metrics + similarity_scores = SearchUtils._get_similarity_scores(query, candidate) + word_scores = SearchUtils._get_word_scores(query, candidate, query_words, candidate_words) + + # Calculate final weighted score + final_score = SearchUtils._calculate_weighted_score(similarity_scores, word_scores, word_ratio) + + # Apply penalties and boosts + final_score = SearchUtils._apply_adjustments(final_score, query, candidate, word_ratio) + + return min(100, max(0, int(final_score))) + + @staticmethod + def _calculate_word_ratio(query_words: set, candidate_words: set) -> float: + """Calculate the ratio of matching words.""" + matching_words = sum(1 for word in query_words if word in candidate_words) + return matching_words / len(query_words) + + @staticmethod + def _get_similarity_scores(query: str, candidate: str) -> dict: + """Get various fuzzy similarity scores.""" + return { + 'ratio': fuzz.ratio(query, candidate), + 'partial': fuzz.partial_ratio(query, candidate), + 'token_set': fuzz.token_set_ratio(query, candidate) + } + + @staticmethod + def _get_word_scores(query: str, candidate: str, query_words: set, candidate_words: set) -> dict: + """Calculate word-based scores.""" + query_tokens = set(query.split()) + candidate_tokens = set(candidate.split()) + + word_intersection = len(query_tokens & candidate_tokens) + word_union = len(query_tokens | candidate_tokens) + jaccard_sim = word_intersection / word_union if word_union > 0 else 0 + + prefix_bonus = SearchUtils._calculate_prefix_bonus(query_words, candidate_words) + + return { + 'jaccard': jaccard_sim, + 'prefix': min(prefix_bonus, 100) + } + + @staticmethod + def _calculate_prefix_bonus(query_words: set, candidate_words: set) -> float: + """Calculate prefix matching bonus.""" + prefix_bonus = 0 + for q_word in query_words: + for c_word in candidate_words: + if c_word.startswith(q_word): + prefix_bonus += (len(q_word) / max(len(c_word), 1)) * 20 + return prefix_bonus + + @staticmethod + def _calculate_weighted_score(similarity_scores: dict, word_scores: dict, word_ratio: float) -> float: + """Calculate the final weighted score.""" + return ( + 0.40 * similarity_scores['token_set'] + + 0.25 * similarity_scores['ratio'] + + 0.15 * similarity_scores['partial'] + + 0.15 * (word_scores['jaccard'] * 100) + + 0.05 * word_scores['prefix'] + ) + + @staticmethod + def _apply_adjustments(score: float, query: str, candidate: str, word_ratio: float) -> float: + """Apply length penalty and word ratio boost.""" + import math + + # Apply length penalty + score *= min(1.0, 0.9 + (0.1 * SearchUtils.length_penalty(query, candidate))) + + # Apply word ratio boost + score = score * (0.7 + (0.3 * word_ratio)) + + return score + +class UniversalSearcher: + """Handles searching for files in configured paths with support for multiple file types.""" + + FILE_TYPE_EXTENSIONS = { + 'Audio': {'.mp4', '.mp3', '.wav', '.opus', '.m4a', '.flac', '.aac', '.ogg'}, + 'Video': {'.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.webm', '.m4v', '.3gp', '.gif'}, + 'Image': {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp', '.svg', '.ico'} + } + + @staticmethod + def _is_valid_file(file_path: Path, filetype: str = None) -> bool: + """Check if file is valid for the specified filetype.""" + if not file_path.is_file(): + return False + + if filetype is None: + # Search all file types + all_extensions = set() + for extensions in UniversalSearcher.FILE_TYPE_EXTENSIONS.values(): + all_extensions.update(extensions) + return file_path.suffix.lower() in all_extensions + + filetype = filetype.capitalize() + if filetype not in UniversalSearcher.FILE_TYPE_EXTENSIONS: + return False + + valid_extensions = UniversalSearcher.FILE_TYPE_EXTENSIONS[filetype] + return file_path.suffix.lower() in valid_extensions + + @staticmethod + def _calculate_match_score(query: str, file_name: str, autocomplete: bool) -> tuple[int, str]: + """Calculate match score for a file.""" + if not query and autocomplete: + return 0, file_name + + if query and len(query) >= 2: + if query in file_name: + logger.debug(f"Found direct match: {file_name} (contains '{query}')") + + score = SearchUtils.combined_score(query, file_name) + logger.debug(f"File: {file_name} - Score: {score}") + + if score >= 30: + return score, file_name + + return -1, file_name + + @staticmethod + async def _search_single_path(path: Path, query: str, max_results: int, autocomplete: bool, filetype: str = None) -> list[tuple[int, str]]: + """Search for files in a single path.""" + matches = [] + + logger.debug(f"Searching in directory: {path} (filetype: {filetype or 'All'})") + + for file_path in path.rglob('*'): + if not UniversalSearcher._is_valid_file(file_path, filetype): + continue + + file_name = file_path.name.lower() + score, _ = UniversalSearcher._calculate_match_score(query, file_name, autocomplete) + + if score >= 0: + matches.append((score, str(file_path.absolute()))) + + if len(matches) >= max_results * 5: + break + + return matches + + @staticmethod + def _get_search_paths(data: dict, filetype: str = None) -> list: + """Get the search paths from settings.""" + return data.get('file_paths', []) + + @classmethod + async def _search_paths(cls, search_paths: list, query: str, max_results: int, autocomplete: bool, filetype: str = None) -> tuple[list, int]: + """Search through the provided paths and return matches and file count.""" + matches = [] + file_count = 0 + + for path_str in search_paths: + path = Path(path_str) + if not path.exists(): + logger.warning(f"Path does not exist: {path}") + continue + + path_matches = await cls._search_single_path(path, query, max_results, autocomplete, filetype) + matches.extend(path_matches) + file_count += len(path_matches) + + if len(matches) >= max_results * 5: + break + + return matches, file_count + + @classmethod + async def search_local(cls, query: str = None, max_results: int = 25, autocomplete: bool = False, filetype: str = None): + """Search for files with fuzzy matching and optional autocomplete support. + + Args: + query: Search query string (can be None for autocomplete) + max_results: Maximum number of results to return + autocomplete: If True, returns results even with empty query + filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all) + + Returns: + List of matching file paths with scores + """ + if not query and not autocomplete: + return [] + + query = (query or '').lower().strip() + logger.info(f"Starting local search for query: '{query}' (filetype: {filetype or 'All'})") + + try: + async with aiofiles.open('settings.json') as f: + content = await f.read() + data = json.loads(content) + + search_paths = cls._get_search_paths(data, filetype) + logger.info(f"Searching in paths: {search_paths}") + + matches, file_count = await cls._search_paths(search_paths, query, max_results, autocomplete, filetype) + + logger.info(f"Scanned {file_count} files total") + + matches.sort(reverse=True, key=lambda x: x[0]) + result = [path for score, path in matches[:max_results]] + + logger.info(f"Search complete. Found {len(result)} matches out of {file_count} files.") + if not result and query: + logger.warning(f"No matches found for query: '{query}'. Try a different search term.") + return result + + except Exception as e: + logger.error(f"Error during local search: {str(e)}", exc_info=True) + return [] + + @classmethod + async def search_local_autocomplete(cls, query: str = None, max_results: int = 10, filetype: str = None): + """Wrapper for autocomplete searches that returns more results and allows empty query. + + Args: + query: Search query string (can be None for autocomplete) + max_results: Maximum number of results to return + filetype: Type of files to search for ('Audio', 'Video', 'Image', or None for all) + """ + # For autocomplete, we want to show more options and allow empty queries + if not query: + return [] + + return await cls.search_local(query, max_results=max_results, autocomplete=True, filetype=filetype) + + @staticmethod + async def _load_baseurl(): + """Load baseurl from settings.json""" + try: + async with aiofiles.open('settings.json') as f: + content = await f.read() + data = json.loads(content) + return data.get('baseurl', 'http://localhost:8080').rstrip('/') + except Exception as e: + logger.error(f"Error loading baseurl: {e}") + return 'http://localhost:8080' + + @staticmethod + async def search_copyparty(query: str, max_results: int = 25): + """Search for files using copyparty API (jPOST method). + + Args: + query: Search query string + max_results: Maximum number of results to return + + Returns: + List of matching file URLs + """ + + + if not query: + return [] + + query = query.strip() + logger.info(f"Starting copyparty search for query: '{query}'") + query = f'name like "*{query}*"' + + try: + baseurl = await UniversalSearcher._load_baseurl() + password = os.getenv("COPYPARTY_R_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() + logger.info(f"Copyparty response: {data}") + 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 it's just a string path/URL + 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. + + 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) + + diff --git a/cogs/vclogs.py b/cogs/vclogs.py index 901b557..9ed7134 100644 --- a/cogs/vclogs.py +++ b/cogs/vclogs.py @@ -7,28 +7,27 @@ import datetime import json import logging import os +import asyncio import discord from discord.ext import commands from gtts import gTTS import aiofiles +from cogs.voice import get_voice_manager logger = logging.getLogger(__name__) - class VcLogs(commands.Cog): def __init__(self, bot): self.bot = bot - self.settings_check() + self._voice_manager = None - settings_file = 'settings.json' + def get_voice_manager(self): + if self._voice_manager is None: + self._voice_manager = get_voice_manager(self.bot) + return self._voice_manager - @staticmethod - def settings_check(): - if not os.path.exists(VcLogs.settings_file): - with open(VcLogs.settings_file, 'w') as f: - json.dump({'volume': '5'}, f, indent=4) - logger.info("settings.json created") + settings_file = 'settings.json' @staticmethod async def handle_state_change(member, channel, state_type, is_enabled, current_time, log_channel): @@ -62,32 +61,34 @@ class VcLogs(commands.Cog): return tts_text, channel async def play_tts(self, channel, text): - tts_file = "temp_tts.mp3" - # Generate TTS audio + voice_mgr = self.get_voice_manager() + + temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'temp') + os.makedirs(temp_dir, exist_ok=True) + tts_file = os.path.join(temp_dir, f"temp_tts_{os.getpid()}.mp3") tts = gTTS(text=text, lang='pl') + logger.info("saving file") tts.save(tts_file) if os.path.exists(VcLogs.settings_file): async with aiofiles.open(VcLogs.settings_file) as f: content = await f.read() data = json.loads(content) - volume = data['volume'] - else: - logger.error("settings.json not found") - - # Connect to the voice channel if not already connected - voice_client = discord.utils.get(self.bot.voice_clients, guild=channel.guild) - if not isinstance(voice_client, discord.VoiceClient): - voice_client = await channel.connect() - elif voice_client.channel != channel: - await voice_client.move_to(channel) - - # Play the TTS audio - if not voice_client.is_playing(): + volume = data.get('volume', 1.0) + + voice_client = await voice_mgr.get_voice_client(channel.guild) + + if voice_client is None: + voice_client = await voice_mgr.join_channel(channel) + + logger.info(f"connected: {voice_client}") + + if voice_client and not voice_client.is_playing(): + logger.info("preparing to play audio") audio_source = discord.FFmpegPCMAudio(tts_file, options=f'-filter:a "volume={volume}"') voice_client.play(audio_source) while voice_client.is_playing(): - await discord.utils.sleep_until(datetime.datetime.now() + datetime.timedelta(seconds=1)) + await asyncio.sleep(0.5) if os.path.exists(tts_file): os.remove(tts_file) @@ -96,10 +97,6 @@ class VcLogs(commands.Cog): ) async def volume(self, ctx: discord.ApplicationContext, volume: int): try: - if not os.path.exists(VcLogs.settings_file): - async with aiofiles.open(VcLogs.settings_file, 'w') as f: - json.dump({'volume': '5'}, f, indent=4) - logger.info("settings.json created") async with aiofiles.open(VcLogs.settings_file, 'w') as f: json.dump({"volume": volume}, f, indent=4) await ctx.followup.send(f"Volume has been set to: {volume}") diff --git a/cogs/voice.py b/cogs/voice.py index dd5b74a..8ee3699 100644 --- a/cogs/voice.py +++ b/cogs/voice.py @@ -10,33 +10,117 @@ import asyncio logger = logging.getLogger(__name__) +_voice_manager = None + +def get_voice_manager(bot): + global _voice_manager + if _voice_manager is None: + _voice_manager = VoiceManager(bot) + return _voice_manager + + +class VoiceManager: + def __init__(self, bot): + self.bot = bot + self._voice_lock = asyncio.Lock() + + async def join_channel(self, channel): + """Join a voice channel by channel object.""" + async with self._voice_lock: + guild = channel.guild + voice_client = discord.utils.get(self.bot.voice_clients, guild=guild) + + if voice_client and voice_client.is_connected(): + if voice_client.channel == channel: + logger.info(f"Already connected to {channel}") + return voice_client + try: + await voice_client.move_to(channel) + logger.info(f"Moved to voice channel: {channel}") + return voice_client + except Exception as e: + logger.error(f"Failed to move voice client: {e}") + try: + await voice_client.disconnect() + except Exception: + pass + + await asyncio.sleep(0.3) + try: + voice_client = await channel.connect(timeout=30.0, reconnect=False) + logger.info(f"Connected to voice channel: {channel}") + return voice_client + except Exception as e: + logger.error(f"Failed to connect to voice channel: {e}") + return None + + async def join_by_channel_id(self, channel_id): + """Join a voice channel by channel ID.""" + channel = self.bot.get_channel(channel_id) + if not channel: + logger.error(f"Channel {channel_id} not found") + return None + if not isinstance(channel, discord.VoiceChannel): + logger.error(f"Channel {channel_id} is not a voice channel") + return None + return await self.join_channel(channel) + + async def get_voice_client(self, guild): + """Get existing voice client for a guild.""" + return discord.utils.get(self.bot.voice_clients, guild=guild) + + async def play_audio(self, channel, audio_source, volume=1.0): + """Play audio in a voice channel.""" + voice_client = await self.join_channel(channel) + if not voice_client: + return False + + if volume != 1.0: + audio_source = discord.PCMVolumeTransformer(audio_source) + audio_source.volume = volume + + voice_client.play(audio_source) + + while voice_client.is_playing(): + await asyncio.sleep(0.1) + return True + + async def stop(self, guild): + """Stop playing and disconnect from voice.""" + voice_client = await self.get_voice_client(guild) + if voice_client: + voice_client.stop() + await voice_client.disconnect() + logger.info(f"Stopped and disconnected from guild {guild}") + + class Voice(commands.Cog): def __init__(self, bot): self.bot = bot + self._voice_lock = asyncio.Lock() async def _validate_user_voice_state(self, ctx): - """Validate that user is in a voice channel.""" if not ctx.author.voice or not ctx.author.voice.channel: await ctx.send("You need to be in a voice channel first!") return False return True async def _create_voice_client(self, ctx): - """Create a new voice client connection.""" - try: - voice_client = await ctx.author.voice.channel.connect(timeout=30.0, reconnect=True) - logger.info(f"Connected to voice channel: {ctx.author.voice.channel}") - return voice_client - except asyncio.TimeoutError: - await ctx.send("Connection to voice channel timed out.") - return None - except discord.ClientException as e: - logger.error(f"Failed to connect to voice channel: {e}") - await ctx.send("Failed to connect to voice channel. Please try again.", ephemeral=True) - return None + async with self._voice_lock: + try: + await asyncio.sleep(0.5) + voice_client = await ctx.author.voice.channel.connect(timeout=30.0, reconnect=False) + logger.info(f"Connected to voice channel: {ctx.author.voice.channel}") + return voice_client + except asyncio.TimeoutError: + await ctx.send("Connection to voice channel timed out.") + return None + except discord.ClientException as e: + logger.error(f"Failed to connect to voice channel: {e}") + await ctx.send("Failed to connect to voice channel. Please try again.", ephemeral=True) + return None async def _move_voice_client(self, voice_client, ctx): - """Move existing voice client to user's channel.""" try: await voice_client.move_to(ctx.author.voice.channel) logger.info(f"Moved to voice channel: {ctx.author.voice.channel}") @@ -50,10 +134,9 @@ class Voice(commands.Cog): return None async def _verify_connection(self, voice_client): - """Verify and repair voice client connection if needed.""" if not voice_client.is_connected(): try: - await voice_client.connect(reconnect=True, timeout=10.0) + await voice_client.connect(reconnect=False, timeout=10.0) return voice_client except Exception as e: logger.error(f"Failed to reconnect to voice: {e}") @@ -61,31 +144,30 @@ class Voice(commands.Cog): return voice_client async def ensure_voice_client(self, ctx): - """Ensure we have a valid voice client connection.""" if not await self._validate_user_voice_state(ctx): return None - + voice_client = discord.utils.get(self.bot.voice_clients, guild=ctx.guild) - - # If no voice client exists, create one + + if voice_client and not voice_client.is_connected(): + try: + await voice_client.disconnect() + except Exception: + pass + voice_client = None + if not voice_client or not voice_client.is_connected(): + if not await self._validate_user_voice_state(ctx): + return None return await self._create_voice_client(ctx) - # If we're in a different channel, move to the user's channel if voice_client.channel != ctx.author.voice.channel: return await self._move_voice_client(voice_client, ctx) - # Verify the connection is healthy return await self._verify_connection(voice_client) async def connect(self, ctx): - """Connect to the user's voice channel with improved connection handling.""" voice_client = await self.ensure_voice_client(ctx) if not voice_client: return None - - return voice_client - - -def setup(bot): - bot.add_cog(Voice(bot)) \ No newline at end of file + return voice_client \ No newline at end of file -- cgit v1.2.3