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 --- src/commands/ai.rs | 76 ++++ src/commands/autocomplete.rs | 128 ++++++ src/commands/cron.rs | 155 ++++++++ src/commands/math.rs | 77 ++++ src/commands/mod.rs | 26 ++ src/commands/music.rs | 414 +++++++++++++++++++ src/commands/qr.rs | 87 ++++ src/commands/trigger.rs | 199 ++++++++++ src/commands/tts.rs | 106 +++++ src/commands/utilities.rs | 918 +++++++++++++++++++++++++++++++++++++++++++ src/commands/web.rs | 122 ++++++ src/config.rs | 145 +++++++ src/copyparty/mod.rs | 267 +++++++++++++ src/database/mod.rs | 344 ++++++++++++++++ src/error.rs | 108 +++++ src/events/mod.rs | 368 +++++++++++++++++ src/lang.rs | 387 ++++++++++++++++++ src/main.rs | 190 +++++++++ src/openrouter/mod.rs | 134 +++++++ src/tasks/mod.rs | 172 ++++++++ src/voice/mod.rs | 404 +++++++++++++++++++ 21 files changed, 4827 insertions(+) create mode 100644 src/commands/ai.rs create mode 100644 src/commands/autocomplete.rs create mode 100644 src/commands/cron.rs create mode 100644 src/commands/math.rs create mode 100644 src/commands/mod.rs create mode 100644 src/commands/music.rs create mode 100644 src/commands/qr.rs create mode 100644 src/commands/trigger.rs create mode 100644 src/commands/tts.rs create mode 100644 src/commands/utilities.rs create mode 100644 src/commands/web.rs create mode 100644 src/config.rs create mode 100644 src/copyparty/mod.rs create mode 100644 src/database/mod.rs create mode 100644 src/error.rs create mode 100644 src/events/mod.rs create mode 100644 src/lang.rs create mode 100644 src/main.rs create mode 100644 src/openrouter/mod.rs create mode 100644 src/tasks/mod.rs create mode 100644 src/voice/mod.rs (limited to 'src') diff --git a/src/commands/ai.rs b/src/commands/ai.rs new file mode 100644 index 0000000..a75f164 --- /dev/null +++ b/src/commands/ai.rs @@ -0,0 +1,76 @@ +use crate::Data; +use crate::error::Error; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// Query the AI senator via OpenRouter +#[poise::command(slash_command)] +pub async fn senator( + ctx: Context<'_>, + #[description = "Wiadomość do wysłania"] text: String, + #[description = "Przemówić w kanale głosowym?"] tts_vc: Option, +) -> crate::error::Result<()> { + ctx.defer().await?; + + let api_key = ctx + .data() + .config + .openrouter_api_key + .as_ref() + .ok_or_else(|| Error::Custom(lang::L_OPENROUTER_NO_KEY.into()))?; + let model = ctx + .data() + .config + .openrouter_model + .as_deref() + .unwrap_or("cognitivecomputations/dolphin-mistral-24b-venice-edition:free"); + + let client = crate::openrouter::OpenRouterClient::new(api_key.clone(), model.to_string()); + let response = client.chat(lang::SENATOR_SYSTEM_PROMPT, &text).await?; + + if tts_vc.unwrap_or(false) { + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + let channel_id = ctx + .serenity_context() + .cache + .guild(guild_id) + .and_then(|g| { + g.voice_states + .get(&ctx.author().id) + .and_then(|vs| vs.channel_id) + }) + .ok_or_else(|| lang::err_not_in_voice())?; + + let tts_url = format!( + "https://translate.google.com/translate_tts?ie=UTF-8&q={}&tl=pl&client=tw-ob", + crate::commands::tts::urlencoding(&response) + ); + let audio_data = ctx + .data() + .http_client + .get(&tts_url) + .send() + .await? + .bytes() + .await?; + + let call = ctx + .data() + .voice_manager + .connect(guild_id, channel_id) + .await?; + let input = crate::voice::decode_audio(&audio_data)?; + let mut call_lock = call.lock().await; + if call_lock.current_channel().is_some() { + call_lock.play_input(input); + } + } + + ctx.say(response).await?; + Ok(()) +} + +pub fn commands() -> Vec> { + vec![senator()] +} diff --git a/src/commands/autocomplete.rs b/src/commands/autocomplete.rs new file mode 100644 index 0000000..497aec9 --- /dev/null +++ b/src/commands/autocomplete.rs @@ -0,0 +1,128 @@ +use poise::serenity_prelude as serenity; + +use crate::Data; +use crate::copyparty::{html_decode, percent_decode}; +use crate::error::Error; + +type Context<'a> = poise::Context<'a, Data, Error>; + +const MAX_CHOICE_LEN: usize = 99; + +fn filename_label(raw_path: &str) -> Option { + let raw_name = raw_path.rsplit('/').next()?; + if raw_name.is_empty() { + return None; + } + let decoded = html_decode(&percent_decode(raw_name)); + if decoded.len() > MAX_CHOICE_LEN { + let mut s: String = decoded.chars().take(MAX_CHOICE_LEN - 1).collect(); + s.push('\u{2026}'); + Some(s) + } else { + Some(decoded) + } +} + +pub async fn audio_file<'a>( + ctx: Context<'_>, + partial: &'a str, +) -> Vec { + eprintln!("[DEBUG] autocomplete audio_file(partial={partial:?})"); + let results = ctx + .data() + .copyparty + .search(partial, 25, &["mp3", "wav", "ogg", "flac", "m4a", "opus"]) + .await; + match results { + Ok(hits) => { + eprintln!("[DEBUG] audio_file got {} raw hits", hits.len()); + let mut choices = Vec::new(); + for (i, hit) in hits.into_iter().enumerate() { + let rp = match &hit.rp { + Some(p) => p.clone(), + None => continue, + }; + eprintln!( + "[DEBUG] hit[{i}]: rp={rp:?}, url={:?}, name={:?}, path={:?}", + hit.url, hit.name, hit.path + ); + let Some(label) = filename_label(&rp) else { + continue; + }; + ctx.data() + .ac_stash(ctx.author().id, label.clone(), hit) + .await; + eprintln!("[DEBUG] -> cached as {label:?}"); + choices.push(serenity::AutocompleteChoice::new(label.clone(), rp)); + if choices.len() >= 25 { + break; + } + } + eprintln!("[DEBUG] audio_file returning {} choices", choices.len()); + choices + } + Err(e) => { + eprintln!("[DEBUG] audio_file search failed: {e}"); + vec![] + } + } +} + +pub async fn any_file<'a>(ctx: Context<'_>, partial: &'a str) -> Vec { + eprintln!("[DEBUG] autocomplete any_file(partial={partial:?})"); + let results = ctx.data().copyparty.search(partial, 25, &[]).await; + match results { + Ok(hits) => { + eprintln!("[DEBUG] any_file got {} raw hits", hits.len()); + let mut choices = Vec::new(); + for (i, hit) in hits.into_iter().enumerate() { + let rp = match &hit.rp { + Some(p) => p.clone(), + None => continue, + }; + eprintln!( + "[DEBUG] hit[{i}]: rp={rp:?}, url={:?}, name={:?}, path={:?}", + hit.url, hit.name, hit.path + ); + let Some(label) = filename_label(&rp) else { + continue; + }; + ctx.data() + .ac_stash(ctx.author().id, label.clone(), hit) + .await; + eprintln!("[DEBUG] -> cached as {label:?}"); + choices.push(serenity::AutocompleteChoice::new(label.clone(), label)); + if choices.len() >= 25 { + break; + } + } + eprintln!("[DEBUG] any_file returning {} choices", choices.len()); + choices + } + Err(e) => { + eprintln!("[DEBUG] any_file search failed: {e}"); + vec![] + } + } +} + +pub async fn trigger_name<'a>(ctx: Context<'_>, partial: &'a str) -> Vec { + eprintln!("[DEBUG] autocomplete trigger_name(partial={partial:?})"); + let db_pool = &ctx.data().db_pool; + match crate::database::list_triggers(db_pool).await { + Ok(triggers) => { + let names: Vec<_> = triggers + .into_iter() + .map(|t| t.name) + .filter(|name| partial.is_empty() || name.starts_with(partial)) + .take(25) + .collect(); + eprintln!("[DEBUG] trigger_name returning {} names", names.len()); + names + } + Err(e) => { + eprintln!("[DEBUG] trigger_name query failed: {e}"); + vec![] + } + } +} diff --git a/src/commands/cron.rs b/src/commands/cron.rs new file mode 100644 index 0000000..48f6ff1 --- /dev/null +++ b/src/commands/cron.rs @@ -0,0 +1,155 @@ +use poise::serenity_prelude as serenity; + +use super::autocomplete::audio_file; +use crate::Data; +use crate::database; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// Cron job management +#[poise::command( + slash_command, + subcommands("dodaj_audio", "dodaj_wiadomość", "pokaż", "usuń", "pauza", "wznów",) +)] +pub async fn cron(_ctx: Context<'_>) -> Result<()> { + Ok(()) +} + +/// Schedule audio playback +#[poise::command(slash_command)] +pub async fn dodaj_audio( + ctx: Context<'_>, + #[description = "Cron expression (e.g. \"37 21 * * *\" = every day at 21:37)"] cron_expression: String, + #[description = "Target voice channel"] channel: serenity::Channel, + #[description = "Copyparty URL of audio file"] + #[autocomplete = "audio_file"] + audio_filename: String, +) -> Result<()> { + let channel_id = match channel { + serenity::Channel::Guild(ch) => ch.id.get() as i64, + _ => return Err(lang::err_bad_channel_type()), + }; + + database::add_cron_job( + &ctx.data().db_pool, + &cron_expression, + "audio", + channel_id, + Some(&audio_filename), + None, + ) + .await?; + + ctx.say(lang::fmt_cron_audio_queued(&cron_expression)) + .await?; + Ok(()) +} + +/// Schedule a periodic message +#[poise::command(slash_command)] +pub async fn dodaj_wiadomość( + ctx: Context<'_>, + #[description = "Cron expression (e.g. \"37 21 * * *\" = every day at 21:37)"] cron_expression: String, + #[description = "Docelowy kanał tekstowy"] channel: serenity::Channel, + #[description = "Zawartość wiadomości"] message: String, +) -> Result<()> { + let channel_id = match channel { + serenity::Channel::Guild(ch) => ch.id.get() as i64, + _ => return Err(lang::err_bad_channel_type()), + }; + + database::add_cron_job( + &ctx.data().db_pool, + &cron_expression, + "wiadomość", + channel_id, + None, + Some(&message), + ) + .await?; + + ctx.say(lang::L_CRON_QUEUED).await?; + Ok(()) +} + +/// List all cron jobs +#[poise::command(slash_command)] +pub async fn pokaż(ctx: Context<'_>) -> Result<()> { + let jobs = database::list_cron_jobs(&ctx.data().db_pool).await?; + if jobs.is_empty() { + ctx.say(lang::L_CRON_NONE).await?; + return Ok(()); + } + + let mut embed = serenity::CreateEmbed::default() + .title(lang::L_CRON_TITLE) + .color(0x00aaff); + + for job in &jobs { + let status = if job.active == 1 { + lang::L_CRON_ACTIVE + } else { + lang::L_CRON_PAUSED + }; + let desc = match job.r#type.as_str() { + "audio" => lang::fmt_cron_label_audio(job.audio_filename.as_deref().unwrap_or("?")), + _ => lang::fmt_cron_label_msg(job.message.as_deref().unwrap_or("?")), + }; + embed = embed.field( + lang::fmt_cron_job_status(job.id, status), + lang::fmt_cron_field(&job.cron_expression, job.channel_id, &desc), + false, + ); + } + + ctx.send(poise::CreateReply::default().embed(embed)).await?; + Ok(()) +} + +/// Remove a cron job +#[poise::command(slash_command)] +pub async fn usuń( + ctx: Context<'_>, + #[description = "ID pracy do usunięcia"] job_id: i64, +) -> Result<()> { + if database::delete_cron_job(&ctx.data().db_pool, job_id).await? { + ctx.say(lang::fmt_cron_deleted(job_id)).await?; + } else { + ctx.say(lang::fmt_cron_not_found(job_id)).await?; + } + Ok(()) +} + +/// Pause a cron job +#[poise::command(slash_command)] +pub async fn pauza( + ctx: Context<'_>, + #[description = "ID pracy do zapauzowania"] job_id: i64, +) -> Result<()> { + if database::set_cron_job_active(&ctx.data().db_pool, job_id, false).await? { + ctx.say(lang::fmt_cron_paused(job_id)).await?; + } else { + ctx.say(lang::fmt_cron_not_found(job_id)).await?; + } + Ok(()) +} + +/// Resume a paused cron job +#[poise::command(slash_command)] +pub async fn wznów( + ctx: Context<'_>, + #[description = "ID pracy cron do wznowienia"] job_id: i64, +) -> Result<()> { + if database::set_cron_job_active(&ctx.data().db_pool, job_id, true).await? { + ctx.say(lang::fmt_cron_resumed(job_id)).await?; + } else { + ctx.say(lang::fmt_cron_not_found(job_id)).await?; + } + Ok(()) +} + +pub fn commands() -> Vec> { + vec![cron()] +} diff --git a/src/commands/math.rs b/src/commands/math.rs new file mode 100644 index 0000000..43a7de8 --- /dev/null +++ b/src/commands/math.rs @@ -0,0 +1,77 @@ +use crate::lang; +use crate::Data; +use crate::error::{Error, Result}; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// Evaluate a math expression via qalc +#[poise::command(slash_command)] +pub async fn math( + ctx: Context<'_>, + #[description = "Ekspresja matematyczn"] equation: String, +) -> Result<()> { + if equation.len() > 1000 { + ctx.say(lang::L_MATH_TOO_LONG).await?; + return Ok(()); + } + + if !equation.is_ascii() { + ctx.say(lang::L_MATH_NOT_ASCII).await?; + return Ok(()); + } + + let mut depth = 0i32; + for c in equation.chars() { + match c { + '(' | '[' | '{' => depth += 1, + ')' | ']' | '}' => depth -= 1, + _ => {} + } + if depth < 0 { + ctx.say(lang::L_MATH_UNBALANCED).await?; + return Ok(()); + } + } + if depth != 0 { + ctx.say(lang::L_MATH_UNBALANCED).await?; + return Ok(()); + } + + let dangerous = [ + "eval", + "exec", + "os.", + "subprocess", + "import", + "__", + "system", + "shutil", + "typecast", + "shell", + "fork", + "execve", + ]; + let lower = equation.to_lowercase(); + if dangerous.iter().any(|d| lower.contains(d)) { + ctx.say(lang::L_MATH_DANGEROUS).await?; + return Ok(()); + } + + let output = tokio::process::Command::new("qalc") + .arg(&equation) + .output() + .await + .map_err(|e| Error::Custom(lang::fmt_math_qalc_err(&e)))?; + + let result = String::from_utf8_lossy(&output.stdout).trim().to_string(); + if result.is_empty() { + ctx.say(lang::L_MATH_NO_RESULT).await?; + } else { + ctx.say(lang::fmt_math_result(&result)).await?; + } + Ok(()) +} + +pub fn commands() -> Vec> { + vec![math()] +} diff --git a/src/commands/mod.rs b/src/commands/mod.rs new file mode 100644 index 0000000..9d6a1d0 --- /dev/null +++ b/src/commands/mod.rs @@ -0,0 +1,26 @@ +pub mod music; +pub mod tts; +pub mod ai; +pub mod web; +pub mod math; +pub mod qr; +pub mod cron; +pub mod trigger; +pub mod utilities; +pub mod autocomplete; + +use crate::Data; + +pub fn all_commands() -> Vec> { + let mut cmds = Vec::new(); + cmds.extend(music::commands()); + cmds.extend(tts::commands()); + cmds.extend(ai::commands()); + cmds.extend(web::commands()); + cmds.extend(math::commands()); + cmds.extend(qr::commands()); + cmds.extend(cron::commands()); + cmds.extend(trigger::commands()); + cmds.extend(utilities::commands()); + cmds +} diff --git a/src/commands/music.rs b/src/commands/music.rs new file mode 100644 index 0000000..313577d --- /dev/null +++ b/src/commands/music.rs @@ -0,0 +1,414 @@ +use poise::serenity_prelude as serenity; +use tracing::{error, info}; + +use super::autocomplete::audio_file; +use crate::Data; +use crate::copyparty::{html_decode, percent_decode}; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +// ── Source classification ────────────────────────────────────────── + +/// What kind of audio source are we dealing with? +/// +/// This mirrors the Python original's split: `EXTERNAL_URL_PATTERN` +/// (youtube/soundcloud/bandcamp) goes through yt-dlp download, while +/// *every other* URL is streamed directly via ffmpeg (radio, HLS, +/// playlists, direct media files). +#[derive(Debug, Clone, PartialEq)] +enum SourceKind { + /// YouTube, SoundCloud, Twitch, Bandcamp — platforms that don't serve + /// raw audio at a direct URL. Needs yt-dlp to extract + download the + /// media to memory before decoding. + YtDlp, + /// Any HTTP(S) URL: internet radio (Icecast/Shoutcast), HLS `.m3u8`, + /// `.m3u`/`.pls` playlists, Copyparty files, Discord attachments, or + /// directly-accessible media files. Streamed via an ffmpeg subprocess + /// — NO full download. ffmpeg handles playlist parsing, Icecast + /// metadata, reconnection, and codec/container detection universally. + Stream, +} + +/// Play music from Copyparty, URL, or file upload +#[poise::command(slash_command)] +pub async fn grajcuj( + ctx: Context<'_>, + #[description = "Plik do odtworzenia"] + #[autocomplete = "audio_file"] + melodia: Option, + #[description = "Bezpośredni plik z twojego systemu"] file: Option, + #[description = "Chujowa kompresja!?"] compressed: Option, +) -> Result<()> { + let compressed = compressed.unwrap_or(false); + eprintln!( + "[DEBUG] play() reached, melodia={:?}, file={:?}, compressed={compressed}", + melodia, + file.as_ref().map(|f| &f.filename) + ); + let result = play_impl(ctx, melodia, file, compressed).await; + if let Err(e) = &result { + eprintln!("[DEBUG] play() error: {e}"); + error!("play command failed: {e}"); + } + result +} + +struct ResolvedSource { + url: String, + display_name: String, + kind: SourceKind, +} + +fn clean_name(raw: &str) -> String { + html_decode(&percent_decode(raw)) +} + +// ── URL classification helpers ────────────────────────────────────── + +/// URLs that require yt-dlp extraction (platforms that don't serve +/// raw audio at a direct URL). These MUST be downloaded first. +/// +/// Matches the Python original's `EXTERNAL_URL_PATTERN`: +/// `youtube|youtu.be|soundcloud|bandcamp` (plus twitch). +fn needs_ytdlp(url: &str) -> bool { + let lower = url.to_lowercase(); + lower.contains("youtube.com/") + || lower.contains("youtu.be/") + || lower.contains("twitch.tv/") + || lower.contains("soundcloud.com/") + || lower.contains("bandcamp.com/") +} + +/// Heuristic: does this URL *look* like a live internet radio stream? +/// (Used for logging / display only — dispatch is controlled by `SourceKind`.) +fn is_live_stream_hint(url: &str) -> bool { + let lower = url.to_lowercase(); + // Common Icecast/Shoutcast mount patterns + lower.contains("/stream") + || lower.contains("/live") + || lower.contains("/radio") + || lower.contains(";stream") + || lower.contains("icecast") + || lower.contains("shoutcast") + || lower.ends_with(".m3u8") + || lower.ends_with(".m3u") + || lower.ends_with(".pls") + || lower.contains(":8000/") // common Icecast port + || lower.contains(":8080/") // common alt port +} + +async fn resolve_source( + ctx: Context<'_>, + melodia: Option, + file: Option, +) -> Result { + // ── File attachment ────────────────────────────────────────── + // Discord attachment URLs are direct media file URLs — ffmpeg streams + // them without a full download. + if let Some(file) = file { + return Ok(ResolvedSource { + url: file.url.clone(), + display_name: file.filename.clone(), + kind: SourceKind::Stream, + }); + } + + let name = melodia.ok_or_else(|| Error::Custom(lang::L_MUSIC_NEED_NAME.into()))?; + + // ── Autocomplete cache (per-user: (UserId, filename) → SearchHit) ─ + if let Some(hit) = ctx.data().ac_lookup(ctx.author().id, &name).await { + let path = hit.url_path().unwrap_or(""); + let path = if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + }; + let url = ctx.data().copyparty.build_url(&path); + let display_name = clean_name(hit.filename().unwrap_or(&name)); + return Ok(ResolvedSource { + url, + display_name, + kind: SourceKind::Stream, + }); + } + + // ── Direct Copyparty path (starts with '/') ────────────────── + if name.starts_with('/') { + let url = ctx.data().copyparty.build_url(&name); + let display_name = clean_name(name.rsplit('/').next().unwrap_or(&name)); + return Ok(ResolvedSource { + url, + display_name, + kind: SourceKind::Stream, + }); + } + + // ── Direct HTTP(S) URL ─────────────────────────────────────── + if name.starts_with("http://") || name.starts_with("https://") { + let url = name.clone(); + let display_name = clean_name(url.rsplit('/').next().unwrap_or(&url)); + + // yt-dlp platforms → download; everything else (radio, HLS, + // playlists, Copyparty, direct media) → ffmpeg stream. + let kind = if needs_ytdlp(&url) { + SourceKind::YtDlp + } else { + SourceKind::Stream + }; + return Ok(ResolvedSource { + url, + display_name, + kind, + }); + } + + // ── Fallback: search Copyparty ─────────────────────────────── + info!(query = %name, "searching copyparty for audio"); + let results = ctx + .data() + .copyparty + .search(&name, 5, &["mp3", "wav", "ogg", "flac", "m4a", "opus"]) + .await + .map_err(|e| { + error!(query = %name, error = %e, "copyparty search failed"); + Error::Custom(lang::fmt_music_search_failed(&e)) + })?; + let hit = results + .first() + .ok_or_else(|| Error::Custom(lang::L_MUSIC_NO_AUDIO_ON_CP.into()))?; + let path = hit + .url_path() + .ok_or_else(|| Error::Custom(lang::L_MUSIC_NO_URL_IN_RESULT.into()))?; + let path = if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + }; + let url = ctx.data().copyparty.build_url(&path); + let display_name = clean_name(hit.filename().unwrap_or("unknown")); + Ok(ResolvedSource { + url, + display_name, + kind: SourceKind::Stream, + }) +} + +/// Strip `?pw=...` / `&pw=...` query params from a URL for safe logging. +fn redact_url(url: &str) -> &str { + let Some(q) = url.find('?') else { + return url; + }; + let before = &url[..q]; + let after = &url[q..]; + // Strip `?pw=...` or `&pw=...` + if after.strip_prefix("?pw=").is_some() { + return before; + } + if let Some(idx) = after.find("&pw=") { + return &url[..q + idx]; + } + url +} + +/// Strip tracking/playlist params (`&...`) from YouTube URLs. +/// yt-dlp only needs `?v=ID`; extra `&` params can cause failures. +pub(crate) fn sanitize_yt_url(url: &str) -> String { + let lower = url.to_lowercase(); + if lower.contains("youtube.com/") || lower.contains("youtu.be/") { + if let Some(pos) = url.find('&') { + return url[..pos].to_string(); + } + } + url.to_string() +} + +// ── Playlist resolution is handled by ffmpeg natively ────────────── +// ffmpeg reads `.m3u`/`.m3u8`/`.pls` playlists directly (HLS segment +// fetching, PLS/M3U entry parsing) — no manual resolution needed. +// This mirrors the Python bot's `FFmpegPCMAudio(source, ...)` approach. + +async fn play_impl( + ctx: Context<'_>, + melodia: Option, + file: Option, + compressed: bool, +) -> Result<()> { + ctx.defer().await?; + + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + + let channel_id = ctx + .serenity_context() + .cache + .guild(guild_id) + .and_then(|g| { + g.voice_states + .get(&ctx.author().id) + .and_then(|vs| vs.channel_id) + }) + .ok_or_else(|| lang::err_not_in_voice())?; + + let src = resolve_source(ctx, melodia, file).await?; + + let call; + let input; + + if compressed { + // ── Compressed path: download → compress → decode ────────── + // Defer stopping current track until compression is done. + let audio_bytes = match src.kind { + SourceKind::YtDlp => { + let clean_url = sanitize_yt_url(&src.url); + info!(url = %clean_url, "downloading via yt-dlp for compression"); + crate::voice::ytdlp_download(&clean_url).await? + } + SourceKind::Stream => { + if is_live_stream_hint(&src.url) { + ctx.say(lang::L_MUSIC_CANT_COMPRESS_LIVE).await?; + // Fall through to normal streaming below + let live = true; + info!(url = %redact_url(&src.url), live, "live stream — compression skipped, streaming via ffmpeg"); + // Stop current, connect, stream + { + let mut map = ctx.data().music_tracks.lock().await; + if let Some(handle) = map.remove(&guild_id.get()) { + let _ = handle.stop(); + } + } + let (call_res, input_res) = tokio::join!( + ctx.data().voice_manager.connect(guild_id, channel_id), + crate::voice::ffmpeg_stream_input(&src.url) + ); + call = call_res?; + input = input_res?; + // skip the rest of the compressed path + let handle = { + let mut call_lock = call.lock().await; + call_lock.play_input(input) + }; + ctx.data() + .music_tracks + .lock() + .await + .insert(guild_id.get(), handle); + ctx.say(lang::fmt_music_playing(&src.display_name)).await?; + return Ok(()); + } + info!(url = %redact_url(&src.url), "downloading for compression"); + crate::voice::download_url(&ctx.data().http_client, &src.url).await? + } + }; + + info!("degrading audio ({} bytes)", audio_bytes.len()); + let degraded = crate::voice::degrade_audio(&audio_bytes).await?; + + // Deferred stop: only interrupt current track now that we're ready. + { + let mut map = ctx.data().music_tracks.lock().await; + if let Some(handle) = map.remove(&guild_id.get()) { + let _ = handle.stop(); + } + } + call = ctx + .data() + .voice_manager + .connect(guild_id, channel_id) + .await?; + input = degraded; + } else { + // ── Normal path: stop first, then stream/download ────────── + { + let mut map = ctx.data().music_tracks.lock().await; + if let Some(handle) = map.remove(&guild_id.get()) { + let _ = handle.stop(); + } + } + + match src.kind { + SourceKind::YtDlp => { + let clean_url = sanitize_yt_url(&src.url); + info!(url = %clean_url, "downloading via yt-dlp"); + let audio_bytes = crate::voice::ytdlp_download(&clean_url).await?; + let (call_res, input_res) = tokio::join!( + ctx.data().voice_manager.connect(guild_id, channel_id), + tokio::task::spawn_blocking(move || crate::voice::decode_audio(&audio_bytes)) + ); + call = call_res?; + input = input_res??; + } + SourceKind::Stream => { + let live = is_live_stream_hint(&src.url); + info!(url = %redact_url(&src.url), live, "streaming via ffmpeg subprocess"); + let (call_res, input_res) = tokio::join!( + ctx.data().voice_manager.connect(guild_id, channel_id), + crate::voice::ffmpeg_stream_input(&src.url) + ); + call = call_res?; + input = input_res?; + } + } + } + + let handle = { + let mut call_lock = call.lock().await; + call_lock.play_input(input) + }; + + ctx.data() + .music_tracks + .lock() + .await + .insert(guild_id.get(), handle); + + ctx.say(lang::fmt_music_playing(&src.display_name)).await?; + Ok(()) +} + +/// Stop playback and disconnect from voice +#[poise::command(slash_command)] +pub async fn stop(ctx: Context<'_>) -> Result<()> { + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + ctx.data().voice_manager.disconnect(guild_id).await?; + ctx.data().music_tracks.lock().await.remove(&guild_id.get()); + ctx.send( + poise::CreateReply::default() + .content(lang::L_MUSIC_STOPPED) + .ephemeral(true), + ) + .await?; + Ok(()) +} + +/// Join your voice channel +#[poise::command(slash_command)] +pub async fn join(ctx: Context<'_>) -> Result<()> { + ctx.defer().await?; + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + + let channel_id = ctx.serenity_context().cache.guild(guild_id).and_then(|g| { + g.voice_states + .get(&ctx.author().id) + .and_then(|vs| vs.channel_id) + }); + + match channel_id { + Some(cid) => match ctx.data().voice_manager.connect(guild_id, cid).await { + Ok(_) => { + ctx.say(lang::fmt_voice_joined(&cid)).await?; + } + Err(e) => { + ctx.say(lang::fmt_voice_join_failed(&e)).await?; + } + }, + None => { + ctx.say(lang::L_VOICE_USER_NOT_IN_VC).await?; + } + } + Ok(()) +} + +pub fn commands() -> Vec> { + vec![grajcuj(), stop(), join()] +} diff --git a/src/commands/qr.rs b/src/commands/qr.rs new file mode 100644 index 0000000..26232ee --- /dev/null +++ b/src/commands/qr.rs @@ -0,0 +1,87 @@ +use poise::serenity_prelude as serenity; + +use crate::Data; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// QR group +#[poise::command(slash_command, subcommands("dekodowanie", "generowanie",))] +pub async fn qr(_ctx: Context<'_>) -> Result<()> { + Ok(()) +} + +/// Generate a QR code from text +#[poise::command(slash_command)] +pub async fn generowanie( + ctx: Context<'_>, + #[description = "Zawartość kodu QR"] text: String, +) -> Result<()> { + ctx.defer().await?; + + let code = qrcode::QrCode::new(text.as_bytes())?; + let image = code.render::>().build(); + + let scaled = image::imageops::resize( + &image, + image.width() * 10, + image.height() * 10, + image::imageops::Nearest, + ); + + let mut buf = std::io::Cursor::new(Vec::new()); + scaled.write_to(&mut buf, image::ImageFormat::Png)?; + + ctx.send( + poise::CreateReply::default() + .content(lang::L_QR_GENERATED) + .attachment(serenity::CreateAttachment::bytes( + buf.into_inner(), + "qrcode.png", + )), + ) + .await?; + Ok(()) +} + +/// Decode a QR code from an uploaded image +#[poise::command(slash_command)] +pub async fn dekodowanie( + ctx: Context<'_>, + #[description = "Obraz zawierający kod QR"] image: serenity::Attachment, +) -> Result<()> { + ctx.defer().await?; + + let img_data = ctx + .data() + .http_client + .get(&image.url) + .send() + .await? + .bytes() + .await?; + let img = image::load_from_memory(&img_data)?; + let gray = img.to_luma8(); + + let decoded = decode_qr_simple(&gray); + match decoded { + Some(text) => { + ctx.say(lang::fmt_qr_decoded(&text)).await?; + } + None => { + ctx.say(lang::L_QR_DECODE_FAILED).await?; + } + } + Ok(()) +} + +fn decode_qr_simple(img: &image::ImageBuffer, Vec>) -> Option { + rxing::helpers::detect_in_luma(img.as_raw().to_vec(), img.width(), img.height(), None) + .ok() + .map(|r| r.getText().to_owned()) +} + +pub fn commands() -> Vec> { + vec![qr()] +} diff --git a/src/commands/trigger.rs b/src/commands/trigger.rs new file mode 100644 index 0000000..7453e03 --- /dev/null +++ b/src/commands/trigger.rs @@ -0,0 +1,199 @@ +use poise::serenity_prelude as serenity; + +use super::autocomplete::trigger_name; +use crate::Data; +use crate::database; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// Trigger keyword management +#[poise::command( + slash_command, + subcommands("create", "add", "remove", "delete", "list", "help",) +)] +pub async fn trigger(_ctx: Context<'_>) -> Result<()> { + Ok(()) +} + +/// Create a new trigger group +#[poise::command(slash_command)] +pub async fn create( + ctx: Context<'_>, + #[description = "Logical group name"] + #[autocomplete = "trigger_name"] + name: String, +) -> Result<()> { + match database::create_trigger(&ctx.data().db_pool, &name).await { + Ok(_) => { + ctx.data().reload_triggers().await; + ctx.say(lang::fmt_trigger_created(&name)).await?; + } + Err(e) => { + ctx.say(lang::fmt_trigger_failed(&e)).await?; + } + } + Ok(()) +} + +/// Add keywords and/or responses to a trigger group +#[poise::command(slash_command)] +pub async fn add( + ctx: Context<'_>, + #[description = "Existing group name"] + #[autocomplete = "trigger_name"] + name: String, + #[description = "Keywords separated by \\"] trigger_keywords: Option, + #[description = "Responses separated by \\"] response: Option, +) -> Result<()> { + let t = database::get_trigger_by_name(&ctx.data().db_pool, &name) + .await? + .ok_or_else(|| lang::err_trigger_not_found())?; + + let mut modified = false; + if let Some(kw) = trigger_keywords { + for keyword in kw.split('\\') { + let k = keyword.trim(); + if !k.is_empty() { + database::add_keyword(&ctx.data().db_pool, t.id, k).await?; + modified = true; + } + } + } + if let Some(r) = response { + for resp in r.split('\\') { + let r = resp.trim(); + if !r.is_empty() { + database::add_response(&ctx.data().db_pool, t.id, r).await?; + modified = true; + } + } + } + + if modified { + ctx.data().reload_triggers().await; + } + ctx.say(lang::L_TRIGGER_UPDATED).await?; + Ok(()) +} + +/// Remove keyword or response from a group +#[poise::command(slash_command)] +pub async fn remove( + ctx: Context<'_>, + #[description = "Group name"] + #[autocomplete = "trigger_name"] + name: String, + #[description = "Keyword to remove"] trigger_keyword: Option, + #[description = "Response to remove"] response: Option, +) -> Result<()> { + let t = database::get_trigger_by_name(&ctx.data().db_pool, &name) + .await? + .ok_or_else(|| lang::err_trigger_not_found())?; + + let mut modified = false; + if let Some(kw) = trigger_keyword { + database::remove_keyword(&ctx.data().db_pool, t.id, &kw).await?; + modified = true; + } + if let Some(r) = response { + database::remove_response(&ctx.data().db_pool, t.id, &r).await?; + modified = true; + } + + if modified { + ctx.data().reload_triggers().await; + } + ctx.say(lang::L_TRIGGER_UPDATED).await?; + Ok(()) +} + +/// Delete an entire trigger group +#[poise::command(slash_command)] +pub async fn delete( + ctx: Context<'_>, + #[description = "Group name"] + #[autocomplete = "trigger_name"] + name: String, +) -> Result<()> { + let t = database::get_trigger_by_name(&ctx.data().db_pool, &name) + .await? + .ok_or_else(|| lang::err_trigger_not_found())?; + + database::delete_trigger(&ctx.data().db_pool, t.id).await?; + ctx.data().reload_triggers().await; + ctx.say(lang::fmt_trigger_deleted(&name)).await?; + Ok(()) +} + +/// List keywords and responses for a group +#[poise::command(slash_command)] +pub async fn list( + ctx: Context<'_>, + #[description = "Group name"] + #[autocomplete = "trigger_name"] + name: String, +) -> Result<()> { + let t = database::get_trigger_by_name(&ctx.data().db_pool, &name) + .await? + .ok_or_else(|| lang::err_trigger_not_found())?; + + let keywords = database::get_keywords(&ctx.data().db_pool, t.id).await?; + let responses = database::get_responses(&ctx.data().db_pool, t.id).await?; + + let mut embed = serenity::CreateEmbed::default() + .title(lang::fmt_trigger_title(&name)) + .color(0x00aaff); + + embed = embed.field( + "Keywords", + if keywords.is_empty() { + lang::L_TRIGGER_KW_NONE.into() + } else { + keywords + .iter() + .map(|r| format!("{r}")) + .collect::>() + .join("\n") + }, + false, + ); + embed = embed.field( + "Responses", + if responses.is_empty() { + lang::L_TRIGGER_KW_NONE.into() + } else { + responses + .iter() + .map(|r| format!("{r}")) + .collect::>() + .join("\n") + }, + false, + ); + + ctx.send(poise::CreateReply::default().embed(embed)).await?; + Ok(()) +} + +/// Display trigger command usage info +#[poise::command(slash_command)] +pub async fn help(ctx: Context<'_>) -> Result<()> { + let help = [ + "**Trigger Commands**", + "`/trigger create ` — Create new trigger group", + "`/trigger add ` — Add keywords/responses (separate with \\)", + "`/trigger remove ` — Remove keywords/responses", + "`/trigger delete ` — Delete entire group", + "`/trigger list ` — View group details", + ] + .join("\n"); + + ctx.say(help).await?; + Ok(()) +} + +pub fn commands() -> Vec> { + vec![trigger()] +} diff --git a/src/commands/tts.rs b/src/commands/tts.rs new file mode 100644 index 0000000..ee13596 --- /dev/null +++ b/src/commands/tts.rs @@ -0,0 +1,106 @@ +use poise::serenity_prelude as serenity; + +use crate::Data; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +#[derive(poise::ChoiceParameter)] +pub enum OutputMode { + #[name = "File"] + File, + #[name = "Voice Channel"] + Vc, +} + +/// Generate Google TTS (Polish) and output to file or voice channel +#[poise::command(slash_command)] +pub async fn tts( + ctx: Context<'_>, + #[description = "Text to synthesize"] text: String, + #[description = "Output destination (file or vc)"] output: Option, +) -> Result<()> { + ctx.defer().await?; + let lang = "pl"; + let tts_url = format!( + "https://translate.google.com/translate_tts?ie=UTF-8&q={}&tl={}&client=tw-ob", + urlencoding(&text), + lang + ); + + let output = output.unwrap_or(OutputMode::Vc); + + let audio_data = ctx + .data() + .http_client + .get(&tts_url) + .send() + .await? + .bytes() + .await?; + + match output { + OutputMode::File => { + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + let path = format!( + "{}/tts_{}.mp3", + crate::config::TEMP_PATH, + uuid::Uuid::new_v4() + ); + tokio::fs::write(&path, &audio_data).await?; + ctx.send( + poise::CreateReply::default() + .content(lang::L_TTS_SAVED) + .attachment(serenity::CreateAttachment::path(&path).await.unwrap()), + ) + .await?; + } + OutputMode::Vc => { + let guild_id = ctx.guild_id().ok_or_else(|| lang::err_not_in_guild())?; + let channel_id = ctx + .serenity_context() + .cache + .guild(guild_id) + .and_then(|g| { + g.voice_states + .get(&ctx.author().id) + .and_then(|vs| vs.channel_id) + }) + .ok_or_else(|| lang::err_not_in_voice())?; + + let call = ctx + .data() + .voice_manager + .connect(guild_id, channel_id) + .await?; + let input = crate::voice::decode_audio(&audio_data)?; + let mut call_lock = call.lock().await; + if call_lock.current_channel().is_some() { + call_lock.play_input(input); + } + ctx.say(lang::L_TTS_PLAYING).await?; + } + } + Ok(()) +} + +pub fn urlencoding(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for &byte in s.as_bytes() { + match byte { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'_' | b'-' | b'.' | b'~' => { + out.push(byte as char) + } + b' ' => out.push('+'), + _ => out.push_str(&format!("%{:02X}", byte)), + } + } + out +} + +pub fn commands() -> Vec> { + vec![tts()] +} diff --git a/src/commands/utilities.rs b/src/commands/utilities.rs new file mode 100644 index 0000000..10ce403 --- /dev/null +++ b/src/commands/utilities.rs @@ -0,0 +1,918 @@ +use poise::serenity_prelude as serenity; +use tracing::warn; + +use super::autocomplete::any_file; +use super::music::sanitize_yt_url; +use crate::Data; +use crate::copyparty::{html_decode, percent_decode}; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +// Above this size (or when `serwer` is forced) the result is uploaded to +// copyparty instead of being attached to the Discord reply. Discord's own +// attachment limit is 25 MiB for non-Nitro, but 10 MiB matches the Python +// original and keeps replies snappy. +const LARGE_FILE_THRESHOLD: usize = 10 * 1024 * 1024; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum FileType { + Video, + Audio, + Image, + Unknown, +} + +/// Classify a file by its extension. +/// +/// Mirrors Python's `mimetypes.guess_type`-based `filetype()` helper without +/// pulling in a mime crate: the set of extensions below covers everything the +/// original bot handled. +fn file_type(filename: &str) -> FileType { + let Some((_, ext)) = filename.rsplit_once('.') else { + return FileType::Unknown; + }; + if ext.is_empty() || ext.contains('/') { + return FileType::Unknown; + } + match ext.to_ascii_lowercase().as_str() { + "mp4" | "avi" | "mkv" | "mov" | "webm" | "flv" | "wmv" | "m4v" | "mpg" | "mpeg" | "3gp" + | "ts" => FileType::Video, + "mp3" | "wav" | "ogg" | "flac" | "m4a" | "opus" | "aac" | "wma" | "oga" => FileType::Audio, + "jpg" | "jpeg" | "png" | "gif" | "bmp" | "webp" | "tiff" | "tif" | "avif" => { + FileType::Image + } + _ => FileType::Unknown, + } +} + +/// Lowercased extension without the leading dot, or `None` if there is none. +fn ext_of(filename: &str) -> Option { + let (_, ext) = filename.rsplit_once('.')?; + if ext.is_empty() || ext.contains('/') { + None + } else { + Some(ext.to_ascii_lowercase()) + } +} + +/// Filename stem: everything before the final dot of the basename. +fn stem_of(filename: &str) -> &str { + let base = filename.rsplit('/').next().unwrap_or(filename); + match base.rsplit_once('.') { + Some((stem, _)) if !stem.is_empty() => stem, + _ => base, + } +} + +fn temp_path_for(name: &str) -> String { + format!("{}/{}", crate::config::TEMP_PATH, name) +} + +/// Best-effort removal of temp files; ignores missing files. +async fn cleanup(paths: &[&str]) { + for p in paths { + let _ = tokio::fs::remove_file(p).await; + } +} + +/// Resolve the input source — either an autocomplete-picked copyparty file or +/// a Discord attachment — downloading the bytes and returning them alongside +/// the original filename. +/// +/// Replies with an appropriate user-facing message and returns `Ok(None)` when +/// no valid source was supplied, so callers can early-return without sending a +/// second message. +async fn resolve_input( + ctx: &Context<'_>, + plik: Option<&str>, + zalacznik: Option<&serenity::Attachment>, +) -> Result, String)>> { + if let Some(name) = plik { + // Autocomplete-selected copyparty file: the choice value is the decoded + // filename, stashed into the per-user ac_cache by `any_file`. + let hit = match ctx.data().ac_lookup(ctx.author().id, name).await { + Some(h) => h, + None => { + ctx.say(lang::L_UTIL_PICK_FROM_AUTOCOMPLETE).await?; + return Ok(None); + } + }; + let path = hit.url_path().unwrap_or(""); + let path = if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + }; + let full_url = ctx.data().copyparty.build_url(&path); + let resp = ctx.data().http_client.get(&full_url).send().await?; + if !resp.status().is_success() { + ctx.say(lang::L_UTIL_DOWNLOAD_FAILED).await?; + return Ok(None); + } + let bytes = resp.bytes().await?.to_vec(); + return Ok(Some((bytes, name.to_string()))); + } + + if let Some(att) = zalacznik { + let resp = ctx.data().http_client.get(&att.url).send().await?; + if !resp.status().is_success() { + ctx.say(lang::L_UTIL_DOWNLOAD_FAILED).await?; + return Ok(None); + } + let bytes = resp.bytes().await?.to_vec(); + return Ok(Some((bytes, att.filename.clone()))); + } + + ctx.say(lang::L_UTIL_NEED_INPUT).await?; + Ok(None) +} + +/// Output extension the compressor produces for a given input. +/// +/// - Video keeps its container extension (ffmpeg re-muxes in place). +/// - Audio is always re-encoded to Opus in an `.ogg` container. +/// - JPEG/GIF keep their extension; other images are rewritten to `.jpg`. +fn compress_output_ext(ft: FileType, original: &str) -> String { + let orig = ext_of(original); + match ft { + FileType::Video => orig.unwrap_or_else(|| "mp4".to_string()), + FileType::Audio => "ogg".to_string(), + FileType::Image => match orig.as_deref() { + Some("jpg" | "jpeg" | "gif") => orig.unwrap(), + _ => "jpg".to_string(), + }, + FileType::Unknown => String::new(), + } +} + +/// Build the ffmpeg/magick command for extreme-low-quality compression. +/// Returns `None` for unsupported (`Unknown`) file types. +fn build_compress_command( + ft: FileType, + input: &str, + output: &str, +) -> Option { + match ft { + FileType::Video => { + let mut c = tokio::process::Command::new("ffmpeg"); + c.arg("-y") + .arg("-i") + .arg(input) + .arg("-c:v") + .arg("libx264") + .arg("-crf") + .arg("51") + .arg("-b:v") + .arg("100") + .arg("-vf") + .arg("fps=10,scale=trunc(iw*0.5/2)*2:trunc(ih*0.5/2)*2") + .arg("-preset") + .arg("ultrafast") + .arg("-c:a") + .arg("libopus") + .arg("-b:a") + .arg("500") + .arg(output); + Some(c) + } + FileType::Audio => { + let mut c = tokio::process::Command::new("ffmpeg"); + c.arg("-y") + .arg("-i") + .arg(input) + .arg("-c:a") + .arg("libopus") + .arg("-b:a") + .arg("500") + .arg("-ar") + .arg("12000") + .arg("-ac") + .arg("1") + .arg(output); + Some(c) + } + FileType::Image => { + let mut c = tokio::process::Command::new("magick"); + c.arg(input); + match ext_of(input).as_deref() { + Some("gif") => { + c.arg("-coalesce") + .arg("-resize") + .arg("30%") + .arg("-colors") + .arg("32") + .arg("-strip") + .arg("-layers") + .arg("Optimize"); + } + // jpg/jpeg stay jpg; other images are rewritten to .jpg by + // `compress_output_ext` and magick infers the encoder from the + // output extension. + _ => { + c.arg("-strip") + .arg("-resize") + .arg("30%") + .arg("-quality") + .arg("5"); + } + } + c.arg(output); + Some(c) + } + FileType::Unknown => None, + } +} + +enum DeliverKind { + Compress, + Convert, +} + +/// Compress a single in-memory blob (an attachment body) into a compressed +/// file, returning the encoded bytes and the output filename. +/// +/// Mirrors the per-attachment path of Python's `Compression` message command, +/// including sequential `TEMP_PATH` reuse so identical attachment stems don't +/// collide (each iteration writes, encodes, then `cleanup`s before the next). +async fn compress_attachment_bytes( + ft: FileType, + original: &str, + input_bytes: &[u8], +) -> Option<(Vec, String)> { + let stem = stem_of(original).to_string(); + let out_ext = compress_output_ext(ft, original); + let out_filename = format!("compressed_{stem}.{out_ext}"); + let input_path = temp_path_for(original); + let output_path = temp_path_for(&out_filename); + let cleanup_paths = [input_path.as_str(), output_path.as_str()]; + + if tokio::fs::write(&input_path, input_bytes).await.is_err() { + return None; + } + let mut cmd = match build_compress_command(ft, &input_path, &output_path) { + Some(c) => c, + None => { + cleanup(&cleanup_paths).await; + return None; + } + }; + let res = match cmd.output().await { + Ok(out) if out.status.success() => tokio::fs::read(&output_path).await.ok(), + _ => None, + }; + cleanup(&cleanup_paths).await; + res.map(|b| (b, out_filename)) +} + +/// Strip characters that Discord/coping filenames can't safely carry. Mirrors +/// Python's `Downloader._sanitize_filename`. +fn sanitize_filename(name: &str) -> String { + const BAD: &[char] = &['<', '>', ':', '"', '/', '\\', '|', '?', '*']; + name.chars() + .filter(|c| !BAD.contains(c)) + .collect::() + .trim() + .to_string() +} + +/// Extract `https://` URLs containing a path (host + `/...`) from a message. +/// Faithful port of Python's `Downloader.URL_PATTERN` regex used by the +/// message-context `download` command — only URLs with at least one `/` after +/// the host are matched, and `http://` links are intentionally ignored. +fn extract_urls(content: &str) -> Vec { + let bytes = content.as_bytes(); + let mut urls = Vec::new(); + let mut i = 0; + while let Some(pos) = content[i..].find("https://") { + let start = i + pos; + let after = start + "https://".len(); + + // Consume the host (until `/` or whitespace). + let mut j = after; + while j < bytes.len() && !bytes[j].is_ascii_whitespace() && bytes[j] != b'/' { + j += 1; + } + + if j < bytes.len() && bytes[j] == b'/' { + // Consume the rest of the URL (until whitespace). + let mut k = j; + while k < bytes.len() && !bytes[k].is_ascii_whitespace() { + k += 1; + } + urls.push(content[start..k].to_string()); + i = k; + } else { + // No path component — not matched by the regex; advance past it. + i = after; + } + } + urls +} + +/// Locate the single file yt-dlp produced in `dir`, picking the most recently +/// modified one (matches Python's `os.listdir`-first fallback behaviour while +/// being deterministic enough for the per-call unique subdir). +async fn find_downloaded_file(dir: &str) -> Option<(String, String)> { + let mut rd = tokio::fs::read_dir(dir).await.ok()?; + let mut newest: Option<(std::time::SystemTime, String)> = None; + while let Ok(Some(entry)) = rd.next_entry().await { + let meta = match entry.metadata().await { + Ok(m) => m, + Err(_) => continue, + }; + if !meta.is_file() { + continue; + } + let mtime = match meta.modified() { + Ok(t) => t, + Err(_) => continue, + }; + if let Some((t, _)) = &newest { + if *t >= mtime { + continue; + } + } + newest = Some((mtime, entry.path().to_string_lossy().into_owned())); + } + newest.map(|(_, p)| { + let fname = std::path::Path::new(&p) + .file_name() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_default(); + (p, fname) + }) +} + +/// Hand the produced download back to the user: attach when below the size +/// threshold (and upload isn't forced), otherwise push to copyparty and reply +/// with the download URL. Mirrors Python's `Downloader` slash `/download`. +async fn deliver_download( + ctx: &Context<'_>, + bytes: Vec, + filename: String, + catbox: bool, +) -> Result<()> { + if bytes.len() > LARGE_FILE_THRESHOLD || catbox { + match ctx.data().copyparty.upload(&filename, bytes).await { + Ok(url) => { + ctx.say(lang::fmt_download_uploaded(&url)).await?; + } + Err(e) => { + tracing::warn!("copyparty upload failed: {e}"); + ctx.say(lang::L_UTIL_UPLOAD_FAILED).await?; + } + } + } else { + ctx.send( + poise::CreateReply::default() + .attachment(serenity::CreateAttachment::bytes(bytes, filename)), + ) + .await?; + } + Ok(()) +} + +/// Run yt-dlp on `url` and deliver the resulting file. The caller is expected +/// to have already called `ctx.defer()`. Each invocation uses a fresh unique +/// subdirectory under `TEMP_PATH` (matching Python's per-call +/// `tempfile.TemporaryDirectory`) so concurrent downloads never collide. +async fn run_download( + ctx: &Context<'_>, + url: &str, + catbox: bool, + fformat: Option<&str>, + filename_user: Option<&str>, +) -> Result<()> { + let fmt = fformat.unwrap_or("mp4"); + let is_audio = matches!(fmt, "mp3" | "wav" | "m4a"); + let dl_dir = format!("{}/dl_{}", crate::config::TEMP_PATH, uuid::Uuid::new_v4()); + tokio::fs::create_dir_all(&dl_dir).await.ok(); + + let out_template = match filename_user { + Some(name) => { + // Custom filename: sanitize, cap at 100 chars, let yt-dlp append ext. + let clean: String = sanitize_filename(name).chars().take(100).collect(); + format!("{dl_dir}/{clean}.%(ext)s") + } + None => format!("{dl_dir}/%(title).100s.%(ext)s"), + }; + + let mut cmd = tokio::process::Command::new("yt-dlp"); + cmd.arg("-f") + .arg("bestaudio/bestvideo+bestaudio") + .arg("--merge-output-format") + .arg("mp4") + .arg("-o") + .arg(&out_template); + if is_audio { + cmd.arg("--extract-audio").arg("--audio-format").arg(fmt); + } + cmd.arg(sanitize_yt_url(url)); + + let output = cmd + .output() + .await + .map_err(|e| Error::Custom(lang::fmt_ytdlp_err(&e)))?; + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + // Keep the full yt-dlp trace in the logs; the user only sees the + // short `fmt_download_failed` (translated) message. + warn!(url = %url, "yt-dlp failed:\n{}", stderr.trim_end()); + let _ = tokio::fs::remove_dir_all(&dl_dir).await; + ctx.say(lang::fmt_download_failed(&stderr)).await?; + return Ok(()); + } + + let Some((actual_path, basename)) = find_downloaded_file(&dl_dir).await else { + let _ = tokio::fs::remove_dir_all(&dl_dir).await; + ctx.say(lang::fmt_download_failed("no file produced")) + .await?; + return Ok(()); + }; + + // Determine the display filename for the upload/attachment. + let final_name = match filename_user { + Some(name) => { + let clean: String = sanitize_filename(name).chars().take(100).collect(); + let ext = std::path::Path::new(&basename) + .extension() + .map(|e| format!(".{}", e.to_string_lossy())) + .unwrap_or_default(); + format!("{clean}{ext}") + } + None => basename, + }; + + let bytes = match tokio::fs::read(&actual_path).await { + Ok(b) => b, + Err(e) => { + let _ = tokio::fs::remove_dir_all(&dl_dir).await; + ctx.say(lang::fmt_download_failed(&e.to_string())).await?; + return Ok(()); + } + }; + + let deliver_result = deliver_download(ctx, bytes, final_name, catbox).await; + let _ = tokio::fs::remove_dir_all(&dl_dir).await; + deliver_result +} + +/// Hand the produced bytes back to the user. +/// +/// Below the size threshold (and when upload isn't forced) the result is +/// attached directly to the reply. Otherwise it's pushed to copyparty and the +/// reply contains the download URL. Upload failures are reported gracefully +/// rather than propagated, matching the Python original. +async fn deliver( + ctx: &Context<'_>, + bytes: Vec, + filename: String, + force_upload: bool, + kind: DeliverKind, +) -> Result<()> { + if bytes.len() > LARGE_FILE_THRESHOLD || force_upload { + match ctx.data().copyparty.upload(&filename, bytes).await { + Ok(url) => { + let msg = match kind { + DeliverKind::Compress => lang::fmt_util_compressed(&url), + DeliverKind::Convert => lang::fmt_util_converted(&url), + }; + ctx.say(msg).await?; + } + Err(e) => { + tracing::warn!("copyparty upload failed: {e}"); + ctx.say(lang::L_UTIL_UPLOAD_FAILED).await?; + } + } + } else { + ctx.send( + poise::CreateReply::default() + .attachment(serenity::CreateAttachment::bytes(bytes, filename)), + ) + .await?; + } + Ok(()) +} + +/// Compress media to extreme low quality +#[poise::command(slash_command, aliases("kompresja"))] +pub async fn compress( + ctx: Context<'_>, + #[description = "File from Copyparty"] + #[autocomplete = "any_file"] + plik: Option, + #[description = "File from Discord"] zalacznik: Option, + #[description = "Force upload to Copyparty"] serwer: Option, +) -> Result<()> { + ctx.defer().await?; + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + + let (input_bytes, original_filename) = + match resolve_input(&ctx, plik.as_deref(), zalacznik.as_ref()).await? { + Some(v) => v, + None => return Ok(()), + }; + + let ft = file_type(&original_filename); + let stem = stem_of(&original_filename).to_string(); + let out_ext = compress_output_ext(ft, &original_filename); + let out_filename = format!("compressed_{stem}.{out_ext}"); + let input_path = temp_path_for(&original_filename); + let output_path = temp_path_for(&out_filename); + let cleanup_paths = [input_path.as_str(), output_path.as_str()]; + + if let Err(e) = tokio::fs::write(&input_path, &input_bytes).await { + tracing::warn!("compress: failed to write input: {e}"); + ctx.say(lang::L_UTIL_COMPRESS_FAILED).await?; + return Ok(()); + } + + let result = match build_compress_command(ft, &input_path, &output_path) { + Some(mut cmd) => cmd.output().await, + None => { + cleanup(&cleanup_paths).await; + ctx.say(lang::L_UTIL_UNSUPPORTED_TYPE).await?; + return Ok(()); + } + }; + + match result { + Ok(out) if out.status.success() => match tokio::fs::read(&output_path).await { + Ok(out_bytes) => { + cleanup(&cleanup_paths).await; + deliver( + &ctx, + out_bytes, + out_filename, + serwer.unwrap_or(false), + DeliverKind::Compress, + ) + .await?; + } + Err(e) => { + tracing::warn!("compress: output missing after encode: {e}"); + cleanup(&cleanup_paths).await; + ctx.say(lang::L_UTIL_COMPRESS_FAILED).await?; + } + }, + _ => { + cleanup(&cleanup_paths).await; + ctx.say(lang::L_UTIL_COMPRESS_FAILED).await?; + } + } + Ok(()) +} + +/// Convert media between formats +#[poise::command(slash_command, aliases("konwersja"))] +pub async fn convert( + ctx: Context<'_>, + #[description = "File from Copyparty"] + #[autocomplete = "any_file"] + plik: Option, + #[description = "File from Discord"] zalacznik: Option, + #[description = "Target format extension (e.g. mp3, wav, avi, png)"] format: String, + #[description = "Force upload to Copyparty"] serwer: Option, +) -> Result<()> { + ctx.defer().await?; + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + + let (input_bytes, original_filename) = + match resolve_input(&ctx, plik.as_deref(), zalacznik.as_ref()).await? { + Some(v) => v, + None => return Ok(()), + }; + + // Normalise the requested extension: lowercase, no leading dot, and reject + // anything containing path separators or extra dots — this keeps it a + // single clean extension component and rules out escaping TEMP_PATH. + let fmt = format.trim().trim_start_matches('.').to_ascii_lowercase(); + if fmt.is_empty() || fmt.contains('/') || fmt.contains('\\') || fmt.contains('.') { + ctx.say(lang::L_UTIL_UNSUPPORTED_TYPE).await?; + return Ok(()); + } + + let ft = file_type(&original_filename); + let stem = stem_of(&original_filename).to_string(); + let out_filename = format!("{stem}.{fmt}"); + let input_path = temp_path_for(&original_filename); + let output_path = temp_path_for(&out_filename); + let cleanup_paths = [input_path.as_str(), output_path.as_str()]; + + if let Err(e) = tokio::fs::write(&input_path, &input_bytes).await { + tracing::warn!("convert: failed to write input: {e}"); + ctx.say(lang::L_UTIL_CONVERT_FAILED).await?; + return Ok(()); + } + + let mut cmd = match ft { + FileType::Video | FileType::Audio => { + let mut c = tokio::process::Command::new("ffmpeg"); + c.arg("-y").arg("-i").arg(&input_path).arg(&output_path); + c + } + FileType::Image => { + let mut c = tokio::process::Command::new("magick"); + c.arg(&input_path).arg(&output_path); + c + } + FileType::Unknown => { + cleanup(&cleanup_paths).await; + ctx.say(lang::L_UTIL_UNSUPPORTED_TYPE).await?; + return Ok(()); + } + }; + + match cmd.output().await { + Ok(out) if out.status.success() => match tokio::fs::read(&output_path).await { + Ok(out_bytes) => { + cleanup(&cleanup_paths).await; + deliver( + &ctx, + out_bytes, + out_filename, + serwer.unwrap_or(false), + DeliverKind::Convert, + ) + .await?; + } + Err(e) => { + tracing::warn!("convert: output missing after encode: {e}"); + cleanup(&cleanup_paths).await; + ctx.say(lang::L_UTIL_CONVERT_FAILED).await?; + } + }, + Ok(out) => { + let stderr = String::from_utf8_lossy(&out.stderr); + cleanup(&cleanup_paths).await; + ctx.say(lang::fmt_util_convert_err(&stderr)).await?; + } + Err(e) => { + cleanup(&cleanup_paths).await; + ctx.say(lang::fmt_util_convert_err(&e.to_string())).await?; + } + } + Ok(()) +} + +/// Upload files from Discord to Copyparty +#[poise::command(slash_command)] +pub async fn upload( + ctx: Context<'_>, + #[description = "File to upload"] zalacznik: serenity::Attachment, +) -> Result<()> { + ctx.defer().await?; + + let data = ctx + .data() + .http_client + .get(&zalacznik.url) + .send() + .await? + .bytes() + .await?; + let file_url = ctx + .data() + .copyparty + .upload(&zalacznik.filename, data.to_vec()) + .await?; + ctx.say(lang::fmt_uploaded(&file_url)).await?; + Ok(()) +} + +/// Search Copyparty file server and return a direct URL +#[poise::command(slash_command)] +pub async fn search( + ctx: Context<'_>, + #[description = "Partial filename to search"] + #[autocomplete = "any_file"] + filename: String, +) -> Result<()> { + // Autocomplete cache: per-user (UserId, filename) → SearchHit. + if let Some(hit) = ctx.data().ac_lookup(ctx.author().id, &filename).await { + let name = html_decode(&percent_decode(hit.filename().unwrap_or(&filename))); + let path = hit.url_path().unwrap_or(""); + let path = if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + }; + let full_url = format!("{}{}", ctx.data().config.copyparty_url, path); + ctx.say(lang::fmt_search_result(&name, &full_url)).await?; + return Ok(()); + } + + // Direct copyparty path from manual input (starts with /) + if filename.starts_with('/') { + let full_url = format!("{}{}", ctx.data().config.copyparty_url, filename); + ctx.say(lang::fmt_search_result(&filename, &full_url)) + .await?; + return Ok(()); + } + + let results = ctx.data().copyparty.search(&filename, 10, &[]).await?; + + if results.is_empty() { + ctx.say(lang::L_SEARCH_NO_RESULTS).await?; + return Ok(()); + } + + let hit = &results[0]; + let name = html_decode(&percent_decode(hit.filename().unwrap_or("unknown"))); + let path = hit.url_path().unwrap_or(""); + let path = if path.starts_with('/') { + path.to_string() + } else { + format!("/{path}") + }; + let full_url = format!("{}{}", ctx.data().config.copyparty_url, path); + + ctx.say(lang::fmt_search_result(&name, &full_url)).await?; + Ok(()) +} + +/// Download media from any yt-dlp-supported URL +#[poise::command(slash_command)] +pub async fn download( + ctx: Context<'_>, + #[description = "Media URL"] url: String, + #[description = "Force upload to Copyparty"] catbox: Option, + #[description = "Output format (mp4, mp3, wav, etc.)"] fformat: Option, + #[description = "Custom output filename"] filename_user: Option, +) -> Result<()> { + ctx.defer().await?; + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + run_download( + &ctx, + &url, + catbox.unwrap_or(false), + fformat.as_deref(), + filename_user.as_deref(), + ) + .await +} + +/// Right-click a message → upload its attachments to Copyparty +/// (mirrors Python's `@commands.message_command(name="upload")`). +/// +/// Display name is not `"upload"` because poise's dispatcher matches commands +/// by name alone without considering `CommandType`, so a context-menu named +/// `"upload"` would shadow — and be shadowed by — the slash `/upload` command +/// (see `poise::dispatch::find_matching_command`). `"Wypchnij załączniki"` +/// avoids the collision while keeping the Polish user-facing label. +#[poise::command(context_menu_command = "Wyślij załączniki")] +pub async fn message_upload(ctx: Context<'_>, message: serenity::Message) -> Result<()> { + if message.attachments.is_empty() { + ctx.say(lang::L_UTIL_NO_ATTACHMENTS).await?; + return Ok(()); + } + ctx.defer().await?; + + let mut urls: Vec = Vec::new(); + let mut failed = 0usize; + for att in &message.attachments { + let resp = match ctx.data().http_client.get(&att.url).send().await { + Ok(r) if r.status().is_success() => r, + _ => { + failed += 1; + continue; + } + }; + let bytes = resp.bytes().await?.to_vec(); + match ctx.data().copyparty.upload(&att.filename, bytes).await { + Ok(url) => urls.push(url), + Err(e) => { + tracing::warn!("upload: copyparty failed for {}: {e}", att.filename); + failed += 1; + } + } + } + + if urls.is_empty() { + ctx.say(lang::L_UTIL_UPLOAD_FAILED).await?; + return Ok(()); + } + + let mut resp = lang::fmt_upload_summary(urls.len()); + for u in &urls { + resp.push_str(u); + resp.push('\n'); + } + if failed > 0 { + resp.push_str(&lang::fmt_upload_partial_failed(failed)); + } + ctx.say(resp).await?; + Ok(()) +} + +/// Right-click a message → compress all of its attachments +/// (mirrors Python's `@commands.message_command(name="kompresja")`). +#[poise::command(context_menu_command = "kompresja")] +pub async fn message_compression(ctx: Context<'_>, message: serenity::Message) -> Result<()> { + if message.attachments.is_empty() { + ctx.say(lang::L_UTIL_NO_ATTACHMENTS).await?; + return Ok(()); + } + ctx.defer().await?; + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + + let total = message.attachments.len(); + let mut attachments_out: Vec = Vec::new(); + let mut made = 0usize; + + for att in &message.attachments { + let resp = match ctx.data().http_client.get(&att.url).send().await { + Ok(r) if r.status().is_success() => r, + _ => continue, + }; + let bytes = match resp.bytes().await { + Ok(b) => b.to_vec(), + Err(_) => continue, + }; + let ft = file_type(&att.filename); + if let Some((out_bytes, out_name)) = + compress_attachment_bytes(ft, &att.filename, &bytes).await + { + attachments_out.push(serenity::CreateAttachment::bytes(out_bytes, out_name)); + made += 1; + } + } + + if attachments_out.is_empty() { + ctx.say(lang::L_UTIL_COMPRESS_FAILED).await?; + return Ok(()); + } + + let mut reply = + poise::CreateReply::default().content(lang::fmt_message_compressed(made, total)); + for a in attachments_out { + reply = reply.attachment(a); + } + ctx.send(reply).await?; + Ok(()) +} + +/// Right-click a message → download every `https://` URL it contains +/// (mirrors Python's `@commands.message_command(name="download")`). +/// +/// Named `"Pobierz z linku"` rather than `"download"` to avoid colliding with +/// the slash `/download` command in poise's name-only dispatcher (see the note +/// on `message_upload`). +#[poise::command(context_menu_command = "Pobierz z linku")] +pub async fn message_download(ctx: Context<'_>, message: serenity::Message) -> Result<()> { + let urls = extract_urls(&message.content); + if urls.is_empty() { + // Faithful to the Python original, which reuses the attachments-missing + // string for the no-URL case. + ctx.say(lang::L_UTIL_NO_ATTACHMENTS).await?; + return Ok(()); + } + ctx.defer().await?; + tokio::fs::create_dir_all(crate::config::TEMP_PATH) + .await + .ok(); + + for url in urls { + if let Err(e) = run_download(&ctx, &url, false, Some("mp4"), None).await { + tracing::warn!("download runner error for {url}: {e}"); + } + } + Ok(()) +} + +/// Echo text to channel +#[poise::command(slash_command)] +pub async fn say(ctx: Context<'_>, #[description = "Message to send"] text: String) -> Result<()> { + ctx.say(&text).await?; + ctx.send( + poise::CreateReply::default() + .content("\u{200b}") + .ephemeral(true), + ) + .await?; + Ok(()) +} + +pub fn commands() -> Vec> { + vec![ + compress(), + convert(), + upload(), + search(), + download(), + say(), + message_upload(), + message_compression(), + message_download(), + ] +} diff --git a/src/commands/web.rs b/src/commands/web.rs new file mode 100644 index 0000000..d68f663 --- /dev/null +++ b/src/commands/web.rs @@ -0,0 +1,122 @@ +use poise::serenity_prelude as serenity; +use scraper::{Html, Selector}; + +use crate::Data; +use crate::error::{Error, Result}; +use crate::lang; + +type Context<'a> = poise::Context<'a, Data, Error>; + +/// Search the web via DuckDuckGo +#[poise::command(slash_command)] +pub async fn web(ctx: Context<'_>, #[description = "Search query"] query: String) -> Result<()> { + ctx.defer().await?; + + let url = format!( + "https://html.duckduckgo.com/html/?q={}", + urlencoding(&query) + ); + let html = ctx + .data() + .http_client + .get(&url) + .send() + .await? + .text() + .await?; + + let results = { + let document = Html::parse_document(&html); + + let result_selector = Selector::parse(".result").unwrap(); + let title_selector = Selector::parse(".result__title a").unwrap(); + let snippet_selector = Selector::parse(".result__snippet").unwrap(); + + let mut results = Vec::new(); + for result in document.select(&result_selector).take(5) { + let title = result + .select(&title_selector) + .next() + .map(|a| a.text().collect::()) + .unwrap_or_default(); + + let url = result + .select(&title_selector) + .next() + .and_then(|a| a.attr("href")) + .unwrap_or("") + .to_string(); + + let snippet = result + .select(&snippet_selector) + .next() + .map(|s| s.text().collect::()) + .unwrap_or_default(); + + let clean_url = if let Some(encoded) = url.split("uddg=").nth(1) { + let encoded = encoded.split('&').next().unwrap_or(encoded); + url_decode(encoded) + } else { + url.to_string() + }; + + results.push((title, clean_url, snippet)); + } + results + }; + + if results.is_empty() { + ctx.say(lang::L_WEB_NO_RESULTS).await?; + return Ok(()); + } + + let mut embed = serenity::CreateEmbed::default() + .title(lang::fmt_web_title(&query)) + .color(0x00aaff); + + for (i, (title, url, snippet)) in results.iter().enumerate() { + embed = embed.field( + format!("{}. {title}", i + 1), + format!("{url}\n{snippet}"), + false, + ); + } + + ctx.send(poise::CreateReply::default().embed(embed)).await?; + Ok(()) +} + +fn url_decode(s: &str) -> String { + let mut result = String::new(); + let mut chars = s.chars(); + while let Some(c) = chars.next() { + match c { + '+' => result.push(' '), + '%' => { + let hi = chars.next().and_then(|c| c.to_digit(16)); + let lo = chars.next().and_then(|c| c.to_digit(16)); + if let (Some(hi), Some(lo)) = (hi, lo) { + result.push(char::from_u32((hi << 4) | lo).unwrap_or('�')); + } else { + result.push('%'); + } + } + _ => result.push(c), + } + } + result +} + +fn urlencoding(s: &str) -> String { + s.chars() + .map(|c| match c { + 'A'..='Z' | 'a'..='z' | '0'..='9' | '_' | '-' | '.' | '~' => c.to_string(), + ' ' => "+".to_string(), + _ => format!("%{:02X}", c as u8), + }) + .collect() +} + +pub fn commands() -> Vec> { + vec![web()] +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..96a6849 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,145 @@ +use std::env; + +pub const TEMP_PATH: &str = "/tmp/senator_sperma"; + +#[derive(Debug, Clone)] +pub struct Config { + pub token: String, + pub status_channel_id: Option, + pub assistant_keyword: String, + pub copyparty_url: String, + pub copyparty_url_upload: String, + pub copyparty_password: Option, + pub openrouter_api_key: Option, + pub openrouter_model: Option, + pub database_path: String, +} + +struct EnvVar { + name: &'static str, + required: bool, + default: Option<&'static str>, + description: &'static str, +} + +const ENV_VARS: &[EnvVar] = &[ + EnvVar { + name: "TOKEN", + required: true, + default: None, + description: "Discord bot token", + }, + EnvVar { + name: "STATUS_CHANNEL_ID", + required: false, + default: None, + description: "status channel ID", + }, + EnvVar { + name: "ASSISTANT_KEYWORD", + required: false, + default: Some("jarvis"), + description: "keyword for AI assistant", + }, + EnvVar { + name: "COPYPARTY_URL", + required: false, + default: Some("http://localhost:3939"), + description: "Copyparty base URL", + }, + EnvVar { + name: "COPYPARTY_URL_UPLOAD", + required: false, + default: Some("http://localhost:3939"), + description: "Copyparty upload URL", + }, + EnvVar { + name: "COPYPARTY_PASSWORD", + required: false, + default: None, + description: "Copyparty auth password", + }, + EnvVar { + name: "OPENROUTER_API_KEY", + required: false, + default: None, + description: "OpenRouter API key (disables AI if missing)", + }, + EnvVar { + name: "OPENROUTER_MODEL", + required: false, + default: Some("cognitivecomputations/dolphin-mistral-24b-venice-edition:free"), + description: "OpenRouter model", + }, + EnvVar { + name: "SENATOR_DB_PATH", + required: false, + default: Some("data/senator.db"), + description: "SQLite database path", + }, + EnvVar { + name: "RUST_LOG", + required: false, + default: Some("info"), + description: "tracing/logging level filter", + }, +]; + +pub fn check_env_vars() { + let mut missing_required = Vec::new(); + let mut missing_optional = Vec::new(); + + for var in ENV_VARS { + match env::var(var.name) { + Ok(_) => continue, + Err(_) => { + if var.required { + tracing::error!("MISSING REQUIRED: {} ({})", var.name, var.description); + missing_required.push(var.name); + } else { + match var.default { + Some(default) => { + tracing::warn!("{} not set, using default: {}", var.name, default) + } + None => tracing::warn!("{} not set ({})", var.name, var.description), + } + missing_optional.push(var.name); + } + } + } + } + + if !missing_required.is_empty() { + tracing::error!( + "Missing required environment variables: {}. The application will not function correctly.", + missing_required.join(", ") + ); + } + + if !missing_optional.is_empty() { + tracing::warn!( + "Missing optional environment variables: {}. Some features will be disabled or use defaults.", + missing_optional.join(", ") + ); + } +} + +impl Config { + pub fn from_env() -> Self { + Self { + token: env::var("TOKEN").expect("Missing TOKEN"), + status_channel_id: env::var("STATUS_CHANNEL_ID") + .ok() + .and_then(|v| v.parse().ok()), + assistant_keyword: env::var("ASSISTANT_KEYWORD").unwrap_or_else(|_| "jarvis".into()), + copyparty_url: env::var("COPYPARTY_URL") + .unwrap_or_else(|_| "http://localhost:3939".into()), + copyparty_url_upload: env::var("COPYPARTY_URL_UPLOAD") + .unwrap_or_else(|_| "http://localhost:3939".into()), + copyparty_password: env::var("COPYPARTY_PASSWORD").ok(), + openrouter_api_key: env::var("OPENROUTER_API_KEY").ok(), + openrouter_model: env::var("OPENROUTER_MODEL").ok(), + database_path: env::var("SENATOR_DB_PATH").unwrap_or_else(|_| "data/senator.db".into()), + } + } +} diff --git a/src/copyparty/mod.rs b/src/copyparty/mod.rs new file mode 100644 index 0000000..00bfd53 --- /dev/null +++ b/src/copyparty/mod.rs @@ -0,0 +1,267 @@ +use reqwest::Client; +use serde::Deserialize; +use tracing::warn; + +use crate::error::Result; +use crate::lang; + +#[derive(Clone)] +pub struct CopypartyClient { + pub base_url: String, + pub upload_url: String, + pub password: Option, + client: Client, +} + +/// Decode percent-encoded copyparty paths (e.g. `%20` → space, `%C5%BA` → `ź`). +pub fn percent_decode(s: &str) -> String { + let mut buf: Vec = Vec::with_capacity(s.len()); + let bytes = s.as_bytes(); + let mut i = 0; + while i < bytes.len() { + let b = bytes[i]; + if b == b'%' && i + 2 < bytes.len() { + if let (Some(hi), Some(lo)) = (hex_nibble(bytes[i + 1]), hex_nibble(bytes[i + 2])) { + buf.push(hi << 4 | lo); + i += 3; + continue; + } + } + if b == b'+' { + buf.push(b' '); + } else { + buf.push(b); + } + i += 1; + } + String::from_utf8_lossy(&buf).into_owned() +} + +/// Decode common HTML entities in a display string. +/// Handles named (`&`, `<`) and numeric (`{`, `{`) forms. +/// Uses a byte buffer (like `percent_decode`) to preserve multi-byte UTF-8. +pub fn html_decode(s: &str) -> String { + let mut buf: Vec = Vec::with_capacity(s.len()); + let bytes = s.as_bytes(); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == b'&' { + let rest = &s[i..]; + if let Some(end) = rest.find(';') { + let entity = &rest[..=end]; + match entity { + "&" => buf.extend_from_slice(b"&"), + "<" => buf.extend_from_slice(b"<"), + ">" => buf.extend_from_slice(b">"), + """ => buf.extend_from_slice(b"\""), + "'" => buf.extend_from_slice(b"'"), + " " => buf.extend_from_slice(b" "), + _ if entity.as_bytes().len() > 3 && entity.as_bytes()[1] == b'#' => { + let (num_str, radix) = + if entity.as_bytes()[2] == b'x' || entity.as_bytes()[2] == b'X' { + (&entity[3..entity.len() - 1], 16u32) + } else { + (&entity[2..entity.len() - 1], 10u32) + }; + let decoded = u32::from_str_radix(num_str, radix) + .ok() + .and_then(std::char::from_u32); + if let Some(c) = decoded { + let mut tmp = [0u8; 4]; + buf.extend_from_slice(c.encode_utf8(&mut tmp).as_bytes()); + i += entity.len(); + continue; + } + buf.extend_from_slice(entity.as_bytes()); + } + _ => buf.extend_from_slice(entity.as_bytes()), + } + i += entity.len(); + continue; + } + } + buf.push(bytes[i]); + i += 1; + } + String::from_utf8_lossy(&buf).into_owned() +} + +fn hex_nibble(b: u8) -> Option { + match b { + b'0'..=b'9' => Some(b - b'0'), + b'A'..=b'F' => Some(b - b'A' + 10), + b'a'..=b'f' => Some(b - b'a' + 10), + _ => None, + } +} + +/// A search result from copyparty's `/?srch` API. +/// +/// **Important**: `/?srch` responses use `rp` (relative path like `/music/foo.mp3`). +/// The `name` / `url` / `path` fields are typically only present in directory +/// listing (`?ls`) responses, not in search. Use the convenience accessors +/// below instead of reading fields directly. +#[derive(Debug, Clone, Deserialize)] +pub struct SearchHit { + /// Relative path including filename (e.g. `/music/foo.mp3`). + /// This is the primary field populated by `/?srch`. + #[serde(default)] + pub rp: Option, + /// Path component (directory listing only) + #[serde(default)] + pub path: Option, + /// URL fragment for building a download URL + #[serde(default)] + pub url: Option, + /// Bare filename (directory listing only) + #[serde(default)] + pub name: Option, +} + +impl SearchHit { + /// Best-effort filename extraction: tries `name`, then `rp` tail, then `url` tail. + pub fn filename(&self) -> Option<&str> { + self.name + .as_deref() + .or_else(|| self.rp.as_deref().and_then(|p| p.rsplit('/').next())) + .or_else(|| self.url.as_deref().and_then(|u| u.rsplit('/').next())) + } + + /// Best-effort URL path for building a full download URL. + /// Tries `url`, then `rp`. + pub fn url_path(&self) -> Option<&str> { + self.url.as_deref().or_else(|| self.rp.as_deref()) + } +} + +#[derive(Debug, Deserialize)] +pub struct UploadResponse { + pub fileurl: Option, +} + +impl CopypartyClient { + pub fn new(base_url: String, upload_url: String, password: Option) -> Self { + Self { + base_url, + upload_url, + password, + client: Client::new(), + } + } + + /// Build a full Copyparty URL from a path, appending `?pw=PASSWORD` + /// if a password is configured. Path can be absolute (e.g. `/music/foo.mp3`) + /// or relative (`music/foo.mp3`). + pub fn build_url(&self, path: &str) -> String { + let prefixed = if path.starts_with('/') { + format!("{}{}", self.base_url, path) + } else { + format!("{}/{}", self.base_url, path) + }; + if let Some(pw) = &self.password { + format!("{prefixed}?pw={pw}") + } else { + prefixed + } + } + + /// Search copyparty via the jPOST `/?srch` endpoint. + /// + /// Uses `serde_json::Value` for flexible response parsing (mirroring the Python + /// `_extract_files_from_response` helper), because the exact JSON shape can vary + /// across copyparty versions. + pub async fn search( + &self, + query: &str, + max_results: u32, + ext_filter: &[&str], + ) -> Result> { + let tokens: Vec<&str> = query.split_whitespace().collect(); + let conditions: Vec = tokens + .iter() + .map(|t| format!("(path like \"*{t}*\" or name like \"*{t}*\")")) + .collect(); + + let mut q = conditions.join(" and "); + if !ext_filter.is_empty() { + let ext_conditions: Vec = ext_filter + .iter() + .map(|e| format!("name like \"*.{e}\"")) + .collect(); + let ext_q = ext_conditions.join(" or "); + // When query has no tokens (e.g. empty autocomplete partial), + // avoid wrapping an empty string in parens → "() and ..." is invalid. + if q.is_empty() { + q = ext_q; + } else { + q = format!("({q}) and ({ext_q})"); + } + } + + let payload = serde_json::json!({"n": max_results, "q": q}); + + let mut req = self + .client + .post(format!("{}/?srch", self.base_url)) + .json(&payload); + if let Some(pw) = &self.password { + req = req.header("PW", pw); + } + + let resp = req.send().await?; + let status = resp.status(); + let body_text = resp.text().await?; + + if !status.is_success() { + let snippet: String = body_text.chars().take(200).collect(); + return Err(crate::error::Error::Custom(lang::fmt_cp_search_http( + &status, &snippet, + ))); + } + + let root: serde_json::Value = serde_json::from_str(&body_text).map_err(|e| { + let snippet: String = body_text.chars().take(200).collect(); + crate::error::Error::Custom(lang::fmt_cp_invalid_json(&e, &snippet)) + })?; + + // Extract hits array – copyparty wraps results in {"hits": [...]} + let hits_val = match &root { + serde_json::Value::Object(map) => map.get("hits"), + _ => None, + }; + + let hits: Vec = match hits_val { + Some(serde_json::Value::Array(arr)) => arr + .iter() + .filter_map(|v| SearchHit::deserialize(v).ok()) + .collect(), + _ => { + warn!( + "copyparty search response missing 'hits' array; body starts: {:?}", + body_text.chars().take(100).collect::() + ); + Vec::new() + } + }; + + Ok(hits) + } + + pub async fn upload(&self, file_name: &str, data: Vec) -> Result { + let part = reqwest::multipart::Part::bytes(data).file_name(file_name.to_string()); + let form = reqwest::multipart::Form::new().part("f", part); + + let mut req = self + .client + .post(format!("{}/?j=1&ck=b2", self.upload_url)) + .multipart(form); + + if let Some(pw) = &self.password { + req = req.header("Cookie", format!("cppwd={pw}")); + } + + let resp: UploadResponse = req.send().await?.json().await?; + resp.fileurl + .ok_or_else(|| crate::error::Error::Custom(lang::L_CP_UPLOAD_NO_URL.into())) + } +} diff --git a/src/database/mod.rs b/src/database/mod.rs new file mode 100644 index 0000000..53f4bf4 --- /dev/null +++ b/src/database/mod.rs @@ -0,0 +1,344 @@ +use chrono::{DateTime, Utc}; +use sqlx::FromRow; +use sqlx::SqlitePool; + +use crate::error::{Error, Result}; + +pub type DbPool = SqlitePool; + +pub async fn init_pool(path: &str) -> Result { + if let Some(parent) = std::path::Path::new(path).parent() { + tokio::fs::create_dir_all(parent).await.map_err(Error::Io)?; + } + let pool = SqlitePool::connect(&format!("sqlite:{path}?mode=rwc")).await?; + run_migrations(&pool).await?; + Ok(pool) +} + +async fn run_migrations(pool: &DbPool) -> Result<()> { + sqlx::query( + "CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS voice_activity ( + user_id TEXT PRIMARY KEY, + join_time REAL NOT NULL DEFAULT 0, + total_active REAL NOT NULL DEFAULT 0, + total_afk REAL NOT NULL DEFAULT 0, + is_afk INTEGER NOT NULL DEFAULT 0, + afk_start REAL NOT NULL DEFAULT 0, + active_start REAL NOT NULL DEFAULT 0, + channel_name TEXT NOT NULL DEFAULT '', + last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS cron_jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + cron_expression TEXT NOT NULL, + type TEXT NOT NULL CHECK(type IN ('audio', 'wiadomość')), + channel_id INTEGER NOT NULL, + active INTEGER NOT NULL DEFAULT 1, + audio_filename TEXT, + message TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS triggers ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT UNIQUE NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS trigger_keywords ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trigger_id INTEGER NOT NULL REFERENCES triggers(id) ON DELETE CASCADE, + keyword TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE(trigger_id, keyword) + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS trigger_responses ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trigger_id INTEGER NOT NULL REFERENCES triggers(id) ON DELETE CASCADE, + response_text TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )", + ) + .execute(pool) + .await?; + + Ok(()) +} + +// -- Voice Activity -- +#[derive(Debug, Clone, FromRow)] +pub struct VoiceActivity { + pub user_id: String, + pub join_time: f64, + pub total_active: f64, + pub total_afk: f64, + pub is_afk: i32, + pub afk_start: f64, + pub active_start: f64, + pub channel_name: String, + #[allow(dead_code)] + pub last_updated: Option>, +} + +#[allow(dead_code)] +pub async fn get_voice_activity(pool: &DbPool, user_id: &str) -> Result> { + Ok( + sqlx::query_as("SELECT * FROM voice_activity WHERE user_id = ?") + .bind(user_id) + .fetch_optional(pool) + .await?, + ) +} + +pub async fn upsert_voice_activity(pool: &DbPool, va: &VoiceActivity) -> Result<()> { + sqlx::query( + "INSERT INTO voice_activity (user_id, join_time, total_active, total_afk, is_afk, afk_start, active_start, channel_name) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id) DO UPDATE SET + join_time = excluded.join_time, + total_active = excluded.total_active, + total_afk = excluded.total_afk, + is_afk = excluded.is_afk, + afk_start = excluded.afk_start, + active_start = excluded.active_start, + channel_name = excluded.channel_name, + last_updated = CURRENT_TIMESTAMP", + ) + .bind(&va.user_id) + .bind(va.join_time) + .bind(va.total_active) + .bind(va.total_afk) + .bind(va.is_afk) + .bind(va.afk_start) + .bind(va.active_start) + .bind(&va.channel_name) + .execute(pool) + .await?; + Ok(()) +} + +// -- Cron Jobs -- +#[derive(Debug, Clone, FromRow)] +pub struct CronJob { + pub id: i64, + pub cron_expression: String, + pub r#type: String, + pub channel_id: i64, + pub active: i32, + pub audio_filename: Option, + pub message: Option, + #[allow(dead_code)] + pub created_at: Option>, +} + +#[allow(dead_code)] +pub async fn list_cron_jobs(pool: &DbPool) -> Result> { + Ok(sqlx::query_as("SELECT * FROM cron_jobs ORDER BY id") + .fetch_all(pool) + .await?) +} + +#[allow(dead_code)] +pub async fn list_active_cron_jobs(pool: &DbPool) -> Result> { + Ok(sqlx::query_as("SELECT * FROM cron_jobs WHERE active = 1") + .fetch_all(pool) + .await?) +} + +pub async fn add_cron_job( + pool: &DbPool, + cron_expression: &str, + r#type: &str, + channel_id: i64, + audio_filename: Option<&str>, + message: Option<&str>, +) -> Result { + let result = sqlx::query( + "INSERT INTO cron_jobs (cron_expression, type, channel_id, audio_filename, message) VALUES (?, ?, ?, ?, ?)", + ) + .bind(cron_expression) + .bind(r#type) + .bind(channel_id) + .bind(audio_filename) + .bind(message) + .execute(pool) + .await?; + Ok(result.last_insert_rowid()) +} + +pub async fn delete_cron_job(pool: &DbPool, id: i64) -> Result { + let result = sqlx::query("DELETE FROM cron_jobs WHERE id = ?") + .bind(id) + .execute(pool) + .await?; + Ok(result.rows_affected() > 0) +} + +pub async fn set_cron_job_active(pool: &DbPool, id: i64, active: bool) -> Result { + let result = sqlx::query("UPDATE cron_jobs SET active = ? WHERE id = ?") + .bind(if active { 1 } else { 0 }) + .bind(id) + .execute(pool) + .await?; + Ok(result.rows_affected() > 0) +} + +// -- Triggers -- +#[derive(Debug, Clone, FromRow)] +pub struct Trigger { + pub id: i64, + pub name: String, + pub created_at: Option>, + pub last_updated: Option>, +} + +#[allow(dead_code)] +#[derive(Debug, Clone, FromRow)] +pub struct TriggerKeyword { + pub id: i64, + pub trigger_id: i64, + pub keyword: String, +} + +#[allow(dead_code)] +#[derive(Debug, Clone, FromRow)] +pub struct TriggerResponse { + pub id: i64, + pub trigger_id: i64, + pub response_text: String, +} + +pub async fn create_trigger(pool: &DbPool, name: &str) -> Result { + let result = sqlx::query("INSERT INTO triggers (name) VALUES (?)") + .bind(name) + .execute(pool) + .await?; + Ok(result.last_insert_rowid()) +} + +pub async fn delete_trigger(pool: &DbPool, id: i64) -> Result { + let result = sqlx::query("DELETE FROM triggers WHERE id = ?") + .bind(id) + .execute(pool) + .await?; + Ok(result.rows_affected() > 0) +} + +pub async fn get_trigger_by_name(pool: &DbPool, name: &str) -> Result> { + Ok(sqlx::query_as("SELECT * FROM triggers WHERE name = ?") + .bind(name) + .fetch_optional(pool) + .await?) +} + +pub async fn list_triggers(pool: &DbPool) -> Result> { + Ok(sqlx::query_as("SELECT * FROM triggers ORDER BY name") + .fetch_all(pool) + .await?) +} + +pub async fn add_keyword(pool: &DbPool, trigger_id: i64, keyword: &str) -> Result<()> { + sqlx::query("INSERT OR IGNORE INTO trigger_keywords (trigger_id, keyword) VALUES (?, ?)") + .bind(trigger_id) + .bind(keyword) + .execute(pool) + .await?; + update_trigger_timestamp(pool, trigger_id).await +} + +pub async fn remove_keyword(pool: &DbPool, trigger_id: i64, keyword: &str) -> Result { + let result = sqlx::query("DELETE FROM trigger_keywords WHERE trigger_id = ? AND keyword = ?") + .bind(trigger_id) + .bind(keyword) + .execute(pool) + .await?; + update_trigger_timestamp(pool, trigger_id).await?; + Ok(result.rows_affected() > 0) +} + +pub async fn add_response(pool: &DbPool, trigger_id: i64, response: &str) -> Result<()> { + sqlx::query("INSERT INTO trigger_responses (trigger_id, response_text) VALUES (?, ?)") + .bind(trigger_id) + .bind(response) + .execute(pool) + .await?; + update_trigger_timestamp(pool, trigger_id).await +} + +pub async fn remove_response(pool: &DbPool, trigger_id: i64, response: &str) -> Result { + let result = + sqlx::query("DELETE FROM trigger_responses WHERE trigger_id = ? AND response_text = ?") + .bind(trigger_id) + .bind(response) + .execute(pool) + .await?; + update_trigger_timestamp(pool, trigger_id).await?; + Ok(result.rows_affected() > 0) +} + +pub async fn get_keywords(pool: &DbPool, trigger_id: i64) -> Result> { + let rows: Vec<(String,)> = + sqlx::query_as("SELECT keyword FROM trigger_keywords WHERE trigger_id = ?") + .bind(trigger_id) + .fetch_all(pool) + .await?; + Ok(rows.into_iter().map(|r| r.0).collect()) +} + +pub async fn get_responses(pool: &DbPool, trigger_id: i64) -> Result> { + let rows: Vec<(String,)> = + sqlx::query_as("SELECT response_text FROM trigger_responses WHERE trigger_id = ?") + .bind(trigger_id) + .fetch_all(pool) + .await?; + Ok(rows.into_iter().map(|r| r.0).collect()) +} + +pub async fn load_all_triggers_with_keywords_and_responses( + pool: &DbPool, +) -> Result, Vec)>> { + let triggers = list_triggers(pool).await?; + let mut result = Vec::new(); + for t in triggers { + let keywords = get_keywords(pool, t.id).await?; + let responses = get_responses(pool, t.id).await?; + result.push((t, keywords, responses)); + } + Ok(result) +} + +async fn update_trigger_timestamp(pool: &DbPool, id: i64) -> Result<()> { + sqlx::query("UPDATE triggers SET last_updated = CURRENT_TIMESTAMP WHERE id = ?") + .bind(id) + .execute(pool) + .await?; + Ok(()) +} diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..d90d2e9 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,108 @@ +use std::fmt; + +#[derive(Debug)] +pub enum Error { + Sqlx(sqlx::Error), + Reqwest(reqwest::Error), + Serenity(serenity::Error), + SongbirdJoin(songbird::error::JoinError), + SongbirdControl(songbird::tracks::ControlError), + Io(std::io::Error), + Cron(cron::error::Error), + Image(image::ImageError), + Qr(qrcode::types::QrError), + Scraper(String), + Config(String), + TaskJoin(String), + Custom(String), +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + use crate::lang; + match self { + Self::Sqlx(e) => write!(f, "{}: {e}", lang::L_ERR_DATABASE), + Self::Reqwest(e) => write!(f, "{}: {e}", lang::L_ERR_HTTP), + Self::Serenity(e) => write!(f, "{}: {e}", lang::L_ERR_DISCORD), + Self::SongbirdJoin(e) => write!(f, "{}: {e}", lang::L_ERR_VOICE_JOIN), + Self::SongbirdControl(e) => write!(f, "{}: {e}", lang::L_ERR_VOICE_CONTROL), + Self::Io(e) => write!(f, "{}: {e}", lang::L_ERR_IO), + Self::Cron(e) => write!(f, "{}: {e}", lang::L_ERR_CRON), + Self::Image(e) => write!(f, "{}: {e}", lang::L_ERR_IMAGE), + Self::Qr(e) => write!(f, "{}: {e}", lang::L_ERR_QR), + Self::Scraper(e) => write!(f, "{}: {e}", lang::L_ERR_SCRAPER), + Self::Config(e) => write!(f, "{}: {e}", lang::L_ERR_CONFIG), + Self::TaskJoin(e) => write!(f, "{}: {e}", lang::L_ERR_TASK_JOIN), + Self::Custom(e) => write!(f, "{e}"), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Sqlx(e) => Some(e), + Self::Reqwest(e) => Some(e), + Self::Serenity(e) => Some(e), + Self::SongbirdJoin(e) => Some(e), + Self::SongbirdControl(e) => Some(e), + Self::Io(e) => Some(e), + Self::Image(e) => Some(e), + Self::Qr(e) => Some(e), + _ => None, + } + } +} + +impl From for Error { + fn from(e: sqlx::Error) -> Self { + Self::Sqlx(e) + } +} +impl From for Error { + fn from(e: reqwest::Error) -> Self { + Self::Reqwest(e) + } +} +impl From for Error { + fn from(e: serenity::Error) -> Self { + Self::Serenity(e) + } +} +impl From for Error { + fn from(e: songbird::error::JoinError) -> Self { + Self::SongbirdJoin(e) + } +} +impl From for Error { + fn from(e: songbird::tracks::ControlError) -> Self { + Self::SongbirdControl(e) + } +} +impl From for Error { + fn from(e: std::io::Error) -> Self { + Self::Io(e) + } +} +impl From for Error { + fn from(e: cron::error::Error) -> Self { + Self::Cron(e) + } +} +impl From for Error { + fn from(e: image::ImageError) -> Self { + Self::Image(e) + } +} +impl From for Error { + fn from(e: qrcode::types::QrError) -> Self { + Self::Qr(e) + } +} +impl From for Error { + fn from(e: tokio::task::JoinError) -> Self { + Self::TaskJoin(e.to_string()) + } +} + +pub type Result = std::result::Result; diff --git a/src/events/mod.rs b/src/events/mod.rs new file mode 100644 index 0000000..457ebcd --- /dev/null +++ b/src/events/mod.rs @@ -0,0 +1,368 @@ +use chrono::{Local, Utc}; +use poise::serenity_prelude as serenity; +use rand::seq::IndexedRandom; +use tracing::{info, warn}; + +use crate::Data; +use crate::database::{self, VoiceActivity}; +use crate::error::Result; +use crate::lang; + +pub async fn handle_ready(_ctx: &serenity::Context) -> Result<()> { + info!("{}", lang::L_EV_READY); + Ok(()) +} + +fn format_duration(seconds: f64) -> String { + let total = seconds as u64; + let hours = total / 3600; + let minutes = (total % 3600) / 60; + let secs = total % 60; + if hours > 0 { + format!("{hours}h {minutes}m {secs}s") + } else if minutes > 0 { + format!("{minutes}m {secs}s") + } else { + format!("{secs}s") + } +} + +async fn play_tts_in_channel( + _ctx: &serenity::Context, + data: &Data, + guild_id: serenity::GuildId, + channel_id: serenity::ChannelId, + text: &str, +) { + let tts_url = format!( + "https://translate.google.com/translate_tts?ie=UTF-8&q={}&tl=pl&client=tw-ob", + crate::commands::tts::urlencoding(text), + ); + + let audio_data = match data.http_client.get(&tts_url).send().await { + Ok(resp) => match resp.bytes().await { + Ok(b) => b, + Err(e) => { + warn!("Failed to read TTS audio bytes: {e}"); + return; + } + }, + Err(e) => { + warn!("Failed to fetch TTS audio: {e}"); + return; + } + }; + + let call = match data.voice_manager.connect(guild_id, channel_id).await { + Ok(c) => c, + Err(e) => { + warn!("Failed to connect to voice for TTS: {e}"); + return; + } + }; + + let input = match crate::voice::decode_audio(&audio_data) { + Ok(i) => i, + Err(e) => { + warn!("Failed to decode TTS audio: {e}"); + return; + } + }; + + let mut call_lock = call.lock().await; + if call_lock.current_channel().is_some() { + call_lock.play_input(input); + } +} + +pub async fn handle_voice_state_update( + ctx: &serenity::Context, + data: &Data, + old: Option, + new: &serenity::VoiceState, +) -> Result<()> { + let log_channel = data.config.status_channel_id.map(serenity::ChannelId::new); + let current_ts = Utc::now().timestamp(); + let current_time = Local::now().format("%Y-%m-%d %H:%M:%S").to_string(); + + let user_id = new.user_id; + let guild_id = new.guild_id; + + // skip bots (Python behavior) + let member = guild_id + .and_then(|gid| ctx.cache.guild(gid)) + .and_then(|g| g.members.get(&user_id).cloned()); + if let Some(ref m) = member { + if m.user.bot { + return Ok(()); + } + } + + let member_name = member + .as_ref() + .map(|m| m.user.name.clone()) + .unwrap_or_default(); + let display_name = member + .as_ref() + .map(|m| m.display_name().to_string()) + .unwrap_or_default(); + + let now = current_ts as f64; + + let user_id_str = user_id.to_string(); + let mut va = database::get_voice_activity(&data.db_pool, &user_id_str) + .await? + .unwrap_or(VoiceActivity { + user_id: user_id_str.clone(), + join_time: 0.0, + total_active: 0.0, + total_afk: 0.0, + is_afk: 0, + afk_start: 0.0, + active_start: 0.0, + channel_name: String::new(), + last_updated: None, + }); + + let old_channel = old.as_ref().and_then(|v| v.channel_id); + let new_channel = new.channel_id; + + let mut tts_text: Option = None; + let mut tts_channel: Option = None; + + // ── JOIN ── + if old_channel.is_none() && new_channel.is_some() { + let cid = new_channel.unwrap(); + let is_afk = new.self_mute || new.self_deaf; + + va.join_time = now; + va.total_active = 0.0; + va.total_afk = 0.0; + va.is_afk = if is_afk { 1 } else { 0 }; + va.afk_start = if is_afk { now } else { 0.0 }; + va.active_start = if !is_afk { now } else { 0.0 }; + va.channel_name = guild_id + .and_then(|gid| ctx.cache.guild(gid)) + .map(|g| { + g.channels + .get(&cid) + .map(|c| c.name.clone()) + .unwrap_or_default() + }) + .unwrap_or_default(); + database::upsert_voice_activity(&data.db_pool, &va).await?; + info!("User {user_id} joined voice"); + + if let Some(lc) = log_channel { + let msg = format!( + "[{current_time}] {member_name} dołączył na {} ", + va.channel_name + ); + lc.say(ctx, &msg).await.ok(); + } + tts_text = Some(lang::fmt_ev_joined(&display_name)); + tts_channel = Some(cid); + } + // ── LEAVE ── + else if old_channel.is_some() && new_channel.is_none() { + let ocid = old_channel.unwrap(); + let channel_name = guild_id + .and_then(|gid| ctx.cache.guild(gid)) + .and_then(|g| g.channels.get(&ocid).map(|c| c.name.clone())) + .unwrap_or_default(); + + if va.join_time > 0.0 { + if va.is_afk == 1 { + va.total_afk += now - va.afk_start; + } else { + va.total_active += (now - va.active_start).max(0.0); + } + + let active_str = format_duration(va.total_active); + let afk_str = format_duration(va.total_afk); + + if let Some(lc) = log_channel { + let msg = format!( + "[{current_time}] {member_name} rozłączył się z {} \n\ + Aktywny: {active_str} | AFK: {afk_str}", + channel_name, + ); + lc.say(ctx, &msg).await.ok(); + } + info!("User {user_id} left voice"); + + database::upsert_voice_activity(&data.db_pool, &va).await?; + } else { + if let Some(lc) = log_channel { + let msg = format!( + "[{current_time}] {member_name} rozłączył się z {channel_name} " + ); + lc.say(ctx, &msg).await.ok(); + } + info!("User {user_id} left voice (no data)"); + } + + tts_text = Some(lang::fmt_ev_left(&display_name)); + tts_channel = Some(ocid); + } + // ── AFK STATE CHANGE (only for tracked users already in channel) ── + else if new_channel.is_some() && va.join_time > 0.0 { + let is_afk = new.self_mute || new.self_deaf; + let was_afk = va.is_afk == 1; + + if was_afk && !is_afk { + // Return from AFK + let afk_duration = now - va.afk_start; + va.total_afk += afk_duration; + va.is_afk = 0; + va.afk_start = 0.0; + va.active_start = now; + database::upsert_voice_activity(&data.db_pool, &va).await?; + + let afk_str = format_duration(afk_duration); + // Python behavior: after.self_mute is always false here so always "odgłuchnął" + let activity = if new.self_mute { + "odciszył" + } else { + "odgłuchnął" + }; + + if let Some(lc) = log_channel { + let msg = format!( + "[{current_time}] {member_name} {activity} się na {} \n\ + AFK: {afk_str}", + va.channel_name, + ); + lc.say(ctx, &msg).await.ok(); + } + info!("User {user_id} returned from AFK"); + } else if !was_afk && is_afk { + // Went AFK + let active_duration = if va.active_start > 0.0 { + (now - va.active_start).max(0.0) + } else { + 0.0 + }; + va.total_active += active_duration; + va.is_afk = 1; + va.afk_start = now; + va.active_start = 0.0; + database::upsert_voice_activity(&data.db_pool, &va).await?; + + let active_str = format_duration(active_duration); + let activity = if new.self_mute { + "wyciszył" + } else { + "ogłuchnął" + }; + + if let Some(lc) = log_channel { + let msg = format!( + "[{current_time}] {member_name} {activity} się na {} \n\ + Aktywny: {active_str}", + va.channel_name, + ); + lc.say(ctx, &msg).await.ok(); + } + info!("User {user_id} went AFK"); + } + } + + // ── TTS for mute/deafen (deaf takes priority with early return) ── + if old.as_ref().map_or(false, |v| v.self_deaf) != new.self_deaf { + let phrase = if new.self_deaf { + lang::fmt_ev_deafened(&display_name) + } else { + lang::fmt_ev_undeafened(&display_name) + }; + if let Some(gid) = guild_id { + if let Some(cid) = tts_channel.or(new_channel) { + play_tts_in_channel(ctx, data, gid, cid, &phrase).await; + } + } + return Ok(()); + } + + if old.as_ref().map_or(false, |v| v.self_mute) != new.self_mute { + let phrase = if new.self_mute { + lang::fmt_ev_muted(&display_name) + } else { + lang::fmt_ev_unmuted(&display_name) + }; + if let Some(gid) = guild_id { + if let Some(cid) = tts_channel.or(new_channel) { + play_tts_in_channel(ctx, data, gid, cid, &phrase).await; + } + } + return Ok(()); + } + + // ── Play join/leave TTS if no mute/deafen change ── + if let (Some(text), Some(cid)) = (tts_text, tts_channel) { + if let Some(gid) = guild_id { + play_tts_in_channel(ctx, data, gid, cid, &text).await; + } + } + + Ok(()) +} + +pub async fn handle_message( + ctx: &serenity::Context, + data: &Data, + new_message: &serenity::Message, +) -> Result<()> { + if new_message.author.id == ctx.cache.current_user().id { + return Ok(()); + } + + let content_lower = new_message.content.to_lowercase(); + let response_to_send = { + let triggers = data.triggers.lock().await; + let mut chosen = None; + for (_trigger, keywords, responses) in triggers.iter() { + if keywords.iter().any(|k| content_lower.contains(k)) { + chosen = responses.choose(&mut rand::rng()).cloned(); + break; + } + } + chosen + }; + if let Some(response) = response_to_send { + new_message.channel_id.say(ctx, response).await.ok(); + } + + let is_mention = new_message.mentions_user_id(ctx.cache.current_user().id); + let has_keyword = content_lower.contains(&data.config.assistant_keyword); + + if (is_mention || has_keyword) && data.config.openrouter_api_key.is_some() { + let api_key = data.config.openrouter_api_key.clone().unwrap(); + let model = data.config.openrouter_model.clone().unwrap_or_else(|| { + "cognitivecomputations/dolphin-mistral-24b-venice-edition:free".into() + }); + + let client = crate::openrouter::OpenRouterClient::new(api_key, model); + let assistant_prompt = lang::ASSISTANT_SYSTEM_PROMPT; + + let mut messages = Vec::new(); + let mut current_msg = Some(new_message.clone()); + for _ in 0..5 { + if let Some(msg) = current_msg { + messages.push(("user".into(), msg.content.clone())); + current_msg = match msg.message_reference.and_then(|r| r.message_id) { + Some(ref_id) => msg.channel_id.message(ctx, ref_id).await.ok(), + None => None, + }; + } else { + break; + } + } + messages.reverse(); + + if let Ok(response) = client.chat_with_history(assistant_prompt, messages).await { + new_message.reply(ctx, response).await.ok(); + } + } + + Ok(()) +} diff --git a/src/lang.rs b/src/lang.rs new file mode 100644 index 0000000..626f387 --- /dev/null +++ b/src/lang.rs @@ -0,0 +1,387 @@ +// ┌──────────────────────────────────────────────────────────────────┐ +// │ lang.rs — All user-facing messages live here. │ +// │ Translate the bot by editing this single file. │ +// │ Naming: L__ for simple &str; │ +// │ fn fmt__(...) -> String for parameterized. │ +// └──────────────────────────────────────────────────────────────────┘ + +// ── Error display prefixes (used in error.rs Display impl) ──────────── + +pub const L_ERR_DATABASE: &str = "Database error"; +pub const L_ERR_HTTP: &str = "HTTP error"; +pub const L_ERR_DISCORD: &str = "Discord error"; +pub const L_ERR_VOICE_JOIN: &str = "Voice join error"; +pub const L_ERR_VOICE_CONTROL: &str = "Voice control error"; +pub const L_ERR_IO: &str = "IO error"; +pub const L_ERR_CRON: &str = "Cron error"; +pub const L_ERR_IMAGE: &str = "Image error"; +pub const L_ERR_QR: &str = "QR error"; +pub const L_ERR_SCRAPER: &str = "Scraper error"; +pub const L_ERR_CONFIG: &str = "Config error"; +pub const L_ERR_TASK_JOIN: &str = "Task join error"; + +// ── Common / shared error messages ──────────────────────────────────── + +/// "Nie jestem w gildii" — bot must be invoked inside a guild +pub const L_NOT_IN_GUILD: &str = "Nie jestem w gildii"; +/// "Nie jesteś na kanale głosowym" — user must be in a voice channel +pub const L_NOT_IN_VOICE: &str = "Nie jesteś na kanale głosowym"; +/// "Zły typ kanału" — wrong channel variant +pub const L_BAD_CHANNEL_TYPE: &str = "Zły typ kanału"; +/// "Grupa triggerów nie znaleziona" — trigger group not found +pub const L_TRIGGER_NOT_FOUND: &str = "Grupa triggerów nie znaleziona"; + +// ── AI / senator command ────────────────────────────────────────────── + +pub const SENATOR_SYSTEM_PROMPT: &str = "Jesteś Senatorem Steavenem Sperma Armstrongiem. Jesteś rasistą, seksistą i homofobem. Odpowiadasz w języku polskim. Odpowiedź ma maksymalnie 100 słów. Nie używaj formatowania markdown."; + +pub const L_OPENROUTER_NO_KEY: &str = "Klucz API OpenRouter nie jest skonfigurowany w .env"; + +// ── Cron command ────────────────────────────────────────────────────── + +pub const L_CRON_QUEUED: &str = "Praca cron zakolejkowana"; +pub const L_CRON_NONE: &str = "Brak zakolejkowanych prac cron"; +pub const L_CRON_TITLE: &str = "Prace cron"; +pub const L_CRON_ACTIVE: &str = "✅ Aktywna"; +pub const L_CRON_PAUSED: &str = "⏸️ Zapauzowana"; + +pub fn fmt_cron_audio_queued(cron: &str) -> String { + format!("Praca audio zakolejkowana z cron'em `{cron}`") +} +pub fn fmt_cron_label_audio(name: &str) -> String { + format!("Audio: {name}") +} +pub fn fmt_cron_label_msg(msg: &str) -> String { + format!("Wiadomość: {msg}") +} +pub fn fmt_cron_job_status(id: i64, status: &str) -> String { + format!("ID {id} — {status}") +} +pub fn fmt_cron_field(cron: &str, chan: i64, desc: &str) -> String { + format!("`{cron}` — <#{chan}>\n{desc}") +} +pub fn fmt_cron_deleted(id: i64) -> String { + format!("Praca cron z ID {id} została usunięta") +} +pub fn fmt_cron_not_found(id: i64) -> String { + format!("Praca z ID {id} nie została znaleziona") +} +pub fn fmt_cron_paused(id: i64) -> String { + format!("Praca cron z ID {id} została zapauzowana") +} +pub fn fmt_cron_resumed(id: i64) -> String { + format!("Praca cron z ID {id} została wznowiona") +} + +// ── Math command ────────────────────────────────────────────────────── + +pub const L_MATH_TOO_LONG: &str = "Ekspresja za długa (max 1000 znaków"; +pub const L_MATH_NOT_ASCII: &str = "Tylko znaki ASCII dozwolone"; +pub const L_MATH_UNBALANCED: &str = "Nawiasy są niezbalansowane"; +pub const L_MATH_DANGEROUS: &str = "Ekspresja zawiera groźne słowa"; +pub const L_MATH_NO_RESULT: &str = "Brak rezultatu"; + +pub fn fmt_math_qalc_err(e: &dyn std::fmt::Display) -> String { + format!("qalc nie jest dostępny: {e}") +} +pub fn fmt_math_result(r: &str) -> String { + format!("```\n{r}\n```") +} + +// ── Music command ───────────────────────────────────────────────────── + +pub const L_MUSIC_NEED_NAME: &str = "Daj mi nazwę pliku, URL albo plik"; +pub const L_MUSIC_NO_AUDIO_ON_CP: &str = "Brak powiązanego audio na copyparty"; +pub const L_MUSIC_NO_URL_IN_RESULT: &str = "Wynik nie ma URL"; +pub const L_MUSIC_CANT_COMPRESS_LIVE: &str = + "Nie można skompresować transmisji na żywo, odtwarzam normalnie"; +pub const L_MUSIC_STOPPED: &str = "wyszed"; +pub const L_VOICE_USER_NOT_IN_VC: &str = "You are not in a voice channel"; + +pub fn fmt_music_search_failed(e: &dyn std::fmt::Display) -> String { + format!("Szukanie na copyparty zawiodło: {e}") +} +pub fn fmt_music_playing(name: &str) -> String { + format!("Grajcowanie: {name}") +} +pub fn fmt_voice_joined(cid: &dyn std::fmt::Display) -> String { + format!("Dołączon <#{cid}>") +} +pub fn fmt_voice_join_failed(e: &dyn std::fmt::Display) -> String { + format!("Nie udało się połączyć z VC: {e}") +} + +// ── TTS command ─────────────────────────────────────────────────────── + +pub const L_TTS_SAVED: &str = "TTS zapisany"; +pub const L_TTS_PLAYING: &str = "Grom TTS na kanale głosowym"; + +// ── Trigger command ─────────────────────────────────────────────────── + +pub const L_TRIGGER_UPDATED: &str = "Trigger zaktualizowany"; +pub const L_TRIGGER_KW_NONE: &str = "Brak"; + +pub fn fmt_trigger_created(name: &str) -> String { + format!("Grupa trigger '{name}' utworzona") +} +pub fn fmt_trigger_failed(e: &dyn std::fmt::Display) -> String { + format!("Nie udało się: {e}") +} +pub fn fmt_trigger_deleted(name: &str) -> String { + format!("Grupa trigger '{name}' usunięta") +} +pub fn fmt_trigger_title(name: &str) -> String { + format!("Trigger: {name}") +} + +// ── Utility commands ────────────────────────────────────────────────── + +pub const L_UTIL_NEED_INPUT: &str = "Podaj plik z copyparty lub załącznik"; +pub const L_UTIL_PICK_FROM_AUTOCOMPLETE: &str = "Wybierz plik z podpowiedzi"; +pub const L_UTIL_DOWNLOAD_FAILED: &str = "Nie udało się pobrać pliku"; +pub const L_UTIL_UNSUPPORTED_TYPE: &str = "Nieobsługiwany typ pliku"; +pub const L_UTIL_COMPRESS_FAILED: &str = "Kompresja nie powiodła się"; +pub const L_UTIL_CONVERT_FAILED: &str = "Konwersja nie powiodła się"; +pub const L_UTIL_UPLOAD_FAILED: &str = "Nie udało się wysłać na serwer"; +pub const L_UTIL_NO_ATTACHMENTS: &str = "Brak załączników w wiadomości"; +pub const L_SEARCH_NO_RESULTS: &str = "No results found"; + +pub fn fmt_util_compressed(url: &str) -> String { + format!("Skompresowano\n{url}") +} +pub fn fmt_util_converted(url: &str) -> String { + format!("Skonwertowano\n{url}") +} +pub fn fmt_util_convert_err(stderr: &str) -> String { + format!("Błąd konwersji: {stderr}") +} + +pub fn fmt_uploaded(url: &str) -> String { + format!("Wysłon: {url}") +} +pub fn fmt_download_uploaded(url: &str) -> String { + format!("pobioron i wysyłon: {url}") +} +pub fn fmt_upload_summary(count: usize) -> String { + format!("Wysłano {count} plików:\n") +} +pub fn fmt_upload_partial_failed(count: usize) -> String { + format!("\n\nNie wysłano {count} plików") +} +pub fn fmt_message_compressed(made: usize, total: usize) -> String { + format!("Skompresowano {made} z {total} plików:") +} +pub fn fmt_search_result(name: &str, url: &str) -> String { + format!("**{name}**\n{url}") +} +pub fn fmt_download_failed(stderr: &str) -> String { + format!( + "Pobieranie się nie powiodło: {}", + translate_ytdlp_stderr(stderr) + ) +} +pub fn fmt_ytdlp_err(e: &dyn std::fmt::Display) -> String { + format!("yt-dlp nie mógło się uruchomić: {e}") +} +pub fn fmt_ytdlp_stderr(stderr: &str) -> String { + translate_ytdlp_stderr(stderr) +} + +/// Translate yt-dlp's verbose stderr into a short, user-friendly Polish +/// message. yt-dlp emits lots of `[youtube] Extracting URL…`, `WARNING:` +/// and `ERROR:` noise that is meaningless to a normal Discord user; this +/// recognises the handful of failure modes users actually hit and falls +/// back to the first `ERROR:`/`WARNING:` line (stripped of the +/// `ERROR: [extractor] id: ` prefix) when nothing matches, so we never +/// show the full multi-line trace. +pub fn translate_ytdlp_stderr(stderr: &str) -> String { + let lower = stderr.to_ascii_lowercase(); + + // YouTube's anti-bot sign-in wall. yt-dlp can't bypass this without + // browser cookies configured on the host — surface that to the user + // rather than the cryptic "Sign in to confirm you're not a bot" line. + if lower.contains("sign in to confirm you") + || lower.contains("confirm you're not a bot") + || lower.contains("confirm you are not a bot") + { + return "YouTube uznaje nas za bota i blokuje pobieranie. Spróbuj ponownie później albo podaj inny link.".into(); + } + + // Generic extractor fallback means the URL didn't match any known + // site — combined with "Requested format is not available" this is + // almost always "the user pasted a non-video URL into /download". + if lower.contains("requested format is not available") + || lower.contains("url could be a direct video link") + || lower.contains("falling back on generic information extractor") + { + return "Ten link nie wygląda na wideo obsługiwane przez yt-dlp. Upewnij się, że to adres do filmu (YouTube / SoundCloud / Bandcamp / Twitch).".into(); + } + + if lower.contains("private video") { + return "Ten film jest prywatny i nie można go pobrać.".into(); + } + if lower.contains("video unavailable") + || lower.contains("video has been removed") + || lower.contains("this video is unavailable") + { + return "Ten film jest niedostępny lub został usunięty.".into(); + } + if lower.contains("members-only") + || lower.contains("premium-only") + || lower.contains("channel's members") + { + return "Ten film jest dostępny tylko dla subskrybujących / członków kanału.".into(); + } + if lower.contains("not available in your country") + || lower.contains("geo-restricted") + || lower.contains("geo restricted") + { + return "Ten film jest zablokowany regionalnie i nie jest dostępny w naszym regionie." + .into(); + } + if lower.contains("age-restricted") + || lower.contains("age restricted") + || lower.contains("inappropriate for some users") + { + return "Ten film jest ograniczony wiekowo i wymaga uwierzytelnienia.".into(); + } + if lower.contains("http error 404") || lower.contains(" http error 404 ") { + return "Nie znaleziono takiego adresu (404).".into(); + } + if lower.contains("unable to extract") && lower.contains("video data") { + return "Nie udało się wyciągnąć danych filmu — strona mogła się zmienić albo link jest niepełny.".into(); + } + + // Fallback: keep only the first ERROR: line and strip the + // `ERROR: [extractor] id: ` prefix so the message is short and + // free of yt-dlp internals. + for line in stderr.lines() { + if let Some(rest) = line.strip_prefix("ERROR:") { + let rest = rest.trim(); + // Drop leading `[extractor] video_id: ` noise if present. + let msg = rest + .strip_prefix(|c: char| c == '[') + .and_then(|r| r.find(']').map(|i| &r[i + 1..])) + .map(|s| { + s.trim_start_matches(|c: char| { + c.is_alphanumeric() || c == '_' || c == '-' || c == ':' + }) + .trim() + }) + .filter(|s| !s.is_empty()) + .unwrap_or(rest); + return format!("Pobieranie się nie powiodło: {msg}"); + } + } + for line in stderr.lines() { + if let Some(rest) = line.strip_prefix("WARNING:") { + return format!("Pobieranie się nie powiodło: {}", rest.trim()); + } + } + + "Pobieranie się nie powiodło.".into() +} + +// ── Web command ─────────────────────────────────────────────────────── + +pub const L_WEB_NO_RESULTS: &str = "Brak rezultatów"; + +pub fn fmt_web_title(query: &str) -> String { + format!("Wyniki wyszukiwania dla: {query}") +} + +// ── QR command ──────────────────────────────────────────────────────── + +pub const L_QR_GENERATED: &str = "Kod QR wygenerowany:"; +pub const L_QR_DECODE_FAILED: &str = "Nie udało się zdekodować kodu QR."; + +pub fn fmt_qr_decoded(text: &str) -> String { + format!("```\n{text}\n```") +} + +// ── Copyparty client ────────────────────────────────────────────────── + +pub const L_CP_UPLOAD_NO_URL: &str = "Wysłanie zawiodło, brak URL w odpowiedzi"; + +pub fn fmt_cp_search_http(status: &dyn std::fmt::Display, snippet: &str) -> String { + format!("copyparty zwróciło status HTTP {status}: {snippet}") +} +pub fn fmt_cp_invalid_json(e: &dyn std::fmt::Display, snippet: &str) -> String { + format!("copyparty zwróciło nielegalny JSON. ({e}): {snippet}") +} + +// ── OpenRouter ──────────────────────────────────────────────────────── + +pub fn fmt_openrouter_err(status: &dyn std::fmt::Display, text: &str) -> String { + format!("Błąd API OpenRouter. {status}: {text}") +} + +// ── Voice / audio decoding ──────────────────────────────────────────── + +pub const L_VOICE_NO_TRACK: &str = "Brak kawałka"; +pub const L_VOICE_NO_CODEC: &str = "Brak parametrów codec'a audio"; + +pub fn fmt_voice_remove_failed(e: &dyn std::fmt::Display) -> String { + format!("Nie udało się usunąć klienta głosowego: {e}") +} +pub fn fmt_voice_probe_failed(e: &dyn std::fmt::Display) -> String { + format!("Próbkowanie symphonia zawiodło: {e}") +} +pub fn fmt_voice_decoder_failed(e: &dyn std::fmt::Display) -> String { + format!("Nie udało się utworzyć dekodera audio: {e}") +} +pub fn fmt_voice_ytdlp_launch(e: &dyn std::fmt::Display) -> String { + format!("yt-dlp nie mogło się uruchomić: {e}") +} +pub fn fmt_voice_ffmpeg_launch(e: &dyn std::fmt::Display) -> String { + format!("ffmpeg nie mogło się uruchomić: {e}") +} + +// ── Events ──────────────────────────────────────────────────────────── + +pub const L_EV_READY: &str = "ŁUKASZ SPERMA AKTYWACJA"; +pub const L_EV_VOICE_STATE_ERROR: &str = "Błąd stanu głosu"; +pub const L_EV_MESSAGE_HANDLER_ERROR: &str = "Błąd ustanawiacza głosu"; + +// ── Event TTS phrases ───────────────────────────────────────────────── + +pub fn fmt_ev_joined(name: &str) -> String { + format!("{name} sie zespawnił") +} +pub fn fmt_ev_left(name: &str) -> String { + format!("{name} jest chujem") +} +pub fn fmt_ev_deafened(name: &str) -> String { + format!("{name} jest bity przez ojca") +} +pub fn fmt_ev_undeafened(name: &str) -> String { + format!("{name} uciekł od ojca") +} +pub fn fmt_ev_muted(name: &str) -> String { + format!("{name} idzie robić jajecznice") +} +pub fn fmt_ev_unmuted(name: &str) -> String { + format!("{name} zrobił jajecznice") +} + +// ── AI assistant (events/message handler) ───────────────────────────── + +pub const ASSISTANT_SYSTEM_PROMPT: &str = + "Jesteś asystentem. Odpowiadasz po polsku, jesteś pomocny, nie używasz emoji."; + +// ── Convenience helpers for repeated Error::Custom patterns ──────────── + +use crate::error::Error; + +pub fn err_not_in_guild() -> Error { + Error::Custom(L_NOT_IN_GUILD.into()) +} +pub fn err_not_in_voice() -> Error { + Error::Custom(L_NOT_IN_VOICE.into()) +} +pub fn err_trigger_not_found() -> Error { + Error::Custom(L_TRIGGER_NOT_FOUND.into()) +} +pub fn err_bad_channel_type() -> Error { + Error::Custom(L_BAD_CHANNEL_TYPE.into()) +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000..4a89315 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,190 @@ +use std::collections::HashMap; + +use poise::serenity_prelude as serenity; +use songbird::SerenityInit; +use tracing_subscriber::filter::EnvFilter; + +mod commands; +mod config; +mod copyparty; +mod database; +mod error; +mod events; +mod lang; +mod openrouter; +mod tasks; +mod voice; + +use database::DbPool; + +pub struct Data { + pub db_pool: DbPool, + pub config: config::Config, + pub voice_manager: voice::VoiceManager, + pub copyparty: copyparty::CopypartyClient, + pub http_client: reqwest::Client, + pub volume: std::sync::atomic::AtomicU64, + pub triggers: tokio::sync::Mutex, Vec)>>, + /// Autocomplete lookup, keyed per-user to avoid cross-command collisions. + /// + /// Mirrors Python's per-cog `_cached_urls[user_id]` map: when two users + /// search the same filename simultaneously, each invocation must resolve its + /// own stashed `SearchHit`, so the cache is keyed by `(UserId, filename)`. + pub ac_cache: tokio::sync::Mutex>, + /// Active music track per guild (only one music source at a time). + pub music_tracks: tokio::sync::Mutex>, +} + +impl Data { + pub async fn reload_triggers(&self) { + let triggers = database::load_all_triggers_with_keywords_and_responses(&self.db_pool) + .await + .unwrap_or_default(); + *self.triggers.lock().await = triggers; + } + + pub async fn ac_stash( + &self, + user_id: serenity::UserId, + key: String, + hit: copyparty::SearchHit, + ) { + self.ac_cache.lock().await.insert((user_id, key), hit); + } + + pub async fn ac_lookup( + &self, + user_id: serenity::UserId, + key: &str, + ) -> Option { + self.ac_cache + .lock() + .await + .get(&(user_id, key.to_string())) + .cloned() + } +} + +#[allow(dead_code)] +type Error = error::Error; + +#[tokio::main] +async fn main() { + let _ = rustls::crypto::ring::default_provider().install_default(); + + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::builder() + .parse(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into())) + .unwrap(), + ) + .init(); + + dotenvy::dotenv().ok(); + config::check_env_vars(); + let config = config::Config::from_env(); + + let db_pool = database::init_pool(&config.database_path) + .await + .expect("Failed to initialize database"); + + let http_client = reqwest::Client::new(); + + let copyparty = copyparty::CopypartyClient::new( + config.copyparty_url.clone(), + config.copyparty_url_upload.clone(), + config.copyparty_password.clone(), + ); + + let triggers = database::load_all_triggers_with_keywords_and_responses(&db_pool) + .await + .unwrap_or_default(); + + let commands_list = commands::all_commands(); + let config_for_client = config.clone(); + + let framework = poise::Framework::builder() + .options(poise::FrameworkOptions { + commands: commands_list, + event_handler: |ctx, event, _framework, user_data| { + Box::pin(async move { + handle_event(ctx, event, user_data).await; + Ok(()) + }) + }, + ..Default::default() + }) + .setup(move |ctx, _ready, framework| { + let config = config.clone(); + let db_pool = db_pool.clone(); + let http_client = http_client.clone(); + let copyparty = copyparty.clone(); + let triggers = triggers.clone(); + Box::pin(async move { + let vm = voice::VoiceManager::new(ctx).await; + + let data = Data { + config, + db_pool, + voice_manager: vm, + copyparty, + http_client, + volume: std::sync::atomic::AtomicU64::new(1.0f64.to_bits()), + triggers: tokio::sync::Mutex::new(triggers), + ac_cache: tokio::sync::Mutex::new(HashMap::new()), + music_tracks: tokio::sync::Mutex::new(HashMap::new()), + }; + + // Spawn cron scheduler loop. + { + let cron_ctx = crate::tasks::CronContext { + pool: data.db_pool.clone(), + http: ctx.http.clone(), + voice_manager: data.voice_manager.clone(), + copyparty: data.copyparty.clone(), + }; + tokio::spawn(async move { + crate::tasks::cron_scheduler_loop(cron_ctx).await; + }); + } + + poise::builtins::register_globally(ctx, &framework.options().commands).await?; + + events::handle_ready(ctx).await.map_err(|e| { + tracing::error!("on_ready handler: {e}"); + e + })?; + + Ok(data) + }) + }) + .build(); + + let mut client = + serenity::ClientBuilder::new(&config_for_client.token, serenity::GatewayIntents::all()) + .framework(framework) + .register_songbird() + .await + .expect("Error creating client"); + + client.start().await.expect("Error starting client"); +} + +async fn handle_event(ctx: &serenity::Context, event: &serenity::FullEvent, user_data: &Data) { + match event { + serenity::FullEvent::Ready { + data_about_bot: _, .. + } => {} + serenity::FullEvent::VoiceStateUpdate { old, new } => { + events::handle_voice_state_update(ctx, user_data, old.clone(), new) + .await + .unwrap_or_else(|e| tracing::error!("{}: {e}", lang::L_EV_VOICE_STATE_ERROR)); + } + serenity::FullEvent::Message { new_message } => { + events::handle_message(ctx, user_data, new_message) + .await + .unwrap_or_else(|e| tracing::error!("{}: {e}", lang::L_EV_MESSAGE_HANDLER_ERROR)); + } + _ => {} + } +} diff --git a/src/openrouter/mod.rs b/src/openrouter/mod.rs new file mode 100644 index 0000000..31cee85 --- /dev/null +++ b/src/openrouter/mod.rs @@ -0,0 +1,134 @@ +use reqwest::Client; +use serde::{Deserialize, Serialize}; + +use crate::error::Result; +use crate::lang; + +pub struct OpenRouterClient { + api_key: String, + model: String, + client: Client, +} + +#[derive(Serialize)] +struct ChatRequest { + model: String, + messages: Vec, + max_tokens: u32, +} + +#[derive(Serialize)] +struct Message { + role: String, + content: String, +} + +#[derive(Deserialize)] +struct ChatResponse { + choices: Vec, +} + +#[derive(Deserialize)] +struct Choice { + message: MessageContent, +} + +#[derive(Deserialize)] +struct MessageContent { + content: String, +} + +impl OpenRouterClient { + pub fn new(api_key: String, model: String) -> Self { + Self { + api_key, + model, + client: Client::new(), + } + } + + pub async fn chat(&self, system_prompt: &str, user_message: &str) -> Result { + let body = ChatRequest { + model: self.model.clone(), + messages: vec![ + Message { + role: "system".into(), + content: system_prompt.to_string(), + }, + Message { + role: "user".into(), + content: user_message.to_string(), + }, + ], + max_tokens: 500, + }; + + let resp = self + .client + .post("https://openrouter.ai/api/v1/chat/completions") + .header("Authorization", format!("Bearer {}", self.api_key)) + .json(&body) + .send() + .await?; + + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(crate::error::Error::Custom(lang::fmt_openrouter_err( + &status, &text, + ))); + } + + let chat_resp: ChatResponse = resp.json().await?; + Ok(chat_resp + .choices + .into_iter() + .next() + .map(|c| c.message.content) + .unwrap_or_default()) + } + + pub async fn chat_with_history( + &self, + system_prompt: &str, + messages: Vec<(String, String)>, // (role, content) pairs + ) -> Result { + let mut msg_vec = vec![Message { + role: "system".into(), + content: system_prompt.to_string(), + }]; + for (role, content) in messages { + msg_vec.push(Message { role, content }); + } + + let body = ChatRequest { + model: self.model.clone(), + messages: msg_vec, + max_tokens: 500, + }; + + let resp = self + .client + .post("https://openrouter.ai/api/v1/chat/completions") + .header("Authorization", format!("Bearer {}", self.api_key)) + .json(&body) + .send() + .await?; + + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + return Err(crate::error::Error::Custom(lang::fmt_openrouter_err( + &status, &text, + ))); + } + + let chat_resp: ChatResponse = resp.json().await?; + Ok(chat_resp + .choices + .into_iter() + .next() + .map(|c| c.message.content) + .unwrap_or_default()) + } +} diff --git a/src/tasks/mod.rs b/src/tasks/mod.rs new file mode 100644 index 0000000..f3f8811 --- /dev/null +++ b/src/tasks/mod.rs @@ -0,0 +1,172 @@ +use std::sync::Arc; + +use chrono::{Duration, Timelike, Utc}; +use cron::Schedule; +use poise::serenity_prelude as serenity; +use std::str::FromStr; +use tokio::time::sleep; +use tracing::{error, info}; + +use crate::copyparty::CopypartyClient; +use crate::database::{self, DbPool}; +use crate::voice::VoiceManager; + +/// Shared resources the cron scheduler needs. +#[derive(Clone)] +pub struct CronContext { + pub pool: DbPool, + pub http: Arc, + pub voice_manager: VoiceManager, + pub copyparty: CopypartyClient, +} + +/// Normalize user-friendly cron expressions to the 7-field format the `cron` crate expects. +/// Supports: 5-field (min hour dom month dow) and 6-field (sec min hour dom month dow). +fn parse_cron(raw: &str) -> Result { + let fields: Vec<&str> = raw.split_whitespace().collect(); + let normalized = match fields.len() { + 7 => raw.to_string(), + 6 => format!("{} *", raw), // sec min hour dom month dow → +year + 5 => format!("0 {} *", raw), // min hour dom month dow → +sec +year + _ => raw.to_string(), // let the crate reject it + }; + Schedule::from_str(&normalized) +} + +/// Check window: look for cron fires in the last WINDOW_SECONDS. +/// 61s gives 1s overlap with 60s tick to avoid missing exact-boundary fires. +const WINDOW_SECONDS: i64 = 61; +/// Tick interval for the scheduler loop. +const TICK_SECONDS: u64 = 60; + +pub async fn check_and_execute_jobs(ctx: &CronContext) { + let jobs = match database::list_active_cron_jobs(&ctx.pool).await { + Ok(jobs) => jobs, + Err(e) => { + error!("Failed to load cron jobs: {e}"); + return; + } + }; + + let now = Utc::now(); + let window_start = now - Duration::seconds(WINDOW_SECONDS); + + for job in jobs { + let schedule = match parse_cron(&job.cron_expression) { + Ok(s) => s, + Err(e) => { + error!("Invalid cron expression '{}': {e}", job.cron_expression); + continue; + } + }; + + // Check if any fire time fell within the last WINDOW_SECONDS. + let mut has_fired = false; + for next_fire in schedule.after(&window_start) { + if next_fire > now { + break; + } + has_fired = true; + break; // one fire per window is enough + } + + if !has_fired { + continue; + } + + info!("Executing cron job {}", job.id); + match job.r#type.as_str() { + "audio" => execute_audio_job(ctx, &job).await, + "wiadomość" => execute_message_job(ctx, &job).await, + _ => {} + } + } +} + +async fn execute_message_job(ctx: &CronContext, job: &crate::database::CronJob) { + let Some(ref msg_text) = job.message else { + return; + }; + let channel_id = serenity::ChannelId::new(job.channel_id as u64); + if let Err(e) = channel_id.say(&ctx.http, msg_text).await { + error!("Cron job {} failed to send message: {e}", job.id); + } else { + info!("Cron job {} sent message to {}", job.id, job.channel_id); + } +} + +async fn execute_audio_job(ctx: &CronContext, job: &crate::database::CronJob) { + let Some(ref audio_path) = job.audio_filename else { + error!("Cron audio job {} has no audio_filename", job.id); + return; + }; + + let channel_id = serenity::ChannelId::new(job.channel_id as u64); + + // Resolve guild + build URL concurrently. + let url = ctx.copyparty.build_url(audio_path); + info!("Cron audio job {} streaming via ffmpeg: {url}", job.id); + let guild_channel_fut = channel_id.to_channel(&ctx.http); + let input_fut = crate::voice::ffmpeg_stream_input(&url); + + let (channel_result, input_result) = tokio::join!(guild_channel_fut, input_fut); + + let guild_id = match channel_result { + Ok(serenity::Channel::Guild(ch)) => ch.guild_id, + other => { + error!( + "Cron audio job {}: channel {} is not a guild voice channel ({:?})", + job.id, + job.channel_id, + other.map(|_| ()) + ); + return; + } + }; + + let input = match input_result { + Ok(i) => i, + Err(e) => { + error!( + "Cron audio job {} ffmpeg stream failed ({url}): {e}", + job.id + ); + return; + } + }; + + // Connect to voice and play. + let call = match ctx.voice_manager.connect(guild_id, channel_id).await { + Ok(c) => c, + Err(e) => { + error!("Cron audio job {} voice connect failed: {e}", job.id); + return; + } + }; + + { + let mut call_lock = call.lock().await; + call_lock.play_input(input); + } + + info!( + "Cron audio job {} playing '{}' (url: {url}) in guild {}", + job.id, audio_path, guild_id + ); +} + +pub async fn cron_scheduler_loop(ctx: CronContext) { + // Align first tick to the start of the next minute, then run every TICK_SECONDS. + let now = Utc::now(); + let next_minute = (now + Duration::minutes(1)) + .with_second(0) + .and_then(|t| t.with_nanosecond(0)) + .unwrap_or(now); + let initial_delay = (next_minute - now).to_std().unwrap_or_default(); + sleep(initial_delay).await; + + loop { + check_and_execute_jobs(&ctx).await; + sleep(std::time::Duration::from_secs(TICK_SECONDS)).await; + } +} diff --git a/src/voice/mod.rs b/src/voice/mod.rs new file mode 100644 index 0000000..0d5b89e --- /dev/null +++ b/src/voice/mod.rs @@ -0,0 +1,404 @@ +use std::io::Cursor; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; + +use serenity::all::{ChannelId, GuildId}; +use songbird::input::{AsyncAdapterStream, AsyncReadOnlySource, RawAdapter}; +use songbird::{Call, Songbird}; +use symphonia::core::audio::sample::Sample; +use symphonia::core::codecs::audio::AudioDecoderOptions; +use symphonia::core::formats::probe::Hint; +use symphonia::core::formats::{FormatOptions, TrackType}; +use symphonia::core::io::MediaSourceStream; +use symphonia::core::meta::MetadataOptions; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use crate::error::{Error, Result}; +use crate::lang; + +#[derive(Clone)] +pub struct VoiceManager { + pub songbird: Arc, +} + +impl VoiceManager { + pub async fn new(ctx: &serenity::client::Context) -> Self { + Self { + songbird: songbird::get(ctx) + .await + .expect("Songbird not registered via register_songbird()"), + } + } + + pub async fn connect( + &self, + guild_id: GuildId, + channel_id: ChannelId, + ) -> Result>> { + let mut last_err = None; + for attempt in 1..=3 { + match self.songbird.join(guild_id, channel_id).await { + Ok(call) => { + tokio::time::sleep(Duration::from_millis(500)).await; + return Ok(call); + } + Err(e) => { + warn!( + "Voice join attempt {attempt}/3 failed for guild {}: {e}", + guild_id + ); + if let songbird::error::JoinError::Driver(ref driver_err) = e { + warn!("Driver error details: {driver_err:?}"); + } + last_err = Some(e); + tokio::time::sleep(Duration::from_secs(1)).await; + } + } + } + Err(Error::SongbirdJoin(last_err.unwrap())) + } + + pub async fn disconnect(&self, guild_id: GuildId) -> Result<()> { + self.songbird + .remove(guild_id) + .await + .map_err(|e| Error::Custom(lang::fmt_voice_remove_failed(&e)))?; + Ok(()) + } + + pub fn get_call(&self, guild_id: GuildId) -> Option>> { + self.songbird.get(guild_id) + } +} + +// ── Audio decoding ───────────────────────────────────────────────────── + +/// Decode an in-memory audio buffer into raw f32 stereo PCM for songbird. +/// Supports mp3, m4a/aac, flac, ogg, wav, opus via symphonia auto-detection. +pub fn decode_audio(data: &[u8]) -> Result { + let mss = MediaSourceStream::new(Box::new(Cursor::new(data.to_vec())), Default::default()); + decode_stream(mss) +} + +fn decode_stream(mss: MediaSourceStream) -> Result { + let hint = Hint::new(); + let fmt_opts = FormatOptions::default(); + let meta_opts = MetadataOptions::default(); + + let mut format = symphonia::default::get_probe() + .probe(&hint, mss, fmt_opts, meta_opts) + .map_err(|e| Error::Custom(lang::fmt_voice_probe_failed(&e)))?; + + let track = format + .default_track(TrackType::Audio) + .ok_or_else(|| Error::Custom(lang::L_VOICE_NO_TRACK.into()))?; + let track_id = track.id; + + let codec_params = track + .codec_params + .as_ref() + .and_then(|p| p.audio()) + .ok_or_else(|| Error::Custom(lang::L_VOICE_NO_CODEC.into()))?; + + let sample_rate = codec_params.sample_rate.unwrap_or(48000); + let channels = codec_params + .channels + .clone() + .map(|c| c.count() as u32) + .unwrap_or(2); + + let dec_opts = AudioDecoderOptions::default(); + let mut decoder = symphonia::default::get_codecs() + .make_audio_decoder(codec_params, &dec_opts) + .map_err(|e| Error::Custom(lang::fmt_voice_decoder_failed(&e)))?; + + let mut pcm_f32: Vec = Vec::new(); + + loop { + match format.next_packet() { + Ok(Some(packet)) => { + if packet.track_id != track_id { + continue; + } + if let Ok(buf) = decoder.decode(&packet) { + let n = buf.samples_interleaved(); + let start = pcm_f32.len(); + pcm_f32.resize(start + n, f32::MID); + buf.copy_to_slice_interleaved(&mut pcm_f32[start..]); + } + } + Ok(None) => break, + Err(_) => break, + } + } + + let pcm_bytes: Vec = pcm_f32.iter().flat_map(|&s| s.to_le_bytes()).collect(); + let cursor = Cursor::new(pcm_bytes); + let adapter = RawAdapter::new(cursor, sample_rate, channels); + + Ok(adapter.into()) +} + +// ── yt-dlp helper ────────────────────────────────────────────────────── + +/// Download audio via yt-dlp, collect stdout into memory, return the raw bytes. +/// Works for both finite media and live streams (yt-dlp handles the difference). +pub async fn ytdlp_download(url: &str) -> Result> { + let output = tokio::process::Command::new("yt-dlp") + .args(["-f", "bestaudio", "-o", "-", "--no-playlist", url]) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .output() + .await + .map_err(|e| Error::Custom(lang::fmt_voice_ytdlp_launch(&e)))?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + // Keep the full yt-dlp trace in the logs for debugging — the user + // only ever sees the short, translated `fmt_ytdlp_stderr` message. + warn!(url = %url, "yt-dlp failed:\n{}", stderr.trim_end()); + return Err(Error::Custom(lang::fmt_ytdlp_stderr(&stderr))); + } + + Ok(output.stdout) +} + +// ── Compression helpers ─────────────────────────────────────────────── + +/// Download a URL into memory via reqwest. +pub async fn download_url(client: &reqwest::Client, url: &str) -> Result> { + let resp = client + .get(url) + .send() + .await + .map_err(|e| Error::Custom(format!("HTTP GET failed for {url}: {e}")))?; + let bytes = resp + .bytes() + .await + .map_err(|e| Error::Custom(format!("Failed to read body for {url}: {e}")))?; + Ok(bytes.to_vec()) +} + +/// DEBUG: bypass encode/decode — just pipe through ffmpeg as PCM. +/// If this plays, the streaming pipeline works; bug is in Opus roundtrip. +pub async fn degrade_audio(input: &[u8]) -> Result { + use std::process::Stdio; + + let mut child = tokio::process::Command::new("ffmpeg") + .args([ + "-y", "-i", "pipe:0", "-vn", "-f", "f32le", "-ac", "2", "-ar", "8000", "-b:a", "500", + "pipe:1", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + .map_err(|e| Error::Custom(format!("ffmpeg passthrough spawn: {e}")))?; + + let mut stdin = child.stdin.take().unwrap(); + let stdout = child + .stdout + .take() + .ok_or_else(|| Error::Custom("stdout not captured".into()))?; + + let owned_input = input.to_vec(); + tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let _ = stdin.write_all(&owned_input).await; + }); + + let ffmpeg_child = FfmpegChild { + _child: child, + stdout, + }; + + let async_source = AsyncReadOnlySource::new(ffmpeg_child); + let adapter = AsyncAdapterStream::new(Box::new(async_source), 64 * 1024); + let raw = RawAdapter::new(adapter, 48000, 2); + Ok(raw.into()) +} + +// ── HTTP streaming via songbird's built-in HttpRequest (legacy) ───── + +/// Create a songbird `Input` that streams audio directly from an HTTP(S) URL. +/// Works for both finite files and infinite Icecast/Shoutcast live streams. +/// +/// Superseded by `ffmpeg_stream_input` for the `/play` command (which needs +/// HLS/playlist/Icecast-metadata handling that symphonia can't provide), but +/// retained for other callers that want lightweight symphonia-based streaming. +#[allow(dead_code)] +pub fn http_stream_input(client: reqwest::Client, url: &str) -> songbird::input::Input { + info!(%url, "streaming via HttpRequest"); + songbird::input::HttpRequest::new(client, url.to_string()).into() +} + +// ── Live streaming via ffmpeg subprocess ───────────────────────────── + +/// Adapter holding an ffmpeg child process and its stdout pipe. +/// +/// Implements `AsyncRead` by delegating to `ChildStdout`. The `Child` is kept +/// alive for the lifetime of the stream and killed on drop via +/// `kill_on_drop(true)`, so stopping playback terminates ffmpeg. +/// +/// `Sync`: the `Child` handle is only touched on `Drop` (a single, one-shot +/// operation); `ChildStdout` is only read by the single `AsyncAdapterSink` +/// task spawned inside `AsyncAdapterStream`. No field is accessed concurrently. +struct FfmpegChild { + _child: tokio::process::Child, + stdout: tokio::process::ChildStdout, +} + +impl tokio::io::AsyncRead for FfmpegChild { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stdout).poll_read(cx, buf) + } +} + +/// Stream an HTTP(S) URL live via an ffmpeg subprocess, producing a songbird +/// `Input` of raw f32 PCM at 48 kHz stereo — NO full download. +/// +/// ffmpeg handles every streaming scenario the Python bot relied on it for: +/// Icecast/Shoutcast radio (incl. inline metadata), HLS `.m3u8` (segment +/// fetching), `.m3u`/`.pls` playlists, and direct media files. The +/// `-reconnect` flags let ffmpeg recover from dropped connections. +/// +/// Byte path: `ffmpeg stdout` → `AsyncReadOnlySource` (unseekable async +/// media source) → `AsyncAdapterStream` (async→sync ring buffer) → +/// `RawAdapter` (prepends `SbirdRaw` f32 header) → songbird `Input`. +pub async fn ffmpeg_stream_input(url: &str) -> Result { + use std::process::Stdio; + + let mut child = tokio::process::Command::new("ffmpeg") + .args([ + "-reconnect", + "1", + "-reconnect_streamed", + "1", + "-reconnect_delay_max", + "5", + "-i", + url, + "-vn", + "-f", + "f32le", + "-ar", + "48000", + "-ac", + "2", + "pipe:1", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .map_err(|e| Error::Custom(lang::fmt_voice_ffmpeg_launch(&e)))?; + + let stdout = child + .stdout + .take() + .ok_or_else(|| Error::Custom("ffmpeg stdout was not captured".into()))?; + + // Drain stderr in background so ffmpeg doesn't block; log first 4 KiB on drop. + if let Some(stderr) = child.stderr.take() { + let url_owned = url.to_string(); + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + let mut buf = Vec::with_capacity(4096); + let mut reader = tokio::io::BufReader::new(stderr); + let _ = reader.read_to_end(&mut buf).await; + let text = String::from_utf8_lossy(&buf); + if !text.is_empty() { + warn!(url = %url_owned, "ffmpeg stderr: {}", text.trim_end()); + } + }); + } + + let ffmpeg_child = FfmpegChild { + _child: child, + stdout, + }; + + let async_source = AsyncReadOnlySource::new(ffmpeg_child); + let adapter = AsyncAdapterStream::new(Box::new(async_source), 64 * 1024); + let raw = RawAdapter::new(adapter, 48000, 2); + Ok(raw.into()) +} + +/// Smoke-test an HTTP stream URL: creates a raw reqwest GET, reads the first +/// chunk directly from the async bytes stream (bypasses AsyncAdapterStream). +/// Logs whether data arrives — isolates the hang to songbird's sync adapter. +#[allow(dead_code)] +pub async fn smoke_test_stream(client: reqwest::Client, url: String) { + use futures_util::StreamExt; + + info!(%url, "smoke_test: sending GET"); + let resp = match client.get(url.as_str()).send().await { + Ok(r) => r, + Err(e) => { + warn!(%url, error = %e, "smoke_test: GET FAILED"); + return; + } + }; + info!(%url, status = %resp.status(), "smoke_test: response received"); + + let mut stream = resp.bytes_stream(); + info!(%url, "smoke_test: awaiting first chunk"); + match stream.next().await { + Some(Ok(chunk)) => { + info!(%url, len = chunk.len(), "smoke_test: first chunk arrived"); + } + Some(Err(e)) => { + warn!(%url, error = %e, "smoke_test: chunk error"); + } + None => { + info!(%url, "smoke_test: stream ended immediately (0 bytes)"); + } + } +} + +/// Diagnostic: fetch headers and full body for a URL, logging key metrics. +/// Helps debug symphonia probe EOF issues (Content-Length vs actual bytes). +/// +/// **Warning**: hangs forever on infinite streams (Icecast). Finite URLs only. +#[allow(dead_code)] +pub async fn debug_fetch(client: &reqwest::Client, url: &str) { + match client.get(url).send().await { + Ok(resp) => { + let status = resp.status(); + let headers = resp.headers().clone(); + let content_length = headers + .get("content-length") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()); + let content_type = headers.get("content-type").and_then(|v| v.to_str().ok()); + + info!( + %url, %status, + ?content_length, ?content_type, + "HTTP response headers" + ); + + match resp.bytes().await { + Ok(body) => { + let actual_len = body.len() as u64; + let reported_len = content_length.unwrap_or(0); + info!( + %url, reported = reported_len, actual = actual_len, + delta = actual_len as i64 - reported_len as i64, + "buffered .bytes(): reported vs actual" + ); + } + Err(e) => warn!(%url, error = %e, ".bytes() failed"), + } + } + Err(e) => warn!(%url, error = %e, "HTTP GET failed"), + } +} -- cgit v1.2.3