Twitter Bot: tokio and futures are hard. Not yet working

.dockerignore

@@ -0,0 +1,3 @@
+input/
+output/
+target/

.gitignore

@@ -0,0 +1,9 @@
+*~
+.idea/
+input/
+output/
+run_cargo.sh
+run_docker.sh
+secret.yaml
+secrets.yaml
+target/

Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "twitter-vocodes-bot"
+edition = "2018"
+version = "0.0.1"
+authors = [
+  "Brandon Thomas <bt@brand.io>",
+  "Brandon Thomas <echelon@gmail.com>",
+]
+publish = false
+
+[[bin]]
+name = "twitter-vocodes-bot"
+path = "src/main.rs"
+
+[dependencies]
+anyhow = "1.0.31"
+chrono = { version = "0.4", features = ["serde"] }
+crossbeam = "0.7.3"
+easyenv = "0.0.2"
+egg-mode = "0.15"
+env_logger = "0.7.1"
+futures = "0.3"
+log = "0.4.11"
+serde = "1.0.104"
+serde_derive = "1.0.104"
+serde_json = "1.0.55"
+subprocess = "0.2.6"
+tokio = { version = "0.2.8", features = ["time", "rt-core", "macros", "blocking", "rt-threaded"] }
+
+# From egg-mode
+yansi = "0.5.0"
+structopt = "0.3.13"

Dockerfile

@@ -0,0 +1,43 @@
+FROM ubuntu:focal as build
+
+ARG DEBIAN_FRONTEND=noninteractive
+
+WORKDIR /tmp
+
+RUN apt-get update \
+    && apt-get install -y \
+        build-essential \
+        curl \
+        gdb \
+        libssl-dev \
+        libssl1.1 \
+        pkg-config \
+        unzip \
+        vim \
+        wget
+
+RUN apt-cache search libssl | grep libssl
+
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs \
+    | sh  -s -- --default-toolchain stable -y
+
+COPY Cargo.toml .
+
+# Prefetch dependencies so code iteration is fast
+RUN $HOME/.cargo/bin/cargo fetch
+
+# Pre-build the dependencies.
+# Code changes won't make us rebuild them.
+RUN set -x \
+  && mkdir -p src \
+  && echo 'fn main() { println!("rebuild me!"); }' > src/main.rs \
+  && $HOME/.cargo/bin/cargo build
+
+COPY src/ src/
+
+# If the timestamps are older on actual src/ files than
+# cached "fake build" above, nothing gets built.
+RUN touch src/main.rs
+
+RUN $HOME/.cargo/bin/cargo build
+

README.md

@@ -0,0 +1,3 @@
+twitter-vocodes-bot
+===================
+A bot for responding to tweets.

src/common.rs

