457 lines
15 KiB
Rust
457 lines
15 KiB
Rust
use crate::config::{AccessType, Config};
|
|
use crate::file_access::{
|
|
FileAccess, LocalFileAccess, SmbAccessConfig, SmbFileAccess, SynologyFileAccess,
|
|
};
|
|
use crate::processor::{cleanup_finished_jobs, fetch_all_jobs, process_spreadsheet};
|
|
use crate::synology::SynologyClient;
|
|
use axum::{
|
|
extract::{Path, State},
|
|
http::{header, HeaderMap, 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<Mutex<Option<chrono::DateTime<chrono::Local>>>>,
|
|
pub source_access: Arc<dyn FileAccess>,
|
|
pub dest_access: Arc<dyn FileAccess>,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct JobInfo {
|
|
pub uid: String,
|
|
pub outfile_name: String,
|
|
pub state: String,
|
|
pub created_at: Option<String>,
|
|
pub updated_at: Option<String>,
|
|
}
|
|
|
|
impl JobInfo {
|
|
fn from_nexrender_json(job: &serde_json::Value) -> Option<Self> {
|
|
let uid = job.get("uid")?.as_str()?.to_string();
|
|
let state = job.get("state")?.as_str()?.to_string();
|
|
|
|
let outfile_name = job
|
|
.get("actions")
|
|
.and_then(|a| a.get("postrender"))
|
|
.and_then(|p| p.as_array())
|
|
.and_then(|arr| {
|
|
arr.iter()
|
|
.find_map(|action| {
|
|
action
|
|
.get("module")
|
|
.and_then(|m| m.as_str())
|
|
.filter(|&m| m == "@nexrender/action-copy")
|
|
.and_then(|_| action.get("output").and_then(|o| o.as_str()))
|
|
})
|
|
.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(|| 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 source_access = create_file_access(&config.source, &config).await?;
|
|
let dest_access = create_file_access(&config.destination, &config).await?;
|
|
|
|
let web_port = config.web_port;
|
|
|
|
let state = AppState {
|
|
config,
|
|
last_generation: Arc::new(Mutex::new(None)),
|
|
source_access,
|
|
dest_access,
|
|
};
|
|
|
|
let app = Router::new()
|
|
.route("/", get(index_page))
|
|
.route("/favicon.ico", get(favicon))
|
|
.route("/static/style.css", get(style_css))
|
|
.route("/static/fontawesome/all.min.css", get(fontawesome_css))
|
|
.route(
|
|
"/static/fontawesome/webfonts/fa-solid-900.woff2",
|
|
get(fa_solid_woff2),
|
|
)
|
|
.route(
|
|
"/static/fontawesome/webfonts/fa-regular-400.woff2",
|
|
get(fa_regular_woff2),
|
|
)
|
|
.route(
|
|
"/static/fontawesome/webfonts/fa-brands-400.woff2",
|
|
get(fa_brands_woff2),
|
|
)
|
|
.route("/assets/logo.png", get(logo_png))
|
|
.route("/api/jobs", get(list_jobs))
|
|
.route("/api/generate", post(generate_jobs))
|
|
.route("/api/cleanup", post(cleanup_jobs))
|
|
.route("/api/status", get(get_status))
|
|
.route("/api/jobs/stop-all", post(stop_all_jobs))
|
|
.route("/api/videos", get(list_videos))
|
|
.route("/api/videos/{filename}", get(stream_video))
|
|
.route("/api/videos/{filename}/move", post(move_video))
|
|
.route("/api/videos/{filename}/status", get(get_video_status))
|
|
.layer(TraceLayer::new_for_http())
|
|
.with_state(state);
|
|
|
|
let addr: SocketAddr = format!("0.0.0.0:{}", web_port).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 create_file_access(
|
|
access_config: &crate::config::AccessConfig,
|
|
nas_config: &Config,
|
|
) -> anyhow::Result<Arc<dyn FileAccess>> {
|
|
match access_config.access_type {
|
|
AccessType::Local => {
|
|
let path = access_config.local_path.as_ref().unwrap();
|
|
Ok(Arc::new(LocalFileAccess::new(path)))
|
|
}
|
|
AccessType::Smb => {
|
|
let url = access_config.smb_url.as_ref().unwrap().clone();
|
|
let user = access_config.smb_user.clone();
|
|
let password = access_config.smb_pass.clone();
|
|
let domain = access_config.smb_domain.clone();
|
|
let smb_config = SmbAccessConfig {
|
|
url,
|
|
user,
|
|
password,
|
|
domain,
|
|
};
|
|
let access: Arc<dyn FileAccess> = Arc::new(SmbFileAccess::new(smb_config));
|
|
Ok(access)
|
|
}
|
|
AccessType::Synology => {
|
|
let mut client = SynologyClient::new(&nas_config.nas_fqdn);
|
|
client
|
|
.login(&nas_config.nas_user, &nas_config.nas_pass)
|
|
.await?;
|
|
let base_path = access_config.synology_path.as_ref().unwrap().clone();
|
|
Ok(Arc::new(SynologyFileAccess::new(client, &base_path)))
|
|
}
|
|
}
|
|
}
|
|
|
|
// Статические ресурсы
|
|
async fn index_page() -> Html<&'static str> {
|
|
Html(include_str!("static/index.html"))
|
|
}
|
|
|
|
async fn style_css() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "text/css")],
|
|
include_str!("static/style.css"),
|
|
)
|
|
}
|
|
|
|
async fn fontawesome_css() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "text/css")],
|
|
include_str!("static/fontawesome/all.min.css"),
|
|
)
|
|
}
|
|
|
|
async fn logo_png() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "image/png")],
|
|
include_bytes!("../assets/logo.png").as_slice(),
|
|
)
|
|
}
|
|
|
|
async fn favicon() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "image/x-icon")],
|
|
include_bytes!("static/favicon.ico").as_slice(),
|
|
)
|
|
}
|
|
|
|
// Управление заданиями Nexrender
|
|
async fn list_jobs(State(state): State<AppState>) -> Result<Json<Vec<JobInfo>>, 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<JobInfo> = jobs_json
|
|
.iter()
|
|
.filter_map(JobInfo::from_nexrender_json)
|
|
.collect();
|
|
|
|
Ok(Json(jobs))
|
|
}
|
|
|
|
async fn generate_jobs(State(state): State<AppState>) -> Result<impl IntoResponse, AppError> {
|
|
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<AppState>) -> Result<impl IntoResponse, AppError> {
|
|
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<AppState>) -> Result<Json<serde_json::Value>, 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))
|
|
}
|
|
|
|
async fn stop_all_jobs(State(state): State<AppState>) -> Result<impl IntoResponse, AppError> {
|
|
use reqwest::Client;
|
|
|
|
let client = Client::new();
|
|
let api_url = &state.config.nexrender_api_url;
|
|
|
|
let jobs = fetch_all_jobs(api_url)
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
|
|
|
|
let mut stopped = 0;
|
|
for job in jobs {
|
|
let state = job.get("state").and_then(|s| s.as_str()).unwrap_or("");
|
|
if state == "queued" || state == "started" || state == "processing" {
|
|
if let Some(uid) = job.get("uid").and_then(|u| u.as_str()) {
|
|
let _ = client.delete(&format!("{}/{}", api_url, uid)).send().await;
|
|
stopped += 1;
|
|
log::info!("Stopped job: {}", uid);
|
|
}
|
|
}
|
|
}
|
|
|
|
log::info!("Stopped {} active jobs", stopped);
|
|
Ok((StatusCode::OK, format!("Stopped {} jobs", stopped)))
|
|
}
|
|
|
|
// Видео
|
|
async fn list_videos(State(state): State<AppState>) -> Result<Json<Vec<String>>, AppError> {
|
|
let files = state
|
|
.source_access
|
|
.list_files()
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
|
|
let videos: Vec<String> = files
|
|
.into_iter()
|
|
.filter(|name| {
|
|
let lower = name.to_lowercase();
|
|
lower.ends_with(".mp4") || lower.ends_with(".mov") || lower.ends_with(".avi")
|
|
})
|
|
.collect();
|
|
Ok(Json(videos))
|
|
}
|
|
|
|
async fn stream_video(
|
|
Path(filename): Path<String>,
|
|
State(state): State<AppState>,
|
|
headers: axum::http::HeaderMap,
|
|
) -> Result<impl IntoResponse, AppError> {
|
|
log::info!("Stream video requested: {}", filename);
|
|
let file_data = state
|
|
.source_access
|
|
.read_file(&filename)
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::NOT_FOUND, format!("File not found: {}", e)))?;
|
|
|
|
let file_size = file_data.len() as u64;
|
|
let range = headers.get(header::RANGE).and_then(|v| v.to_str().ok());
|
|
|
|
let (status, content_range, start, end) = if let Some(range_str) = range {
|
|
let range_str = range_str.trim_start_matches("bytes=");
|
|
let parts: Vec<&str> = range_str.split('-').collect();
|
|
let start: u64 = parts[0].parse().unwrap_or(0);
|
|
let end: u64 = if parts.len() > 1 && !parts[1].is_empty() {
|
|
parts[1].parse().unwrap_or(file_size - 1)
|
|
} else {
|
|
file_size - 1
|
|
};
|
|
(
|
|
StatusCode::PARTIAL_CONTENT,
|
|
Some(format!("bytes {}-{}/{}", start, end, file_size)),
|
|
start,
|
|
end,
|
|
)
|
|
} else {
|
|
(StatusCode::OK, None, 0, file_size - 1)
|
|
};
|
|
|
|
let mut resp_headers = HeaderMap::new();
|
|
resp_headers.insert(header::CONTENT_TYPE, "video/mp4".parse().unwrap());
|
|
if let Some(content_range) = content_range {
|
|
resp_headers.insert(header::CONTENT_RANGE, content_range.parse().unwrap());
|
|
}
|
|
let len = end - start + 1;
|
|
resp_headers.insert(header::CONTENT_LENGTH, len.to_string().parse().unwrap());
|
|
resp_headers.insert(header::ACCEPT_RANGES, "bytes".parse().unwrap());
|
|
|
|
let body = file_data[start as usize..=end as usize].to_vec();
|
|
Ok((status, resp_headers, body))
|
|
}
|
|
|
|
async fn move_video(
|
|
Path(filename): Path<String>,
|
|
State(state): State<AppState>,
|
|
axum::extract::Query(params): axum::extract::Query<std::collections::HashMap<String, String>>,
|
|
) -> Result<impl IntoResponse, AppError> {
|
|
let remove = params.get("remove").map(|s| s == "true").unwrap_or(false);
|
|
|
|
let data = state
|
|
.source_access
|
|
.read_file(&filename)
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::NOT_FOUND, format!("File not found: {}", e)))?;
|
|
|
|
// Определяем целевую подпапку по имени файла
|
|
let subfolder = if filename.contains("_Today") {
|
|
"Today"
|
|
} else if filename.contains("_Tomorrow") {
|
|
"Tomorrow"
|
|
} else {
|
|
"Data"
|
|
};
|
|
let dest_filename = format!("{}/{}", subfolder, filename);
|
|
|
|
// Проверяем, существует ли уже файл в целевой папке
|
|
if state.dest_access.read_file(&dest_filename).await.is_ok() {
|
|
return Ok((
|
|
StatusCode::OK,
|
|
format!("File {} already exists in {} folder.", filename, subfolder),
|
|
));
|
|
}
|
|
|
|
// Запись
|
|
state
|
|
.dest_access
|
|
.write_file(&dest_filename, &data)
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
|
|
|
|
let mut message = format!(
|
|
"File {} moved successfully to {} folder.",
|
|
filename, subfolder
|
|
);
|
|
|
|
if remove {
|
|
state
|
|
.source_access
|
|
.delete_file(&filename)
|
|
.await
|
|
.map_err(|e| AppError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
|
|
message.push_str(" Please delete the source file manually if needed.");
|
|
}
|
|
|
|
Ok((StatusCode::OK, message))
|
|
}
|
|
|
|
// Шрифты Font Awesome
|
|
async fn fa_solid_woff2() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "font/woff2")],
|
|
include_bytes!("static/fontawesome/webfonts/fa-solid-900.woff2").as_slice(),
|
|
)
|
|
}
|
|
|
|
async fn fa_regular_woff2() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "font/woff2")],
|
|
include_bytes!("static/fontawesome/webfonts/fa-regular-400.woff2").as_slice(),
|
|
)
|
|
}
|
|
|
|
async fn fa_brands_woff2() -> impl IntoResponse {
|
|
(
|
|
[(header::CONTENT_TYPE, "font/woff2")],
|
|
include_bytes!("static/fontawesome/webfonts/fa-brands-400.woff2").as_slice(),
|
|
)
|
|
}
|
|
|
|
async fn get_video_status(
|
|
Path(filename): Path<String>,
|
|
State(state): State<AppState>,
|
|
) -> Result<Json<serde_json::Value>, AppError> {
|
|
let subfolder = if filename.contains("_Today") {
|
|
"Today"
|
|
} else if filename.contains("_Tomorrow") {
|
|
"Tomorrow"
|
|
} else {
|
|
"Data"
|
|
};
|
|
let dest_filename = format!("{}/{}", subfolder, filename);
|
|
let exists = state.dest_access.read_file(&dest_filename).await.is_ok();
|
|
Ok(Json(serde_json::json!({
|
|
"exists_in_dest": exists,
|
|
"subfolder": subfolder
|
|
})))
|
|
}
|
|
|
|
struct AppError(StatusCode, String);
|
|
|
|
impl IntoResponse for AppError {
|
|
fn into_response(self) -> axum::response::Response {
|
|
(self.0, self.1).into_response()
|
|
}
|
|
}
|