v0.2.0
This commit is contained in:
434
src/processor.rs
Normal file
434
src/processor.rs
Normal file
@@ -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<String>,
|
||||
pub rows: Vec<HashMap<String, String>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct ExcelWorkbook {
|
||||
pub sheets: Vec<SheetData>,
|
||||
}
|
||||
|
||||
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<Vec<(String, String)>> {
|
||||
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<ExcelWorkbook> {
|
||||
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<ExcelWorkbook> {
|
||||
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<Data>,
|
||||
) -> Result<SheetData> {
|
||||
let (row_count, col_count) = range.get_size();
|
||||
let mut data_matrix: Vec<Vec<String>> = 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<String> = 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<HashMap<String, String>> = 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<Vec<(String, String)>> {
|
||||
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<String, String> = 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<String, String> = 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<JobData> = 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<Value> = 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<Vec<Value>> {
|
||||
let client = Client::new();
|
||||
let response = client.get(api_url).send().await?;
|
||||
if response.status().is_success() {
|
||||
let jobs: Vec<Value> = response.json().await?;
|
||||
Ok(jobs)
|
||||
} else {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user