Something went wrong. Try again.
Source code for my personal quote bot project.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314use bsky_sdk::api::app::bsky::feed::post;use bsky_sdk::api::types::string::Datetime;use bsky_sdk::{BskyAgent, api::types::Object};
use glob::glob;use grep::{regex, searcher::sinks};use rand::seq::SliceRandom;use redis::aio::ConnectionManagerConfig;
use std::{sync::Arc, time::Duration};use tokio_cron_scheduler::{Job, JobScheduler};
use redis::AsyncCommands;
const DEFAULT_QUEUE: &str = "queue:default";const EVENT_QUEUE: &str = "queue:event";
// See https://cron.help for what these strings meanconst POSTING_INTERVAL_CRON: &str = "0 0,30 * * * *";const POSTING_INTERVAL_DEBUG: &str = "1/10 * * * * *";const EVENT_UPDATE_INTERVAL: &str = "55 23 * * *";
const POSTING_RETRIES: i32 = 5;
fn prepare_post<I: Into<String>>(text: I) -> post::RecordData { post::RecordData { text: text.into(), created_at: Datetime::now(), embed: None, entities: None, facets: None, labels: None, langs: None, reply: None, tags: None, }}
#[derive(Clone, Debug)]struct QuoteFilter { path: String, content: String, dates: Vec<String>,}
impl QuoteFilter { pub async fn get_quote( &self, mut con: impl redis::aio::ConnectionLike + AsyncCommands + Clone, ) -> Result<String, ()> { // 1: Attempt to read from the event (priority) queue let event_quote: Option<String> = con.lpop(EVENT_QUEUE, None).await.ok(); if let Some(quote) = event_quote { return Ok(quote); }
// 2: Otherwise, we read from the regular queue, repopulating it if it's empty self.reshuffle_quotes(con.clone(), DEFAULT_QUEUE).await?; con.lpop(DEFAULT_QUEUE, None).await.map_err(|_| ()) }
async fn reshuffle_quotes( &self, mut con: impl redis::aio::ConnectionLike + AsyncCommands, output_queue: &str, ) -> Result<(), ()> { let len: u64 = con.llen(output_queue).await.map_err(|_| ())?; // NOTE: The following assumes the queue hasn't been repopulated by any other client // in-between the call to llen and the execution of the pipeline. // Hopefully won't be a problem :) if len == 0 { let mut file_contents = self.read_files();
{ let mut rand = rand::rng(); file_contents.shuffle(&mut rand); }
let mut pipeline = redis::pipe(); for file_contents in file_contents.into_iter() { pipeline.lpush(output_queue, file_contents.as_str()); } let _: () = pipeline.query_async(&mut con).await.map_err(|_| ())?; }
Ok(()) }
fn read_files(&self) -> Vec<String> { let matcher = regex::RegexMatcher::new(&self.content).unwrap(); let mut searcher = grep::searcher::Searcher::new(); let mut results = Vec::new();
for file in glob(&self.path).unwrap() { let file = match file { Ok(file) => file, Err(_) => continue, };
let mut matched = false; let sink = sinks::Lossy(|_lnum, _line| { matched = true; Ok(false) });
let search_result = searcher.search_path(&matcher, &file, sink); if !matched || search_result.is_err() { continue; }
let contents = std::fs::read_to_string(file).unwrap(); results.push(contents.trim().to_string()); }
results }}
#[derive(Clone)]struct RedisState { con_manager: redis::aio::ConnectionManager,}
impl RedisState { pub async fn new(url: String) -> Result<Self, ()> { let redis = redis::Client::open(url).map_err(|_| ())?; let config = ConnectionManagerConfig::new() .set_response_timeout(std::time::Duration::from_secs(10)) .set_number_of_retries(3); let con_manager = redis::aio::ConnectionManager::new_with_config(redis, config) .await .map_err(|_| ())?;
Ok(RedisState { con_manager }) }
pub async fn fetch_quote(&self, filter: &QuoteFilter) -> Result<String, ()> { loop { match filter.get_quote(self.con_manager.clone()).await { Ok(text) => return Ok(text), Err(_) => eprintln!("Error fetching quote from redis storage. Retrying..."), }; } }}
#[derive(Clone)]struct BlueskyState { bsky_agent: BskyAgent, bsky_session: Object<bsky_sdk::api::com::atproto::server::create_session::OutputData>,}
impl BlueskyState { pub async fn new_session(username: String, password: String) -> Result<Self, ()> { let agent = BskyAgent::builder().build().await.map_err(|_| ())?; let session = agent.login(username, password).await.map_err(|_| ())?;
Ok(Self { bsky_agent: agent, bsky_session: session, }) }
pub async fn submit_post(self, post: String) -> Result<(), ()> { let post = prepare_post(post.as_str());
for current_try in 0..POSTING_RETRIES { if let Err(e) = self.bsky_agent.create_record(post.clone()).await { eprintln!("Could not post quote: `{e}`"); eprintln!("Attempting to refresh login...");
if let Err(e) = self .bsky_agent .resume_session(self.bsky_session.clone()) .await { eprintln!("Failed to resume sessions due to following error: {e}") } } else { if current_try > 0 { eprintln!("Successfully posted quote on retry #{current_try}"); } return Ok(()); } }
Err(()) }}
#[derive(Clone)]struct State { redis: RedisState, bsky_session: Option<BlueskyState>,}
impl State { pub fn redis(&self) -> &RedisState { &self.redis }
pub fn bsky(&self) -> Option<&BlueskyState> { self.bsky_session.as_ref() }}
#[tokio::main]async fn main() -> Result<(), Box<dyn std::error::Error>> { let debug_mode = std::env::var("DEBUG").unwrap_or("0".to_string()) == "1"; let use_bsky = std::env::var("USE_BLUESKY").unwrap_or("0".to_string()) == "1";
let redis_state = RedisState::new(std::env::var("REDIS_URL").unwrap_or("redis://localhost".to_string())) .await .expect("Initial redis connection failure"); let bsky_state = if use_bsky { Some( BlueskyState::new_session( std::env::var("BLUESKY_USERNAME").expect("Bluesky username not supplied"), std::env::var("BLUESKY_PASSWORD") .expect("Bluesky application password not supplied"), ) .await .expect("Could not connect to Bluesky with supplied credentials"), ) } else { None };
let app_state = Arc::new(State { redis: redis_state, bsky_session: bsky_state, });
let sched = JobScheduler::new().await?;
/* let event_filter = Arc::new(QuoteFilter { content: r"\b(?i:mother|mommy|mama|mom)\b".to_string(), path: "test/**/*.txt".to_string(), dates: vec![], }); */
let regular_filter = Arc::new(QuoteFilter { content: r".*".to_string(), path: if !debug_mode { "quotes/**/*.txt".to_string() } else { "test/**/*.txt".to_string() }, dates: vec![], });
let posting_interval = if !debug_mode { POSTING_INTERVAL_CRON } else { POSTING_INTERVAL_DEBUG };
let post_job = Job::new_async(posting_interval, move |_uuid, _| { let filter = regular_filter.clone(); let app_state = app_state.clone();
Box::pin(async move { // We try fetching a new quote from our redis storage until we succeed let text = match app_state.redis().fetch_quote(&filter).await { Ok(text) => text, Err(_) => { eprintln!("Error fetching quote from redis storage."); return; } };
if let Some(bsky) = app_state.bsky() { if let Err(_) = bsky.clone().submit_post(text).await { eprintln!("Error posting to bluesky."); return; } } else { // Let's just print the quote! println!("{}\n", text); } }) })?;
// Add async job sched.add(post_job).await?;
// sched // .add(Job::new_async(EVENT_UPDATE_INTERVAL, move |_uuid, _| { // let filter = event_filter.clone(); // let con = con_event_monitor.clone(); // let _agent = agent_event_monitor.clone(); // Can be used later to e.g. update profile
// Box::pin(async move { // // For testing purposes, let's always upload events // reshuffle_quotes(&filter, con.clone(), EVENT_QUEUE) // .await // .unwrap(); // }) // })?) // .await?;
sched .start() .await .expect("Error starting tokio scheduler. Shutting down..."); loop { tokio::time::sleep(Duration::from_secs(10)).await; }}