diff --git a/src/db/metadata.rs b/src/db/metadata.rs index 37bb747..66c1bb9 100644 --- a/src/db/metadata.rs +++ b/src/db/metadata.rs @@ -1,6 +1,7 @@ use super::Database; use crate::models::*; -use sqlx::{Acquire, Row, Sqlite, Transaction}; +use sqlx::{Row, Sqlite, Transaction}; +use std::ops::DerefMut; use uuid::Uuid; impl Database { @@ -160,13 +161,13 @@ impl Database { /// 与 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 { + .bind(name).fetch_optional(tx.deref_mut()).await { return r.get::("id"); } sqlx::query("INSERT OR IGNORE INTO persons (name, display_name) VALUES (?, ?)") - .bind(name).bind(display_name).execute(&mut *tx).await.ok(); + .bind(name).bind(display_name).execute(tx.deref_mut()).await.ok(); if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?") - .bind(name).fetch_optional(&mut *tx).await { + .bind(name).fetch_optional(tx.deref_mut()).await { return r.get::("id"); } 0 @@ -179,7 +180,7 @@ impl Database { 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(); + .bind(person_id).bind(alias_type).bind(alias_name).execute(tx.deref_mut()).await.ok(); } pub async fn insert_person_album(&self, person_id: i64, album_id: &str) { @@ -189,7 +190,7 @@ impl Database { 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(); + .bind(person_id).bind(album_id).execute(tx.deref_mut()).await.ok(); } pub async fn delete_person_albums(&self, person_id: i64) { @@ -206,33 +207,43 @@ impl Database { /// 把 src_id 的所有相册关联、别名移动到 dst_id，然后删除 src_id pub async fn merge_person_into(&self, src_id: i64, dst_id: i64) { + // 整个合并过程是一个原子操作，用事务包裹避免逐条 autocommit + let mut tx = match self.pool.begin().await { + Ok(t) => t, + Err(_) => return, + }; + // 1. 移动相册关联 sqlx::query("INSERT OR IGNORE INTO person_albums (person_id, album_id) SELECT ?, album_id FROM person_albums WHERE person_id = ?") - .bind(dst_id).bind(src_id).execute(&self.pool).await.ok(); + .bind(dst_id).bind(src_id).execute(tx.deref_mut()).await.ok(); // 2. 移动别名 sqlx::query("INSERT OR IGNORE INTO person_aliases (person_id, alias_type, alias_name) SELECT ?, alias_type, alias_name FROM person_aliases WHERE person_id = ?") - .bind(dst_id).bind(src_id).execute(&self.pool).await.ok(); + .bind(dst_id).bind(src_id).execute(tx.deref_mut()).await.ok(); // 3. 把 src 的原名也作为别名加入 dst - if let Some(src_person) = self.get_person_by_id(src_id).await { - // 原始名作为别名 - self.insert_person_alias(dst_id, "merged_original", &src_person.name).await; - // 如果有 display_name 也作为别名 - if let Some(dn) = &src_person.display_name { - self.insert_person_alias(dst_id, "merged_display", dn).await; + if let Some(r) = sqlx::query("SELECT id, name, display_name FROM persons WHERE id = ?") + .bind(src_id).fetch_optional(&self.pool).await.ok().flatten() + { + let src_name: String = r.get("name"); + let src_display: Option = r.get("display_name"); + self.insert_person_alias_with_tx(&mut tx, dst_id, "merged_original", &src_name).await; + if let Some(ref dn) = src_display { + self.insert_person_alias_with_tx(&mut tx, dst_id, "merged_display", dn).await; } } // 4. 删除 src 的关联和别名 - self.delete_person_albums(src_id).await; - sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(src_id).execute(&self.pool).await.ok(); + sqlx::query("DELETE FROM person_albums WHERE person_id = ?").bind(src_id).execute(tx.deref_mut()).await.ok(); + sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(src_id).execute(tx.deref_mut()).await.ok(); // 5. 删除 src 人物 - sqlx::query("DELETE FROM persons WHERE id = ?").bind(src_id).execute(&self.pool).await.ok(); + sqlx::query("DELETE FROM persons WHERE id = ?").bind(src_id).execute(tx.deref_mut()).await.ok(); // 6. 更新 dst 的统计数据 - self.update_person_counts(dst_id).await; + self.update_person_counts_with_tx(&mut tx, dst_id).await; + + tx.commit().await.ok(); } pub async fn get_person_by_id(&self, id: i64) -> Option { @@ -251,7 +262,7 @@ impl Database { 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(); + .bind(person_name_original).bind(person_name_cleaned).bind(album_id).execute(tx.deref_mut()).await.ok(); } /// 清空该人物所有关联相册的 person 字段（当该人物被确认为非人物后调用） @@ -276,7 +287,7 @@ impl Database { // 单条 DELETE + 一条多值 INSERT sqlx::query("DELETE FROM album_tags WHERE album_id = ?") .bind(album_id) - .execute(&mut *tx) + .execute(tx.deref_mut()) .await.ok(); // 构建多值 INSERT, 避免逐条执行 @@ -290,7 +301,7 @@ impl Database { for tag in &tags { q = q.bind(album_id).bind(tag); } - q.execute(&mut *tx).await.ok(); + q.execute(tx.deref_mut()).await.ok(); } tx.commit().await.ok(); @@ -300,11 +311,11 @@ 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(); + .bind(tags_json).bind(all_words_json).bind(album_id).execute(tx.deref_mut()).await.ok(); sqlx::query("DELETE FROM album_tags WHERE album_id = ?") .bind(album_id) - .execute(&mut *tx) + .execute(tx.deref_mut()) .await.ok(); if let Ok(tags) = serde_json::from_str::>(tags_json) { @@ -318,7 +329,7 @@ impl Database { for tag in &tags { q = q.bind(album_id).bind(tag); } - q.execute(&mut *tx).await.ok(); + q.execute(tx.deref_mut()).await.ok(); } } } @@ -336,11 +347,11 @@ impl Database { 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(); + ).bind(person_id).bind(person_id).bind(person_id).execute(tx.deref_mut()).await.ok(); } /// 获取所有相册 ID 集合（用于迁移） pub async fn get_all_album_ids(&self) -> Vec { sqlx::query_scalar::<_, String>("SELECT id FROM albums").fetch_all(&self.pool).await.unwrap_or_default() } -} \ No newline at end of file +} diff --git a/src/engine/pipeline.rs b/src/engine/pipeline.rs index 6a76bd6..0e367b4 100644 --- a/src/engine/pipeline.rs +++ b/src/engine/pipeline.rs @@ -17,7 +17,6 @@ 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; // ════════════════════════════════════════════════