GeneralBots/gbserver
2
0
Fork 0
Code Issues 7 Pull requests Projects Releases Packages Wiki Activity Actions

- Logging review.
Some checks failed
GBCI / build (push) Has been cancelled
Details

Browse source
This commit is contained in:
Rodrigo Rodriguez (Pragmatismo) 2025-09-17 20:11:56 -03:00
parent f10522bf76
commit e5dfcf2659
25 changed files with 298 additions and 281 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
Cargo.lock generated
View file

@ -2705,9 +2705,9 @@ dependencies = [
[[package]]
name = "log"
version = "0.4.27"
version = "0.4.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94"
checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432"
dependencies = [
"value-bag",
]

2
Cargo.toml
View file

@ -34,7 +34,7 @@ lettre = { version = "0.10", features = [
"tokio1",
"tokio1-native-tls",
] }
log = "0.4"
log = "0.4.28"
mailparse = "0.13"
minio = { git = "https://github.com/minio/minio-rs", branch = "master" }
native-tls = "0.2"

9
src/main.rs
View file

@ -1,4 +1,5 @@
use actix_web::middleware::Logger;
use log::info;
use std::sync::Arc;
use actix_web::{web, App, HttpServer};
@ -8,11 +9,12 @@ use services::{config::*, file::*};
use sqlx::PgPool;
use crate::services::automation::AutomationService;
use crate::services::email::{get_emails, list_emails, save_click, send_email};
use crate::services::email::{
get_emails, get_latest_email_from, list_emails, save_click, save_draft, send_email,
};
use crate::services::llm::{chat, chat_stream};
use crate::services::llm_local::ensure_llama_servers_running;
use crate::services::llm_local::{chat_completions_local, embeddings_local};
use crate::services::llm_provider::chat_completions;
use crate::services::web_automation::{initialize_browser_pool, BrowserPool};
mod models;
@ -22,6 +24,7 @@ mod services;
async fn main() -> std::io::Result<()> {
dotenv().ok();
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
info!("Starting General Bots 6.0...");
let config = AppConfig::from_env();
let db_url = config.database_url();
@ -84,7 +87,9 @@ async fn main() -> std::io::Result<()> {
.service(chat_stream)
.service(chat_completions_local)
.service(chat)
.service(save_draft)
.service(embeddings_local)
.service(get_latest_email_from)
})
.bind((config.server.host.clone(), config.server.port))?
.run()

19
src/services/automation.rs
View file

@ -4,10 +4,10 @@ use crate::services::state::AppState;
use chrono::Datelike;
use chrono::Timelike;
use chrono::{DateTime, Utc};
use log::{error, info};
use std::path::Path;
use tokio::time::Duration;
use uuid::Uuid;
pub struct AutomationService {
state: AppState, // Use web::Data directly
scripts_dir: String,
@ -30,7 +30,7 @@ impl AutomationService {
interval.tick().await;
if let Err(e) = self.run_cycle(&mut last_check).await {
eprintln!("Automation cycle error: {}", e);
error!("Automation cycle error: {}", e);
}
}
})
@ -94,7 +94,7 @@ impl AutomationService {
}
}
Err(e) => {
eprintln!("Error checking changes for table {}: {}", table, e);
error!("Error checking changes for table {}: {}", table, e);
}
}
}
@ -128,8 +128,9 @@ impl AutomationService {
)
.execute(pool)
.await
{
eprintln!(
error!(
"Failed to update last_triggered for automation {}: {}",
automation_id, e
);
@ -177,20 +178,20 @@ impl AutomationService {
let full_path = Path::new(&self.scripts_dir).join(param);
match tokio::fs::read_to_string(&full_path).await {
Ok(script_content) => {
println!("Executing action with param: {}", param);
info!("Executing action with param: {}", param);
let script_service = ScriptService::new(&self.state.clone());
match script_service.compile(&script_content) {
Ok(ast) => match script_service.run(&ast) {
Ok(result) => println!("Script executed successfully: {:?}", result),
Err(e) => eprintln!("Error executing script: {}", e),
Ok(result) => info!("Script executed successfully: {:?}", result),
Err(e) => error!("Error executing script: {}", e),
},
Err(e) => eprintln!("Error compiling script: {}", e),
Err(e) => error!("Error compiling script: {}", e),
}
}
Err(e) => {
eprintln!("Failed to execute action {}: {}", full_path.display(), e);
error!("Failed to execute action {}: {}", full_path.display(), e);
}
}
}

10
src/services/email.rs
View file

@ -1,4 +1,6 @@
use crate::services::{config::EmailConfig, state::AppState};
use log::info;
use actix_web::error::ErrorInternalServerError;
use actix_web::http::header::ContentType;
use actix_web::{web, HttpResponse, Result};
@ -263,7 +265,7 @@ pub async fn save_email_draft(
Ok(chrono::Utc::now().timestamp().to_string())
}
pub async fn fetch_latest_email_from_sender(
async fn fetch_latest_email_from_sender(
email_config: &EmailConfig,
from_email: &str,
) -> Result<String, Box<dyn std::error::Error>> {
@ -479,9 +481,9 @@ pub async fn send_email(
) -> Result<HttpResponse, actix_web::Error> {
let (to, subject, body) = payload.into_inner();
println!("To: {}", to);
println!("Subject: {}", subject);
println!("Body: {}", body);
info!("To: {}", to);
info!("Subject: {}", subject);
info!("Body: {}", body);
// Send via SMTP
internal_send_email(&state.config.clone().unwrap().email, &to, &subject, &body).await;

34
src/services/keywords/create_site.rs
View file

@ -1,9 +1,11 @@
use log::info;
use rhai::Dynamic;
use rhai::Engine;
use std::error::Error;
use std::fs;
use std::path::{ PathBuf};
use std::io::Read;
use std::path::PathBuf;
use crate::services::state::AppState;
use crate::services::utils;
@ -12,10 +14,7 @@ pub fn create_site_keyword(state: &AppState, engine: &mut Engine) {
let state_clone = state.clone();
engine
.register_custom_syntax(
&[
"CREATE_SITE", "$expr$", ",", "$expr$", ",", "$expr$",
],
&["CREATE_SITE", "$expr$", ",", "$expr$", ",", "$expr$"],
true,
move |context, inputs| {
if inputs.len() < 3 {
@ -25,9 +24,13 @@ pub fn create_site_keyword(state: &AppState, engine: &mut Engine) {
let alias = context.eval_expression_tree(&inputs[0])?;
let template_dir = context.eval_expression_tree(&inputs[1])?;
let prompt = context.eval_expression_tree(&inputs[2])?;
let config = state_clone.config.as_ref().expect("Config must be initialized").clone();
let config = state_clone
.config
.as_ref()
.expect("Config must be initialized")
.clone();
let fut = create_site(&config, alias, template_dir, prompt);
let result =
tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(fut))
@ -55,16 +58,17 @@ async fn create_site(
// Process all HTML files in template directory
let mut combined_content = String::new();
for entry in fs::read_dir(&template_path).map_err(|e| e.to_string())? {
let entry = entry.map_err(|e| e.to_string())?;
let path = entry.path();
if path.extension().map_or(false, |ext| ext == "html") {
let mut file = fs::File::open(&path).map_err(|e| e.to_string())?;
let mut contents = String::new();
file.read_to_string(&mut contents).map_err(|e| e.to_string())?;
file.read_to_string(&mut contents)
.map_err(|e| e.to_string())?;
combined_content.push_str(&contents);
combined_content.push_str("\n\n--- TEMPLATE SEPARATOR ---\n\n");
}
@ -78,13 +82,13 @@ async fn create_site(
);
// Call LLM with the combined prompt
println!("Asking LLM to create site.");
info!("Asking LLM to create site.");
let llm_result = utils::call_llm(&full_prompt, &config.ai).await?;
// Write the generated HTML file
let index_path = alias_path.join("index.html");
fs::write(index_path, llm_result).map_err(|e| e.to_string())?;
println!("Site created at: {}", alias_path.display());
info!("Site created at: {}", alias_path.display());
Ok(alias_path.to_string_lossy().into_owned())
}
}

13
src/services/keywords/find.rs
View file

@ -1,14 +1,14 @@
use log::{error, info};
use rhai::Dynamic;
use rhai::Engine;
use serde_json::{json, Value};
use sqlx::{PgPool};
use sqlx::PgPool;
use crate::services::state::AppState;
use crate::services::utils;
use crate::services::utils::row_to_json;
use crate::services::utils::to_array;
pub fn find_keyword(state: &AppState, engine: &mut Engine) {
let db = state.db_custom.clone();
@ -48,7 +48,7 @@ pub async fn execute_find(
filter_str: &str,
) -> Result<Value, String> {
// Changed to String error like your Actix code
println!(
info!(
"Starting execute_find with table: {}, filter: {}",
table_str, filter_str
);
@ -59,7 +59,7 @@ pub async fn execute_find(
"SELECT * FROM {} WHERE {} LIMIT 10",
table_str, where_clause
);
println!("Executing query: {}", query);
info!("Executing query: {}", query);
// Use the same simple pattern as your Actix code - no timeout wrapper
let rows = sqlx::query(&query)
@ -67,11 +67,11 @@ pub async fn execute_find(
.fetch_all(pool)
.await
.map_err(|e| {
eprintln!("SQL execution error: {}", e);
error!("SQL execution error: {}", e);
e.to_string()
})?;
println!("Query successful, got {} rows", rows.len());
info!("Query successful, got {} rows", rows.len());
let mut results = Vec::new();
for row in rows {
@ -85,4 +85,3 @@ pub async fn execute_find(
"results": results
}))
}

6
src/services/keywords/for_next.rs
View file

@ -1,9 +1,9 @@
use crate::services::state::AppState;
use log::info;
use rhai::Dynamic;
use rhai::Engine;
use crate::services::state::AppState;
pub fn for_keyword(_state: &AppState, engine: &mut Engine) {
engine
.register_custom_syntax(&["EXIT", "FOR"], false, |_context, _inputs| {
Err("EXIT FOR".into())
@ -34,7 +34,7 @@ pub fn for_keyword(_state: &AppState, engine: &mut Engine) {
let collection = context.eval_expression_tree(&inputs[1])?;
// Debug: Print the collection type
println!("Collection type: {}", collection.type_name());
info!("Collection type: {}", collection.type_name());
let ccc = collection.clone();
// Convert to array - with proper error handling
let array = match collection.into_array() {

6
src/services/keywords/get.rs
View file

@ -1,3 +1,5 @@
use log::info;
use crate::services::state::AppState;
use reqwest::{self, Client};
use rhai::{Dynamic, Engine};
@ -31,7 +33,7 @@ pub fn get_keyword(_state: &AppState, engine: &mut Engine) {
};
if modified_url.starts_with("https://") {
println!("HTTPS GET request: {}", modified_url);
info!("HTTPS GET request: {}", modified_url);
let fut = execute_get(&modified_url);
let result =
@ -57,7 +59,7 @@ pub fn get_keyword(_state: &AppState, engine: &mut Engine) {
}
pub async fn execute_get(url: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
println!("Starting execute_get with URL: {}", url);
info!("Starting execute_get with URL: {}", url);
// Create a client that ignores invalid certificates
let client = Client::builder()

48
src/services/keywords/get_website.rs
View file

@ -1,4 +1,5 @@
use crate::services::{state::AppState, web_automation::BrowserPool};
use log::info;
use rhai::{Dynamic, Engine};
use std::error::Error;
use std::sync::Arc;
@ -15,18 +16,11 @@ pub fn get_website_keyword(state: &AppState, engine: &mut Engine) {
false,
move |context, inputs| {
let search_term = context.eval_expression_tree(&inputs[0])?.to_string();
println!(
"GET WEBSITE executed - Search: '{}'",
search_term
);
info!("GET WEBSITE executed - Search: '{}'", search_term);
let browser_pool_clone = browser_pool.clone();
let fut = execute_headless_browser_search(
browser_pool_clone,
&search_term
);
let fut = execute_headless_browser_search(browser_pool_clone, &search_term);
let result =
tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(fut))
@ -40,26 +34,22 @@ pub fn get_website_keyword(state: &AppState, engine: &mut Engine) {
pub async fn execute_headless_browser_search(
browser_pool: Arc<BrowserPool>, // Adjust path as needed
search_term: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
println!(
"Starting headless browser search: '{}' ",
search_term
);
search_term: &str,
) -> Result<String, Box<dyn Error + Send + Sync>> {
info!("Starting headless browser search: '{}' ", search_term);
let search_term = search_term.to_string();
let result = browser_pool
.with_browser(|driver| {
Box::pin(async move { perform_search(driver, &search_term).await })
})
.with_browser(|driver| Box::pin(async move { perform_search(driver, &search_term).await }))
.await?;
Ok(result)
}
async fn perform_search(
driver: WebDriver,
search_term: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
search_term: &str,
) -> Result<String, Box<dyn Error + Send + Sync>> {
// Navigate to DuckDuckGo
driver.goto("https://duckduckgo.com").await?;
@ -96,13 +86,13 @@ async fn extract_search_results(
// Modern DuckDuckGo (as seen in the HTML)
"a[data-testid='result-title-a']", // Primary result links
"a[data-testid='result-extras-url-link']", // URL links in results
"a.eVNpHGjtxRBq_gLOfGDr", // Class-based selector for result titles
"a.Rn_JXVtoPVAFyGkcaXyK", // Class-based selector for URL links
".ikg2IXiCD14iVX7AdZo1 a", // Heading container links
".OQ_6vPwNhCeusNiEDcGp a", // URL container links
"a.eVNpHGjtxRBq_gLOfGDr", // Class-based selector for result titles
"a.Rn_JXVtoPVAFyGkcaXyK", // Class-based selector for URL links
".ikg2IXiCD14iVX7AdZo1 a", // Heading container links
".OQ_6vPwNhCeusNiEDcGp a", // URL container links
// Fallback selectors
".result__a", // Classic DuckDuckGo
"a.result-link", // Alternative
".result__a", // Classic DuckDuckGo
"a.result-link", // Alternative
".result a[href]", // Generic result links
];
@ -111,11 +101,11 @@ async fn extract_search_results(
for element in elements {
if let Ok(Some(href)) = element.attr("href").await {
// Filter out internal and non-http links
if href.starts_with("http")
if href.starts_with("http")
&& !href.contains("duckduckgo.com")
&& !href.contains("duck.co")
&& !results.contains(&href) {
&& !results.contains(&href)
{
// Get the display URL for verification
let display_url = if let Ok(text) = element.text().await {
text.trim().to_string()
@ -140,4 +130,4 @@ async fn extract_search_results(
results.dedup();
Ok(results)
}
}

42
src/services/keywords/llm_keyword.rs
View file

@ -1,28 +1,30 @@
use rhai::{Dynamic, Engine};
use log::info;
use crate::services::{state::AppState, utils::call_llm};
use rhai::{Dynamic, Engine};
pub fn llm_keyword(state: &AppState, engine: &mut Engine) {
let ai_config = state.config.clone().unwrap().ai.clone();
engine.register_custom_syntax(
&["LLM", "$expr$"], // Syntax: LLM "text to process"
false, // Expression, not statement
move |context, inputs| {
let text = context.eval_expression_tree(&inputs[0])?;
let text_str = text.to_string();
engine
.register_custom_syntax(
&["LLM", "$expr$"], // Syntax: LLM "text to process"
false, // Expression, not statement
move |context, inputs| {
let text = context.eval_expression_tree(&inputs[0])?;
let text_str = text.to_string();
println!("LLM processing text: {}", text_str);
// Use the same pattern as GET
info!("LLM processing text: {}", text_str);
let fut = call_llm(
&text_str, &ai_config);
let result = tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(fut)
}).map_err(|e| format!("LLM call failed: {}", e))?;
Ok(Dynamic::from(result))
}
).unwrap();
// Use the same pattern as GET
let fut = call_llm(&text_str, &ai_config);
let result =
tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(fut))
.map_err(|e| format!("LLM call failed: {}", e))?;
Ok(Dynamic::from(result))
},
)
.unwrap();
}

5
src/services/keywords/on.rs
View file

@ -1,3 +1,4 @@
use log::{error, info};
use rhai::Dynamic;
use rhai::Engine;
use serde_json::{json, Value};
@ -54,7 +55,7 @@ pub async fn execute_on_trigger(
table: &str,
script_name: &str,
) -> Result<Value, String> {
println!(
info!(
"Starting execute_on_trigger with kind: {:?}, table: {}, script_name: {}",
kind, table, script_name
);
@ -71,7 +72,7 @@ pub async fn execute_on_trigger(
.execute(pool)
.await
.map_err(|e| {
eprintln!("SQL execution error: {}", e);
error!("SQL execution error: {}", e);
e.to_string()
})?;

4
src/services/keywords/print.rs
View file

@ -1,10 +1,10 @@
use log::info;
use rhai::Dynamic;
use rhai::Engine;
use crate::services::state::AppState;
pub fn print_keyword(_state: &AppState, engine: &mut Engine) {
// PRINT command
engine
.register_custom_syntax(
@ -12,7 +12,7 @@ pub fn print_keyword(_state: &AppState, engine: &mut Engine) {
true, // Statement
|context, inputs| {
let value = context.eval_expression_tree(&inputs[0])?;
println!("{}", value);
info!("{}", value);
Ok(Dynamic::UNIT)
},
)

4
src/services/keywords/prompt.md
View file

@ -49,7 +49,7 @@ pub async fn execute_{keyword_name}(
pool: &PgPool,
{params_with_types}
) -> Result<Value, Box<dyn std::error::Error>> {
println!("Executing {keyword_name} with: {debug_params}");
info!("Executing {keyword_name} with: {debug_params}");
let result = sqlx::query(
"{sql_query_with_i32_enum}"
@ -113,7 +113,7 @@ pub async fn execute_set_schedule(
cron: &str,
script_name: &str,
) -> Result<Value, Box<dyn std::error::Error>> {
println!("Executing schedule: {}, {}", cron, script_name);
info!("Executing schedule: {}, {}", cron, script_name);
let result = sqlx::query(
"INSERT INTO system_automations

7
src/services/keywords/set.rs
View file

@ -1,3 +1,4 @@
use log::{error, info};
use rhai::Dynamic;
use rhai::Engine;
use serde_json::{json, Value};
@ -47,7 +48,7 @@ pub async fn execute_set(
filter_str: &str,
updates_str: &str,
) -> Result<Value, String> {
println!(
info!(
"Starting execute_set with table: {}, filter: {}, updates: {}",
table_str, filter_str, updates_str
);
@ -65,7 +66,7 @@ pub async fn execute_set(
"UPDATE {} SET {} WHERE {}",
table_str, set_clause, where_clause
);
println!("Executing query: {}", query);
info!("Executing query: {}", query);
// Build query with proper parameter binding
let mut query = sqlx::query(&query);
@ -81,7 +82,7 @@ pub async fn execute_set(
}
let result = query.execute(pool).await.map_err(|e| {
eprintln!("SQL execution error: {}", e);
error!("SQL execution error: {}", e);
e.to_string()
})?;

3
src/services/keywords/set_schedule.rs
View file

@ -1,3 +1,4 @@
use log::info;
use rhai::Dynamic;
use rhai::Engine;
use serde_json::{json, Value};
@ -39,7 +40,7 @@ pub async fn execute_set_schedule(
cron: &str,
script_name: &str,
) -> Result<Value, Box<dyn std::error::Error>> {
println!(
info!(
"Starting execute_set_schedule with cron: {}, script_name: {}",
cron, script_name
);

75
src/services/keywords/wait.rs
View file

@ -1,39 +1,46 @@
use rhai::{Dynamic, Engine};
use crate::services::state::AppState;
use log::info;
use rhai::{Dynamic, Engine};
use std::thread;
use std::time::Duration;
pub fn wait_keyword(_state: &AppState, engine: &mut Engine) {
engine.register_custom_syntax(
&["WAIT", "$expr$"],
false, // Expression, not statement
move |context, inputs| {
let seconds = context.eval_expression_tree(&inputs[0])?;
// Convert to number (handle both int and float)
let duration_secs = if seconds.is::<i64>() {
seconds.cast::<i64>() as f64
} else if seconds.is::<f64>() {
seconds.cast::<f64>()
} else {
return Err(format!("WAIT expects a number, got: {}", seconds).into());
};
if duration_secs < 0.0 {
return Err("WAIT duration cannot be negative".into());
}
// Cap maximum wait time to prevent abuse (e.g., 5 minutes max)
let capped_duration = if duration_secs > 300.0 { 300.0 } else { duration_secs };
println!("WAIT {} seconds (thread sleep)", capped_duration);
// Use thread::sleep to block only the current thread, not the entire server
let duration = Duration::from_secs_f64(capped_duration);
thread::sleep(duration);
println!("WAIT completed after {} seconds", capped_duration);
Ok(Dynamic::from(format!("Waited {} seconds", capped_duration)))
}
).unwrap();
}
engine
.register_custom_syntax(
&["WAIT", "$expr$"],
false, // Expression, not statement
move |context, inputs| {
let seconds = context.eval_expression_tree(&inputs[0])?;
// Convert to number (handle both int and float)
let duration_secs = if seconds.is::<i64>() {
seconds.cast::<i64>() as f64
} else if seconds.is::<f64>() {
seconds.cast::<f64>()
} else {
return Err(format!("WAIT expects a number, got: {}", seconds).into());
};
if duration_secs < 0.0 {
return Err("WAIT duration cannot be negative".into());
}
// Cap maximum wait time to prevent abuse (e.g., 5 minutes max)
let capped_duration = if duration_secs > 300.0 {
300.0
} else {
duration_secs
};
info!("WAIT {} seconds (thread sleep)", capped_duration);
// Use thread::sleep to block only the current thread, not the entire server
let duration = Duration::from_secs_f64(capped_duration);
thread::sleep(duration);
info!("WAIT completed after {} seconds", capped_duration);
Ok(Dynamic::from(format!("Waited {} seconds", capped_duration)))
},
)
.unwrap();
}

36
src/services/llm-email.md
View file

@ -49,18 +49,18 @@ pub async fn chat(
) -> Result<HttpResponse> {
let azure_config = from_config(&state.config.clone().unwrap().ai);
let open_ai = OpenAI::new(azure_config);
// Define available tools based on context
let tools = get_available_tools(&request.context);
// Build the prompt with context and available tools
let system_prompt = build_system_prompt(&request.context, &tools);
let user_message = format!("{}\n\nUser input: {}", system_prompt, request.input);
let response = match open_ai.invoke(&user_message).await {
Ok(res) => res,
Err(err) => {
eprintln!("Error invoking API: {}", err);
error!("Error invoking API: {}", err);
return Err(actix_web::error::ErrorInternalServerError(
"Failed to invoke OpenAI API",
));
@ -80,7 +80,7 @@ pub async fn chat(
fn get_available_tools(context: &Option<AppContext>) -> Vec<ToolDefinition> {
let mut tools = Vec::new();
if let Some(ctx) = context {
if let Some(view_type) = &ctx.view_type {
match view_type.as_str() {
@ -99,7 +99,7 @@ fn get_available_tools(context: &Option<AppContext>) -> Vec<ToolDefinition> {
"required": ["content"]
}),
});
tools.push(ToolDefinition {
name: "forwardEmail".to_string(),
description: "Forward the current email to specified recipients".to_string(),
@ -124,13 +124,13 @@ fn get_available_tools(context: &Option<AppContext>) -> Vec<ToolDefinition> {
}
}
}
tools
}
fn build_system_prompt(context: &Option<AppContext>, tools: &[ToolDefinition]) -> String {
let mut prompt = String::new();
if let Some(ctx) = context {
if let Some(view_type) = &ctx.view_type {
match view_type.as_str() {
@ -143,11 +143,11 @@ fn build_system_prompt(context: &Option<AppContext>, tools: &[ToolDefinition]) -
Labels: {:?}\n\n",
email_ctx.subject, email_ctx.id, email_ctx.labels
));
if let Some(from) = &email_ctx.from {
prompt.push_str(&format!("From: {}\n", from));
}
if let Some(body) = &email_ctx.body {
prompt.push_str(&format!("Body: {}\n", body));
}
@ -157,7 +157,7 @@ fn build_system_prompt(context: &Option<AppContext>, tools: &[ToolDefinition]) -
}
}
}
if !tools.is_empty() {
prompt.push_str("\nAvailable tools:\n");
for tool in tools {
@ -166,7 +166,7 @@ fn build_system_prompt(context: &Option<AppContext>, tools: &[ToolDefinition]) -
tool.name, tool.description, tool.parameters
));
}
prompt.push_str(
"If you need to use a tool, respond with:\n\
TOOL_CALL: tool_name\n\
@ -175,7 +175,7 @@ fn build_system_prompt(context: &Option<AppContext>, tools: &[ToolDefinition]) -
Otherwise, just provide a normal response.\n"
);
}
prompt
}
@ -183,19 +183,19 @@ fn parse_tool_calls(response: &str) -> Option<Vec<ToolCall>> {
if !response.contains("TOOL_CALL:") {
return None;
}
let mut tool_calls = Vec::new();
let lines: Vec<&str> = response.lines().collect();
let mut i = 0;
while i < lines.len() {
if lines[i].starts_with("TOOL_CALL:") {
let tool_name = lines[i].replace("TOOL_CALL:", "").trim().to_string();
// Look for parameters in the next line
if i + 1 < lines.len() && lines[i + 1].starts_with("PARAMETERS:") {
let params_str = lines[i + 1].replace("PARAMETERS:", "").trim();
if let Ok(parameters) = serde_json::from_str::<serde_json::Value>(params_str) {
tool_calls.push(ToolCall {
tool_name,
@ -206,10 +206,10 @@ fn parse_tool_calls(response: &str) -> Option<Vec<ToolCall>> {
}
i += 1;
}
if tool_calls.is_empty() {
None
} else {
Some(tool_calls)
}
}
}

8
src/services/llm.rs
View file

@ -1,3 +1,5 @@
use log::error;
use actix_web::{
web::{self, Bytes},
HttpResponse, Responder,
@ -8,7 +10,7 @@ use langchain_rust::{
chain::{Chain, LLMChainBuilder},
fmt_message, fmt_template,
language_models::llm::LLM,
llm::{openai::OpenAI},
llm::openai::OpenAI,
message_formatter,
prompt::HumanMessagePromptTemplate,
prompt_args,
@ -16,7 +18,7 @@ use langchain_rust::{
template_fstring,
};
use crate::services::{ state::AppState, utils::azure_from_config};
use crate::services::{state::AppState, utils::azure_from_config};
#[derive(serde::Deserialize)]
struct ChatRequest {
@ -71,7 +73,7 @@ pub async fn chat(
let response_text = match open_ai.invoke(&prompt).await {
Ok(res) => res,
Err(err) => {
eprintln!("Error invoking API: {}", err);
error!("Error invoking API: {}", err);
return Err(actix_web::error::ErrorInternalServerError(
"Failed to invoke OpenAI API",
));

69
src/services/llm_local.rs
View file

@ -1,10 +1,9 @@
use actix_web::{post, web, HttpRequest, HttpResponse, Result};
use dotenv::dotenv;
use log::{error, info};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::env;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::time::{sleep, Duration};
// OpenAI-compatible request/response structures
@ -59,7 +58,7 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
let llm_local = env::var("LLM_LOCAL").unwrap_or_else(|_| "false".to_string());
if llm_local.to_lowercase() != "true" {
println!("ℹ️ LLM_LOCAL is not enabled, skipping local server startup");
info!("ℹ️ LLM_LOCAL is not enabled, skipping local server startup");
return Ok(());
}
@ -71,19 +70,19 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
let llm_model_path = env::var("LLM_MODEL_PATH").unwrap_or_else(|_| "".to_string());
let embedding_model_path = env::var("EMBEDDING_MODEL_PATH").unwrap_or_else(|_| "".to_string());
println!("🚀 Starting local llama.cpp servers...");
println!("📋 Configuration:");
println!(" LLM URL: {}", llm_url);
println!(" Embedding URL: {}", embedding_url);
println!(" LLM Model: {}", llm_model_path);
println!(" Embedding Model: {}", embedding_model_path);
info!("🚀 Starting local llama.cpp servers...");
info!("📋 Configuration:");
info!(" LLM URL: {}", llm_url);
info!(" Embedding URL: {}", embedding_url);
info!(" LLM Model: {}", llm_model_path);
info!(" Embedding Model: {}", embedding_model_path);
// Check if servers are already running
let llm_running = is_server_running(&llm_url).await;
let embedding_running = is_server_running(&embedding_url).await;
if llm_running && embedding_running {
println!("✅ Both LLM and Embedding servers are already running");
info!("✅ Both LLM and Embedding servers are already running");
return Ok(());
}
@ -91,25 +90,25 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
let mut tasks = vec![];
if !llm_running && !llm_model_path.is_empty() {
println!("🔄 Starting LLM server...");
info!("🔄 Starting LLM server...");
tasks.push(tokio::spawn(start_llm_server(
llama_cpp_path.clone(),
llm_model_path.clone(),
llm_url.clone(),
)));
} else if llm_model_path.is_empty() {
println!("⚠️ LLM_MODEL_PATH not set, skipping LLM server");
info!("⚠️ LLM_MODEL_PATH not set, skipping LLM server");
}
if !embedding_running && !embedding_model_path.is_empty() {
println!("🔄 Starting Embedding server...");
info!("🔄 Starting Embedding server...");
tasks.push(tokio::spawn(start_embedding_server(
llama_cpp_path.clone(),
embedding_model_path.clone(),
embedding_url.clone(),
)));
} else if embedding_model_path.is_empty() {
println!("⚠️ EMBEDDING_MODEL_PATH not set, skipping Embedding server");
info!("⚠️ EMBEDDING_MODEL_PATH not set, skipping Embedding server");
}
// Wait for all server startup tasks
@ -118,7 +117,7 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
}
// Wait for servers to be ready with verbose logging
println!("⏳ Waiting for servers to become ready...");
info!("⏳ Waiting for servers to become ready...");
let mut llm_ready = llm_running || llm_model_path.is_empty();
let mut embedding_ready = embedding_running || embedding_model_path.is_empty();
@ -129,7 +128,7 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
while attempts < max_attempts && (!llm_ready || !embedding_ready) {
sleep(Duration::from_secs(2)).await;
println!(
info!(
"🔍 Checking server health (attempt {}/{})...",
attempts + 1,
max_attempts
@ -137,26 +136,26 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
if !llm_ready && !llm_model_path.is_empty() {
if is_server_running(&llm_url).await {
println!(" ✅ LLM server ready at {}", llm_url);
info!(" ✅ LLM server ready at {}", llm_url);
llm_ready = true;
} else {
println!(" ❌ LLM server not ready yet");
info!(" ❌ LLM server not ready yet");
}
}
if !embedding_ready && !embedding_model_path.is_empty() {
if is_server_running(&embedding_url).await {
println!(" ✅ Embedding server ready at {}", embedding_url);
info!(" ✅ Embedding server ready at {}", embedding_url);
embedding_ready = true;
} else {
println!(" ❌ Embedding server not ready yet");
info!(" ❌ Embedding server not ready yet");
}
}
attempts += 1;
if attempts % 10 == 0 {
println!(
info!(
"⏰ Still waiting for servers... (attempt {}/{})",
attempts, max_attempts
);
@ -164,7 +163,7 @@ pub async fn ensure_llama_servers_running() -> Result<(), Box<dyn std::error::Er
}
if llm_ready && embedding_ready {
println!("🎉 All llama.cpp servers are ready and responding!");
info!("🎉 All llama.cpp servers are ready and responding!");
Ok(())
} else {
let mut error_msg = "❌ Servers failed to start within timeout:".to_string();
@ -279,7 +278,7 @@ pub async fn chat_completions_local(
.timeout(Duration::from_secs(120)) // 2 minute timeout
.build()
.map_err(|e| {
eprintln!("Error creating HTTP client: {}", e);
error!("Error creating HTTP client: {}", e);
actix_web::error::ErrorInternalServerError("Failed to create HTTP client")
})?;
@ -290,7 +289,7 @@ pub async fn chat_completions_local(
.send()
.await
.map_err(|e| {
eprintln!("Error calling llama.cpp server: {}", e);
error!("Error calling llama.cpp server: {}", e);
actix_web::error::ErrorInternalServerError("Failed to call llama.cpp server")
})?;
@ -298,7 +297,7 @@ pub async fn chat_completions_local(
if status.is_success() {
let llama_response: LlamaCppResponse = response.json().await.map_err(|e| {
eprintln!("Error parsing llama.cpp response: {}", e);
error!("Error parsing llama.cpp response: {}", e);
actix_web::error::ErrorInternalServerError("Failed to parse llama.cpp response")
})?;
@ -331,7 +330,7 @@ pub async fn chat_completions_local(
.await
.unwrap_or_else(|_| "Unknown error".to_string());
eprintln!("Llama.cpp server error ({}): {}", status, error_text);
error!("Llama.cpp server error ({}): {}", status, error_text);
let actix_status = actix_web::http::StatusCode::from_u16(status.as_u16())
.unwrap_or(actix_web::http::StatusCode::INTERNAL_SERVER_ERROR);
@ -352,7 +351,7 @@ pub struct EmbeddingRequest {
pub input: Vec<String>,
pub model: String,
#[serde(default)]
pub encoding_format: Option<String>,
pub _encoding_format: Option<String>,
}
// Custom deserializer to handle both string and array inputs
@ -432,7 +431,7 @@ struct LlamaCppEmbeddingRequest {
// FIXED: Handle the stupid nested array format
#[derive(Debug, Deserialize)]
struct LlamaCppEmbeddingResponseItem {
pub index: usize,
pub _index: usize,
pub embedding: Vec<Vec<f32>>, // This is the fucked up part - embedding is an array of arrays
}
@ -452,7 +451,7 @@ pub async fn embeddings_local(
.timeout(Duration::from_secs(120))
.build()
.map_err(|e| {
eprintln!("Error creating HTTP client: {}", e);
error!("Error creating HTTP client: {}", e);
actix_web::error::ErrorInternalServerError("Failed to create HTTP client")
})?;
@ -472,7 +471,7 @@ pub async fn embeddings_local(
.send()
.await
.map_err(|e| {
eprintln!("Error calling llama.cpp server for embedding: {}", e);
error!("Error calling llama.cpp server for embedding: {}", e);
actix_web::error::ErrorInternalServerError(
"Failed to call llama.cpp server for embedding",
)
@ -483,15 +482,15 @@ pub async fn embeddings_local(
if status.is_success() {
// First, get the raw response text for debugging
let raw_response = response.text().await.map_err(|e| {
eprintln!("Error reading response text: {}", e);
error!("Error reading response text: {}", e);
actix_web::error::ErrorInternalServerError("Failed to read response")
})?;
// Parse the response as a vector of items with nested arrays
let llama_response: Vec<LlamaCppEmbeddingResponseItem> =
serde_json::from_str(&raw_response).map_err(|e| {
eprintln!("Error parsing llama.cpp embedding response: {}", e);
eprintln!("Raw response: {}", raw_response);
error!("Error parsing llama.cpp embedding response: {}", e);
error!("Raw response: {}", raw_response);
actix_web::error::ErrorInternalServerError(
"Failed to parse llama.cpp embedding response",
)
@ -517,7 +516,7 @@ pub async fn embeddings_local(
index,
});
} else {
eprintln!("No embedding data returned for input: {}", input_text);
error!("No embedding data returned for input: {}", input_text);
return Ok(HttpResponse::InternalServerError().json(serde_json::json!({
"error": {
"message": format!("No embedding data returned for input {}", index),
@ -531,7 +530,7 @@ pub async fn embeddings_local(
.await
.unwrap_or_else(|_| "Unknown error".to_string());
eprintln!("Llama.cpp server error ({}): {}", status, error_text);
error!("Llama.cpp server error ({}): {}", status, error_text);
let actix_status = actix_web::http::StatusCode::from_u16(status.as_u16())
.unwrap_or(actix_web::http::StatusCode::INTERNAL_SERVER_ERROR);

12
src/services/llm_provider.rs
View file

@ -1,3 +1,5 @@
use log::info;
use actix_web::{post, web, HttpRequest, HttpResponse, Result};
use dotenv::dotenv;
use regex::Regex;
@ -38,9 +40,9 @@ struct Choice {
async fn chat_completions(body: web::Bytes, _req: HttpRequest) -> Result<HttpResponse> {
// Always log raw POST data
if let Ok(body_str) = std::str::from_utf8(&body) {
println!("POST Data: {}", body_str);
info!("POST Data: {}", body_str);
} else {
println!("POST Data (binary): {:?}", body);
info!("POST Data (binary): {:?}", body);
}
dotenv().ok();
@ -72,7 +74,7 @@ async fn chat_completions(body: web::Bytes, _req: HttpRequest) -> Result<HttpRes
);
let body_str = std::str::from_utf8(&body).unwrap_or("");
println!("Original POST Data: {}", body_str);
info!("Original POST Data: {}", body_str);
// Remove the problematic params
let re =
@ -80,7 +82,7 @@ async fn chat_completions(body: web::Bytes, _req: HttpRequest) -> Result<HttpRes
let cleaned = re.replace_all(body_str, "");
let cleaned_body = web::Bytes::from(cleaned.to_string());
println!("Cleaned POST Data: {}", cleaned);
info!("Cleaned POST Data: {}", cleaned);
// Send request to Azure
let client = Client::new();
@ -100,7 +102,7 @@ async fn chat_completions(body: web::Bytes, _req: HttpRequest) -> Result<HttpRes
.map_err(actix_web::error::ErrorInternalServerError)?;
// Log the raw response
println!("Raw Azure response: {}", raw_response);
info!("Raw Azure response: {}", raw_response);
if status.is_success() {
Ok(HttpResponse::Ok().body(raw_response))

3
src/services/script.rs
View file

@ -12,6 +12,7 @@ use crate::services::keywords::set::set_keyword;
use crate::services::keywords::set_schedule::set_schedule_keyword;
use crate::services::keywords::wait::wait_keyword;
use crate::services::state::AppState;
use log::info;
use rhai::{Dynamic, Engine, EvalAltResult};
pub struct ScriptService {
@ -134,7 +135,7 @@ impl ScriptService {
/// Preprocesses BASIC-style script to handle semicolon-free syntax
pub fn compile(&self, script: &str) -> Result<rhai::AST, Box<EvalAltResult>> {
let processed_script = self.preprocess_basic_script(script);
println!("Processed Script:\n{}", processed_script);
info!("Processed Script:\n{}", processed_script);
match self.engine.compile(&processed_script) {
Ok(ast) => Ok(ast),
Err(parse_error) => Err(Box::new(EvalAltResult::from(parse_error))),

5
src/services/utils.rs
View file

@ -1,6 +1,7 @@
use crate::services::config::AIConfig;
use langchain_rust::llm::OpenAI;
use langchain_rust::{language_models::llm::LLM, llm::AzureConfig};
use log::error;
use log::{debug, warn};
use rhai::{Array, Dynamic};
use serde_json::{json, Value};
@ -22,7 +23,7 @@ use tokio::io::AsyncWriteExt;
pub fn azure_from_config(config: &AIConfig) -> AzureConfig {
AzureConfig::new()
.with_api_base(&config.endpoint)
.with_api_base(&config.endpoint)
.with_api_key(&config.key)
.with_api_version(&config.version)
.with_deployment_id(&config.instance)
@ -42,7 +43,7 @@ pub async fn call_llm(
match open_ai.invoke(&prompt).await {
Ok(response_text) => Ok(response_text),
Err(err) => {
eprintln!("Error invoking LLM API: {}", err);
error!("Error invoking LLM API: {}", err);
Err(Box::new(std::io::Error::new(
std::io::ErrorKind::Other,
"Failed to invoke LLM API",

148
src/services/web_automation.rs
View file

@ -1,5 +1,6 @@
// wget https://dl.google.com/linux/direct/google-chrome-stable_current_amd64.deb
// sudo dpkg -i google-chrome-stable_current_amd64.deb
use log::info;
use crate::services::utils;
@ -93,29 +94,26 @@ impl BrowserSetup {
Err("Brave browser not found. Please install Brave first.".into())
}
async fn setup_chromedriver() -> Result<String, Box<dyn std::error::Error>> {
// Create chromedriver directory in executable's parent directory
let mut chromedriver_dir = env::current_exe()?
.parent()
.unwrap()
.to_path_buf();
chromedriver_dir.push("chromedriver");
// Ensure the directory exists
if !chromedriver_dir.exists() {
fs::create_dir(&chromedriver_dir).await?;
}
async fn setup_chromedriver() -> Result<String, Box<dyn std::error::Error>> {
// Create chromedriver directory in executable's parent directory
let mut chromedriver_dir = env::current_exe()?.parent().unwrap().to_path_buf();
chromedriver_dir.push("chromedriver");
// Determine the final chromedriver path
let chromedriver_path = if cfg!(target_os = "windows") {
chromedriver_dir.join("chromedriver.exe")
} else {
chromedriver_dir.join("chromedriver")
};
// Ensure the directory exists
if !chromedriver_dir.exists() {
fs::create_dir(&chromedriver_dir).await?;
}
// Check if chromedriver exists
if fs::metadata(&chromedriver_path).await.is_err() {
let (download_url, platform) = match (cfg!(target_os = "windows"), cfg!(target_arch = "x86_64")) {
// Determine the final chromedriver path
let chromedriver_path = if cfg!(target_os = "windows") {
chromedriver_dir.join("chromedriver.exe")
} else {
chromedriver_dir.join("chromedriver")
};
// Check if chromedriver exists
if fs::metadata(&chromedriver_path).await.is_err() {
let (download_url, platform) = match (cfg!(target_os = "windows"), cfg!(target_arch = "x86_64")) {
(true, true) => (
"https://storage.googleapis.com/chrome-for-testing-public/138.0.7204.183/win64/chromedriver-win64.zip",
"win64",
@ -138,69 +136,69 @@ async fn setup_chromedriver() -> Result<String, Box<dyn std::error::Error>> {
),
_ => return Err("Unsupported platform".into()),
};
let mut zip_path = std::env::temp_dir();
zip_path.push("chromedriver.zip");
println!("Downloading chromedriver for {}...", platform);
// Download the zip file
utils::download_file(download_url, &zip_path.to_str().unwrap()).await?;
let mut zip_path = std::env::temp_dir();
zip_path.push("chromedriver.zip");
info!("Downloading chromedriver for {}...", platform);
// Extract the zip to a temporary directory first
let mut temp_extract_dir = std::env::temp_dir();
temp_extract_dir.push("chromedriver_extract");
let temp_extract_dir1 = temp_extract_dir.clone();
// Clean up any previous extraction
let _ = fs::remove_dir_all(&temp_extract_dir).await;
fs::create_dir(&temp_extract_dir).await?;
// Download the zip file
utils::download_file(download_url, &zip_path.to_str().unwrap()).await?;
utils::extract_zip_recursive(&zip_path, &temp_extract_dir)?;
// Extract the zip to a temporary directory first
let mut temp_extract_dir = std::env::temp_dir();
temp_extract_dir.push("chromedriver_extract");
let temp_extract_dir1 = temp_extract_dir.clone();
// Chrome for Testing zips contain a platform-specific directory
// Find the chromedriver binary in the extracted structure
let mut extracted_binary_path = temp_extract_dir;
extracted_binary_path.push(format!("chromedriver-{}", platform));
extracted_binary_path.push(if cfg!(target_os = "windows") {
"chromedriver.exe"
} else {
"chromedriver"
});
// Clean up any previous extraction
let _ = fs::remove_dir_all(&temp_extract_dir).await;
fs::create_dir(&temp_extract_dir).await?;
// Try to move the file, fall back to copy if cross-device
match fs::rename(&extracted_binary_path, &chromedriver_path).await {
Ok(_) => (),
Err(e) if e.kind() == std::io::ErrorKind::CrossesDevices => {
// Cross-device move failed, use copy instead
fs::copy(&extracted_binary_path, &chromedriver_path).await?;
// Set permissions on the copied file
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&chromedriver_path).await?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&chromedriver_path, perms).await?;
utils::extract_zip_recursive(&zip_path, &temp_extract_dir)?;
// Chrome for Testing zips contain a platform-specific directory
// Find the chromedriver binary in the extracted structure
let mut extracted_binary_path = temp_extract_dir;
extracted_binary_path.push(format!("chromedriver-{}", platform));
extracted_binary_path.push(if cfg!(target_os = "windows") {
"chromedriver.exe"
} else {
"chromedriver"
});
// Try to move the file, fall back to copy if cross-device
match fs::rename(&extracted_binary_path, &chromedriver_path).await {
Ok(_) => (),
Err(e) if e.kind() == std::io::ErrorKind::CrossesDevices => {
// Cross-device move failed, use copy instead
fs::copy(&extracted_binary_path, &chromedriver_path).await?;
// Set permissions on the copied file
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&chromedriver_path).await?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&chromedriver_path, perms).await?;
}
}
},
Err(e) => return Err(e.into()),
Err(e) => return Err(e.into()),
}
// Clean up
let _ = fs::remove_file(&zip_path).await;
let _ = fs::remove_dir_all(temp_extract_dir1).await;
// Set executable permissions (if not already set during copy)
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&chromedriver_path).await?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&chromedriver_path, perms).await?;
}
}
// Clean up
let _ = fs::remove_file(&zip_path).await;
let _ = fs::remove_dir_all(temp_extract_dir1).await;
// Set executable permissions (if not already set during copy)
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&chromedriver_path).await?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&chromedriver_path, perms).await?;
}
Ok(chromedriver_path.to_string_lossy().to_string())
}
Ok(chromedriver_path.to_string_lossy().to_string())
}
}
// Modified BrowserPool initialization

7
src/tests/integration_file_list_test.rs
View file

@ -16,7 +16,6 @@ use std::str::FromStr;
use tempfile::NamedTempFile;
use tokio_stream::StreamExt;
#[tokio::test]
async fn test_successful_file_listing() -> Result<(), Box<dyn std::error::Error>> {
@ -61,7 +60,7 @@ async fn test_successful_file_listing() -> Result<(), Box<dyn std::error::Error>
let app_state = web::Data::new(AppState {
minio_client: Some(minio_client.clone()),
config: None,
db_pool: None
db_pool: None,
});
let app = test::init_service(App::new().app_data(app_state.clone()).service(list_file)).await;
@ -97,10 +96,10 @@ async fn test_successful_file_listing() -> Result<(), Box<dyn std::error::Error>
match result {
Ok(resp) => {
for item in resp.contents {
println!("{:?}", item);
info!("{:?}", item);
}
}
Err(e) => println!("Error: {:?}", e),
Err(e) => info!("Error: {:?}", e),
}
}