825 lines
39 KiB
Rust
825 lines
39 KiB
Rust
// 訊息管道: a bot answers people in Slack, Discord or Telegram (the original: Slack / Microsoft
|
||
// Teams / Discord apps, plus Telegram-style messaging channels). The original's servers receive
|
||
// those messages; a desktop app has no public address, so here each channel is polled with the
|
||
// person's own bot token, and the bot's final answer of that turn is posted back (in a thread /
|
||
// as a reply to the message).
|
||
|
||
use serde::{Deserialize, Serialize};
|
||
use serde_json::{json, Value};
|
||
use std::collections::HashMap;
|
||
use std::sync::atomic::{AtomicU64, Ordering};
|
||
use std::sync::mpsc::Sender;
|
||
use std::sync::{Arc, Mutex, OnceLock};
|
||
use std::time::Duration;
|
||
|
||
/// How often Slack and Discord are checked (estimated; within both APIs' rate limits).
|
||
pub const POLL_SECS: u64 = 3;
|
||
/// Telegram long-poll wait.
|
||
pub const TELEGRAM_WAIT_SECS: u64 = 25;
|
||
/// Longest single message each service takes.
|
||
pub const DISCORD_MAX: usize = 2000;
|
||
pub const TELEGRAM_MAX: usize = 4096;
|
||
pub const SLACK_MAX: usize = 39_000;
|
||
|
||
#[derive(Clone, Copy, Serialize, Deserialize, Debug, PartialEq, Eq, Hash)]
|
||
pub enum Kind {
|
||
Slack,
|
||
Discord,
|
||
Telegram,
|
||
Teams,
|
||
/// Mac「訊息」(iMessage / SMS).
|
||
IMessage,
|
||
/// Mail on this Mac.
|
||
Email,
|
||
/// 專屬號碼: texts to the person's Twilio number.
|
||
Sms,
|
||
/// X: @mentions, answered as replies.
|
||
X,
|
||
}
|
||
|
||
impl Kind {
|
||
pub fn label(self) -> &'static str {
|
||
match self {
|
||
Kind::Slack => "Slack",
|
||
Kind::Discord => "Discord",
|
||
Kind::Telegram => "Telegram",
|
||
Kind::Teams => "Teams",
|
||
Kind::IMessage => "訊息",
|
||
Kind::Email => "郵件",
|
||
Kind::Sms => "簡訊（Twilio）",
|
||
Kind::X => "X",
|
||
}
|
||
}
|
||
pub fn token_hint(self) -> &'static str {
|
||
match self {
|
||
Kind::Slack => "Bot User OAuth Token（xoxb-…）。在 api.slack.com/apps 建立 App，加上 channels:history、groups:history、im:history、chat:write 權限並安裝到工作區，再把 App 邀請進頻道。",
|
||
Kind::Discord => "Bot Token。在 discord.com/developers 建立 Application → Bot，打開 Message Content Intent，再用 OAuth2 把 Bot 加進你的伺服器。",
|
||
Kind::Telegram => "跟 @BotFather 說 /newbot 拿到的 Token（123456:ABC…）。",
|
||
Kind::IMessage => "不用 Token：用這台 Mac 的「訊息」。讀新訊息要給 AnyBot「完全取用磁碟」權限（系統設定 → 隱私權與安全性），傳訊息時 macOS 會問一次能否控制「訊息」。",
|
||
Kind::Email => "不用 Token：用這台 Mac 的「郵件」收信和回信。macOS 會問一次能否控制「郵件」。",
|
||
Kind::Sms => "「Account SID:Auth Token」（在 console.twilio.com 首頁）。先在 Twilio 買一個號碼；Bot 會回覆傳到這個號碼的簡訊，也能用它打電話和傳簡訊（每次都先問你）。",
|
||
Kind::X => "X 的使用者存取權杖（OAuth 2.0，需要 tweet.read、tweet.write、users.read 權限；在 developer.x.com 建立 App 取得）。Bot 會回覆提到你的貼文，也能用你的帳號發文（每次都先問你）。讀取提及需要 X API 付費方案。",
|
||
Kind::Teams => "不用 Token：填你的 Microsoft Entra 應用程式（用戶端）ID，按「登入 Microsoft」用裝置碼登入。應用程式要在「驗證」打開「允許公用用戶端流程」，並加上 Chat.ReadWrite、ChannelMessage.Read.All、ChannelMessage.Send 權限（頻道訊息可能要管理員同意）。",
|
||
}
|
||
}
|
||
pub fn target_hint(self) -> &'static str {
|
||
match self {
|
||
Kind::Slack => "頻道 ID（例如 C0123ABCD，在頻道資訊最下面）",
|
||
Kind::Discord => "頻道 ID（開發者模式下對頻道按右鍵 → 複製頻道 ID）",
|
||
Kind::Telegram => "只回這個聊天 ID（留空 = 任何人私訊 Bot 都回）",
|
||
Kind::IMessage => "只回這個電話或 Apple ID（例如 +886912345678；留空 = 任何人）",
|
||
Kind::Email => "只看這個信箱名稱（留空 = 收件匣）",
|
||
Kind::Sms => "你的 Twilio 號碼（例如 +15551234567）",
|
||
Kind::X => "（留空）",
|
||
Kind::Teams => "chat:<聊天 ID>，或 <團隊 ID>/<頻道 ID>（在 Teams 對聊天或頻道「取得連結」可以看到）",
|
||
}
|
||
}
|
||
}
|
||
|
||
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
|
||
pub struct Channel {
|
||
pub id: u64,
|
||
pub kind: Kind,
|
||
pub name: String,
|
||
/// The AnyBot bot that answers.
|
||
pub bot_id: u64,
|
||
pub token: String,
|
||
/// Slack / Discord channel id; Telegram chat id (empty = any).
|
||
#[serde(default)]
|
||
pub target: String,
|
||
#[serde(default = "yes")]
|
||
pub enabled: bool,
|
||
/// Teams: the Entra app (client) id and tenant (empty = organizations); `token` is the refresh token.
|
||
#[serde(default)]
|
||
pub client_id: String,
|
||
#[serde(default)]
|
||
pub tenant: String,
|
||
}
|
||
|
||
fn yes() -> bool {
|
||
true
|
||
}
|
||
|
||
/// Where an answer goes back.
|
||
#[derive(Clone, Debug, PartialEq)]
|
||
pub struct ReplyTo {
|
||
pub kind: Kind,
|
||
pub token: String,
|
||
pub target: String,
|
||
/// Slack thread ts / Discord message id / Telegram message id.
|
||
pub thread: String,
|
||
pub channel_name: String,
|
||
}
|
||
|
||
#[derive(Clone, Debug)]
|
||
pub struct Incoming {
|
||
pub bot_id: u64,
|
||
pub from: String,
|
||
pub text: String,
|
||
pub reply: ReplyTo,
|
||
}
|
||
|
||
#[derive(Clone, Default, Debug)]
|
||
pub struct Status {
|
||
pub ok: bool,
|
||
pub note: String,
|
||
pub received: u64,
|
||
}
|
||
|
||
pub fn statuses() -> &'static Mutex<HashMap<u64, Status>> {
|
||
static S: OnceLock<Mutex<HashMap<u64, Status>>> = OnceLock::new();
|
||
S.get_or_init(|| Mutex::new(HashMap::new()))
|
||
}
|
||
|
||
fn set_status(id: u64, ok: bool, note: &str, got: u64) {
|
||
let mut m = statuses().lock().unwrap();
|
||
let s = m.entry(id).or_default();
|
||
s.ok = ok;
|
||
s.note = note.to_string();
|
||
s.received += got;
|
||
}
|
||
|
||
fn agent(timeout: u64) -> ureq::Agent {
|
||
ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(timeout))).http_status_as_error(false).build().into()
|
||
}
|
||
|
||
/// Seconds to wait after a 429 ("rate:N" from get_json), bounded so a bogus value can't park
|
||
/// the channel forever.
|
||
fn rate_secs(e: &str) -> u64 {
|
||
e.split(':').nth(1).and_then(|s| s.parse().ok()).unwrap_or(10).clamp(1, 300)
|
||
}
|
||
|
||
fn get_json(url: &str, auth: Option<&str>, timeout: u64) -> Result<Value, String> {
|
||
let mut r = agent(timeout).get(url);
|
||
if let Some(a) = auth {
|
||
r = r.header("Authorization", a);
|
||
}
|
||
let mut resp = r.call().map_err(|e| e.to_string())?;
|
||
let status = resp.status().as_u16();
|
||
let text = resp.body_mut().read_to_string().unwrap_or_default();
|
||
if status == 429 {
|
||
// Carry the server's own "wait N seconds" (Telegram and Slack both send it); a fixed
|
||
// 5-second retry used to hammer the API and stay rate-limited.
|
||
let ra = serde_json::from_str::<Value>(&text).ok().and_then(|v| v["parameters"]["retry_after"].as_u64()).unwrap_or(10);
|
||
return Err(format!("rate:{ra}"));
|
||
}
|
||
if !(200..300).contains(&status) {
|
||
return Err(format!("HTTP {status}：{}", text.chars().take(160).collect::<String>()));
|
||
}
|
||
serde_json::from_str(&text).map_err(|e| e.to_string())
|
||
}
|
||
|
||
fn post_json(url: &str, auth: Option<&str>, body: &Value) -> Result<Value, String> {
|
||
let mut r = agent(30).post(url).header("Content-Type", "application/json; charset=utf-8");
|
||
if let Some(a) = auth {
|
||
r = r.header("Authorization", a);
|
||
}
|
||
let mut resp = r.send(body.to_string().as_bytes()).map_err(|e| e.to_string())?;
|
||
let status = resp.status().as_u16();
|
||
let text = resp.body_mut().read_to_string().unwrap_or_default();
|
||
if !(200..300).contains(&status) {
|
||
return Err(format!("HTTP {status}：{}", text.chars().take(160).collect::<String>()));
|
||
}
|
||
Ok(serde_json::from_str(&text).unwrap_or(Value::Null))
|
||
}
|
||
|
||
/// Split long answers on line breaks so each piece fits.
|
||
pub fn split(text: &str, max: usize) -> Vec<String> {
|
||
let mut out = vec![];
|
||
let mut cur = String::new();
|
||
for line in text.split_inclusive('\n') {
|
||
if cur.chars().count() + line.chars().count() > max && !cur.is_empty() {
|
||
out.push(std::mem::take(&mut cur));
|
||
}
|
||
if line.chars().count() > max {
|
||
let chars: Vec<char> = line.chars().collect();
|
||
for c in chars.chunks(max) {
|
||
out.push(c.iter().collect());
|
||
}
|
||
} else {
|
||
cur.push_str(line);
|
||
}
|
||
}
|
||
if !cur.trim().is_empty() {
|
||
out.push(cur);
|
||
}
|
||
out
|
||
}
|
||
|
||
/// Telegram's address (a test server can stand in for it).
|
||
fn tg() -> String {
|
||
std::env::var("ANYBOT_TELEGRAM_API").unwrap_or_else(|_| "https://api.telegram.org".into())
|
||
}
|
||
|
||
/// Check a token and say who it is.
|
||
pub fn test(kind: Kind, token: &str) -> Result<String, String> {
|
||
let t = token.trim();
|
||
match kind {
|
||
Kind::Slack => {
|
||
let v = get_json("https://slack.com/api/auth.test", Some(&format!("Bearer {t}")), 15)?;
|
||
if v["ok"].as_bool() == Some(true) {
|
||
Ok(format!("已連上 Slack：{}（{}）", v["user"].as_str().unwrap_or(""), v["team"].as_str().unwrap_or("")))
|
||
} else {
|
||
Err(format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")))
|
||
}
|
||
}
|
||
Kind::Discord => {
|
||
let v = get_json("https://discord.com/api/v10/users/@me", Some(&format!("Bot {t}")), 15)?;
|
||
Ok(format!("已連上 Discord：{}", v["username"].as_str().unwrap_or("")))
|
||
}
|
||
Kind::Teams => Err("Teams 請用「登入 Microsoft」".into()),
|
||
Kind::Sms => crate::phone::twilio_test(t),
|
||
Kind::X => crate::phone::x_me(t).map(|(_, u)| format!("已連上 X：@{u}")),
|
||
Kind::IMessage => imessage_rows(0).map(|_| "可以讀「訊息」了。".to_string()),
|
||
Kind::Email => {
|
||
let (o, ok) = crate::macapps::run("mac_mail_accounts", "{}");
|
||
if ok { Ok("可以用「郵件」了。".into()) } else { Err(o) }
|
||
}
|
||
Kind::Telegram => {
|
||
let v = get_json(&format!("{}/bot{t}/getMe", tg()), None, 15)?;
|
||
if v["ok"].as_bool() == Some(true) {
|
||
Ok(format!("已連上 Telegram：@{}", v["result"]["username"].as_str().unwrap_or("")))
|
||
} else {
|
||
Err(format!("Telegram 說：{}", v["description"].as_str().unwrap_or("不明錯誤")))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Post the bot's answer back where the message came from.
|
||
pub fn reply(to: &ReplyTo, text: &str) -> Result<(), String> {
|
||
let t = to.token.trim();
|
||
match to.kind {
|
||
// `thread` carries our own number (texts) or the post being answered (X).
|
||
Kind::Sms => crate::phone::send_sms(t, &to.thread, &to.target, text)?,
|
||
Kind::X => {
|
||
let mut reply_to = to.thread.clone();
|
||
for part in split(text, crate::phone::X_MAX - 20) {
|
||
let url = crate::phone::x_post(t, &part, &reply_to)?;
|
||
reply_to = url.rsplit('/').next().unwrap_or("").to_string();
|
||
}
|
||
}
|
||
Kind::IMessage => {
|
||
let (o, ok) = crate::macapps::run(crate::macapps::SEND_MESSAGE, &json!({"to": to.target, "text": text}).to_string());
|
||
if !ok {
|
||
return Err(o);
|
||
}
|
||
}
|
||
Kind::Email => {
|
||
// `thread` carries the subject.
|
||
let subject = if to.thread.to_lowercase().starts_with("re:") { to.thread.clone() } else { format!("Re: {}", to.thread) };
|
||
let (o, ok) = crate::macapps::run("mac_send_mail", &json!({"to": [to.target], "subject": subject, "body": text}).to_string());
|
||
if !ok {
|
||
return Err(o);
|
||
}
|
||
}
|
||
Kind::Teams => {
|
||
// `token` carries the access token for Teams replies.
|
||
for part in split(text, 25_000) {
|
||
crate::teams::send(t, &to.target, &to.thread, &part)?;
|
||
}
|
||
}
|
||
Kind::Slack => {
|
||
for part in split(text, SLACK_MAX) {
|
||
let v = post_json("https://slack.com/api/chat.postMessage", Some(&format!("Bearer {t}")), &json!({"channel": to.target, "text": part, "thread_ts": to.thread}))?;
|
||
if v["ok"].as_bool() != Some(true) {
|
||
return Err(format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")));
|
||
}
|
||
}
|
||
}
|
||
Kind::Discord => {
|
||
for (i, part) in split(text, DISCORD_MAX).into_iter().enumerate() {
|
||
let mut body = json!({"content": part, "allowed_mentions": {"parse": []}});
|
||
if i == 0 && !to.thread.is_empty() {
|
||
body["message_reference"] = json!({"message_id": to.thread, "fail_if_not_exists": false});
|
||
}
|
||
post_json(&format!("https://discord.com/api/v10/channels/{}/messages", to.target), Some(&format!("Bot {t}")), &body)?;
|
||
}
|
||
}
|
||
Kind::Telegram => {
|
||
for (i, part) in split(text, TELEGRAM_MAX).into_iter().enumerate() {
|
||
let mut body = json!({"chat_id": to.target, "text": part});
|
||
if i == 0 && !to.thread.is_empty() {
|
||
body["reply_to_message_id"] = json!(to.thread.parse::<i64>().unwrap_or(0));
|
||
}
|
||
let v = post_json(&format!("{}/bot{t}/sendMessage", tg()), None, &body)?;
|
||
if v["ok"].as_bool() == Some(false) {
|
||
return Err(format!("Telegram 說：{}", v["description"].as_str().unwrap_or("")));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
// ---------- reading new messages ----------
|
||
|
||
/// Slack: messages newer than `cursor` (a ts), oldest first, people only. Returns (new cursor, items).
|
||
pub fn slack_new(history: &Value, cursor: &str, own_user: &str) -> (String, Vec<(String, String, String)>) {
|
||
let mut items: Vec<(String, String, String)> = vec![];
|
||
let mut newest = cursor.to_string();
|
||
if let Some(msgs) = history["messages"].as_array() {
|
||
for m in msgs.iter().rev() {
|
||
let ts = m["ts"].as_str().unwrap_or("").to_string();
|
||
if ts.parse::<f64>().unwrap_or(0.0) <= cursor.parse::<f64>().unwrap_or(0.0) {
|
||
continue;
|
||
}
|
||
if ts.parse::<f64>().unwrap_or(0.0) > newest.parse::<f64>().unwrap_or(0.0) {
|
||
newest = ts.clone();
|
||
}
|
||
let is_bot = m.get("bot_id").is_some() || m["subtype"].is_string() || m["user"].as_str() == Some(own_user);
|
||
if is_bot {
|
||
continue;
|
||
}
|
||
let text = m["text"].as_str().unwrap_or("").replace(&format!("<@{own_user}>"), "").trim().to_string();
|
||
if !text.is_empty() {
|
||
items.push((m["user"].as_str().unwrap_or("someone").to_string(), text, m["thread_ts"].as_str().unwrap_or(&ts).to_string()));
|
||
}
|
||
}
|
||
}
|
||
(newest, items)
|
||
}
|
||
|
||
/// Discord: messages after `cursor` (a snowflake), oldest first, people only.
|
||
pub fn discord_new(list: &Value, cursor: &str) -> (String, Vec<(String, String, String)>) {
|
||
let mut items = vec![];
|
||
let mut newest = cursor.to_string();
|
||
let n = |s: &str| s.parse::<u128>().unwrap_or(0);
|
||
if let Some(msgs) = list.as_array() {
|
||
let mut msgs: Vec<&Value> = msgs.iter().collect();
|
||
msgs.sort_by_key(|m| n(m["id"].as_str().unwrap_or("0")));
|
||
for m in msgs {
|
||
let id = m["id"].as_str().unwrap_or("0").to_string();
|
||
if n(&id) <= n(cursor) {
|
||
continue;
|
||
}
|
||
if n(&id) > n(&newest) {
|
||
newest = id.clone();
|
||
}
|
||
if m["author"]["bot"].as_bool() == Some(true) {
|
||
continue;
|
||
}
|
||
let text = m["content"].as_str().unwrap_or("").trim().to_string();
|
||
if !text.is_empty() {
|
||
let who = m["author"]["global_name"].as_str().or(m["author"]["username"].as_str()).unwrap_or("someone").to_string();
|
||
items.push((who, text, id));
|
||
}
|
||
}
|
||
}
|
||
(newest, items)
|
||
}
|
||
|
||
/// Telegram: (next offset, [(from, text, chat id, message id)]).
|
||
pub fn telegram_new(updates: &Value, offset: i64, only_chat: &str) -> (i64, Vec<(String, String, String, String)>) {
|
||
let mut next = offset;
|
||
let mut items = vec![];
|
||
if let Some(us) = updates["result"].as_array() {
|
||
for u in us {
|
||
next = next.max(u["update_id"].as_i64().unwrap_or(0) + 1);
|
||
let m = &u["message"];
|
||
let chat = m["chat"]["id"].as_i64().map(|c| c.to_string()).unwrap_or_default();
|
||
if chat.is_empty() || (!only_chat.trim().is_empty() && only_chat.trim() != chat) {
|
||
continue;
|
||
}
|
||
if let Some(text) = m["text"].as_str() {
|
||
let who = m["from"]["first_name"].as_str().or(m["from"]["username"].as_str()).unwrap_or("someone").to_string();
|
||
items.push((who, text.to_string(), chat, m["message_id"].as_i64().unwrap_or(0).to_string()));
|
||
}
|
||
}
|
||
}
|
||
(next, items)
|
||
}
|
||
|
||
/// "王小明 <ming@example.com>" → "ming@example.com"
|
||
pub fn address_of(sender: &str) -> String {
|
||
match (sender.rfind('<'), sender.rfind('>')) {
|
||
(Some(a), Some(b)) if b > a => sender[a + 1..b].trim().to_string(),
|
||
_ => sender.trim().to_string(),
|
||
}
|
||
}
|
||
|
||
fn chat_db() -> String {
|
||
dirs::home_dir().unwrap_or_default().join("Library/Messages/chat.db").display().to_string()
|
||
}
|
||
|
||
/// Messages after `after` (ROWID): (id, text, handle, from me).
|
||
pub fn imessage_rows(after: i64) -> Result<Vec<(i64, String, String, bool)>, String> {
|
||
let q = format!("SELECT m.ROWID AS id, m.text AS text, h.id AS handle, m.is_from_me AS me FROM message m LEFT JOIN handle h ON m.handle_id = h.ROWID WHERE m.ROWID > {after} ORDER BY m.ROWID DESC LIMIT 50");
|
||
let out = std::process::Command::new("sqlite3").args(["-readonly", "-json", &chat_db(), &q]).output().map_err(|e| e.to_string())?;
|
||
if !out.status.success() {
|
||
let e = String::from_utf8_lossy(&out.stderr).to_string();
|
||
return Err(if e.contains("authorization denied") || e.contains("unable to open") {
|
||
"讀不到「訊息」：請給 AnyBot（或「終端機」）「完全取用磁碟」權限。".into()
|
||
} else {
|
||
e
|
||
});
|
||
}
|
||
let v: Value = serde_json::from_slice(&out.stdout).unwrap_or(json!([]));
|
||
let mut rows: Vec<(i64, String, String, bool)> = v
|
||
.as_array()
|
||
.into_iter()
|
||
.flatten()
|
||
.map(|r| (r["id"].as_i64().unwrap_or(0), r["text"].as_str().unwrap_or("").to_string(), r["handle"].as_str().unwrap_or("").to_string(), r["me"].as_i64() == Some(1)))
|
||
.collect();
|
||
rows.reverse();
|
||
Ok(rows)
|
||
}
|
||
|
||
// ---------- the pollers ----------
|
||
|
||
static GENERATION: AtomicU64 = AtomicU64::new(0);
|
||
|
||
/// (Re)start one poller per enabled channel; older pollers stop by themselves.
|
||
pub fn restart(channels: Vec<Channel>, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send + Sync>) {
|
||
let gen = GENERATION.fetch_add(1, Ordering::SeqCst) + 1;
|
||
// The bot's phone and X tools use the same accounts.
|
||
{
|
||
let mut a = crate::phone::accounts().lock().unwrap();
|
||
a.twilio = channels.iter().find(|c| c.enabled && c.kind == Kind::Sms && !c.token.trim().is_empty()).map(|c| (c.token.trim().to_string(), c.target.trim().to_string()));
|
||
a.x = channels.iter().find(|c| c.enabled && c.kind == Kind::X && !c.token.trim().is_empty()).map(|c| c.token.trim().to_string());
|
||
}
|
||
for c in channels.into_iter().filter(|c| c.enabled && !c.token.trim().is_empty()) {
|
||
let tx = tx.clone();
|
||
let repaint = repaint.clone();
|
||
std::thread::spawn(move || poll(c, gen, tx, repaint));
|
||
}
|
||
}
|
||
|
||
fn alive(gen: u64) -> bool {
|
||
GENERATION.load(Ordering::SeqCst) == gen
|
||
}
|
||
|
||
fn poll(c: Channel, gen: u64, tx: Sender<Incoming>, repaint: Arc<dyn Fn() + Send + Sync>) {
|
||
let token = c.token.trim().to_string();
|
||
let deliver = |from: String, text: String, target: String, thread: String| {
|
||
let _ = tx.send(Incoming { bot_id: c.bot_id, from, text, reply: ReplyTo { kind: c.kind, token: token.clone(), target, thread, channel_name: c.name.clone() } });
|
||
set_status(c.id, true, "收到新訊息", 1);
|
||
repaint();
|
||
};
|
||
match c.kind {
|
||
Kind::Slack => {
|
||
let auth = format!("Bearer {token}");
|
||
let own = get_json("https://slack.com/api/auth.test", Some(&auth), 15).ok().and_then(|v| v["user_id"].as_str().map(String::from)).unwrap_or_default();
|
||
let mut cursor = String::new();
|
||
while alive(gen) {
|
||
let url = format!("https://slack.com/api/conversations.history?channel={}&limit=20{}", c.target.trim(), if cursor.is_empty() { String::new() } else { format!("&oldest={cursor}") });
|
||
match get_json(&url, Some(&auth), 20) {
|
||
Ok(v) if v["ok"].as_bool() == Some(true) => {
|
||
if cursor.is_empty() {
|
||
// Start from now: older messages aren't answered. (An empty channel
|
||
// has no ts to stand on — "0" made the next round replay history.)
|
||
cursor = v["messages"].get(0).and_then(|m| m["ts"].as_str()).map(String::from).unwrap_or_else(|| chrono::Utc::now().timestamp().to_string());
|
||
set_status(c.id, true, "已連上，等待新訊息", 0);
|
||
} else {
|
||
let (next, items) = slack_new(&v, &cursor, &own);
|
||
cursor = next;
|
||
for (from, text, thread) in items {
|
||
deliver(from, text, c.target.trim().to_string(), thread);
|
||
}
|
||
}
|
||
}
|
||
Ok(v) => set_status(c.id, false, &format!("Slack 說：{}", v["error"].as_str().unwrap_or("不明錯誤")), 0),
|
||
Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))),
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
std::thread::sleep(Duration::from_secs(POLL_SECS));
|
||
}
|
||
}
|
||
Kind::Discord => {
|
||
let auth = format!("Bot {token}");
|
||
let mut cursor = String::new();
|
||
while alive(gen) {
|
||
let url = if cursor.is_empty() {
|
||
format!("https://discord.com/api/v10/channels/{}/messages?limit=1", c.target.trim())
|
||
} else {
|
||
format!("https://discord.com/api/v10/channels/{}/messages?after={cursor}&limit=50", c.target.trim())
|
||
};
|
||
match get_json(&url, Some(&auth), 20) {
|
||
Ok(v) => {
|
||
if cursor.is_empty() {
|
||
cursor = v[0]["id"].as_str().unwrap_or("0").to_string();
|
||
set_status(c.id, true, "已連上，等待新訊息", 0);
|
||
} else {
|
||
let (next, items) = discord_new(&v, &cursor);
|
||
cursor = next;
|
||
for (from, text, id) in items {
|
||
deliver(from, text, c.target.trim().to_string(), id);
|
||
}
|
||
}
|
||
}
|
||
Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))),
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
std::thread::sleep(Duration::from_secs(POLL_SECS));
|
||
}
|
||
}
|
||
Kind::Teams => {
|
||
let mut cursor = String::new();
|
||
let mut me = String::new();
|
||
while alive(gen) {
|
||
match crate::teams::access(c.id, &c.client_id, &c.tenant, &token) {
|
||
Ok(access) => {
|
||
if me.is_empty() {
|
||
me = crate::teams::graph_get(&access, "/me").ok().and_then(|v| v["id"].as_str().map(String::from)).unwrap_or_default();
|
||
}
|
||
match crate::teams::graph_get(&access, &crate::teams::messages_path(&c.target)) {
|
||
Ok(v) => {
|
||
if cursor.is_empty() {
|
||
cursor = crate::teams::new_messages(&v, "", "").0;
|
||
if cursor.is_empty() {
|
||
cursor = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%S").to_string();
|
||
}
|
||
set_status(c.id, true, "已連上，等待新訊息", 0);
|
||
} else {
|
||
let (next, items) = crate::teams::new_messages(&v, &cursor, &me);
|
||
cursor = next;
|
||
for (from, text, id) in items {
|
||
let _ = tx.send(Incoming { bot_id: c.bot_id, from, text, reply: ReplyTo { kind: Kind::Teams, token: access.clone(), target: c.target.trim().to_string(), thread: if c.target.starts_with("chat:") { String::new() } else { id }, channel_name: c.name.clone() } });
|
||
set_status(c.id, true, "收到新訊息", 1);
|
||
repaint();
|
||
}
|
||
}
|
||
}
|
||
Err(e) if e.starts_with("rate") => std::thread::sleep(Duration::from_secs(rate_secs(&e))),
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
}
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
std::thread::sleep(Duration::from_secs(POLL_SECS + 2));
|
||
}
|
||
}
|
||
Kind::IMessage => {
|
||
let mut cursor: Option<i64> = None;
|
||
while alive(gen) {
|
||
match imessage_rows(cursor.unwrap_or(0)) {
|
||
Ok(rows) => {
|
||
let newest = rows.iter().map(|r| r.0).max();
|
||
match cursor {
|
||
None => {
|
||
cursor = Some(newest.unwrap_or(0));
|
||
set_status(c.id, true, "已連上，等待新訊息", 0);
|
||
}
|
||
Some(_) => {
|
||
for (id, text, handle, me) in rows {
|
||
cursor = Some(id.max(cursor.unwrap_or(0)));
|
||
if me || text.trim().is_empty() || (!c.target.trim().is_empty() && handle != c.target.trim()) {
|
||
continue;
|
||
}
|
||
deliver(handle.clone(), text, handle, String::new());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
std::thread::sleep(Duration::from_secs(POLL_SECS));
|
||
}
|
||
}
|
||
Kind::Email => {
|
||
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
|
||
let mut first = true;
|
||
while alive(gen) {
|
||
let (o, ok) = crate::macapps::run("mac_read_mail", &json!({"mailbox": c.target.trim(), "unreadOnly": true, "days": 3, "limit": 20}).to_string());
|
||
if ok {
|
||
let list: Value = serde_json::from_str(o.trim_start_matches("郵件：")).unwrap_or(json!([]));
|
||
for m in list.as_array().into_iter().flatten() {
|
||
let id = m["id"].as_str().unwrap_or("").to_string();
|
||
if !seen.insert(id.clone()) || first {
|
||
continue;
|
||
}
|
||
// Read the whole message to answer it.
|
||
let (full, ok) = crate::macapps::run("mac_read_mail", &json!({"id": id}).to_string());
|
||
let body: Value = if ok { serde_json::from_str(&full).unwrap_or(Value::Null) } else { Value::Null };
|
||
let from = m["from"].as_str().unwrap_or("").to_string();
|
||
let subject = m["subject"].as_str().unwrap_or("").to_string();
|
||
let text = format!("主旨：{subject}\n{}", body["body"].as_str().unwrap_or(""));
|
||
deliver(from.clone(), text, address_of(&from), subject);
|
||
}
|
||
if first {
|
||
set_status(c.id, true, "已連上，等待新郵件", 0);
|
||
}
|
||
first = false;
|
||
} else {
|
||
set_status(c.id, false, &o, 0);
|
||
}
|
||
std::thread::sleep(Duration::from_secs(30));
|
||
}
|
||
}
|
||
Kind::Sms => {
|
||
let number = c.target.trim().to_string();
|
||
let mut seen = std::collections::HashSet::new();
|
||
let mut first = true;
|
||
while alive(gen) {
|
||
let path = format!("Messages.json?To={}&PageSize=20", number.replace('+', "%2B"));
|
||
match crate::phone::twilio_get(&token, &path) {
|
||
Ok(v) => {
|
||
let items = crate::phone::sms_new(&v, &mut seen);
|
||
if first {
|
||
// Older texts aren't answered.
|
||
first = false;
|
||
set_status(c.id, true, "已連上，等待簡訊", 0);
|
||
} else {
|
||
for (_, from, body) in items {
|
||
deliver(from.clone(), body, from, number.clone());
|
||
}
|
||
}
|
||
}
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
std::thread::sleep(Duration::from_secs(POLL_SECS));
|
||
}
|
||
}
|
||
Kind::X => {
|
||
let me = match crate::phone::x_me(&token) {
|
||
Ok(m) => m,
|
||
Err(e) => {
|
||
set_status(c.id, false, &e, 0);
|
||
return;
|
||
}
|
||
};
|
||
let mut since = String::new();
|
||
let mut first = true;
|
||
while alive(gen) {
|
||
match crate::phone::x_mentions(&token, &me.0, &since) {
|
||
Ok((newest, items)) => {
|
||
since = newest;
|
||
if first {
|
||
first = false;
|
||
set_status(c.id, true, &format!("已連上 @{}，等待提及", me.1), 0);
|
||
} else {
|
||
for (id, who, text) in items {
|
||
deliver(format!("@{who}"), text, who, id);
|
||
}
|
||
}
|
||
}
|
||
Err(e) => set_status(c.id, false, &e, 0),
|
||
}
|
||
// X's free and basic plans allow few reads: once a minute.
|
||
for _ in 0..60 {
|
||
if !alive(gen) {
|
||
break;
|
||
}
|
||
std::thread::sleep(Duration::from_secs(1));
|
||
}
|
||
}
|
||
}
|
||
Kind::Telegram => {
|
||
// Skip what's waiting from before: start after the latest update. If this acknowledge
|
||
// fails, offset 0 would replay the whole backlog on the first poll — retry it.
|
||
let mut offset = loop {
|
||
match get_json(&format!("{}/bot{token}/getUpdates?offset=-1", tg()), None, 15) {
|
||
Ok(v) => break v["result"].as_array().and_then(|a| a.last()).and_then(|u| u["update_id"].as_i64()).map(|u| u + 1).unwrap_or(1),
|
||
Err(e) => {
|
||
set_status(c.id, false, &format!("正在連上 Telegram：{e}"), 0);
|
||
if !alive(gen) {
|
||
return;
|
||
}
|
||
std::thread::sleep(Duration::from_secs(rate_secs(&e)));
|
||
}
|
||
}
|
||
};
|
||
set_status(c.id, true, "已連上，等待新訊息", 0);
|
||
while alive(gen) {
|
||
let url = format!("{}/bot{token}/getUpdates?offset={offset}&timeout={TELEGRAM_WAIT_SECS}", tg());
|
||
match get_json(&url, None, TELEGRAM_WAIT_SECS + 10) {
|
||
Ok(v) => {
|
||
let (next, items) = telegram_new(&v, offset, &c.target);
|
||
offset = next;
|
||
if !alive(gen) {
|
||
break;
|
||
}
|
||
for (from, text, chat, mid) in items {
|
||
deliver(from, text, chat, mid);
|
||
}
|
||
}
|
||
Err(e) => {
|
||
set_status(c.id, false, &e, 0);
|
||
// A 429 says how long to back off; hammering every 5 s keeps it limited.
|
||
std::thread::sleep(Duration::from_secs(if e.starts_with("rate") { rate_secs(&e) } else { 5 }));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn slack_skips_bots_and_old_messages() {
|
||
let v = json!({"ok": true, "messages": [
|
||
{"ts": "1700000003.000", "user": "UBOT", "text": "my own"},
|
||
{"ts": "1700000002.500", "bot_id": "B1", "text": "other bot"},
|
||
{"ts": "1700000002.000", "user": "U1", "text": "<@UBOT> 幫我查天氣", "thread_ts": "1700000001.000"},
|
||
{"ts": "1700000001.500", "user": "U2", "text": "嗨"},
|
||
{"ts": "1700000000.000", "user": "U3", "text": "old"}
|
||
]});
|
||
let (cur, items) = slack_new(&v, "1700000001.000", "UBOT");
|
||
assert_eq!(cur, "1700000003.000");
|
||
assert_eq!(items, vec![("U2".into(), "嗨".into(), "1700000001.500".into()), ("U1".into(), "幫我查天氣".into(), "1700000001.000".into())]);
|
||
}
|
||
|
||
#[test]
|
||
fn discord_orders_and_skips_bots() {
|
||
let v = json!([
|
||
{"id": "30", "content": "second", "author": {"username": "amy"}},
|
||
{"id": "25", "content": "bot", "author": {"username": "b", "bot": true}},
|
||
{"id": "20", "content": "first", "author": {"username": "bob", "global_name": "Bob"}}
|
||
]);
|
||
let (cur, items) = discord_new(&v, "10");
|
||
assert_eq!(cur, "30");
|
||
assert_eq!(items.iter().map(|i| i.1.as_str()).collect::<Vec<_>>(), vec!["first", "second"]);
|
||
assert_eq!(items[0].0, "Bob");
|
||
}
|
||
|
||
#[test]
|
||
fn telegram_filters_chat() {
|
||
let v = json!({"ok": true, "result": [
|
||
{"update_id": 7, "message": {"message_id": 1, "chat": {"id": 42}, "from": {"first_name": "小明"}, "text": "hi"}},
|
||
{"update_id": 8, "message": {"message_id": 2, "chat": {"id": 99}, "from": {"first_name": "x"}, "text": "no"}}
|
||
]});
|
||
let (next, items) = telegram_new(&v, 0, "42");
|
||
assert_eq!(next, 9);
|
||
assert_eq!(items, vec![("小明".into(), "hi".into(), "42".into(), "1".into())]);
|
||
assert_eq!(telegram_new(&v, 0, "").1.len(), 2);
|
||
}
|
||
|
||
/// A whole round trip against a stand-in Telegram server: a message arrives, the answer goes back.
|
||
#[test]
|
||
fn telegram_round_trip() {
|
||
use std::io::{Read, Write};
|
||
let server = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||
let port = server.local_addr().unwrap().port();
|
||
std::env::set_var("ANYBOT_TELEGRAM_API", format!("http://127.0.0.1:{port}"));
|
||
let sent = Arc::new(Mutex::new(Vec::<String>::new()));
|
||
let s2 = sent.clone();
|
||
std::thread::spawn(move || {
|
||
let mut served = 0;
|
||
for conn in server.incoming().flatten() {
|
||
let mut c = conn;
|
||
// Read the whole request: head, then as much body as Content-Length says.
|
||
c.set_read_timeout(Some(Duration::from_secs(3))).ok();
|
||
let mut got = vec![];
|
||
let mut buf = vec![0u8; 16384];
|
||
loop {
|
||
let n = c.read(&mut buf).unwrap_or(0);
|
||
got.extend_from_slice(&buf[..n]);
|
||
let text = String::from_utf8_lossy(&got).to_string();
|
||
if let Some(h) = text.find("\r\n\r\n") {
|
||
let len = text[..h].lines().find_map(|l| l.to_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))).unwrap_or(0);
|
||
if got.len() >= h + 4 + len {
|
||
break;
|
||
}
|
||
}
|
||
if n == 0 {
|
||
break;
|
||
}
|
||
}
|
||
let req = String::from_utf8_lossy(&got).to_string();
|
||
let body = if req.contains("offset=-1") {
|
||
r#"{"ok":true,"result":[{"update_id":100,"message":{"message_id":1,"chat":{"id":42},"text":"old"}}]}"#.to_string()
|
||
} else if req.contains("getUpdates") {
|
||
served += 1;
|
||
if served == 1 {
|
||
r#"{"ok":true,"result":[{"update_id":101,"message":{"message_id":5,"chat":{"id":42},"from":{"first_name":"小明"},"text":"明天天氣？"}}]}"#.to_string()
|
||
} else {
|
||
std::thread::sleep(Duration::from_millis(300));
|
||
r#"{"ok":true,"result":[]}"#.to_string()
|
||
}
|
||
} else {
|
||
s2.lock().unwrap().push(req.clone());
|
||
r#"{"ok":true,"result":{}}"#.to_string()
|
||
};
|
||
let _ = c.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes());
|
||
}
|
||
});
|
||
let (tx, rx) = std::sync::mpsc::channel();
|
||
let ch = Channel { id: 1, kind: Kind::Telegram, name: "測試".into(), bot_id: 7, token: "T0K".into(), target: String::new(), enabled: true, client_id: String::new(), tenant: String::new() };
|
||
restart(vec![ch], tx, Arc::new(|| {}));
|
||
let got = rx.recv_timeout(Duration::from_secs(10)).expect("message arrives");
|
||
GENERATION.fetch_add(1, Ordering::SeqCst);
|
||
assert_eq!((got.bot_id, got.from.as_str(), got.text.as_str(), got.reply.target.as_str(), got.reply.thread.as_str()), (7, "小明", "明天天氣？", "42", "5"));
|
||
reply(&got.reply, "明天晴天").unwrap();
|
||
let reqs = sent.lock().unwrap().clone();
|
||
assert!(reqs.iter().any(|r| r.contains("/botT0K/sendMessage") && r.contains("明天晴天") && r.contains("\"reply_to_message_id\":5")), "{reqs:?}");
|
||
std::env::remove_var("ANYBOT_TELEGRAM_API");
|
||
}
|
||
|
||
#[test]
|
||
fn email_addresses_are_found() {
|
||
assert_eq!(address_of("王小明 <ming@example.com>"), "ming@example.com");
|
||
assert_eq!(address_of("ming@example.com"), "ming@example.com");
|
||
}
|
||
|
||
#[test]
|
||
fn long_answers_are_split() {
|
||
let text = "a\n".repeat(3000);
|
||
let parts = split(&text, DISCORD_MAX);
|
||
assert!(parts.len() >= 3 && parts.iter().all(|p| p.chars().count() <= DISCORD_MAX));
|
||
assert_eq!(parts.concat(), text);
|
||
}
|
||
}
|