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

update

Browse Source
This commit is contained in:
lychang
2026-07-20 00:50:45 +08:00
parent b1f0c2f36a
commit da412c1c76
7 changed files with 118 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/db/album.rs
View File

@@ -98,6 +98,7 @@ impl Database {
// ── 写入 ──
pub async fn create_album(&self, album: CreateAlbum) -> Result<Uuid, sqlx::Error> {
let _guard = self.write_lock.lock().await;
let id = Uuid::new_v4();
let now = chrono::Utc::now().to_rfc3339();
sqlx::query(
@@ -108,6 +109,7 @@ impl Database {
}
pub async fn delete_album(&self, album_id: Uuid) -> Result<(), sqlx::Error> {
let _guard = self.write_lock.lock().await;
sqlx::query("DELETE FROM albums WHERE id = ?").bind(album_id.to_string()).execute(&self.pool).await?;
Ok(())
}

19
src/db/connection.rs
View File

@@ -1,9 +1,12 @@
use sqlx::{SqlitePool, sqlite::SqlitePoolOptions};
use crate::config::Config;
use tokio::sync::Mutex;
pub struct Database {
pub pool: SqlitePool,
pub(crate) pool: SqlitePool,
pub config: Config,
/// 写互斥锁：单协程串行化所有写操作，消除 WAL 锁争用
pub(crate) write_lock: Mutex<()>,
}
impl Database {
@@ -22,26 +25,28 @@ impl Database {
}
let pool = SqlitePoolOptions::new()
.max_connections(10)
.min_connections(2)
.max_connections(2) // 写串行化后无需过多连接
.min_connections(1)
.acquire_timeout(std::time::Duration::from_secs(5))
.idle_timeout(std::time::Duration::from_secs(300))
.max_lifetime(std::time::Duration::from_secs(1800))
.connect(url)
.await?;
// SQLite 性能优化
// SQLite 高并发优化 PRAGMA
sqlx::query("PRAGMA journal_mode=WAL;").execute(&pool).await?;
sqlx::query("PRAGMA wal_autocheckpoint=4000;").execute(&pool).await?;
sqlx::query("PRAGMA synchronous=NORMAL;").execute(&pool).await?;
sqlx::query("PRAGMA cache_size=-64000;").execute(&pool).await?;
sqlx::query("PRAGMA busy_timeout=5000;").execute(&pool).await?; // 等待 5s 避免 SQLITE_BUSY
sqlx::query("PRAGMA wal_autocheckpoint=1000;").execute(&pool).await?; // 小 WAL → 短 checkpoint
sqlx::query("PRAGMA journal_size_limit=67108864;").execute(&pool).await?; // WAL 上限 64MB
sqlx::query("PRAGMA cache_size=-64000;").execute(&pool).await?; // 64KB 页缓存
sqlx::query("PRAGMA mmap_size=268435456;").execute(&pool).await?;
sqlx::query("PRAGMA temp_store=MEMORY;").execute(&pool).await?;
sqlx::query("PRAGMA foreign_keys=ON;").execute(&pool).await?;
sqlx::migrate!("./migrations").run(&pool).await?;
let db = Database { pool, config: config.clone() };
let db = Database { pool, config: config.clone(), write_lock: Mutex::new(()) };
db.ensure_image_count_field().await?;
Ok(db)
}

3
src/db/favorite.rs
View File

@@ -46,6 +46,7 @@ impl Database {
}
pub async fn add_favorite(&self, user_id: &str, fav: CreateFavorite) -> Result<Uuid, sqlx::Error> {
let _guard = self.write_lock.lock().await;
let id = Uuid::new_v4();
let now = chrono::Utc::now().to_rfc3339();
sqlx::query("INSERT INTO favorites (id, user_id, image_id, album_id, created_at) VALUES (?, ?, ?, ?, ?)")
@@ -55,11 +56,13 @@ impl Database {
}
pub async fn remove_favorite(&self, favorite_id: Uuid) -> Result<(), sqlx::Error> {
let _guard = self.write_lock.lock().await;
sqlx::query("DELETE FROM favorites WHERE id = ?").bind(favorite_id.to_string()).execute(&self.pool).await?;
Ok(())
}
pub async fn remove_favorite_by_image(&self, user_id: &str, image_id: Uuid) -> Result<(), sqlx::Error> {
let _guard = self.write_lock.lock().await;
sqlx::query("DELETE FROM favorites WHERE user_id = ? AND image_id = ?")
.bind(user_id).bind(image_id.to_string()).execute(&self.pool).await?;
Ok(())

1
src/db/image.rs
View File

@@ -8,6 +8,7 @@ pub use crate::models::image::Image;
impl Database {
pub async fn add_images(&self, images: Vec<CreateImage>) -> Result<Vec<Uuid>, sqlx::Error> {
let _guard = self.write_lock.lock().await;
if images.is_empty() { return Ok(Vec::new()); }
let now = chrono::Utc::now().to_rfc3339();
let mut ids = Vec::new();

67
src/db/metadata.rs
View File

@@ -1,6 +1,6 @@
use super::Database;
use crate::models::*;
use sqlx::{Acquire, Row};
use sqlx::{Acquire, Row, Sqlite, Transaction};
use uuid::Uuid;
impl Database {
@@ -157,22 +157,48 @@ impl Database {
self.get_person_by_name(name).await.map(|p| p.id).unwrap_or(0)
}
/// 与 insert_person 相同，但使用已有的事务
pub async fn insert_person_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, name: &str, display_name: Option<&str>) -> i64 {
if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?")
.bind(name).fetch_optional(&mut *tx).await {
return r.get::<i64, _>("id");
}
sqlx::query("INSERT OR IGNORE INTO persons (name, display_name) VALUES (?, ?)")
.bind(name).bind(display_name).execute(&mut *tx).await.ok();
if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?")
.bind(name).fetch_optional(&mut *tx).await {
return r.get::<i64, _>("id");
}
0
}
pub async fn insert_person_alias(&self, person_id: i64, alias_type: &str, alias_name: &str) {
sqlx::query("INSERT OR IGNORE INTO person_aliases (person_id, alias_type, alias_name) VALUES (?, ?, ?)")
.bind(person_id).bind(alias_type).bind(alias_name).execute(&self.pool).await.ok();
}
pub async fn insert_person_alias_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64, alias_type: &str, alias_name: &str) {
sqlx::query("INSERT OR IGNORE INTO person_aliases (person_id, alias_type, alias_name) VALUES (?, ?, ?)")
.bind(person_id).bind(alias_type).bind(alias_name).execute(&mut *tx).await.ok();
}
pub async fn insert_person_album(&self, person_id: i64, album_id: &str) {
sqlx::query("INSERT OR IGNORE INTO person_albums (person_id, album_id) VALUES (?, ?)")
.bind(person_id).bind(album_id).execute(&self.pool).await.ok();
}
pub async fn insert_person_album_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64, album_id: &str) {
sqlx::query("INSERT OR IGNORE INTO person_albums (person_id, album_id) VALUES (?, ?)")
.bind(person_id).bind(album_id).execute(&mut *tx).await.ok();
}
pub async fn delete_person_albums(&self, person_id: i64) {
sqlx::query("DELETE FROM person_albums WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok();
}
/// 彻底删除一个人物（包括别名、相册关联、自身）
pub async fn delete_person_completely(&self, person_id: i64) {
let _guard = self.write_lock.lock().await;
sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok();
sqlx::query("DELETE FROM person_albums WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok();
sqlx::query("DELETE FROM persons WHERE id = ?").bind(person_id).execute(&self.pool).await.ok();
@@ -223,6 +249,11 @@ impl Database {
.bind(person_name_original).bind(person_name_cleaned).bind(album_id).execute(&self.pool).await.ok();
}
pub async fn update_album_person_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, album_id: &str, person_name_original: &str, person_name_cleaned: &str) {
sqlx::query("UPDATE albums SET person_name_original = ?, person_name_cleaned = ?, processed_at = datetime('now') WHERE id = ?")
.bind(person_name_original).bind(person_name_cleaned).bind(album_id).execute(&mut *tx).await.ok();
}
/// 清空该人物所有关联相册的 person 字段（当该人物被确认为非人物后调用）
pub async fn update_album_person_by_person_id(&self, person_id: i64) {
sqlx::query(
@@ -266,6 +297,32 @@ impl Database {
}
}
/// 与 update_album_tags 相同，但使用已有的事务（不启动内部子事务）
pub async fn update_album_tags_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, album_id: &str, tags_json: &str, all_words_json: &str) {
sqlx::query("UPDATE albums SET tags = ?, all_words = ?, processed_at = datetime('now') WHERE id = ?")
.bind(tags_json).bind(all_words_json).bind(album_id).execute(&mut *tx).await.ok();
sqlx::query("DELETE FROM album_tags WHERE album_id = ?")
.bind(album_id)
.execute(&mut *tx)
.await.ok();
if let Ok(tags) = serde_json::from_str::<Vec<String>>(tags_json) {
if !tags.is_empty() {
let placeholders: Vec<String> = tags.iter().map(|_| "(?, ?)".to_string()).collect();
let sql = format!(
"INSERT OR IGNORE INTO album_tags (album_id, tag) VALUES {}",
placeholders.join(", ")
);
let mut q = sqlx::query(&sql);
for tag in &tags {
q = q.bind(album_id).bind(tag);
}
q.execute(&mut *tx).await.ok();
}
}
}
pub async fn update_person_counts(&self, person_id: i64) {
sqlx::query(
r#"UPDATE persons SET total_albums = (SELECT COUNT(*) FROM person_albums WHERE person_id = ?),
@@ -274,6 +331,14 @@ impl Database {
).bind(person_id).bind(person_id).bind(person_id).execute(&self.pool).await.ok();
}
pub async fn update_person_counts_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64) {
sqlx::query(
r#"UPDATE persons SET total_albums = (SELECT COUNT(*) FROM person_albums WHERE person_id = ?),
total_images = (SELECT COALESCE(SUM(a.image_count), 0) FROM albums a JOIN person_albums pa ON a.id = pa.album_id WHERE pa.person_id = ?),
updated_at = datetime('now') WHERE id = ?"#
).bind(person_id).bind(person_id).bind(person_id).execute(&mut *tx).await.ok();
}
/// 获取所有相册 ID 集合（用于迁移）
pub async fn get_all_album_ids(&self) -> Vec<String> {
sqlx::query_scalar::<_, String>("SELECT id FROM albums").fetch_all(&self.pool).await.unwrap_or_default()

3
src/db/migration.rs
View File

@@ -24,6 +24,9 @@ pub struct MigrationStats {
impl Database {
/// 从老数据库迁移元数据
pub async fn migrate_from_old_db(&self, old_db_path: &str) -> Result<MigrationStats, String> {
// 串行化写入：迁移过程中禁止其他写入操作
let _guard = self.write_lock.lock().await;
if !std::path::Path::new(old_db_path).exists() {
return Err(format!("老数据库不存在: {}", old_db_path));
}

43
src/engine/pipeline.rs
View File

@@ -17,6 +17,7 @@ use crate::db::Database;
use crate::engine::title_parser;
use crate::engine::tag_extractor;
use crate::engine::person_merger;
use sqlx::Acquire;
use std::collections::HashSet;
// ════════════════════════════════════════════════
@@ -201,17 +202,20 @@ pub async fn phase_detect_person_in_tags(db: &Database) -> (DirtySet, usize) {
// ── 写入 person_name（直接提取，无需等待 Phase 1）──
let (primary, aliases) = title_parser::split_person_name_with_aliases(person_name);
let final_name = if primary.is_empty() { person_name.clone() } else { primary };
db.update_album_person(&album.id, person_name, &final_name).await;
// 用事务包裹所有写入，避免逐条 autocommit 导致巨慢
let mut tx = db.pool.begin().await.expect("begin tx");
db.update_album_person_with_tx(&mut tx, &album.id, person_name, &final_name).await;
if !title_parser::is_excluded_name(&final_name) {
let pid = db.insert_person(&final_name, None).await;
db.insert_person_album(pid, &album.id).await;
let pid = db.insert_person_with_tx(&mut tx, &final_name, None).await;
db.insert_person_album_with_tx(&mut tx, pid, &album.id).await;
for (at, an) in &aliases {
if !an.is_empty() && an != &final_name {
db.insert_person_alias(pid, at, an).await;
db.insert_person_alias_with_tx(&mut tx, pid, at, an).await;
}
}
db.update_person_counts(pid).await;
db.update_person_counts_with_tx(&mut tx, pid).await;
}
// ── 写入清理后的标签 ──
@@ -223,7 +227,9 @@ pub async fn phase_detect_person_in_tags(db: &Database) -> (DirtySet, usize) {
.filter(|pair| !known_persons.contains(&pair[0].to_lowercase()))
.collect();
let words_json_new = serde_json::to_string(&all_words_new).unwrap_or_else(|_| "[]".to_string());
db.update_album_tags(&album.id, &tags_json_new, &words_json_new).await;
db.update_album_tags_with_tx(&mut tx, &album.id, &tags_json_new, &words_json_new).await;
tx.commit().await.expect("commit tx");
if has_person { dirty.add(&album.id); }
}
@@ -323,18 +329,23 @@ pub async fn phase_parse_dirty(db: &Database, dirty_set: &DirtySet) -> (usize, u
jieba_person
};
// 用事务包裹所有写入，避免逐条 autocommit 导致巨慢
let mut pending_tx: Option<sqlx::Transaction<'_, sqlx::Sqlite>> = None;
if let Some(name) = &best_person {
let (primary, aliases) = title_parser::split_person_name_with_aliases(name);
let final_name = if primary.is_empty() { name.clone() } else { primary };
db.update_album_person(&album.id, name, &final_name).await;
let pid = db.insert_person(&final_name, None).await;
db.insert_person_album(pid, &album.id).await;
let mut tx = db.pool.begin().await.expect("begin tx");
db.update_album_person_with_tx(&mut tx, &album.id, name, &final_name).await;
let pid = db.insert_person_with_tx(&mut tx, &final_name, None).await;
db.insert_person_album_with_tx(&mut tx, pid, &album.id).await;
for (at, an) in &aliases {
if !an.is_empty() && an != &final_name {
db.insert_person_alias(pid, at, an).await;
db.insert_person_alias_with_tx(&mut tx, pid, at, an).await;
}
}
db.update_person_counts(pid).await;
db.update_person_counts_with_tx(&mut tx, pid).await;
pending_tx = Some(tx);
person_updated += 1;
}
@@ -354,8 +365,16 @@ pub async fn phase_parse_dirty(db: &Database, dirty_set: &DirtySet) -> (usize, u
let tags_json = serde_json::to_string(&tag_names).unwrap_or_else(|_| "[]".to_string());
let all_words: Vec<[String; 2]> = unique.iter().map(|(t, tp)| [t.clone(), tp.clone()]).collect();
let words_json = serde_json::to_string(&all_words).unwrap_or_else(|_| "[]".to_string());
db.update_album_tags(&album.id, &tags_json, &words_json).await;
let mut tx = match pending_tx.take() {
Some(t) => t,
None => db.pool.begin().await.expect("begin tx"),
};
db.update_album_tags_with_tx(&mut tx, &album.id, &tags_json, &words_json).await;
tx.commit().await.expect("commit tx");
tag_updated += 1;
} else if let Some(tx) = pending_tx.take() {
// 只有 person 写入，没有标签写入
tx.commit().await.expect("commit tx");
}
}