8 changed files with 380 additions and 371 deletions
@ -0,0 +1,139 @@
@@ -0,0 +1,139 @@
|
||||
//! Tracks WhatsApp message acknowledgements, and alerts if we don't get any.
|
||||
|
||||
use tokio_timer::Interval; |
||||
use whatsappweb::Jid; |
||||
use whatsappweb::message::{ChatMessageContent, MessageAckLevel, MessageAck}; |
||||
use chrono::prelude::*; |
||||
use std::collections::HashMap; |
||||
use futures::sync::mpsc::UnboundedSender; |
||||
use unicode_segmentation::UnicodeSegmentation; |
||||
use std::time::{Instant, Duration}; |
||||
use futures::{Future, Async, Poll, Stream}; |
||||
use failure::Error; |
||||
|
||||
use crate::comm::{ControlBotCommand, InitParameters}; |
||||
|
||||
struct MessageSendStatus { |
||||
ack_level: Option<MessageAckLevel>, |
||||
sent_ts: DateTime<Utc>, |
||||
content: ChatMessageContent, |
||||
destination: Jid, |
||||
alerted: bool, |
||||
alerted_pending: bool, |
||||
} |
||||
pub struct WaAckTracker { |
||||
cb_tx: UnboundedSender<ControlBotCommand>, |
||||
outgoing_messages: HashMap<String, MessageSendStatus>, |
||||
ack_warn: u64, |
||||
ack_warn_pending: u64, |
||||
ack_expiry: u64, |
||||
timer: Interval, |
||||
} |
||||
impl Future for WaAckTracker { |
||||
type Item = ();
|
||||
type Error = Error; |
||||
|
||||
fn poll(&mut self) -> Poll<(), Error> { |
||||
while let Async::Ready(_) = self.timer.poll()? { |
||||
self.check_acks(); |
||||
} |
||||
Ok(Async::NotReady) |
||||
} |
||||
} |
||||
impl WaAckTracker { |
||||
pub fn new<T>(p: &InitParameters<T>) -> Self { |
||||
let cb_tx = p.cm.cb_tx.clone(); |
||||
let ack_ivl = p.cfg.whatsapp.ack_check_interval.unwrap_or(3); |
||||
let ack_warn_ms = p.cfg.whatsapp.ack_warn_ms.unwrap_or(5000); |
||||
let ack_warn_pending_ms = p.cfg.whatsapp.ack_warn_pending_ms.unwrap_or(ack_warn_ms * 2); |
||||
let ack_expiry_ms = p.cfg.whatsapp.ack_expiry_ms.unwrap_or(60000); |
||||
let timer = Interval::new(Instant::now(), Duration::new(ack_ivl, 0)); |
||||
Self { |
||||
ack_warn: ack_warn_ms, |
||||
ack_warn_pending: ack_warn_pending_ms, |
||||
ack_expiry: ack_expiry_ms, |
||||
outgoing_messages: HashMap::new(), |
||||
cb_tx, timer |
||||
} |
||||
} |
||||
pub fn register_send(&mut self, to: Jid, content: ChatMessageContent, mid: String) { |
||||
let mss = MessageSendStatus { |
||||
ack_level: None, |
||||
sent_ts: Utc::now(), |
||||
content, |
||||
destination: to, |
||||
alerted: false, |
||||
alerted_pending: false |
||||
}; |
||||
self.outgoing_messages.insert(mid, mss); |
||||
} |
||||
pub fn print_acks(&mut self) -> Vec<String> { |
||||
let now = Utc::now(); |
||||
let mut lines = vec![]; |
||||
for (mid, mss) in self.outgoing_messages.iter_mut() { |
||||
let delta = now - mss.sent_ts; |
||||
let mut summary = mss.content.quoted_description(); |
||||
if summary.len() > 15 { |
||||
summary = summary.graphemes(true) |
||||
.take(10) |
||||
.chain(std::iter::once("…")) |
||||
.collect(); |
||||
} |
||||
let al: std::borrow::Cow<str> = match mss.ack_level { |
||||
Some(al) => format!("{:?}", al).into(), |
||||
None => "undelivered".into() |
||||
}; |
||||
lines.push(format!("- \"\x1d{}\x1d\" to \x02{}\x02 ({}s ago) is \x02{}\x02",
|
||||
summary, mss.destination.to_string(), delta.num_seconds(), al)); |
||||
lines.push(format!(" (message ID \x11{}\x0f)", mid)); |
||||
} |
||||
if lines.len() == 0 { |
||||
lines.push("No outgoing messages".into()); |
||||
} |
||||
lines |
||||
} |
||||
pub fn on_message_ack(&mut self, ack: MessageAck) { |
||||
if let Some(mss) = self.outgoing_messages.get_mut(&ack.id.0) { |
||||
debug!("Ack known message {} at level: {:?}", ack.id.0, ack.level); |
||||
mss.ack_level = Some(ack.level); |
||||
} |
||||
else { |
||||
debug!("Ack unknown message {} at level: {:?}", ack.id.0, ack.level); |
||||
} |
||||
}
|
||||
fn send_fail<T: Into<String>>(cb_tx: &mut UnboundedSender<ControlBotCommand>, msg: T) { |
||||
cb_tx.unbounded_send(ControlBotCommand::ReportFailure(msg.into())) |
||||
.unwrap(); |
||||
} |
||||
fn check_acks(&mut self) { |
||||
trace!("Checking acks"); |
||||
let now = Utc::now(); |
||||
for (mid, mss) in self.outgoing_messages.iter_mut() { |
||||
let delta = now - mss.sent_ts; |
||||
let delta_ms = delta.num_milliseconds() as u64; |
||||
if mss.ack_level.is_none() { |
||||
if delta_ms >= self.ack_warn && !mss.alerted { |
||||
warn!("Message {} has been un-acked for {} seconds!", mid, delta.num_seconds()); |
||||
Self::send_fail(&mut self.cb_tx, format!("Warning: Sending message ID {} has failed, or is taking longer than usual!", mid)); |
||||
mss.alerted = true; |
||||
} |
||||
} |
||||
if let Some(MessageAckLevel::PendingSend) = mss.ack_level { |
||||
if delta_ms >= self.ack_warn_pending && !mss.alerted_pending { |
||||
warn!("Message {} has been pending for {} seconds!", mid, delta.num_seconds()); |
||||
Self::send_fail(&mut self.cb_tx, format!("Warning: Sending message ID {} is still pending. Is WhatsApp running and connected?", mid)); |
||||
mss.alerted_pending = true; |
||||
} |
||||
} |
||||
} |
||||
let ack_expiry = self.ack_expiry; |
||||
self.outgoing_messages.retain(|_, m| { |
||||
if let Some(MessageAckLevel::PendingSend) | None = m.ack_level { |
||||
// Always retain the failures, so the user knows what happened with them (!)
|
||||
return true; |
||||
} |
||||
let diff_ms = (now - m.sent_ts).num_milliseconds() as u64; |
||||
diff_ms < ack_expiry |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,191 @@
@@ -0,0 +1,191 @@
|
||||
//! WhatsApp message processing tools.
|
||||
|
||||
use whatsappweb::{Jid, MediaType}; |
||||
use chrono::prelude::*; |
||||
use futures::sync::mpsc::UnboundedSender; |
||||
use whatsappweb::message::{ChatMessageContent, QuotedChatMessage, MessageId, Peer}; |
||||
use regex::{Regex, Captures}; |
||||
use huawei_modem::pdu::PduAddress; |
||||
use std::sync::Arc; |
||||
use unicode_segmentation::UnicodeSegmentation; |
||||
|
||||
use crate::comm::WhatsappCommand; |
||||
use crate::store::Store; |
||||
use crate::whatsapp_media::{MediaInfo, self}; |
||||
use crate::util::Result; |
||||
|
||||
pub struct IncomingMessage { |
||||
pub id: MessageId, |
||||
pub peer: Option<Peer>,
|
||||
pub from: Jid, |
||||
pub group: Option<i32>, |
||||
pub content: ChatMessageContent, |
||||
pub quoted: Option<QuotedChatMessage>, |
||||
pub ts: NaiveDateTime |
||||
} |
||||
pub struct ProcessedIncomingMessage { |
||||
pub from: Jid, |
||||
pub text: String, |
||||
pub group: Option<i32>, |
||||
pub ts: NaiveDateTime |
||||
} |
||||
pub struct WaMessageProcessor { |
||||
pub(crate) store: Store, |
||||
pub(crate) media_path: String, |
||||
pub(crate) dl_path: String, |
||||
pub(crate) wa_tx: Arc<UnboundedSender<WhatsappCommand>> |
||||
} |
||||
|
||||
impl WaMessageProcessor { |
||||
fn process_incoming_media(&mut self, id: MessageId, peer: Option<Peer>, from: Jid, group: Option<i32>, ct: ChatMessageContent, ts: NaiveDateTime) -> Result<()> { |
||||
|
||||
let (ty, fi, name) = match ct { |
||||
ChatMessageContent::Image { info, .. } => (MediaType::Image, info, None), |
||||
ChatMessageContent::Video { info, .. } => (MediaType::Video, info, None), |
||||
ChatMessageContent::Audio { info, .. } => (MediaType::Audio, info, None), |
||||
ChatMessageContent::Document { info, filename } => (MediaType::Document, info, Some(filename)), |
||||
_ => unreachable!() |
||||
}; |
||||
let mi = MediaInfo { |
||||
ty, fi, name, peer, ts, |
||||
mi: id, |
||||
from, group, |
||||
path: self.media_path.clone(), |
||||
dl_path: self.dl_path.clone(), |
||||
tx: self.wa_tx.clone() |
||||
}; |
||||
mi.start(); |
||||
Ok(()) |
||||
} |
||||
fn process_wa_text_message<'a>(&mut self, msg: &'a str) -> String { |
||||
lazy_static! { |
||||
static ref BOLD_RE: Regex = Regex::new(r#"\*([^\*]+)\*"#).unwrap(); |
||||
static ref ITALICS_RE: Regex = Regex::new(r#"_([^_]+)_"#).unwrap(); |
||||
static ref MENTIONS_RE: Regex = Regex::new(r#"@(\d+)"#).unwrap(); |
||||
} |
||||
let emboldened = BOLD_RE.replace_all(msg, "\x02$1\x02"); |
||||
let italicised = ITALICS_RE.replace_all(&emboldened, "\x1D$1\x1D"); |
||||
let store = &mut self.store; |
||||
let ret = MENTIONS_RE.replace_all(&italicised, |caps: &Captures| { |
||||
let pdua: PduAddress = caps[0].replace("@", "+").parse().unwrap(); |
||||
match store.get_recipient_by_addr_opt(&pdua) { |
||||
Ok(Some(recip)) => recip.nick, |
||||
Ok(None) => format!("<+{}>", &caps[1]), |
||||
Err(e) => { |
||||
warn!("Error searching for mention recipient: {}", e); |
||||
format!("@{}", &caps[1]) |
||||
} |
||||
} |
||||
}); |
||||
ret.to_string() |
||||
} |
||||
fn jid_to_nick(&mut self, jid: &Jid) -> Result<Option<String>> { |
||||
if let Some(num) = jid.phonenumber() { |
||||
if let Ok(pdua) = num.parse() { |
||||
return Ok(self.store.get_recipient_by_addr_opt(&pdua)? |
||||
.map(|x| x.nick)); |
||||
} |
||||
} |
||||
Ok(None) |
||||
} |
||||
pub fn process_wa_incoming(&mut self, inc: IncomingMessage) -> Result<(Vec<ProcessedIncomingMessage>, bool)> { |
||||
let IncomingMessage { id, peer, from, group, content, ts, quoted } = inc; |
||||
let mut ret = Vec::with_capacity(2); |
||||
let mut is_media = false; |
||||
let text = match content { |
||||
ChatMessageContent::Text(s) => self.process_wa_text_message(&s), |
||||
ChatMessageContent::Unimplemented(mut det) => { |
||||
if det.trim() == "" { |
||||
debug!("Discarding empty unimplemented message."); |
||||
return Ok((ret, is_media)); |
||||
} |
||||
if det.len() > 128 { |
||||
det = det.graphemes(true) |
||||
.take(128) |
||||
.chain(std::iter::once("…")) |
||||
.collect(); |
||||
} |
||||
format!("[\x02\x0304unimplemented\x0f] {}", det) |
||||
}, |
||||
ChatMessageContent::LiveLocation { lat, long, speed, .. } => { |
||||
// FIXME: use write!() maybe
|
||||
let spd = if let Some(s) = speed { |
||||
format!("travelling at {:.02} m/s - https://google.com/maps?q={},{}", s, lat, long) |
||||
} |
||||
else { |
||||
format!("broadcasting live location - https://google.com/maps?q={},{}", lat, long) |
||||
}; |
||||
format!("\x01ACTION is {}\x01", spd) |
||||
}, |
||||
ChatMessageContent::Location { lat, long, name, .. } => { |
||||
let place = if let Some(n) = name { |
||||
format!("at '{}'", n) |
||||
} |
||||
else { |
||||
"somewhere".into() |
||||
}; |
||||
format!("\x01ACTION is {} - https://google.com/maps?q={},{}\x01", place, lat, long) |
||||
}, |
||||
ChatMessageContent::Redaction { mid } => { |
||||
// TODO: make this more useful
|
||||
format!("\x01ACTION redacted message ID \x11{}\x11\x01", mid.0) |
||||
}, |
||||
ChatMessageContent::Contact { display_name, vcard } => { |
||||
match whatsapp_media::store_contact(&self.media_path, &self.dl_path, vcard) { |
||||
Ok(link) => { |
||||
format!("\x01ACTION uploaded a contact for '{}' - {}\x01", display_name, link) |
||||
}, |
||||
Err(e) => { |
||||
warn!("Failed to save contact card: {}", e); |
||||
format!("\x01ACTION uploaded a contact for '{}' (couldn't download)\x01", display_name) |
||||
} |
||||
} |
||||
}, |
||||
mut x @ ChatMessageContent::Image { .. } | |
||||
mut x @ ChatMessageContent::Video { .. } | |
||||
mut x @ ChatMessageContent::Audio { .. } | |
||||
mut x @ ChatMessageContent::Document { .. } => { |
||||
let capt = x.take_caption(); |
||||
self.process_incoming_media(id.clone(), peer.clone(), from.clone(), group, x, ts)?; |
||||
is_media = true; |
||||
if let Some(c) = capt { |
||||
c |
||||
} |
||||
else { |
||||
return Ok((ret, is_media)); |
||||
} |
||||
} |
||||
}; |
||||
if let Some(qm) = quoted { |
||||
let nick = if group.is_some() { |
||||
let nick = self.jid_to_nick(&qm.participant)? |
||||
.unwrap_or(qm.participant.to_string()); |
||||
format!("<{}> ", nick) |
||||
} |
||||
else { |
||||
String::new() |
||||
}; |
||||
let mut message = qm.content.quoted_description(); |
||||
if message.len() > 128 { |
||||
message = message.graphemes(true) |
||||
.take(128) |
||||
.chain(std::iter::once("…")) |
||||
.collect(); |
||||
} |
||||
let quote = format!("\x0315> \x1d{}{}", nick, message); |
||||
ret.push(ProcessedIncomingMessage { |
||||
from: from.clone(), |
||||
text: quote, |
||||
group, |
||||
ts |
||||
}); |
||||
} |
||||
ret.push(ProcessedIncomingMessage { |
||||
from, |
||||
text, |
||||
group, |
||||
ts |
||||
}); |
||||
Ok((ret, is_media)) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue