admin/rust-imager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

update code level

Browse Source
This commit is contained in:
lychang
2026-05-14 00:04:35 +08:00
parent 8f214b587b
commit 73dbb30e27
7 changed files with 128 additions and 4541 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
.trae/skills/image-brotli-processor/SKILL.md Normal file
View File

@@ -0,0 +1,36 @@
---
name: "image-brotli-processor"
description: "将图片从缩略图改为brotli压缩处理。在需要优化图片存储和传输性能时调用，特别是当用户想要减少图片文件大小和提高加载速度时。"
---
# 图片Brotli压缩处理器
这个skill帮助将传统的缩略图系统改为使用brotli压缩的图片处理系统，以提高性能和减少存储空间。
## 功能
- 将图片处理从传统的缩略图改为brotli压缩
- 添加智能尺寸调整和压缩策略
- 提供向后兼容的API端点
- 优化网络传输和存储效率
## 使用场景
- 当用户想要优化图片加载性能时
- 需要减少图片存储空间占用时
- 想要提高网络传输效率时
- 需要保持图片质量的同时减小文件大小时
## 实现步骤
1. 修改图片处理器，添加brotli压缩功能
2. 添加新的brotli图片服务端点
3. 更新路由配置
4. 修改前端图片引用路径
## 技术优势
- **更高的压缩率**：比gzip高20-30%的压缩效率
- **更快的加载速度**：减小文件大小，提高网络传输效率
- **保持图片质量**：在压缩的同时保持视觉质量
- **浏览器支持**：现代浏览器都支持brotli解码

3569
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

3
Cargo.toml
View File

@@ -19,5 +19,4 @@ rand = "0.8"
tracing = "0.1"
tracing-subscriber = "0.3"
thiserror = "1.0"
mimalloc = "0.1.48"
brotli = "3.4"
mimalloc = "0.1.48"

72
src/import.rs
View File

@@ -138,62 +138,34 @@ pub async fn import_existing_data(config: &Config, db: Arc<Database>, _task_queu
return Ok(());
}
// 如果有队列系统，使用队列异步处理相册导入
if let Some(queue) = _task_queue {
println!("🚀 使用队列系统异步导入相册...");
// 使用简单的串行处理方式导入相册
println!("📁 开始导入新相册...");
let mut total_imported = 0;
let mut successful_albums = 0;
for (folder_name, path) in album_folders {
println!("📁 开始导入相册: {}", folder_name);
for (folder_name, path) in album_folders {
println!("📁 添加相册导入任务: {}", folder_name);
// 创建相册导入任务
let relative_path = path.strip_prefix(&config.storage.images_dir)
.unwrap_or(&path)
.to_string_lossy()
.to_string();
let task = crate::queue::TaskFactory::create_album_import_task(
relative_path,
None,
);
// 添加任务到队列
if let Err(e) = queue.add_task(task).await {
eprintln!("❌ 添加相册导入任务失败: {}", e);
} else {
println!("✅ 相册导入任务已添加到队列: {}", folder_name);
}
}
println!("🎉 所有相册导入任务已添加到队列，将由AlbumImportWorker异步处理");
} else {
// 如果没有队列系统，使用原来的串行处理方式
println!("⚠️ 队列系统不可用，使用串行处理方式导入相册...");
let mut total_imported = 0;
let mut successful_albums = 0;
for (folder_name, path) in album_folders {
println!("📁 开始导入相册: {}", folder_name);
match import_single_album(&db, &folder_name, &path).await {
Ok(count) => {
if count > 0 {
successful_albums += 1;
total_imported += count;
println!("✅ 相册 {} 导入完成，共导入 {} 张图片", folder_name, count);
} else {
println!("⚠️ 相册 {} 没有图片需要导入", folder_name);
}
}
Err(e) => {
eprintln!("❌ 相册 {} 导入失败: {}", folder_name, e);
// 继续处理下一个相册，不中断整个导入过程
match import_single_album(&db, &folder_name, &path).await {
Ok(count) => {
if count > 0 {
successful_albums += 1;
total_imported += count;
println!("✅ 相册 {} 导入完成，共导入 {} 张图片", folder_name, count);
} else {
println!("⚠️ 相册 {} 没有图片需要导入", folder_name);
}
}
Err(e) => {
eprintln!("❌ 相册 {} 导入失败: {}", folder_name, e);
// 继续处理下一个相册，不中断整个导入过程
}
}
println!("🎉 导入完成！成功导入 {} 个相册，共 {} 张图片", successful_albums, total_imported);
}
println!("🎉 导入完成！成功导入 {} 个相册，共 {} 张图片", successful_albums, total_imported);
Ok(())
}

29
src/main.rs
View File

@@ -46,37 +46,17 @@ async fn main() -> std::io::Result<()> {
let image_processor = Arc::new(image_processor::ImageProcessor::new(config.storage.clone()));
// 初始化智能任务队列
let task_queue = Arc::new(queue::TaskQueue::new_with_db(
config.queue.max_concurrent_tasks, // 最大并发任务数
config.queue.queue_size_limit, // 队列大小限制
config.queue.high_request_rate_threshold, // 高请求频率阈值
config.queue.idle_request_rate_threshold, // 空闲请求频率阈值
Arc::clone(&db), // 数据库实例
));
// 在后台线程中自动导入新相册（不阻塞HTTP服务启动）
let config_clone = config.clone();
let db_clone = Arc::clone(&db);
let task_queue_clone = Arc::clone(&task_queue);
tokio::spawn(async move {
info!("[后台任务] 开始扫描图片目录并导入新相册...");
match import::import_existing_data(&config_clone, db_clone, Some(task_queue_clone)).await {
match import::import_existing_data(&config_clone, db_clone, None).await {
Ok(_) => info!("[后台任务] 相册导入完成"),
Err(e) => error!("[后台任务] 相册导入失败: {}", e),
}
});
// 启动相册导入Worker
let album_import_worker = queue::AlbumImportWorker::new(
Arc::clone(&task_queue),
Arc::clone(&db),
config.storage.clone(),
);
tokio::spawn(async move {
album_import_worker.start().await;
});
// 初始化相册内存缓存
let albums_cache = Arc::new(tokio::sync::RwLock::new(AlbumsCache::new()));
@@ -118,11 +98,7 @@ async fn main() -> std::io::Result<()> {
}
});
// 启动定时清理过期任务的任务（每小时执行一次）
let task_queue_for_cleanup = Arc::clone(&task_queue);
tokio::spawn(async move {
task_queue_for_cleanup.start_periodic_cleanup(Duration::from_secs(60 * 60)).await; // 1小时
});
println!("API接口:");
println!(" - GET /api/albums - 获取所有相册");
@@ -152,7 +128,6 @@ async fn main() -> std::io::Result<()> {
.app_data(web::Data::new(Arc::clone(&db)))
.app_data(web::Data::new(Arc::clone(&image_processor)))
.app_data(web::Data::new(Arc::clone(&albums_cache)))
// 不再需要任务队列作为app_data，因为已移除缩略图功能
// API路由
.route("/api/albums", web::get().to(routes::get_albums))
.route("/api/search", web::get().to(routes::search_albums))

772
src/queue.rs
View File

@@ -1,281 +1,68 @@
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, AtomicU64, Ordering};
use std::sync::atomic::{AtomicU64, AtomicBool};
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, Notify};
use tokio::time::sleep;
use tracing::{info, warn, error, debug};
use uuid::Uuid;
use image::GenericImageView;
use crate::models::{AlbumIndex, CreateAlbum, CreateImage};
use crate::database::Database;
/// 任务类型
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum TaskType {
ThumbnailGeneration, // 缩略图生成
AlbumImport, // 相册导入
}
/// 任务优先级
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum TaskPriority {
High = 0, // 前端请求的缩略图
Medium = 1, // 相册导入
Low = 2, // 后台质量升级
}
/// 通用任务结构
#[derive(Debug, Clone)]
pub struct Task {
pub id: Uuid,
pub task_type: TaskType,
pub priority: TaskPriority,
pub created_at: Instant,
pub data: TaskData,
}
/// 任务数据
#[derive(Debug, Clone)]
pub enum TaskData {
ThumbnailGeneration {
image_id: Uuid,
original_data: Vec<u8>,
},
AlbumImport {
album_path: String,
album_data: Option<Vec<u8>>,
},
}
use tokio::sync::Notify;
use tracing::{info, debug};
/// 队列统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueStats {
pub pending_tasks: usize,
pub active_tasks: usize,
pub completed_tasks: usize,
pub failed_tasks: usize,
pub queue_size_limit: usize,
pub is_paused: bool,
pub task_type_stats: Vec<TaskTypeStats>,
// HTTP请求频率信息
pub request_frequency: f64,
pub request_count: u64,
}
/// 任务类型统计
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskTypeStats {
pub task_type: TaskType,
pub pending: usize,
pub completed: usize,
pub failed: usize,
}
/// 智能任务队列
/// 简化的任务队列（仅用于记录HTTP请求频率）
pub struct TaskQueue {
high_priority_queue: Mutex<VecDeque<Task>>,
normal_priority_queue: Mutex<VecDeque<Task>>,
active_tasks: AtomicUsize,
completed_tasks: AtomicUsize,
failed_tasks: AtomicUsize,
max_concurrent_tasks: usize,
queue_size_limit: usize,
is_paused: AtomicBool,
notify: Notify,
last_activity: Mutex<Instant>,
// HTTP请求频率监控
request_count: AtomicU64,
last_request_time: Mutex<Instant>,
last_request_time: std::sync::Mutex<Instant>,
request_window: Duration,
// 配置阈值
high_request_rate_threshold: f64,
idle_request_rate_threshold: f64,
// 数据库引用用于定时任务
db: Option<Arc<Database>>,
is_paused: AtomicBool,
notify: Notify,
}
impl TaskQueue {
pub fn new(max_concurrent_tasks: usize, queue_size_limit: usize, high_request_rate_threshold: f64, idle_request_rate_threshold: f64) -> Self {
pub fn new(_max_concurrent_tasks: usize, _queue_size_limit: usize, _high_request_rate_threshold: f64, _idle_request_rate_threshold: f64) -> Self {
Self {
high_priority_queue: Mutex::new(VecDeque::new()),
normal_priority_queue: Mutex::new(VecDeque::new()),
active_tasks: AtomicUsize::new(0),
completed_tasks: AtomicUsize::new(0),
failed_tasks: AtomicUsize::new(0),
max_concurrent_tasks,
queue_size_limit,
is_paused: AtomicBool::new(false),
notify: Notify::new(),
last_activity: Mutex::new(Instant::now()),
// HTTP请求频率监控
request_count: AtomicU64::new(0),
last_request_time: Mutex::new(Instant::now()),
last_request_time: std::sync::Mutex::new(Instant::now()),
request_window: Duration::from_secs(10),
// 配置阈值
high_request_rate_threshold,
idle_request_rate_threshold,
db: None,
}
}
pub fn new_with_db(max_concurrent_tasks: usize, queue_size_limit: usize, high_request_rate_threshold: f64, idle_request_rate_threshold: f64, db: Arc<Database>) -> Self {
Self {
high_priority_queue: Mutex::new(VecDeque::new()),
normal_priority_queue: Mutex::new(VecDeque::new()),
active_tasks: AtomicUsize::new(0),
completed_tasks: AtomicUsize::new(0),
failed_tasks: AtomicUsize::new(0),
max_concurrent_tasks,
queue_size_limit,
is_paused: AtomicBool::new(false),
notify: Notify::new(),
last_activity: Mutex::new(Instant::now()),
}
}
pub fn new_with_db(_max_concurrent_tasks: usize, _queue_size_limit: usize, _high_request_rate_threshold: f64, _idle_request_rate_threshold: f64, _db: Arc<crate::database::Database>) -> Self {
Self {
// HTTP请求频率监控
request_count: AtomicU64::new(0),
last_request_time: Mutex::new(Instant::now()),
last_request_time: std::sync::Mutex::new(Instant::now()),
request_window: Duration::from_secs(10),
// 配置阈值
high_request_rate_threshold,
idle_request_rate_threshold,
db: Some(db),
is_paused: AtomicBool::new(false),
notify: Notify::new(),
}
}
/// 添加任务到队列
pub async fn add_task(&self, task: Task) -> Result<(), String> {
let (_high_queue_len, _normal_queue_len) = {
let high_queue = self.high_priority_queue.lock().await;
let normal_queue = self.normal_priority_queue.lock().await;
// 检查队列总容量是否已满
if high_queue.len() + normal_queue.len() >= self.queue_size_limit {
return Err("队列已满，无法添加新任务".to_string());
}
// 检查是否已有相同ID的任务
let exists_in_high = high_queue.iter().any(|t| t.id == task.id);
let exists_in_normal = normal_queue.iter().any(|t| t.id == task.id);
if exists_in_high || exists_in_normal {
return Err("该任务已在队列中".to_string());
}
(high_queue.len(), normal_queue.len())
};
// 根据优先级添加到相应队列
match task.priority {
TaskPriority::High => {
let mut high_queue = self.high_priority_queue.lock().await;
// 高优先级队列内部也按优先级排序
let insert_position = high_queue
.iter()
.position(|t| t.priority > task.priority)
.unwrap_or(high_queue.len());
high_queue.insert(insert_position, task);
debug!("高优先级任务已添加，当前高优先级队列大小: {}", high_queue.len());
}
_ => { // Medium and Low go to normal priority queue
let mut normal_queue = self.normal_priority_queue.lock().await;
// 按优先级排序
let insert_position = normal_queue
.iter()
.position(|t| t.priority > task.priority)
.unwrap_or(normal_queue.len());
normal_queue.insert(insert_position, task);
debug!("普通优先级任务已添加，当前普通优先级队列大小: {}", normal_queue.len());
}
}
// 更新最后活动时间
*self.last_activity.lock().await = Instant::now();
// 通知工作线程
self.notify.notify_one();
Ok(())
}
/// 从队列获取下一个任务（优先处理高优先级队列）
pub async fn next_task(&self) -> Option<Task> {
// 检查是否暂停
if self.is_paused.load(Ordering::Acquire) {
return None;
}
// 检查并发任务数量
if self.active_tasks.load(Ordering::Acquire) >= self.max_concurrent_tasks {
return None;
}
// 优先从高优先级队列获取任务
{
let mut high_queue = self.high_priority_queue.lock().await;
if let Some(task) = high_queue.pop_front() {
self.active_tasks.fetch_add(1, Ordering::Release);
return Some(task);
}
}
// 如果高优先级队列为空，则从普通优先级队列获取任务
let mut normal_queue = self.normal_priority_queue.lock().await;
if let Some(task) = normal_queue.pop_front() {
self.active_tasks.fetch_add(1, Ordering::Release);
Some(task)
} else {
None
}
}
/// 标记任务完成
pub fn task_completed(&self, success: bool) {
self.active_tasks.fetch_sub(1, Ordering::Release);
if success {
self.completed_tasks.fetch_add(1, Ordering::Release);
} else {
self.failed_tasks.fetch_add(1, Ordering::Release);
}
}
/// 暂停队列处理
pub fn pause(&self) {
self.is_paused.store(true, Ordering::Release);
info!("任务队列已暂停");
}
/// 恢复队列处理
pub fn resume(&self) {
self.is_paused.store(false, Ordering::Release);
self.notify.notify_one();
info!("任务队列已恢复");
}
/// 记录HTTP请求（在每次HTTP请求处理时调用）
pub async fn record_http_request(&self) {
let now = Instant::now();
let mut last_request_time = self.last_request_time.lock().await;
let mut last_request_time = self.last_request_time.lock().unwrap();
// 如果距离上次请求时间超过了窗口时间，重置计数器
if now.duration_since(*last_request_time) > self.request_window {
self.request_count.store(0, Ordering::Release);
self.request_count.store(0, std::sync::atomic::Ordering::Release);
}
// 增加请求计数
self.request_count.fetch_add(1, Ordering::Release);
self.request_count.fetch_add(1, std::sync::atomic::Ordering::Release);
*last_request_time = now;
}
/// 获取当前请求频率（请求数/秒）
pub fn get_request_frequency(&self) -> f64 {
let request_count = self.request_count.load(Ordering::Acquire) as f64;
let request_count = self.request_count.load(std::sync::atomic::Ordering::Acquire) as f64;
let window_seconds = self.request_window.as_secs_f64();
if window_seconds > 0.0 {
@@ -285,522 +72,25 @@ impl TaskQueue {
}
}
/// 检查服务器负载是否过高（基于HTTP请求频率）
pub async fn is_server_busy(&self) -> bool {
let request_frequency = self.get_request_frequency();
// 基于HTTP请求频率判断业务繁忙度
// 高请求频率（超过配置阈值）时认为服务器繁忙
let is_high_request_rate = request_frequency > self.high_request_rate_threshold;
// 如果请求频率高，或者并发任务/队列大小超过阈值，认为服务器繁忙
is_high_request_rate
}
/// 获取队列统计信息
pub async fn get_stats(&self) -> QueueStats {
let (high_queue_len, normal_queue_len, type_stats) = {
let high_queue = self.high_priority_queue.lock().await;
let normal_queue = self.normal_priority_queue.lock().await;
// 按任务类型统计
let mut type_stats = std::collections::HashMap::new();
// 统计高优先级队列中的任务
for task in high_queue.iter() {
let stats = type_stats
.entry(task.task_type.clone())
.or_insert((0, 0, 0));
stats.0 += 1; // pending (high priority)
}
// 统计普通优先级队列中的任务
for task in normal_queue.iter() {
let stats = type_stats
.entry(task.task_type.clone())
.or_insert((0, 0, 0));
stats.0 += 1; // pending (normal priority)
}
(high_queue.len(), normal_queue.len(), type_stats)
};
let task_type_stats = type_stats
.into_iter()
.map(|(task_type, (pending, completed, failed))| TaskTypeStats {
task_type,
pending,
completed,
failed,
})
.collect();
QueueStats {
pending_tasks: high_queue_len + normal_queue_len,
active_tasks: self.active_tasks.load(Ordering::Acquire),
completed_tasks: self.completed_tasks.load(Ordering::Acquire),
failed_tasks: self.failed_tasks.load(Ordering::Acquire),
queue_size_limit: self.queue_size_limit,
is_paused: self.is_paused.load(Ordering::Acquire),
task_type_stats,
// 添加请求频率信息
request_frequency: self.get_request_frequency(),
request_count: self.request_count.load(Ordering::Acquire),
request_count: self.request_count.load(std::sync::atomic::Ordering::Acquire),
}
}
/// 清理过期的任务（超过一定时间的任务）
pub async fn cleanup_expired_tasks(&self, max_age: Duration) {
let now = Instant::now();
// 清理高优先级队列中的过期任务
{
let mut high_queue = self.high_priority_queue.lock().await;
let initial_len = high_queue.len();
high_queue.retain(|task| now.duration_since(task.created_at) <= max_age);
let removed_count = initial_len - high_queue.len();
if removed_count > 0 {
info!("清理了 {} 个过期的高优先级任务", removed_count);
}
}
// 清理普通优先级队列中的过期任务
{
let mut normal_queue = self.normal_priority_queue.lock().await;
let initial_len = normal_queue.len();
normal_queue.retain(|task| now.duration_since(task.created_at) <= max_age);
let removed_count = initial_len - normal_queue.len();
if removed_count > 0 {
info!("清理了 {} 个过期的普通优先级任务", removed_count);
}
}
/// 暂停队列处理
pub fn pause(&self) {
self.is_paused.store(true, std::sync::atomic::Ordering::Release);
info!("任务队列已暂停");
}
/// 等待队列变为空闲状态
pub async fn wait_for_idle(&self, timeout: Duration) -> bool {
let start = Instant::now();
while start.elapsed() < timeout {
sleep(Duration::from_millis(100)).await;
}
false
}
/// 启动定期清理过期任务
pub async fn start_periodic_cleanup(self: Arc<Self>, interval_duration: Duration) {
let queue = Arc::clone(&self);
tokio::spawn(async move {
let mut interval = tokio::time::interval(interval_duration);
loop {
interval.tick().await;
// 清理超过24小时的任务
queue.cleanup_expired_tasks(Duration::from_secs(24 * 3600)).await;
}
});
}
}
/// 缩略图生成Worker
pub struct ThumbnailWorker {
queue: Arc<TaskQueue>,
image_processor: Arc<crate::image_processor::ImageProcessor>,
}
impl ThumbnailWorker {
pub fn new(
queue: Arc<TaskQueue>,
image_processor: Arc<crate::image_processor::ImageProcessor>,
) -> Self {
Self {
queue,
image_processor,
}
}
/// 启动缩略图Worker
pub async fn start(self) {
info!("启动缩略图生成Worker");
loop {
// 检查队列中是否有任务
if let Some(task) = self.queue.next_task().await {
if matches!(task.task_type, TaskType::ThumbnailGeneration) {
debug!("缩略图Worker开始处理任务: {:?}", task.id);
let success = self.process_thumbnail_task(task).await;
self.queue.task_completed(success);
// 在处理完每个任务后检查服务器负载
if self.queue.is_server_busy().await {
info!("服务器负载较高，暂停队列处理");
self.queue.pause();
// 每30秒检查一次服务器状态，直到服务器不忙时恢复队列
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
if !self.queue.is_server_busy().await {
info!("服务器负载恢复正常，恢复队列处理");
self.queue.resume();
break;
}
info!("服务器仍然繁忙，继续等待...");
}
}
} else {
// 如果不是缩略图任务，放回队列
// 根据任务优先级放回相应队列
match task.priority {
TaskPriority::High => {
let mut high_queue = self.queue.high_priority_queue.lock().await;
high_queue.push_front(task);
}
_ => {
let mut normal_queue = self.queue.normal_priority_queue.lock().await;
normal_queue.push_front(task);
}
}
}
} else {
// 队列为空，等待新任务通知
self.queue.notify.notified().await;
}
// 短暂休眠避免CPU占用过高
sleep(Duration::from_millis(100)).await;
}
}
// 缩略图任务处理功能已移除，直接返回成功
async fn process_thumbnail_task(&self, task: Task) -> bool {
if let TaskData::ThumbnailGeneration {
image_id,
original_data: _,
} = task.data
{
info!("缩略图任务已跳过（功能已移除）: {:?}", image_id);
true // 直接返回成功，避免阻塞队列
} else {
warn!("任务类型不匹配，期望缩略图生成任务");
false
}
}
}
/// 相册导入Worker
pub struct AlbumImportWorker {
queue: Arc<TaskQueue>,
db: Arc<crate::database::Database>,
config: crate::config::StorageConfig,
}
impl AlbumImportWorker {
pub fn new(
queue: Arc<TaskQueue>,
db: Arc<crate::database::Database>,
config: crate::config::StorageConfig,
) -> Self {
Self { queue, db, config }
}
/// 启动相册导入Worker
pub async fn start(self) {
info!("启动相册导入Worker");
loop {
// 检查队列中是否有任务
if let Some(task) = self.queue.next_task().await {
if matches!(task.task_type, TaskType::AlbumImport) {
debug!("相册导入Worker开始处理任务: {:?}", task.id);
let success = self.process_album_import_task(task).await;
self.queue.task_completed(success);
// 在处理完每个任务后检查服务器负载
if self.queue.is_server_busy().await {
info!("服务器负载较高，暂停队列处理");
self.queue.pause();
// 每30秒检查一次服务器状态，直到服务器不忙时恢复队列
loop {
tokio::time::sleep(Duration::from_secs(30)).await;
if !self.queue.is_server_busy().await {
info!("服务器负载恢复正常，恢复队列处理");
self.queue.resume();
break;
}
info!("服务器仍然繁忙，继续等待...");
}
}
} else {
// 如果不是相册导入任务，放回队列并减少active_tasks计数器
// 根据任务优先级放回相应队列
match task.priority {
TaskPriority::High => {
let mut high_queue = self.queue.high_priority_queue.lock().await;
high_queue.push_front(task);
}
_ => {
let mut normal_queue = self.queue.normal_priority_queue.lock().await;
normal_queue.push_front(task);
}
}
// 手动减少active_tasks计数器，因为任务没有被处理
self.queue.active_tasks.fetch_sub(1, Ordering::Release);
}
} else {
// 队列为空，等待新任务通知
self.queue.notify.notified().await;
}
// 短暂休眠避免CPU占用过高
sleep(Duration::from_millis(100)).await;
}
}
/// 处理相册导入任务
async fn process_album_import_task(&self, task: Task) -> bool {
if let TaskData::AlbumImport {
album_path,
album_data: _,
} = task.data
{
info!("开始处理相册导入任务: {}", album_path);
// 将相对路径与images_dir组合成完整路径
let full_path = std::path::Path::new(&self.config.images_dir).join(&album_path);
// 检查相册文件夹是否存在
if !full_path.exists() {
error!("相册文件夹不存在: {}", full_path.display());
return false;
}
// 导入单个相册
match self.import_single_album(&album_path, &full_path).await {
Ok(count) => {
info!("相册导入成功: {} (导入 {} 张图片)", album_path, count);
true
}
Err(e) => {
error!("相册导入失败: {} - {}", album_path, e);
false
}
}
} else {
warn!("任务类型不匹配，期望相册导入任务");
false
}
}
/// 导入单个相册（按照import.rs中的逻辑实现）
async fn import_single_album(&self, folder_name: &str, path: &std::path::Path) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
info!("开始导入相册: {}", folder_name);
// 读取index.json文件
let index_file_path = path.join("index.json");
let album_title = if index_file_path.exists() {
match std::fs::read_to_string(&index_file_path) {
Ok(content) => {
match serde_json::from_str::<AlbumIndex>(&content) {
Ok(index) => {
info!("从index.json读取相册标题: {}", index.title);
index.title
}
Err(e) => {
warn!("解析index.json失败: {}, 使用文件夹名作为标题", e);
folder_name.to_string()
}
}
}
Err(e) => {
warn!("读取index.json失败: {}, 使用文件夹名作为标题", e);
folder_name.to_string()
}
}
} else {
info!("index.json不存在，使用文件夹名作为标题");
folder_name.to_string()
};
// 创建新相册
let new_album = CreateAlbum {
title: album_title.clone(),
document: folder_name.to_string(),
description: Some(format!("从文件夹 {} 自动导入的相册", folder_name)),
};
let album_id = self.db.create_album(new_album).await?;
info!("创建相册: {} (ID: {})", album_title, album_id);
// 读取图片URL列表（如果有index.json）
let image_urls = if index_file_path.exists() {
match std::fs::read_to_string(&index_file_path) {
Ok(content) => {
match serde_json::from_str::<AlbumIndex>(&content) {
Ok(index) => {
info!("从index.json读取到 {} 个图片URL", index.images.len());
Some(index.images)
}
Err(_) => None
}
}
Err(_) => None
}
} else {
None
};
// 快速扫描图片文件
let image_files = self.scan_image_files(path, folder_name)?;
if image_files.is_empty() {
info!("相册 {} 中没有找到图片文件", folder_name);
return Ok(0);
}
info!("发现 {} 张图片，开始处理...", image_files.len());
// 并行处理图片信息获取
let mut image_tasks = Vec::new();
for (index, (filename, image_path)) in image_files.iter().enumerate() {
let image_url = image_urls.as_ref().and_then(|urls| {
if urls.len() > index {
Some(urls[index].clone())
} else {
None
}
});
let filename_clone = filename.clone();
let image_path_clone = image_path.clone();
let album_id_clone = album_id;
let folder_name_clone = folder_name.to_string();
let task = tokio::task::spawn_blocking(move || {
// 在阻塞线程中处理图片信息
let (file_size, width, height) = match AlbumImportWorker::get_image_info_static(&image_path_clone) {
Ok(info) => info,
Err(e) => {
eprintln!("获取图片信息失败: {} - {}", image_path_clone.display(), e);
(0, 0, 0)
}
};
CreateImage {
album_id: album_id_clone,
filename: format!("{}/{}", folder_name_clone, filename_clone),
original_url: image_url,
title: Some(format!("图片 {}", index + 1)),
description: Some(format!("从文件夹 {} 导入的图片", folder_name_clone)),
file_size,
width,
height,
mime_type: match image_path_clone.extension().and_then(|s| s.to_str()).unwrap_or("jpg").to_lowercase().as_str() {
"jpg" | "jpeg" => "image/jpeg".to_string(),
"png" => "image/png".to_string(),
"gif" => "image/gif".to_string(),
"webp" => "image/webp".to_string(),
"bmp" => "image/bmp".to_string(),
_ => "image/jpeg".to_string(),
},
}
});
image_tasks.push(task);
}
// 收集所有图片信息
let mut images_to_insert = Vec::new();
for task in image_tasks {
match task.await {
Ok(image) => images_to_insert.push(image),
Err(e) => error!("图片处理任务失败: {}", e),
}
}
// 批量插入图片
if !images_to_insert.is_empty() {
match self.db.add_images(images_to_insert).await {
Ok(ids) => {
info!("成功导入 {} 张图片到相册 {}", ids.len(), folder_name);
// 更新相册图片数量
if let Err(e) = self.db.update_album_image_count(album_id).await {
error!("更新相册图片数量失败: {}", e);
}
Ok(ids.len())
}
Err(e) => {
error!("批量插入图片失败: {}", e);
Ok(0)
}
}
} else {
info!("没有图片需要插入到相册 {}", folder_name);
Ok(0)
}
}
/// 扫描图片文件（从import.rs迁移）
fn scan_image_files(&self, path: &std::path::Path, _folder_name: &str) -> Result<Vec<(String, std::path::PathBuf)>, Box<dyn std::error::Error + Send + Sync>> {
let mut image_files = Vec::new();
for entry in std::fs::read_dir(path)? {
let entry = entry?;
let file_path = entry.path();
if file_path.is_file() {
if let Some(extension) = file_path.extension().and_then(|s| s.to_str()) {
let ext_lower = extension.to_lowercase();
if ext_lower == "jpg" || ext_lower == "jpeg" || ext_lower == "png" ||
ext_lower == "gif" || ext_lower == "webp" || ext_lower == "bmp" {
if let Some(filename) = file_path.file_name().and_then(|s| s.to_str()) {
image_files.push((filename.to_string(), file_path));
}
}
}
}
}
Ok(image_files)
}
/// 静态方法：获取图片信息（用于spawn_blocking）
fn get_image_info_static(image_path: &std::path::Path) -> Result<(i64, i32, i32), Box<dyn std::error::Error + Send + Sync>> {
let metadata = std::fs::metadata(image_path)?;
let file_size = metadata.len() as i64;
// 使用image库获取图片尺寸
let img = image::open(image_path)?;
let (width, height) = img.dimensions();
Ok((file_size, width as i32, height as i32))
}
}
/// 任务工厂 - 用于创建不同类型的任务
pub struct TaskFactory;
impl TaskFactory {
/// 创建相册导入任务
pub fn create_album_import_task(album_path: String, album_data: Option<Vec<u8>>) -> Task {
Task {
id: Uuid::new_v4(),
task_type: TaskType::AlbumImport,
priority: TaskPriority::Medium,
created_at: Instant::now(),
data: TaskData::AlbumImport {
album_path,
album_data,
},
}
/// 恢复队列处理
pub fn resume(&self) {
self.is_paused.store(false, std::sync::atomic::Ordering::Release);
self.notify.notify_one();
info!("任务队列已恢复");
}
}

188
src/routes.rs
View File

@@ -9,50 +9,7 @@ use tracing::error;
use uuid::Uuid;
use crate::queue;
// 记录HTTP请求频率的辅助函数
async fn log_request_frequency(method: &str, path: &str, client_ip: Option<&str>) {
// 这里可以添加请求频率记录逻辑
// 例如：记录到数据库、内存缓存或日志文件
let ip = client_ip.unwrap_or("unknown");
println!("Request: {} {} from {}", method, path, ip);
}
// 统一错误处理宏
macro_rules! handle_db_error {
($result:expr, $error_msg:expr) => {
match $result {
Ok(value) => value,
Err(e) => {
error!("{}: {}", $error_msg, e);
return HttpResponse::InternalServerError().json(serde_json::json!({
"error": $error_msg
}));
}
}
};
}
// 输入验证错误类型
#[derive(Debug)]
enum ValidationError {
QueryTooLong,
InvalidCharacters,
}
// 搜索查询验证函数
fn validate_search_query(query: &str) -> Result<(), ValidationError> {
if query.len() > 100 {
return Err(ValidationError::QueryTooLong);
}
// 允许Unicode字符（包括中文）、ASCII图形字符和空白字符
// 只阻止控制字符和其他潜在危险字符
if query.chars().any(|c| c.is_control() || c == '\0') {
return Err(ValidationError::InvalidCharacters);
}
Ok(())
}
// 收藏功能API路由
@@ -61,7 +18,15 @@ pub async fn get_favorites(db: web::Data<Arc<Database>>) -> impl Responder {
// 这里使用固定用户ID，实际项目中应该从认证信息中获取
let user_id = "default_user".to_string();
let favorites = handle_db_error!(db.get_user_favorites(&user_id).await, "获取收藏列表失败");
let favorites = match db.get_user_favorites(&user_id).await {
Ok(value) => value,
Err(e) => {
error!("获取收藏列表失败: {}", e);
return HttpResponse::InternalServerError().json(serde_json::json!({
"error": "获取收藏列表失败"
}));
}
};
HttpResponse::Ok().json(favorites)
}
@@ -85,10 +50,15 @@ pub async fn check_favorite_status(
// 这里使用固定用户ID，实际项目中应该从认证信息中获取
let user_id = "default_user".to_string();
let status = handle_db_error!(
db.get_favorite_status(&user_id, image_id).await,
"检查收藏状态失败"
);
let status = match db.get_favorite_status(&user_id, image_id).await {
Ok(value) => value,
Err(e) => {
error!("检查收藏状态失败: {}", e);
return HttpResponse::InternalServerError().json(serde_json::json!({
"error": "检查收藏状态失败"
}));
}
};
HttpResponse::Ok().json(status)
}
@@ -183,51 +153,7 @@ pub async fn remove_favorite_by_image(
}
}
pub async fn get_original_image(
db: web::Data<Arc<Database>>,
image_processor: web::Data<Arc<ImageProcessor>>,
path: web::Path<String>,
) -> impl Responder {
let image_id = match Uuid::parse_str(&path.into_inner()) {
Ok(id) => id,
Err(_) => {
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "无效的图片ID"
}));
}
};
match db.get_image(image_id).await {
Ok(Some(image)) => {
let original_path = image_processor.config.images_dir.join(&image.filename);
if original_path.exists() {
match fs::read(&original_path) {
Ok(data) => HttpResponse::Ok().content_type(image.mime_type).body(data),
Err(e) => {
eprintln!("读取原始图片失败: {}", e);
HttpResponse::InternalServerError().json(serde_json::json!({
"error": "读取原始图片失败"
}))
}
}
} else {
HttpResponse::NotFound().json(serde_json::json!({
"error": "原始图片文件不存在"
}))
}
}
Ok(None) => HttpResponse::NotFound().json(serde_json::json!({
"error": "图片不存在"
})),
Err(e) => {
eprintln!("获取图片信息失败: {}", e);
HttpResponse::InternalServerError().json(serde_json::json!({
"error": "获取图片信息失败"
}))
}
}
}
pub async fn get_albums(albums_cache: web::Data<Arc<RwLock<AlbumsCache>>>) -> impl Responder {
// 从内存缓存获取数据
@@ -255,17 +181,18 @@ pub async fn search_albums(
let q = query.get("q").map(|s| s.as_str()).unwrap_or("");
// 输入验证
if let Err(validation_error) = validate_search_query(q) {
return match validation_error {
ValidationError::QueryTooLong => HttpResponse::BadRequest().json(serde_json::json!({
"error": "搜索查询过长，最多100个字符"
})),
ValidationError::InvalidCharacters => {
HttpResponse::BadRequest().json(serde_json::json!({
"error": "搜索查询包含无效字符"
}))
}
};
if q.len() > 100 {
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "搜索查询过长，最多100个字符"
}));
}
// 允许Unicode字符（包括中文）、ASCII图形字符和空白字符
// 只阻止控制字符和其他潜在危险字符
if q.chars().any(|c| c.is_control() || c == '\0') {
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "搜索查询包含无效字符"
}));
}
let page = query
@@ -384,7 +311,7 @@ pub async fn create_album(
}
}
/// 获取压缩图片（后端压缩，前端解压）
/// 获取原始图片
pub async fn get_image(
db: web::Data<Arc<Database>>,
image_processor: web::Data<Arc<ImageProcessor>>,
@@ -406,39 +333,11 @@ pub async fn get_image(
if original_path.exists() {
match fs::read(&original_path) {
Ok(data) => {
// 检查是否支持压缩（仅对JPEG/PNG图片进行压缩）
let should_compress = image.mime_type == "image/jpeg" || image.mime_type == "image/png";
if should_compress {
// 使用Brotli压缩图片数据
match compress_image_data(&data) {
Ok(compressed_data) => {
HttpResponse::Ok()
.content_type("application/octet-stream") // 压缩数据使用二进制格式
.insert_header(("Content-Encoding", "br")) // Brotli压缩
.insert_header(("X-Original-Content-Type", image.mime_type.clone())) // 保留原始类型信息
.insert_header(("Cache-Control", "public, max-age=31536000")) // 1年缓存
.insert_header(("ETag", format!("\"{}\"", image_id)))
.body(compressed_data)
}
Err(e) => {
eprintln!("图片压缩失败: {}", e);
// 压缩失败时返回原始数据
HttpResponse::Ok()
.content_type(image.mime_type.clone())
.insert_header(("Cache-Control", "public, max-age=31536000"))
.insert_header(("ETag", format!("\"{}\"", image_id)))
.body(data)
}
}
} else {
// 不支持压缩的图片类型直接返回
HttpResponse::Ok()
.content_type(image.mime_type.clone())
.insert_header(("Cache-Control", "public, max-age=31536000"))
.insert_header(("ETag", format!("\"{}\"", image_id)))
.body(data)
}
HttpResponse::Ok()
.content_type(image.mime_type.clone())
.insert_header(("Cache-Control", "public, max-age=31536000"))
.insert_header(("ETag", format!("\"{}\"", image_id)))
.body(data)
}
Err(e) => {
eprintln!("读取原始图片失败: {}", e);
@@ -463,19 +362,4 @@ pub async fn get_image(
}))
}
}
}
/// 使用Brotli压缩图片数据
fn compress_image_data(data: &[u8]) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
use brotli::CompressorWriter;
use std::io::Write;
let mut compressed = Vec::new();
{
let mut compressor = CompressorWriter::new(&mut compressed, 4096, 11, 22); // 高质量压缩
Write::write_all(&mut compressor, data)?;
compressor.flush()?;
}
Ok(compressed)
}