@@ -0,0 +1,230 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+//since this is going to get included in examples that might not use everything, clear out warnings
+//that are unnecessary
+#![allow(dead_code)]
+
+use egg_mode;
+use std;
+use std::io::{Read, Write};
+
+pub use yansi::Paint;
+use crate::Secrets;
+
+//This is not an example that can be built with cargo! This is some helper code for the other
+//examples so they can load access keys from the same place.
+
+#[derive(Clone)]
+pub struct Config {
+  pub token: egg_mode::Token,
+  pub user_id: u64,
+  pub screen_name: String,
+}
+
+impl Config {
+  pub async fn load(secrets: Secrets) -> Self {
+    let a1 = Config::load_inner(secrets.clone()).await;
+    if let Some(conf) = a1 {
+      return conf;
+    }
+
+    Config::load_inner(secrets.clone()).await.unwrap()
+  }
+
+  /// This needs to be a separate function so we can retry after creating the
+  /// twitter_settings file. Idealy we would recurse, but that requires boxing
+  /// the output which doesn't seem worthwhile
+  async fn load_inner(secrets: Secrets) -> Option<Self> {
+    //IMPORTANT: make an app for yourself at apps.twitter.com and get your
+    //key/secret into these files; these examples won't work without them
+    //let consumer_key = include_str!("consumer_key").trim();
+    //let consumer_secret = include_str!("consumer_secret").trim();
+
+    // TODO: DO NOT COMMIT
+    let consumer_key = &secrets.api_key;
+    let consumer_secret = &secrets.api_secret_key;
+
+
+    println!("Trying to get connection to Twitter - creating con_token");
+    // "It works when I pass a Token with the Access variant instead of the Bearer variant."
+    let con_token = egg_mode::KeyPair::new(consumer_key.clone(), consumer_secret.clone());
+
+    println!("con_token created");
+
+    let mut config = String::new();
+    let user_id: u64;
+    let username: String;
+    let token: egg_mode::Token;
+
+    //look at all this unwrapping! who told you it was my birthday?
+    //if let Ok(mut f) = std::fs::File::open("twitter_settings") {
+    if true {
+      /*f.read_to_string(&mut config).unwrap();
+
+      let mut iter = config.split('\n');
+
+      username = iter.next().unwrap().to_string();
+      user_id = u64::from_str_radix(&iter.next().unwrap(), 10).unwrap();
+      let access_token = egg_mode::KeyPair::new(
+        iter.next().unwrap().to_string(),
+        iter.next().unwrap().to_string(),
+      );
+       */
+
+      let accessToken = &secrets.access_token;
+      let accessTokenSecret = &secrets.access_token_secret;
+
+      username = secrets.username.to_string();
+      user_id = secrets.user_id;
+
+      let access_token = egg_mode::KeyPair::new(
+        accessToken.to_string(),
+        accessTokenSecret.to_string(),
+      );
+
+
+      token = egg_mode::Token::Access {
+        consumer: con_token,
+        access: access_token,
+      };
+
+      if let Err(err) = egg_mode::auth::verify_tokens(&token).await {
+        println!("We've hit an error using your old tokens: {:?}", err);
+        println!("We'll have to reauthenticate before continuing.");
+        std::fs::remove_file("twitter_settings").unwrap();
+      } else {
+        println!("Welcome back, {}!\n", username);
+      }
+    } else {
+      let request_token = egg_mode::auth::request_token(&con_token, "oob").await.unwrap();
+
+      println!("Go to the following URL, sign in, and give me the PIN that comes back:");
+      println!("{}", egg_mode::auth::authorize_url(&request_token));
+
+      let mut pin = String::new();
+      std::io::stdin().read_line(&mut pin).unwrap();
+      println!("");
+
+      let tok_result = egg_mode::auth::access_token(con_token, &request_token, pin)
+        .await
+        .unwrap();
+
+      token = tok_result.0;
+      user_id = tok_result.1;
+      username = tok_result.2;
+
+      match token {
+        egg_mode::Token::Access {
+          access: ref access_token,
+          ..
+        } => {
+          config.push_str(&username);
+          config.push('\n');
+          config.push_str(&format!("{}", user_id));
+          config.push('\n');
+          config.push_str(&access_token.key);
+          config.push('\n');
+          config.push_str(&access_token.secret);
+        }
+        _ => unreachable!(),
+      }
+
+      let mut f = std::fs::File::create("twitter_settings").unwrap();
+      f.write_all(config.as_bytes()).unwrap();
+
+      println!("Welcome, {}, let's get this show on the road!", username);
+    }
+
+    return Some(Config {
+      token: token,
+      user_id: user_id,
+      screen_name: username,
+    });
+
+    /*//TODO: Is there a better way to query whether a file exists?
+    if std::fs::metadata("twitter_settings").is_ok() {
+      Some(Config {
+        token: token,
+        user_id: user_id,
+        screen_name: username,
+      })
+    } else {
+      None
+    }*/
+  }
+}
+
+pub fn print_tweet(tweet: &egg_mode::tweet::Tweet) {
+  if let Some(ref user) = tweet.user {
+    println!(
+      "{} (@{}) posted at {}",
+      Paint::blue(&user.name),
+      Paint::bold(Paint::blue(&user.screen_name)),
+      tweet.created_at.with_timezone(&chrono::Local)
+    );
+  }
+
+  if let Some(ref screen_name) = tweet.in_reply_to_screen_name {
+    println!("➜ in reply to @{}", Paint::blue(screen_name));
+  }
+
+  if let Some(ref status) = tweet.retweeted_status {
+    println!("{}", Paint::red("Retweet ➜"));
+    print_tweet(status);
+    return;
+  } else {
+    println!("{}", Paint::green(&tweet.text));
+  }
+
+  if let Some(source) = &tweet.source {
+    println!("➜ via {} ({})", source.name, source.url);
+  }
+
+  if let Some(ref place) = tweet.place {
+    println!("➜ from: {}", place.full_name);
+  }
+
+  if let Some(ref status) = tweet.quoted_status {
+    println!("{}", Paint::red("➜ Quoting the following status:"));
+    print_tweet(status);
+  }
+
+  if !tweet.entities.hashtags.is_empty() {
+    println!("➜ Hashtags contained in the tweet:");
+    for tag in &tweet.entities.hashtags {
+      println!("  {}", tag.text);
+    }
+  }
+
+  if !tweet.entities.symbols.is_empty() {
+    println!("➜ Symbols contained in the tweet:");
+    for tag in &tweet.entities.symbols {
+      println!("  {}", tag.text);
+    }
+  }
+
+  if !tweet.entities.urls.is_empty() {
+    println!("➜ URLs contained in the tweet:");
+    for url in &tweet.entities.urls {
+      if let Some(expanded_url) = &url.expanded_url {
+        println!("  {}", expanded_url);
+      }
+    }
+  }
+
+  if !tweet.entities.user_mentions.is_empty() {
+    println!("➜ Users mentioned in the tweet:");
+    for user in &tweet.entities.user_mentions {
+      println!("  {}", Paint::bold(Paint::blue(&user.screen_name)));
+    }
+  }
+
+  if let Some(ref media) = tweet.extended_entities {
+    println!("➜ Media attached to the tweet:");
+    for info in &media.media {
+      println!("  A {:?}", info.media_type);
+    }
+  }
+}

