初步实现 matrix bot

还没实现 Python 侧

ica-rs/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-default = ["ica"]
+default = ["ica", "matrix"]
 ica = ["dep:ed25519", "dep:ed25519-dalek", "dep:hex", "dep:rust_socketio"]
 matrix = ["dep:matrix-sdk", "dep:url"]
 
@@ -14,7 +14,7 @@ matrix = ["dep:matrix-sdk", "dep:url"]
 
 # matrix
 url = { version = "2.5.0", optional = true }
-matrix-sdk = { version = "0.7.1", optional = true }
+matrix-sdk = { version = "0.7.1", optional = true, default-features = false, features = ["rustls-tls"] }
 
 # ica
 ed25519 = { version = "2.2.3", optional = true }

ica-rs/src/config.rs

@@ -73,7 +73,16 @@ impl BotConfig {
         ret
     }
     pub fn new_from_cli() -> Self {
-        let config_file_path = env::args().nth(1).expect("No config path given");
+        // let config_file_path = env::args().nth(1).expect("No config path given");
+        // -c <config_file_path>
+        let mut config_file_path = String::new();
+        let mut args = env::args();
+        while let Some(arg) = args.next() {
+            if arg == "-c" {
+                config_file_path = args.next().expect("No config path given");
+                break;
+            }
+        }
         Self::new_from_path(config_file_path)
     }
 

ica-rs/src/data_struct/ica/messages.rs

@@ -1,7 +1,7 @@
 use crate::data_struct::ica::files::MessageFile;
 use crate::data_struct::ica::{MessageId, RoomId, UserId};
 
-use chrono::{DateTime, NaiveDateTime};
+use chrono::DateTime;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value as JsonValue};
 use tracing::warn;

ica-rs/src/data_struct/ica/messages/msg_trait.rs

@@ -1,6 +1,6 @@
 use std::fmt::Display;
 
-use chrono::{DateTime, NaiveDateTime};
+use chrono::DateTime;
 use serde::{Deserialize, Serialize};
 use serde_json::Value as JsonValue;
 

ica-rs/src/error.rs

@@ -0,0 +1,61 @@
+pub type ClientResult<T, E> = Result<T, E>;
+
+#[derive(Debug)]
+pub enum IcaError {
+    /// Socket IO 链接错误
+    SocketIoError(rust_socketio::Error),
+}
+
+#[derive(Debug)]
+pub enum MatrixError {
+    /// Homeserver Url 错误
+    HomeserverUrlError(url::ParseError),
+    /// Http 请求错误
+    HttpError(matrix_sdk::HttpError),
+    /// Matrix Error
+    MatrixError(matrix_sdk::Error),
+}
+
+impl From<rust_socketio::Error> for IcaError {
+    fn from(e: rust_socketio::Error) -> Self { IcaError::SocketIoError(e) }
+}
+
+impl From<matrix_sdk::Error> for MatrixError {
+    fn from(e: matrix_sdk::Error) -> Self { MatrixError::MatrixError(e) }
+}
+
+impl std::fmt::Display for IcaError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            IcaError::SocketIoError(e) => write!(f, "Socket IO 链接错误: {}", e),
+        }
+    }
+}
+
+impl std::error::Error for IcaError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            IcaError::SocketIoError(e) => Some(e),
+        }
+    }
+}
+
+impl std::fmt::Display for MatrixError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            MatrixError::HomeserverUrlError(e) => write!(f, "Homeserver Url 错误: {}", e),
+            MatrixError::HttpError(e) => write!(f, "Http 请求错误: {}", e),
+            MatrixError::MatrixError(e) => write!(f, "Matrix Error: {}", e),
+        }
+    }
+}
+
+impl std::error::Error for MatrixError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            MatrixError::HomeserverUrlError(e) => Some(e),
+            MatrixError::HttpError(e) => Some(e),
+            MatrixError::MatrixError(e) => Some(e),
+        }
+    }
+}

ica-rs/src/ica.rs

