From 8721353b8dc5d398f27bcf0d045ecc000241287c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BB=D0=B5=D0=BA=D1=81=D0=B5=D0=B9=20=D0=91=D0=B0?= =?UTF-8?q?=D1=80=D0=B0=D0=B1=D0=B0=D0=BD=D0=BE=D0=B2?= Date: Fri, 17 Apr 2026 11:58:16 +0300 Subject: [PATCH] v0.2.0 --- Cargo.lock | 229 +++++++++++++++++++- Cargo.toml | 9 +- src/main.rs | 484 ++---------------------------------------- src/processor.rs | 434 +++++++++++++++++++++++++++++++++++++ src/static/index.html | 214 +++++++++++++++++++ src/web.rs | 186 ++++++++++++++++ 6 files changed, 1093 insertions(+), 463 deletions(-) create mode 100644 src/processor.rs create mode 100644 src/static/index.html create mode 100644 src/web.rs diff --git a/Cargo.lock b/Cargo.lock index 202edea..53222ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,12 +10,15 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" [[package]] name = "ae_anons" -version = "0.1.1" +version = "0.2.0" dependencies = [ "anyhow", + "askama", + "axum", "bytes", "calamine", "chrono", + "clap", "dotenv", "env_logger", "futures", @@ -27,6 +30,8 @@ dependencies = [ "serde_json", "thiserror", "tokio", + "tower", + "tower-http", "urlencoding", ] @@ -113,6 +118,58 @@ dependencies = [ "derive_arbitrary", ] +[[package]] +name = "askama" +version = "0.15.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b8246bcbf8eb97abef10c2d92166449680d41d55c0fc6978a91dec2e3619608" +dependencies = [ + "askama_macros", + "itoa", + "percent-encoding", + "serde", + "serde_json", +] + +[[package]] +name = "askama_derive" +version = "0.15.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f9670bc84a28bb3da91821ef74226949ab63f1265aff7c751634f1dd0e6f97c" +dependencies = [ + "askama_parser", + "basic-toml", + "memchr", + "proc-macro2", + "quote", + "rustc-hash", + "serde", + "serde_derive", + "syn", +] + +[[package]] +name = "askama_macros" +version = "0.15.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0756b45480437dded0565dfc568af62ccce146fb6cfe902e808ba86e445f44f" +dependencies = [ + "askama_derive", +] + +[[package]] +name = "askama_parser" +version = "0.15.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d0af3691ba3af77949c0b5a3925444b85cb58a0184cc7fec16c68ba2e7be868" +dependencies = [ + "rustc-hash", + "serde", + "serde_derive", + "unicode-ident", + "winnow", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -125,12 +182,73 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "base64" version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "basic-toml" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba62675e8242a4c4e806d12f11d136e626e6c8361d6b829310732241652a178a" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "2.11.0" @@ -199,6 +317,46 @@ dependencies = [ "windows-link", ] +[[package]] +name = "clap" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + [[package]] name = "codepage" version = "0.1.2" @@ -519,6 +677,12 @@ version = "0.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + [[package]] name = "http" version = "1.4.0" @@ -552,12 +716,24 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "http-range-header" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c" + [[package]] name = "httparse" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hyper" version = "1.9.0" @@ -572,6 +748,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -869,6 +1046,12 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.0" @@ -1199,6 +1382,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustc-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" + [[package]] name = "rustix" version = "1.1.4" @@ -1338,6 +1527,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -1400,6 +1600,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" @@ -1575,6 +1781,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -1585,14 +1792,24 @@ checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8" dependencies = [ "bitflags", "bytes", + "futures-core", "futures-util", "http", "http-body", + "http-body-util", + "http-range-header", + "httpdate", "iri-string", + "mime", + "mime_guess", + "percent-encoding", "pin-project-lite", + "tokio", + "tokio-util", "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -1613,6 +1830,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-core", ] @@ -1940,6 +2158,15 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09dac053f1cd375980747450bfc7250c264eaae0583872e845c0c7cd578872b5" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.51.0" diff --git a/Cargo.toml b/Cargo.toml index 5c7857d..62118ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ae_anons" -version = "0.1.1" +version = "0.2.0" edition = "2021" license = "MIT" license-file = "LICENSE" @@ -28,6 +28,13 @@ log = "0.4" env_logger = "0.11" bytes = "1.9" +# Web server (updated to latest stable versions) +axum = "0.8.9" +tower = "0.5" +tower-http = { version = "0.6", features = ["fs", "trace"] } +askama = "0.15" +clap = { version = "4.5", features = ["derive"] } + [dev-dependencies] tokio = { version = "1.0", features = ["full", "rt-multi-thread"] } diff --git a/src/main.rs b/src/main.rs index 3a1fbd5..f78a737 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,480 +1,42 @@ mod config; mod nexrender; mod synology; +mod processor; +mod web; -use anyhow::{anyhow, Result}; -use calamine::{Data, Reader, Xlsx}; -use chrono::{Duration, NaiveDate}; +use anyhow::Result; +use clap::Parser; use config::Config; -use log::{debug, error, info}; -use nexrender::{JobData, LogoRegistry}; -use reqwest::Client; -use serde_json::Value; -use std::collections::HashMap; -use std::io::Cursor; -use std::path::Path; -use std::time::Duration as StdDuration; -use synology::SynologyClient; -use tokio::time::sleep; +use log::info; -// Структуры для in-memory данных -#[derive(Debug, Clone)] -pub struct SheetData { - pub name: String, - pub headers: Vec, - pub rows: Vec>, -} +#[derive(Parser)] +#[command(author, version, about, long_about = None)] +struct Cli { + /// Run web server instead of one-time processing + #[arg(short, long)] + web: bool, -#[derive(Debug, Default)] -pub struct ExcelWorkbook { - pub sheets: Vec, + /// Run one-time processing (default if no flags) + #[arg(short, long)] + once: bool, } #[tokio::main] async fn main() -> Result<()> { env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); - info!("Starting AE Anons processor v0.1.1 (in-memory mode)"); + let cli = Cli::parse(); let config = Config::from_env()?; - debug!("Configuration loaded"); - let mut client = SynologyClient::new(&config.nas_fqdn); - client.login(&config.nas_user, &config.nas_pass).await?; - info!("Successfully authenticated with Synology NAS"); - - let info = client.get_info().await?; - info!("Connected to NAS: {}", info.hostname); - - // Скачиваем и парсим Excel в памяти - let workbook = download_and_parse_excel_in_memory(&mut client, &config).await?; - display_workbook_structure(&workbook); - - process_nexrender_jobs(&workbook, &config).await?; - - client.logout().await?; - info!("Session terminated successfully"); - - Ok(()) -} - -async fn download_and_parse_excel_in_memory( - client: &mut SynologyClient, - config: &Config, -) -> Result { - let file_name_full = Path::new(&config.nas_file) - .file_name() - .and_then(|n| n.to_str()) - .unwrap_or("unknown"); - let search_name = file_name_full.replace(".osheet", ""); - let expected_path = Path::new(&config.nas_file) - .parent() - .and_then(|p| p.to_str()) - .unwrap_or(""); - - let api_format_path = expected_path - .replace("Team Folder", "team-folders") - .replace(' ', ""); - - let full_expected_path_api = format!("{}/{}", api_format_path, file_name_full); - info!("Searching for file: {}", full_expected_path_api); - - let search_result = client.search_file_by_name(&search_name).await?; - let items = search_result["data"]["items"] - .as_array() - .ok_or_else(|| anyhow!("No items in search result"))?; - - if items.is_empty() { - return Err(anyhow!("File '{}' not found", file_name_full)); - } - - let exact_match: Vec<&Value> = items - .iter() - .filter(|item| { - item.get("display_path") - .and_then(|p| p.as_str()) - .unwrap_or("") - == full_expected_path_api - }) - .collect(); - - if exact_match.is_empty() { - return Err(anyhow!("File not found at: {}", full_expected_path_api)); - } - - let file = exact_match[0]; - let file_id = file["file_id"] - .as_str() - .ok_or_else(|| anyhow!("Missing file_id"))?; - let actual_file_name = file["name"] - .as_str() - .ok_or_else(|| anyhow!("Missing name"))?; - - info!("Found file: {}", actual_file_name); - info!("Exporting file from Synology Office to Excel format (in-memory)..."); - - // Получаем бинарные данные Excel напрямую в память - let excel_data = client.export_by_file_id(file_id, actual_file_name).await?; - info!("Exported {} bytes to memory", excel_data.len()); - - info!("Parsing Excel workbook from memory..."); - parse_excel_from_bytes(&excel_data) -} - -fn parse_excel_from_bytes(data: &[u8]) -> Result { - let cursor = Cursor::new(data); - let mut workbook: Xlsx<_> = calamine::open_workbook_from_rs(cursor) - .map_err(|e| anyhow!("Failed to open workbook from memory: {}", e))?; - - let sheet_names = workbook.sheet_names().to_vec(); - let mut excel_workbook = ExcelWorkbook::default(); - - for sheet_name in sheet_names { - debug!("Processing sheet: '{}'", sheet_name); - - let range = workbook - .worksheet_range(&sheet_name) - .map_err(|e| anyhow!("Failed to read sheet '{}': {}", sheet_name, e))?; - - let sheet_data = parse_sheet_dynamic_optimized(&sheet_name, &range)?; - excel_workbook.sheets.push(sheet_data); - } - - Ok(excel_workbook) -} - -fn parse_sheet_dynamic_optimized( - sheet_name: &str, - range: &calamine::Range, -) -> Result { - // Предварительно выделяем память для избежания реаллокаций - let (row_count, col_count) = range.get_size(); - let mut data_matrix: Vec> = Vec::with_capacity(row_count); - - // Строим матрицу данных (без enumerate) - for row in range.rows() { - let mut row_data = Vec::with_capacity(col_count); - for cell in row { - row_data.push(cell_to_string_optimized(cell)); - } - data_matrix.push(row_data); - } - if data_matrix.is_empty() { - return Ok(SheetData { - name: sheet_name.to_string(), - headers: Vec::new(), - rows: Vec::new(), - }); - } - - // Извлекаем заголовки с дедупликацией - let headers: Vec = data_matrix[0] - .iter() - .enumerate() - .map(|(idx, header)| { - let h = header.trim().to_string(); - if h.is_empty() { - format!("Column_{}", idx + 1) - } else { - h - } - }) - .collect(); - - // Парсим строки данных - let rows_data: Vec> = data_matrix - .iter() - .skip(1) - .filter_map(|row_values| { - let mut row_map = HashMap::with_capacity(headers.len()); - let mut has_data = false; - - for (col_idx, value) in row_values.iter().enumerate() { - if col_idx < headers.len() && !value.is_empty() { - row_map.insert(headers[col_idx].clone(), value.clone()); - has_data = true; - } - } - - if has_data { - Some(row_map) - } else { - None - } - }) - .take(10000) // Ограничение для безопасности - .collect(); - - Ok(SheetData { - name: sheet_name.to_string(), - headers, - rows: rows_data, - }) -} - -#[inline] -fn cell_to_string_optimized(cell: &Data) -> String { - match cell { - Data::Empty => String::new(), - Data::String(s) => s.clone(), - Data::Float(f) => { - if is_excel_date(*f) { - excel_date_to_string(*f) - } else if f.fract() == 0.0 { - // Используем itoa для целых чисел (опционально) - format!("{:.0}", f) - } else { - // Используем ryu для float (опционально) - f.to_string() - } - } - Data::Int(i) => { - let f = *i as f64; - if is_excel_date(f) { - excel_date_to_string(f) - } else { - i.to_string() - } - } - Data::Bool(b) => { - if *b { - "TRUE".to_string() - } else { - "FALSE".to_string() - } - } - Data::DateTime(dt) => { - let serial = dt.as_f64(); - if is_excel_date(serial) { - excel_date_to_string(serial) - } else { - dt.to_string() - } - } - Data::DateTimeIso(s) => s.clone(), - Data::DurationIso(s) => s.clone(), - Data::Error(e) => format!("{:?}", e), - } -} - -#[inline] -fn is_excel_date(value: f64) -> bool { - (1.0..100000.0).contains(&value) -} - -fn excel_date_to_string(serial: f64) -> String { - let days = serial as i64; - let base = NaiveDate::from_ymd_opt(1899, 12, 30).unwrap(); - if let Some(date) = base.checked_add_signed(Duration::days(days)) { - // Используем метод format напрямую - он публичный - date.format("%d.%m.%Y").to_string() + if cli.web { + info!("Starting AE Anons web server v{}", env!("CARGO_PKG_VERSION")); + web::run_web_server(config).await?; } else { - serial.to_string() - } -} - -fn display_workbook_structure(workbook: &ExcelWorkbook) { - info!("Workbook contains {} sheets", workbook.sheets.len()); - - for (idx, sheet) in workbook.sheets.iter().enumerate() { - debug!( - "Sheet #{}: '{}' | Headers: {} | Rows: {}", - idx + 1, - sheet.name, - sheet.headers.len(), - sheet.rows.len() - ); - } -} - -async fn process_nexrender_jobs(workbook: &ExcelWorkbook, config: &Config) -> Result<()> { - info!("Preparing Nexrender jobs..."); - - let start_sheet = workbook - .get_sheet("Start") - .ok_or_else(|| anyhow!("Sheet 'Start' not found"))?; - let sport_sheet = workbook - .get_sheet("SPORT") - .ok_or_else(|| anyhow!("Sheet 'SPORT' not found"))?; - let teams_sheet = workbook - .get_sheet("TEAMS") - .ok_or_else(|| anyhow!("Sheet 'TEAMS' not found"))?; - let channel_sheet = workbook - .get_sheet("CHANELL") - .ok_or_else(|| anyhow!("Sheet 'CHANELL' not found"))?; - - // Используем предварительное выделение памяти - let packs: HashMap = sport_sheet - .rows - .iter() - .filter_map(|row| Some((row.get("SPORT")?.clone(), row.get("LINK")?.clone()))) - .collect(); - info!("Loaded {} sport packs", packs.len()); - - let mut logos = LogoRegistry::with_capacity(teams_sheet.rows.len()); - for row in &teams_sheet.rows { - if let (Some(team), Some(sport), Some(link)) = - (row.get("TEAM"), row.get("SPORT"), row.get("LINK")) - { - logos.insert(team.clone(), sport.clone(), link.clone()); - } - } - info!("Loaded {} team logos", teams_sheet.rows.len()); - - let channels: HashMap = channel_sheet - .rows - .iter() - .filter_map(|row| Some((row.get("CHANELL")?.clone(), row.get("LINK")?.clone()))) - .collect(); - info!("Loaded {} channel logos", channels.len()); - - // Предварительно выделяем память для jobs - let mut jobs: Vec = Vec::with_capacity(start_sheet.rows.len() * 3); - - for (idx, row) in start_sheet.rows.iter().enumerate() { - if let Some(state) = row.get("STATE") { - if state == "FALSE" { - if let Some(job) = JobData::from_row(row, idx, &packs, &logos, &channels) { - jobs.extend(job.create_variants()); - } - } - } - } - - info!("Generated {} total jobs (including variants)", jobs.len()); - - if jobs.is_empty() { - info!("No jobs with STATE='FALSE' found"); - return Ok(()); - } - - cleanup_finished_jobs(&config.nexrender_api_url).await?; - - let http_client = Client::new(); - let mut submitted_jobs: Vec<(String, String)> = Vec::with_capacity(jobs.len()); - - // Отправляем jobs параллельно для ускорения - let mut tasks = Vec::with_capacity(jobs.len()); - for job in jobs { - let client = http_client.clone(); - let api_url = config.nexrender_api_url.clone(); - let config_clone = config.clone(); // ✅ Клонируем весь конфиг - - tasks.push(tokio::spawn(async move { - let nexrender_job = job.to_nexrender_job(&config_clone); // ✅ Передаём Config - info!("Submitting job: {}", job.outfile_name); - - let response = client.post(&api_url).json(&nexrender_job).send().await?; - - if response.status().is_success() { - let result: Value = response.json().await?; - if let Some(uid) = result.get("uid").and_then(|u| u.as_str()) { - Ok::<_, anyhow::Error>(Some((uid.to_string(), job.outfile_name))) - } else { - Ok(None) - } - } else { - let status = response.status(); - let text = response.text().await.unwrap_or_default(); - error!("Failed to submit job ({}): {}", status, text); - Ok(None) - } - })); - } - - for task in tasks { - match task.await { - Ok(Ok(Some((uid, outfile_name)))) => { - info!("Job submitted: {}", outfile_name); - submitted_jobs.push((uid, outfile_name)); - } - Ok(Ok(None)) => { - debug!("Job submission returned no UID"); - } - Ok(Err(e)) => { - error!("Job submission error: {}", e); - } - Err(e) => { - error!("Task join error: {}", e); - } - } - } - - info!( - "Successfully submitted {} jobs to Nexrender", - submitted_jobs.len() - ); - - if !submitted_jobs.is_empty() { - info!("Monitoring job completion..."); - monitor_jobs(&http_client, &config.nexrender_api_url, submitted_jobs).await?; + // По умолчанию или с флагом --once выполняем однократную обработку + info!("Starting AE Anons processor v{} (one-time mode)", env!("CARGO_PKG_VERSION")); + let submitted = processor::process_spreadsheet(&config).await?; + info!("Submitted {} jobs. Exiting.", submitted.len()); } Ok(()) -} - -// ... остальные функции без изменений ... -impl ExcelWorkbook { - pub fn get_sheet(&self, name: &str) -> Option<&SheetData> { - self.sheets.iter().find(|s| s.name == name) - } -} - -async fn cleanup_finished_jobs(api_url: &str) -> Result<()> { - let client = Client::new(); - let response = client.get(api_url).send().await?; - - if response.status().is_success() { - let jobs: Vec = response.json().await?; - for job in jobs { - if let (Some(uid), Some(state)) = ( - job.get("uid").and_then(|u| u.as_str()), - job.get("state").and_then(|s| s.as_str()), - ) { - if state == "finished" || state == "error" { - let _ = client.delete(&format!("{}/{}", api_url, uid)).send().await; - info!("Cleaned up completed job: {}", uid); - } - } - } - } - Ok(()) -} - -async fn monitor_jobs( - client: &Client, - api_url: &str, - mut pending: Vec<(String, String)>, -) -> Result<()> { - while !pending.is_empty() { - sleep(StdDuration::from_secs(25)).await; - - let mut remaining = Vec::with_capacity(pending.len()); - for (uid, outname) in pending.drain(..) { - let response = client.get(&format!("{}/{}", api_url, uid)).send().await?; - - if response.status().is_success() { - let job: Value = response.json().await?; - let state = job - .get("state") - .and_then(|s| s.as_str()) - .unwrap_or("unknown"); - - match state { - "finished" => { - info!("Job completed: {}", outname) - } - "error" => error!("Job failed: {}", outname), - _ => remaining.push((uid, outname)), - } - } else { - remaining.push((uid, outname)); - } - } - pending = remaining; - if !pending.is_empty() { - info!("Waiting for {} jobs to complete...", pending.len()); - } - } - info!("All jobs completed successfully"); - Ok(()) -} +} \ No newline at end of file diff --git a/src/processor.rs b/src/processor.rs new file mode 100644 index 0000000..a3115d2 --- /dev/null +++ b/src/processor.rs @@ -0,0 +1,434 @@ +use crate::config::Config; +use crate::nexrender::{JobData, LogoRegistry}; +use crate::synology::SynologyClient; +use anyhow::{anyhow, Result}; +use calamine::{Data, Reader, Xlsx}; +use chrono::{Duration, NaiveDate}; +use log::{debug, error, info}; +use reqwest::Client; +use serde_json::Value; +use std::collections::HashMap; +use std::io::Cursor; +use std::path::Path; + +// Добавить после импортов в processor.rs + +#[derive(Debug, Clone)] +pub struct SheetData { + pub name: String, + pub headers: Vec, + pub rows: Vec>, +} + +#[derive(Debug, Default)] +pub struct ExcelWorkbook { + pub sheets: Vec, +} + +impl ExcelWorkbook { + pub fn get_sheet(&self, name: &str) -> Option<&SheetData> { + self.sheets.iter().find(|s| s.name == name) + } +} + +/// Основная функция обработки: скачивает Excel, генерирует задания, отправляет в Nexrender +pub async fn process_spreadsheet(config: &Config) -> Result> { + let mut client = SynologyClient::new(&config.nas_fqdn); + client.login(&config.nas_user, &config.nas_pass).await?; + info!("Successfully authenticated with Synology NAS"); + + let info = client.get_info().await?; + info!("Connected to NAS: {}", info.hostname); + + // Скачиваем и парсим Excel в памяти + let workbook = download_and_parse_excel_in_memory(&mut client, config).await?; + display_workbook_structure(&workbook); + + let submitted = generate_and_submit_jobs(&workbook, config).await?; + + client.logout().await?; + info!("Session terminated successfully"); + + Ok(submitted) +} + +async fn download_and_parse_excel_in_memory( + client: &mut SynologyClient, + config: &Config, +) -> Result { + let file_name_full = Path::new(&config.nas_file) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("unknown"); + let search_name = file_name_full.replace(".osheet", ""); + let expected_path = Path::new(&config.nas_file) + .parent() + .and_then(|p| p.to_str()) + .unwrap_or(""); + + let api_format_path = expected_path + .replace("Team Folder", "team-folders") + .replace(' ', ""); + + let full_expected_path_api = format!("{}/{}", api_format_path, file_name_full); + info!("Searching for file: {}", full_expected_path_api); + + let search_result = client.search_file_by_name(&search_name).await?; + let items = search_result["data"]["items"] + .as_array() + .ok_or_else(|| anyhow!("No items in search result"))?; + + if items.is_empty() { + return Err(anyhow!("File '{}' not found", file_name_full)); + } + + let exact_match: Vec<&Value> = items + .iter() + .filter(|item| { + item.get("display_path") + .and_then(|p| p.as_str()) + .unwrap_or("") + == full_expected_path_api + }) + .collect(); + + if exact_match.is_empty() { + return Err(anyhow!("File not found at: {}", full_expected_path_api)); + } + + let file = exact_match[0]; + let file_id = file["file_id"] + .as_str() + .ok_or_else(|| anyhow!("Missing file_id"))?; + let actual_file_name = file["name"] + .as_str() + .ok_or_else(|| anyhow!("Missing name"))?; + + info!("Found file: {}", actual_file_name); + info!("Exporting file from Synology Office to Excel format (in-memory)..."); + + // Получаем бинарные данные Excel напрямую в память + let excel_data = client.export_by_file_id(file_id, actual_file_name).await?; + info!("Exported {} bytes to memory", excel_data.len()); + + info!("Parsing Excel workbook from memory..."); + parse_excel_from_bytes(&excel_data) +} + +fn parse_excel_from_bytes(data: &[u8]) -> Result { + let cursor = Cursor::new(data); + let mut workbook: Xlsx<_> = calamine::open_workbook_from_rs(cursor) + .map_err(|e| anyhow!("Failed to open workbook from memory: {}", e))?; + + let sheet_names = workbook.sheet_names().to_vec(); + let mut excel_workbook = ExcelWorkbook::default(); + + for sheet_name in sheet_names { + debug!("Processing sheet: '{}'", sheet_name); + + let range = workbook + .worksheet_range(&sheet_name) + .map_err(|e| anyhow!("Failed to read sheet '{}': {}", sheet_name, e))?; + + let sheet_data = parse_sheet_dynamic_optimized(&sheet_name, &range)?; + excel_workbook.sheets.push(sheet_data); + } + + Ok(excel_workbook) +} + +fn parse_sheet_dynamic_optimized( + sheet_name: &str, + range: &calamine::Range, +) -> Result { + let (row_count, col_count) = range.get_size(); + let mut data_matrix: Vec> = Vec::with_capacity(row_count); + + for row in range.rows() { + let mut row_data = Vec::with_capacity(col_count); + for cell in row { + row_data.push(cell_to_string_optimized(cell)); + } + data_matrix.push(row_data); + } + + if data_matrix.is_empty() { + return Ok(SheetData { + name: sheet_name.to_string(), + headers: Vec::new(), + rows: Vec::new(), + }); + } + + // Извлекаем заголовки с дедупликацией + let headers: Vec = data_matrix[0] + .iter() + .enumerate() + .map(|(idx, header)| { + let h = header.trim().to_string(); + if h.is_empty() { + format!("Column_{}", idx + 1) + } else { + h + } + }) + .collect(); + + // Парсим строки данных + let rows_data: Vec> = data_matrix + .iter() + .skip(1) + .filter_map(|row_values| { + let mut row_map = HashMap::with_capacity(headers.len()); + let mut has_data = false; + + for (col_idx, value) in row_values.iter().enumerate() { + if col_idx < headers.len() && !value.is_empty() { + row_map.insert(headers[col_idx].clone(), value.clone()); + has_data = true; + } + } + + if has_data { + Some(row_map) + } else { + None + } + }) + .take(10000) // Ограничение для безопасности + .collect(); + + Ok(SheetData { + name: sheet_name.to_string(), + headers, + rows: rows_data, + }) +} + +#[inline] +fn cell_to_string_optimized(cell: &Data) -> String { + match cell { + Data::Empty => String::new(), + Data::String(s) => s.clone(), + Data::Float(f) => { + if is_excel_date(*f) { + excel_date_to_string(*f) + } else if f.fract() == 0.0 { + // Используем itoa для целых чисел (опционально) + format!("{:.0}", f) + } else { + // Используем ryu для float (опционально) + f.to_string() + } + } + Data::Int(i) => { + let f = *i as f64; + if is_excel_date(f) { + excel_date_to_string(f) + } else { + i.to_string() + } + } + Data::Bool(b) => { + if *b { + "TRUE".to_string() + } else { + "FALSE".to_string() + } + } + Data::DateTime(dt) => { + let serial = dt.as_f64(); + if is_excel_date(serial) { + excel_date_to_string(serial) + } else { + dt.to_string() + } + } + Data::DateTimeIso(s) => s.clone(), + Data::DurationIso(s) => s.clone(), + Data::Error(e) => format!("{:?}", e), + } +} + +#[inline] +fn is_excel_date(value: f64) -> bool { + (1.0..100000.0).contains(&value) +} + +fn excel_date_to_string(serial: f64) -> String { + let days = serial as i64; + let base = NaiveDate::from_ymd_opt(1899, 12, 30).unwrap(); + if let Some(date) = base.checked_add_signed(Duration::days(days)) { + // Используем метод format напрямую - он публичный + date.format("%d.%m.%Y").to_string() + } else { + serial.to_string() + } +} + +fn display_workbook_structure(workbook: &ExcelWorkbook) { + info!("Workbook contains {} sheets", workbook.sheets.len()); + + for (idx, sheet) in workbook.sheets.iter().enumerate() { + debug!( + "Sheet #{}: '{}' | Headers: {} | Rows: {}", + idx + 1, + sheet.name, + sheet.headers.len(), + sheet.rows.len() + ); + } +} + +async fn generate_and_submit_jobs( + workbook: &ExcelWorkbook, + config: &Config, +) -> Result> { + info!("Preparing Nexrender jobs..."); + + let start_sheet = workbook + .get_sheet("Start") + .ok_or_else(|| anyhow!("Sheet 'Start' not found"))?; + let sport_sheet = workbook + .get_sheet("SPORT") + .ok_or_else(|| anyhow!("Sheet 'SPORT' not found"))?; + let teams_sheet = workbook + .get_sheet("TEAMS") + .ok_or_else(|| anyhow!("Sheet 'TEAMS' not found"))?; + let channel_sheet = workbook + .get_sheet("CHANELL") + .ok_or_else(|| anyhow!("Sheet 'CHANELL' not found"))?; + + let packs: HashMap = sport_sheet + .rows + .iter() + .filter_map(|row| Some((row.get("SPORT")?.clone(), row.get("LINK")?.clone()))) + .collect(); + info!("Loaded {} sport packs", packs.len()); + + let mut logos = LogoRegistry::with_capacity(teams_sheet.rows.len()); + for row in &teams_sheet.rows { + if let (Some(team), Some(sport), Some(link)) = + (row.get("TEAM"), row.get("SPORT"), row.get("LINK")) + { + logos.insert(team.clone(), sport.clone(), link.clone()); + } + } + info!("Loaded {} team logos", teams_sheet.rows.len()); + + let channels: HashMap = channel_sheet + .rows + .iter() + .filter_map(|row| Some((row.get("CHANELL")?.clone(), row.get("LINK")?.clone()))) + .collect(); + info!("Loaded {} channel logos", channels.len()); + + let mut jobs: Vec = Vec::with_capacity(start_sheet.rows.len() * 3); + + for (idx, row) in start_sheet.rows.iter().enumerate() { + if let Some(state) = row.get("STATE") { + if state == "FALSE" { + if let Some(job) = JobData::from_row(row, idx, &packs, &logos, &channels) { + jobs.extend(job.create_variants()); + } + } + } + } + + info!("Generated {} total jobs (including variants)", jobs.len()); + + if jobs.is_empty() { + info!("No jobs with STATE='FALSE' found"); + return Ok(Vec::new()); + } + + cleanup_finished_jobs(&config.nexrender_api_url).await?; + + let http_client = Client::new(); + let mut submitted_jobs: Vec<(String, String)> = Vec::with_capacity(jobs.len()); + + let mut tasks = Vec::with_capacity(jobs.len()); + for job in jobs { + let client = http_client.clone(); + let api_url = config.nexrender_api_url.clone(); + let config_clone = config.clone(); + + tasks.push(tokio::spawn(async move { + let nexrender_job = job.to_nexrender_job(&config_clone); + info!("Submitting job: {}", job.outfile_name); + + let response = client.post(&api_url).json(&nexrender_job).send().await?; + + if response.status().is_success() { + let result: Value = response.json().await?; + if let Some(uid) = result.get("uid").and_then(|u| u.as_str()) { + Ok::<_, anyhow::Error>(Some((uid.to_string(), job.outfile_name))) + } else { + Ok(None) + } + } else { + let status = response.status(); + let text = response.text().await.unwrap_or_default(); + error!("Failed to submit job ({}): {}", status, text); + Ok(None) + } + })); + } + + for task in tasks { + match task.await { + Ok(Ok(Some((uid, outfile_name)))) => { + info!("Job submitted: {}", outfile_name); + submitted_jobs.push((uid, outfile_name)); + } + Ok(Ok(None)) => { + debug!("Job submission returned no UID"); + } + Ok(Err(e)) => { + error!("Job submission error: {}", e); + } + Err(e) => { + error!("Task join error: {}", e); + } + } + } + + info!( + "Successfully submitted {} jobs to Nexrender", + submitted_jobs.len() + ); + + Ok(submitted_jobs) +} + +pub async fn cleanup_finished_jobs(api_url: &str) -> Result<()> { + let client = Client::new(); + let response = client.get(api_url).send().await?; + + if response.status().is_success() { + let jobs: Vec = response.json().await?; + for job in jobs { + if let (Some(uid), Some(state)) = ( + job.get("uid").and_then(|u| u.as_str()), + job.get("state").and_then(|s| s.as_str()), + ) { + if state == "finished" || state == "error" { + let _ = client.delete(&format!("{}/{}", api_url, uid)).send().await; + info!("Cleaned up completed job: {}", uid); + } + } + } + } + Ok(()) +} + +pub async fn fetch_all_jobs(api_url: &str) -> Result> { + let client = Client::new(); + let response = client.get(api_url).send().await?; + if response.status().is_success() { + let jobs: Vec = response.json().await?; + Ok(jobs) + } else { + Ok(Vec::new()) + } +} diff --git a/src/static/index.html b/src/static/index.html new file mode 100644 index 0000000..bacd5c9 --- /dev/null +++ b/src/static/index.html @@ -0,0 +1,214 @@ + + + + + AE Anons Control Panel + + + +
+

🎬 AE Anons - Nexrender Job Manager

+ +
+ + + + +
+ +
+ +
+ +
+ + + + + + + + + + + + + + +
Output FileStateCreatedUpdatedUID
Loading...
+
+ + + + \ No newline at end of file diff --git a/src/web.rs b/src/web.rs new file mode 100644 index 0000000..9d93e78 --- /dev/null +++ b/src/web.rs @@ -0,0 +1,186 @@ +use crate::config::Config; +use crate::processor::{cleanup_finished_jobs, fetch_all_jobs, process_spreadsheet}; +use axum::{ + extract::State, + http::StatusCode, + response::{Html, IntoResponse, Json}, + routing::{get, post}, + Router, +}; +use serde::Serialize; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::sync::Mutex; +use tower_http::trace::TraceLayer; + +#[derive(Clone)] +pub struct AppState { + pub config: Config, + pub last_generation: Arc>>>, +} + +#[derive(Serialize)] +pub struct JobInfo { + pub uid: String, + pub outfile_name: String, + pub state: String, + pub created_at: Option, + pub updated_at: Option, +} + +impl JobInfo { + fn from_nexrender_json(job: &serde_json::Value) -> Option { + let uid = job.get("uid")?.as_str()?.to_string(); + let state = job.get("state")?.as_str()?.to_string(); + + // Извлекаем имя выходного файла из postrender actions + let outfile_name = job + .get("actions") + .and_then(|a| a.get("postrender")) + .and_then(|p| p.as_array()) + .and_then(|arr| { + // Ищем действие copy (в нём финальный путь) + arr.iter() + .find_map(|action| { + // Проверяем, что это действие copy + action + .get("module") + .and_then(|m| m.as_str()) + .filter(|&m| m == "@nexrender/action-copy") + .and_then(|_| { + // Извлекаем output из copy + action.get("output").and_then(|o| o.as_str()) + }) + }) + // Если copy не найдено, пробуем encode + .or_else(|| { + arr.iter() + .find_map(|action| action.get("output").and_then(|o| o.as_str())) + }) + }) + .map(|path| { + // Извлекаем только имя файла из полного пути + std::path::Path::new(path) + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or(path) + .to_string() + }) + .unwrap_or_else(|| { + // Если не удалось извлечь, используем UID + format!("job_{}", uid) + }); + + Some(JobInfo { + uid, + outfile_name, + state, + created_at: job + .get("createdAt") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + updated_at: job + .get("updatedAt") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), + }) + } +} + +pub async fn run_web_server(config: Config) -> anyhow::Result<()> { + let state = AppState { + config, + last_generation: Arc::new(Mutex::new(None)), + }; + + let app = Router::new() + .route("/", get(index_page)) + .route("/api/jobs", get(list_jobs)) + .route("/api/generate", post(generate_jobs)) + .route("/api/cleanup", post(cleanup_jobs)) + .route("/api/status", get(get_status)) + .layer(TraceLayer::new_for_http()) + .with_state(state); + + let addr: SocketAddr = "0.0.0.0:3000".parse()?; + log::info!("Web server listening on http://{}", addr); + + let listener = tokio::net::TcpListener::bind(addr).await?; + axum::serve(listener, app).await?; + Ok(()) +} + +// Остальные обработчики остаются без изменений... +async fn index_page() -> Html<&'static str> { + Html(include_str!("static/index.html")) +} + +async fn list_jobs(State(state): State) -> Result>, AppError> { + let jobs_json = fetch_all_jobs(&state.config.nexrender_api_url) + .await + .map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + + let jobs: Vec = jobs_json + .iter() + .filter_map(JobInfo::from_nexrender_json) + .collect(); + + Ok(Json(jobs)) +} + +async fn generate_jobs(State(state): State) -> Result { + let mut last_gen = state.last_generation.lock().await; + if let Some(last) = *last_gen { + let elapsed = chrono::Local::now().signed_duration_since(last); + if elapsed.num_seconds() < 5 { + return Err(AppError( + StatusCode::TOO_MANY_REQUESTS, + "Generation already in progress or too recent".to_string(), + )); + } + } + *last_gen = Some(chrono::Local::now()); + drop(last_gen); + + let config = state.config.clone(); + let last_gen_clone = state.last_generation.clone(); + + tokio::spawn(async move { + match process_spreadsheet(&config).await { + Ok(submitted) => { + log::info!("Generation completed, {} jobs submitted", submitted.len()); + } + Err(e) => { + log::error!("Generation failed: {}", e); + } + } + *last_gen_clone.lock().await = None; + }); + + Ok((StatusCode::ACCEPTED, "Job generation started")) +} + +async fn cleanup_jobs(State(state): State) -> Result { + cleanup_finished_jobs(&state.config.nexrender_api_url) + .await + .map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + Ok((StatusCode::OK, "Cleanup completed")) +} + +async fn get_status(State(state): State) -> Result, AppError> { + let last_gen = *state.last_generation.lock().await; + let status = serde_json::json!({ + "last_generation": last_gen.map(|dt| dt.to_rfc3339()), + "nexrender_api": state.config.nexrender_api_url, + }); + Ok(Json(status)) +} + +// Error handling +struct AppError(StatusCode, String); + +impl IntoResponse for AppError { + fn into_response(self) -> axum::response::Response { + (self.0, self.1).into_response() + } +}