src/main.rs

@@ -0,0 +1,215 @@
+//! A Twitter bot for Vo.codes
+//! Copyright 2020 Brandon Thomas
+
+mod common;
+mod workers;
+
+use anyhow::bail;
+use crate::workers::{spawn_workers, Workload};
+use crossbeam::Receiver;
+use easyenv::{init_env_logger, get_env_num};
+use egg_mode::media::{media_types, upload_media, set_metadata, get_status, ProgressInfo, MediaHandle};
+use egg_mode::stream::StreamMessage;
+use egg_mode::tweet::DraftTweet;
+use futures::TryStreamExt;
+use log::{warn, info, error};
+use std::{env, thread};
+use std::io::{stdout, Write};
+use std::path::PathBuf;
+use std::time::Duration;
+use tokio::time::delay_for;
+use egg_mode::error::Error;
+use tokio::runtime::Builder;
+
+pub type AnyhowResult<T> = anyhow::Result<T>;
+
+#[derive(Clone)]
+pub struct Secrets {
+  pub api_key: String,
+  pub api_secret_key: String,
+  pub access_token: String,
+  pub access_token_secret: String,
+  pub username: String,
+  // Not sure where to get the user id, or what it's used for.
+  pub user_id: u64,
+}
+
+impl Secrets {
+  pub fn load_from_env() -> AnyhowResult<Self> {
+    Ok(Secrets {
+      api_key: env::var("API_KEY")?,
+      api_secret_key: env::var("API_SECRET_KEY")?,
+      access_token: env::var("ACCESS_TOKEN")?,
+      access_token_secret: env::var("ACCESS_TOKEN_SECRET")?,
+      username: env::var("USERNAME")?,
+      user_id: get_env_num::<u64>("USER_ID", 0)?,
+    })
+  }
+}
+
+fn main() -> AnyhowResult<()> {
+  /*let mut runtime = Builder::new()
+    .build()
+    .unwrap()
+    .block_on(async {
+      main_2().await
+    });*/
+
+  let mut runtime = tokio::runtime::Builder::new()
+    .threaded_scheduler()
+    .enable_all()
+    .on_thread_start(|| {
+      println!("thread started");
+    })
+    .build()
+    .unwrap();
+
+  runtime.block_on(main_2());
+
+
+  Ok(())
+}
+
+//#[tokio::main]
+async fn main_2() -> AnyhowResult<()> {
+
+  init_env_logger(None);
+
+  let secrets = Secrets::load_from_env()?;
+
+  info!("Loading configs and logging in...");
+
+  let config = common::Config::load(secrets.clone()).await;
+
+  info!("Configs loaded. Spawning workers...");
+
+  let (sender, receiver) = crossbeam::bounded(5);
+  spawn_workers(config.clone(), receiver, 5).await;
+
+  info!("Streaming tweets...");
+
+  thread::sleep(Duration::from_secs(10));
+
+  let stream = egg_mode::stream::filter()
+    .track(&["vocodesbot"])
+    .language(&["en"])
+    .start(&config.token)
+    .try_for_each(|m| {
+      if let StreamMessage::Tweet(tweet) = m {
+        common::print_tweet(&tweet);
+        println!("──────────────────────────────────────");
+
+        sender.send(Workload { info: tweet.text.clone() });
+
+        info!("Responding...");
+        let fut = async {
+          let path = PathBuf::from("/home/bt/dev/voice/ffmpeg-server/output/out.mp4");
+
+          match respond_with_media(&config, "Hello, this is a response.", path, "some media").await {
+            Err(e) => warn!("An error: {:?}", e),
+            Ok(e) => info!("Response posted."),
+          }
+        };
+
+      } else {
+        println!("Not a message: {:?}", m);
+      }
+
+      futures::future::ok(())
+    });
+
+
+  if let Err(e) = stream.await {
+    let e : egg_mode::error::Error = e;
+    error!("Stream error: {}", &e);
+    match e {
+      Error::BadUrl => {},
+      Error::InvalidResponse(_, _) => {},
+      Error::MissingValue(_) => {},
+      Error::FutureAlreadyCompleted => {},
+      Error::TwitterError(_, _) => {},
+      Error::RateLimit(rate_limit) => {
+        error!("Rate limited. Will reopen at unix timestamp: {}", rate_limit);
+      },
+      Error::MediaError(_) => {},
+      Error::BadStatus(_) => {},
+      Error::NetError(_) => {},
+      Error::TlsError(_) => {},
+      Error::IOError(_) => {},
+      Error::DeserializeError(_) => {},
+      Error::TimestampParseError(_) => {},
+      Error::TimerShutdownError(_) => {},
+      Error::HeaderParseError(_) => {},
+      Error::HeaderConvertError(_) => {},
+    }
+    error!("Disconnected")
+  }
+
+  Ok(())
+}
+
+async fn respond_with_media(
+  config: &common::Config,
+  tweet_text: &str,
+  media_path: PathBuf,
+  media_alt_text: &str) -> Result<(), Box<dyn std::error::Error>>
+{
+  let mut tweet = DraftTweet::new(tweet_text.to_string());
+
+  let handle = upload_media_to_twitter(config, media_path, media_alt_text).await?;
+  tweet.add_media(handle.id.clone());
+
+  tweet.send(&config.token).await?;
+
+  Ok(())
+}
+
+async fn upload_media_to_twitter(
+  config: &common::Config,
+  media_path: PathBuf,
+  media_alt_text: &str) -> AnyhowResult<MediaHandle>
+{
+  println!("Uploading media from '{}'", media_path.display());
+
+  let typ = match media_path.extension().and_then(|os| os.to_str()).unwrap_or("") {
+    "jpg" | "jpeg" => media_types::image_jpg(),
+    "gif" => media_types::image_gif(),
+    "png" => media_types::image_png(),
+    "webp" => media_types::image_webp(),
+    "mp4" => media_types::video_mp4(),
+    _ => {
+      eprintln!("Format not recognized, must be one of [jpg, jpeg, gif, png, webp, mp4]");
+      std::process::exit(1);
+    }
+  };
+
+  let bytes = std::fs::read(media_path)?;
+  let handle = upload_media(&bytes, &typ, &config.token).await?;
+
+  set_metadata(&handle.id, media_alt_text, &config.token).await?;
+
+  println!("Media uploaded");
+
+  // Wait 60 seconds for processing
+  println!("Waiting for media to finish processing...");
+
+  for ct in 0..=60u32 {
+    match get_status(handle.id.clone(), &config.token).await?.progress {
+      None | Some(ProgressInfo::Success) => {
+        println!("\nMedia sucessfully processed");
+        break;
+      }
+      Some(ProgressInfo::Pending(_)) | Some(ProgressInfo::InProgress(_)) => {
+        print!(".");
+        stdout().flush()?;
+        delay_for(Duration::from_secs(1)).await;
+      }
+      Some(ProgressInfo::Failed(err)) => Err(err)?,
+    }
+    if ct == 60 {
+      bail!("Error: timeout");
+    }
+  }
+
+  Ok(handle)
+}