@@ -4,14 +4,19 @@ pub mod events;
 use futures_util::FutureExt;
 use rust_socketio::asynchronous::{Client, ClientBuilder};
 use rust_socketio::{Event, Payload, TransportType};
-use tracing::{event, info, Level};
+use tracing::{event, span, Level};
 
 use crate::config::IcaConfig;
-use crate::{wrap_any_callback, wrap_callback};
+use crate::error::{ClientResult, IcaError};
+use crate::{wrap_any_callback, wrap_callback, StopGetter};
 
-pub async fn start_ica(config: &IcaConfig, stop_reciver: tokio::sync::oneshot::Receiver<()>) {
+pub async fn start_ica(config: &IcaConfig, stop_reciver: StopGetter) -> ClientResult<(), IcaError> {
-    event!(Level::INFO, "ica-async-rs v{} start ica", crate::ICA_VERSION);
+    let span = span!(Level::INFO, "Icalingua Client");
-    let socket = ClientBuilder::new(config.host.clone())
+    let _enter = span.enter();
+
+    event!(Level::INFO, "ica-async-rs v{} initing", crate::ICA_VERSION);
+
+    let socket = match ClientBuilder::new(config.host.clone())
         .transport_type(TransportType::Websocket)
         .on_any(wrap_any_callback!(events::any_event))
         .on("requireAuth", wrap_callback!(client::sign_callback))
@@ -27,27 +32,50 @@ pub async fn start_ica(config: &IcaConfig, stop_reciver: tokio::sync::oneshot::R
         .on("deleteMessage", wrap_callback!(events::delete_message))
         .connect()
         .await
-        .expect("Connection failed");
+    {
-
+        Ok(client) => {
-    info!("Connected");
+            event!(Level::INFO, "socketio connected");
+            client
+        }
+        Err(e) => {
+            event!(Level::ERROR, "socketio connect failed: {}", e);
+            return Err(IcaError::SocketIoError(e));
+        }
+    };
 
     if config.notice_start {
         for room in config.notice_room.iter() {
             let startup_msg = crate::data_struct::ica::messages::SendMessage::new(
-                format!("shenbot v {}\nica-async-rs bot v{}", crate::VERSION, crate::ICA_VERSION),
+                format!("shenbot v {}\nica-async-rs v{}", crate::VERSION, crate::ICA_VERSION),
                 *room,
                 None,
             );
             tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-            info!("发送启动消息到房间: {}", room);
+
+            event!(Level::INFO, "发送启动消息到房间: {}", room);
+
             if let Err(e) =
                 socket.emit("sendMessage", serde_json::to_value(startup_msg).unwrap()).await
             {
-                info!("启动信息发送失败 房间:{}|e:{}", room, e);
+                event!(Level::INFO, "启动信息发送失败 房间:{}|e:{}", room, e);
             }
         }
     }
     // 等待停止信号
     stop_reciver.await.ok();
-    socket.disconnect().await.expect("Disconnect failed");
+    event!(Level::INFO, "socketio client stopping");
+    let disconnect = socket.disconnect().await;
+    match disconnect {
+        Ok(_) => {
+            event!(Level::INFO, "socketio client stopped");
+            Ok(())
+        }
+        Err(e) => {
+            event!(Level::ERROR, "socketio client stopped with error: {}", e);
+            Err(IcaError::SocketIoError(e))
+        }
+    }
+
+    // event!(Level::INFO, "socketio client stopped");
+    // disconnect.into()
 }

ica-rs/src/main.rs

@@ -2,6 +2,7 @@ use std::time::Duration;
 
 mod config;
 mod data_struct;
+mod error;
 #[cfg(feature = "ica")]
 mod ica;
 #[cfg(feature = "matrix")]
@@ -10,7 +11,7 @@ mod py;
 mod status;
 
 use config::BotConfig;
-use tracing::{event, info, Level};
+use tracing::{event, info, span, Level};
 
 pub static mut MAIN_STATUS: status::BotStatus = status::BotStatus {
     config: None,
@@ -20,6 +21,8 @@ pub static mut MAIN_STATUS: status::BotStatus = status::BotStatus {
 
 pub type MainStatus = status::BotStatus;
 
+pub type StopGetter = tokio::sync::oneshot::Receiver<()>;
+
 pub const VERSION: &str = env!("CARGO_PKG_VERSION");
 pub const ICA_VERSION: &str = "1.4.0";
 pub const MATRIX_VERSION: &str = "0.1.0";
@@ -40,8 +43,22 @@ macro_rules! wrap_any_callback {
 
 #[tokio::main]
 async fn main() {
-    tracing_subscriber::fmt().with_max_level(tracing::Level::DEBUG).init();
+    // -d -> debug
-    event!(Level::INFO, "shenbot-async-rs v{} main", VERSION);
+    // none -> info
+    let level = {
+        let args = std::env::args();
+        if args.collect::<Vec<String>>().contains(&"-d".to_string()) {
+            Level::DEBUG
+        } else {
+            Level::INFO
+        }
+    };
+
+    tracing_subscriber::fmt().with_max_level(level).init();
+    let span = span!(Level::INFO, "Shenbot Main");
+    let _enter = span.enter();
+
+    event!(Level::INFO, "shenbot-async-rs v{} starting", VERSION);
 
     let bot_config = BotConfig::new_from_cli();
     MainStatus::static_init(bot_config);
@@ -50,16 +67,30 @@ async fn main() {
     py::init_py();
 
     // 准备一个用于停止 socket 的变量
-    let (send, recv) = tokio::sync::oneshot::channel::<()>();
+    event!(Level::INFO, "启动 ICA");
+    let (ica_send, ica_recv) = tokio::sync::oneshot::channel::<()>();
 
     if bot_config.check_ica() {
-        info!("启动 ica");
+        event!(Level::INFO, "启动 ica");
         let config = bot_config.ica();
         tokio::spawn(async move {
-            ica::start_ica(&config, recv).await;
+            ica::start_ica(&config, ica_recv).await.unwrap();
         });
     } else {
-        info!("未启用 ica");
+        event!(Level::INFO, "未启用 ica");
+    }
+
+    event!(Level::INFO, "启动 Matrix");
+    let (matrix_send, matrix_recv) = tokio::sync::oneshot::channel::<()>();
+
+    if bot_config.check_matrix() {
+        event!(Level::INFO, "启动 Matrix");
+        let config = bot_config.matrix();
+        tokio::spawn(async move {
+            matrix::start_matrix(&config, matrix_recv).await.unwrap();
+        });
+    } else {
+        event!(Level::INFO, "未启用 Matrix");
     }
 
     tokio::time::sleep(Duration::from_secs(2)).await;
@@ -68,7 +99,8 @@ async fn main() {
     let mut input = String::new();
     std::io::stdin().read_line(&mut input).unwrap();
 
-    send.send(()).ok();
+    ica_send.send(()).ok();
+    matrix_send.send(()).ok();
 
     info!("Disconnected");
 }

ica-rs/src/matrix.rs

@@ -1 +1,152 @@
+pub mod events;
 
+use std::str::FromStr;
+
+use matrix_sdk::{
+    config::SyncSettings,
+    ruma::{
+        api::client::message::send_message_event,
+        events::room::message::{
+            AddMentions, ForwardThread, MessageType, OriginalSyncRoomMessageEvent,
+            RoomMessageEventContent,
+        },
+        OwnedRoomId, TransactionId,
+    },
+    Client, Room, RoomState,
+};
+use tracing::{event, span, Level};
+use url::Url;
+
+use crate::config::MatrixConfig;
+use crate::error::{ClientResult, MatrixError};
+use crate::StopGetter;
+
+pub async fn start_matrix(
+    config: &MatrixConfig,
+    stop_reciver: StopGetter,
+) -> ClientResult<(), MatrixError> {
+    let span = span!(Level::INFO, "Matrix Client");
+    let _enter = span.enter();
+
+    let homeserver_url = match Url::parse(&config.home_server) {
+        Ok(url) => url,
+        Err(e) => {
+            event!(Level::ERROR, "Homeserver Url 错误: {}", e);
+            return Err(MatrixError::HomeserverUrlError(e));
+        }
+    };
+    let password = &config.bot_password;
+    let username = &config.bot_id;
+
+    let client = match Client::new(homeserver_url).await {
+        Ok(client) => {
+            event!(Level::INFO, "Logged in as {}", username);
+            client
+        }
+        Err(error) => {
+            event!(Level::ERROR, "Failed to log in as {}: {}", username, error);
+            return Err(MatrixError::HttpError(error));
+        }
+    };
+
+    let display_name = format!("shenbot-matrix v{}", crate::MATRIX_VERSION);
+
+    match client
+        .matrix_auth()
+        .login_username(&username, &password)
+        .initial_device_display_name(&display_name)
+        .await
+    {
+        Ok(_) => {
+            event!(Level::INFO, "Logged in as {}", username);
+        }
+        Err(error) => {
+            event!(Level::ERROR, "Failed to log in as {}: {}", username, error);
+            return Err(MatrixError::MatrixError(error));
+        }
+    }
+
+    // 发送启动消息
+    if config.notice_start {
+        for room in config.notice_room.iter() {
+            let startup_msg = RoomMessageEventContent::text_plain(format!(
+                "shenbot v {}\nmatrix-rs v{} started!",
+                crate::VERSION,
+                crate::MATRIX_VERSION
+            ));
+            let startup_req: send_message_event::v3::Request =
+                send_message_event::v3::Request::new(
+                    OwnedRoomId::from_str(&room).unwrap(),
+                    TransactionId::new(),
+                    &startup_msg,
+                )
+                .unwrap();
+
+            event!(Level::INFO, "发送启动消息到房间: {}", room);
+
+            if let Err(e) = client.send::<send_message_event::v3::Request>(startup_req, None).await
+            {
+                event!(Level::INFO, "启动信息发送失败 房间:{}|e:{}", room, e);
+            }
+        }
+    } else {
+        event!(Level::INFO, "未启用启动消息");
+    }
+
+    client.add_event_handler(on_room_message);
+
+    match client.sync_once(SyncSettings::new()).await {
+        Ok(_) => {
+            event!(Level::INFO, "Synced");
+        }
+        Err(error) => {
+            event!(Level::ERROR, "Failed to sync: {}", error);
+            return Err(MatrixError::MatrixError(error));
+        }
+    }
+
+    client.sync(SyncSettings::default()).await?;
+
+    // while stop_reciver.await.is_err() {
+    //     event!(Level::INFO, "Matrix client is running");
+    //     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+    // }
+
+    event!(Level::INFO, "Matrix is not implemented yet");
+    stop_reciver.await.ok();
+    event!(Level::INFO, "Matrix client stopping");
+    // some stop
+    event!(Level::INFO, "Matrix client stopped");
+    Ok(())
+}
+
+pub async fn on_room_message(event: OriginalSyncRoomMessageEvent, room: Room) {
+    // We only want to listen to joined rooms.
+    if room.state() != RoomState::Joined {
+        return;
+    }
+
+    // We only want to log text messages.
+    let MessageType::Text(msgtype) = &event.content.msgtype else {
+        return;
+    };
+
+    // 匹配消息
+
+    // /bot
+    if msgtype.body == "/bot" {
+        let pong = format!("shenbot v {}\nmatrix-rs v{}", crate::VERSION, crate::MATRIX_VERSION);
+
+        let reply = RoomMessageEventContent::text_plain(pong);
+        let reply = reply.make_reply_to(
+            &event.into_full_event(room.room_id().to_owned()),
+            ForwardThread::No,
+            AddMentions::No,
+        );
+
+        room.send(reply).await.expect("Failed to send message");
+        return;
+    }
+
+    // 发给 Python 处理剩下的
+}

ica-rs/src/matrix/events.rs

@@ -0,0 +1 @@
+