src/workers.rs

@@ -0,0 +1,144 @@
+use crate::{common, AnyhowResult};
+use crossbeam::Receiver;
+use log::{info, error, warn};
+use std::thread;
+use tokio::time::Duration;
+use std::path::PathBuf;
+use egg_mode::media::{MediaHandle, media_types, upload_media, set_metadata, get_status, ProgressInfo};
+use egg_mode::tweet::DraftTweet;
+use anyhow::Error;
+use tokio::runtime::Handle;
+
+pub struct Workload {
+  pub info: String,
+}
+
+
+pub async fn spawn_workers(config: common::Config, receiver: Receiver<Workload>, num_workers: usize) {
+  //thread::sleep(Duration::from_secs(5));
+
+  for i in 0 .. num_workers {
+    let conf = config.clone();
+    let recv = receiver.clone();
+
+    info!("Spawning worker {}", i);
+
+    let handle = Handle::current();
+
+    //tokio::task::spawn(async move {});
+    handle.spawn(async move {
+      info!("a");
+      start_worker(conf, recv, i).await;
+      info!("b");
+    });
+
+
+    //thread::spawn(move || start_worker(conf, recv, i));
+  }
+
+  //joinables.iter().map(|j| j.join());
+}
+
+async fn start_worker(config: common::Config, receiver: Receiver<Workload>, worker_id: usize) {
+  loop {
+    info!("Worker thread loop - listening: {}", worker_id);
+
+    let workload = match receiver.recv() {
+      Err(e) => {
+        warn!("Error receiving from queue: {}", e);
+        thread::sleep(Duration::from_millis(500));
+        continue;
+      },
+      Ok(workload) => workload,
+    };
+
+    info!("Worker {} got tweet: {}", worker_id, &workload.info);
+
+    let path = PathBuf::from("/home/bt/dev/voice/ffmpeg-server/output/out.mp4");
+
+    let r = respond_with_media(&config, "generic response", path, "media");
+
+    match r {
+      Ok(_) => info!("Responded to tweet successfully"),
+      Err(e) => error!("Error responding to tweet: {}", e),
+    }
+  }
+}
+
+fn respond_with_media(
+  config: &common::Config,
+  tweet_text: &str,
+  media_path: PathBuf,
+  media_alt_text: &str) -> AnyhowResult<()>
+{
+
+  // let rt = tokio::runtime::Runtime::new().unwrap();
+  // let mut executor = rt.executor();
+
+  // executor.spawn(async {
+  //   let mut tweet = DraftTweet::new(tweet_text.to_string());
+
+  //   //let handle = upload_media_to_twitter(config, media_path, media_alt_text).await?;
+  //   //tweet.add_media(handle.id.clone());
+
+  //   tweet.send(&config.token).await
+  // });
+
+  // tokio::task::spawn_blocking({
+  //   fut
+  // })?;
+  //futures::executor::block_on(fut)?;
+
+  Ok(())
+}
+
+/*fn upload_media_to_twitter(
+  config: &common::Config,
+  media_path: PathBuf,
+  media_alt_text: &str) -> AnyhowResult<MediaHandle>
+{
+  println!("Uploading media from '{}'", media_path.display());
+
+  let typ = match media_path.extension().and_then(|os| os.to_str()).unwrap_or("") {
+    "jpg" | "jpeg" => media_types::image_jpg(),
+    "gif" => media_types::image_gif(),
+    "png" => media_types::image_png(),
+    "webp" => media_types::image_webp(),
+    "mp4" => media_types::video_mp4(),
+    _ => {
+      eprintln!("Format not recognized, must be one of [jpg, jpeg, gif, png, webp, mp4]");
+      std::process::exit(1);
+    }
+  };
+
+  let bytes = std::fs::read(media_path)?;
+  let handle = upload_media(&bytes, &typ, &config.token).await?;
+
+  set_metadata(&handle.id, media_alt_text, &config.token).await?;
+
+  println!("Media uploaded");
+
+  // Wait 60 seconds for processing
+  println!("Waiting for media to finish processing...");
+
+  for ct in 0..=60u32 {
+    match get_status(handle.id.clone(), &config.token).await?.progress {
+      None | Some(ProgressInfo::Success) => {
+        println!("\nMedia sucessfully processed");
+        break;
+      }
+      Some(ProgressInfo::Pending(_)) | Some(ProgressInfo::InProgress(_)) => {
+        print!(".");
+        stdout().flush()?;
+        delay_for(Duration::from_secs(1)).await;
+      }
+      Some(ProgressInfo::Failed(err)) => Err(err)?,
+    }
+    if ct == 60 {
+      bail!("Error: timeout");
+    }
+  }
+
+  Ok(handle)
+}
+*/