use super::Database; use crate::models::*; use sqlx::{Row, Sqlite, Transaction}; use std::collections::HashMap; use std::ops::DerefMut; use uuid::Uuid; impl Database { // ── 人员查询 ── pub async fn get_all_person_names(&self) -> Vec<(String, i64)> { sqlx::query("SELECT p.name, p.total_albums FROM persons p ORDER BY p.total_albums DESC") .fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| (r.get::("name"), r.get::("total_albums"))).collect() } pub async fn get_all_aliases(&self) -> Vec<(String, String)> { sqlx::query("SELECT a.alias_name, p.name as person_name FROM person_aliases a JOIN persons p ON a.person_id = p.id") .fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| (r.get::("alias_name"), r.get::("person_name"))).collect() } pub async fn get_all_persons_with_counts(&self) -> Vec<(i64, String, i64, i64)> { sqlx::query("SELECT id, name, total_albums, total_images FROM persons ORDER BY total_albums DESC") .fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| (r.get::("id"), r.get::("name"), r.get::("total_albums"), r.get::("total_images"))).collect() } pub async fn get_person_by_name(&self, name: &str) -> Option { let r = sqlx::query("SELECT * FROM persons WHERE name = ?").bind(name).fetch_optional(&self.pool).await.ok()??; Some(Person { id: r.get("id"), name: r.get("name"), display_name: r.get("display_name"), total_albums: r.get("total_albums"), total_images: r.get("total_images"), original_names: r.get("original_names"), created_at: r.get("created_at"), updated_at: r.get("updated_at"), }) } pub async fn get_person_albums(&self, person_id: i64) -> Vec { sqlx::query( r#"SELECT a.id, a.title, a.document, a.description, a.cover_image_id, a.image_count, a.created_at FROM albums a JOIN person_albums pa ON a.id = pa.album_id WHERE pa.person_id = ? ORDER BY a.created_at DESC"# ).bind(person_id).fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| AlbumInfo { id: Uuid::parse_str(&r.get::("id")).unwrap_or_else(|_| Uuid::nil()), title: r.get("title"), document: r.get("document"), description: r.get("description"), cover_image_id: r.get("cover_image_id"), image_count: r.get("image_count"), created_at: r.get("created_at"), }).collect() } pub async fn get_person_aliases(&self, person_id: i64) -> Vec { sqlx::query("SELECT * FROM person_aliases WHERE person_id = ?").bind(person_id) .fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| PersonAlias { id: r.get("id"), person_id: r.get("person_id"), alias_type: r.get("alias_type"), alias_name: r.get("alias_name"), created_at: r.get("created_at"), }).collect() } pub async fn search_persons(&self, query: &str) -> Vec { let pattern = format!("%{}%", query); let rows = sqlx::query( r#"SELECT DISTINCT p.id, p.name, p.display_name, p.total_albums, p.total_images FROM persons p LEFT JOIN person_aliases a ON p.id = a.person_id WHERE p.name LIKE ? OR a.alias_name LIKE ? ORDER BY p.total_albums DESC"# ).bind(&pattern).bind(&pattern).fetch_all(&self.pool).await.unwrap_or_default(); let pids: Vec = rows.iter().map(|r| r.get::("id")).collect(); let aliases = self.batch_get_person_aliases(&pids).await; rows.iter().map(|r| { let pid = r.get::("id"); PersonWithCounts { id: pid, name: r.get("name"), display_name: r.get("display_name"), total_albums: r.get("total_albums"), total_images: r.get("total_images"), aliases: aliases.get(&pid).cloned().unwrap_or_default(), } }).collect() } pub async fn get_all_persons_with_aliases(&self) -> Vec { let rows = sqlx::query("SELECT id, name, display_name, total_albums, total_images FROM persons ORDER BY total_albums DESC") .fetch_all(&self.pool).await.unwrap_or_default(); let pids: Vec = rows.iter().map(|r| r.get::("id")).collect(); let aliases = self.batch_get_person_aliases(&pids).await; rows.iter().map(|r| { let pid = r.get::("id"); PersonWithCounts { id: pid, name: r.get("name"), display_name: r.get("display_name"), total_albums: r.get("total_albums"), total_images: r.get("total_images"), aliases: aliases.get(&pid).cloned().unwrap_or_default(), } }).collect() } /// 批量查询多个人物的别名(避免 N+1) async fn batch_get_person_aliases(&self, pids: &[i64]) -> HashMap> { if pids.is_empty() { return HashMap::new(); } let placeholders: Vec = pids.iter().map(|_| "?".to_string()).collect(); let sql = format!( "SELECT person_id, alias_name FROM person_aliases WHERE person_id IN ({})", placeholders.join(",") ); let mut q = sqlx::query(&sql); for pid in pids { q = q.bind(pid); } let rows = q.fetch_all(&self.pool).await.unwrap_or_default(); let mut map: HashMap> = HashMap::new(); for r in &rows { let pid: i64 = r.get("person_id"); let alias: String = r.get("alias_name"); map.entry(pid).or_default().push(alias); } map } // ── 标签查询 ── pub async fn get_popular_tags(&self, limit: i64) -> Vec { // 从 tag_dict 获取过滤词(停用词、品牌词、排除名) // 注意:必须 clone 到本地变量,不能持 RwLockReadGuard 跨 await let (stop_words, excluded, included) = { let dict = crate::engine::tag_dict::get_tag_dict(); ( dict.stop_words.clone(), dict.excluded_person_names.clone(), dict.included_person_names.clone(), ) }; // 标签自身小写的集合用于快速过滤 let filtered_tags: Vec = sqlx::query( r#"SELECT tag, COUNT(*) as cnt, COUNT(DISTINCT album_id) as album_cnt FROM album_tags GROUP BY tag ORDER BY cnt DESC LIMIT ?"# ).bind(limit * 3).fetch_all(&self.pool).await.unwrap_or_default() .iter().filter_map(|r| { let tag: String = r.get("tag"); let tag_lower = tag.to_lowercase(); // 过滤停用词 if stop_words.contains(&tag_lower) || stop_words.contains(&tag) { return None; } // 过滤品牌/工作室名 if excluded.contains(&tag_lower) || excluded.contains(&tag) { return None; } // 过滤包含人名(如果标签本身就是个人名,也不应该出现在标签中) if included.contains(&tag_lower) || included.contains(&tag) { return None; } Some(TagStat { tag: r.get("tag"), count: r.get("cnt"), album_count: r.get("album_cnt"), }) }).collect(); // 只保留 top limit 个 let mut result: Vec = filtered_tags; result.truncate(limit as usize); result } /// 批量获取所有 tag→albums 映射(供缓存使用) pub async fn get_all_tag_albums(&self) -> HashMap> { let rows = sqlx::query( r#"SELECT t.tag, a.id, a.title, a.document, a.description, a.cover_image_id, a.image_count, a.created_at FROM album_tags t JOIN albums a ON t.album_id = a.id ORDER BY a.created_at DESC"# ).fetch_all(&self.pool).await.unwrap_or_default(); let mut map: HashMap> = HashMap::new(); for r in &rows { let tag: String = r.get("tag"); let album = AlbumInfo { id: Uuid::parse_str(&r.get::("id")).unwrap_or_else(|_| Uuid::nil()), title: r.get("title"), document: r.get("document"), description: r.get("description"), cover_image_id: r.get("cover_image_id"), image_count: r.get("image_count"), created_at: r.get("created_at"), }; map.entry(tag).or_default().push(album); } map } pub async fn get_tag_co_occurrences(&self, tag: &str, limit: i64) -> Vec { let rows = sqlx::query( r#"SELECT t2.tag as co_tag, COUNT(*) as cnt FROM album_tags t1 JOIN album_tags t2 ON t1.album_id = t2.album_id WHERE t1.tag = ? AND t2.tag != ? GROUP BY t2.tag ORDER BY cnt DESC LIMIT ?"# ).bind(tag).bind(tag).bind(limit).fetch_all(&self.pool).await.unwrap_or_default(); let total = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM album_tags WHERE tag = ?") .bind(tag).fetch_one(&self.pool).await.unwrap_or(1); rows.iter().map(|r| { let cnt: i64 = r.get("cnt"); TagCoOccurrence { tag: tag.to_string(), co_tag: r.get("co_tag"), count: cnt, score: cnt as f64 / total as f64, } }).collect() } pub async fn search_albums_by_tag(&self, tag: &str) -> Vec { sqlx::query( r#"SELECT a.id, a.title, a.document, a.description, a.cover_image_id, a.image_count, a.created_at FROM albums a JOIN album_tags t ON a.id = t.album_id WHERE t.tag = ? ORDER BY a.created_at DESC"# ).bind(tag).fetch_all(&self.pool).await.unwrap_or_default() .iter().map(|r| AlbumInfo { id: Uuid::parse_str(&r.get::("id")).unwrap_or_else(|_| Uuid::nil()), title: r.get("title"), document: r.get("document"), description: r.get("description"), cover_image_id: r.get("cover_image_id"), image_count: r.get("image_count"), created_at: r.get("created_at"), }).collect() } // ── 统计 ── pub async fn get_metadata_stats(&self) -> MetadataStats { MetadataStats { total_persons: sqlx::query_scalar("SELECT COUNT(*) FROM persons").fetch_one(&self.pool).await.unwrap_or(0), total_aliases: sqlx::query_scalar("SELECT COUNT(*) FROM person_aliases").fetch_one(&self.pool).await.unwrap_or(0), total_albums_with_person: sqlx::query_scalar("SELECT COUNT(*) FROM albums WHERE person_name_original IS NOT NULL AND person_name_original != ''").fetch_one(&self.pool).await.unwrap_or(0), total_albums_without_person: sqlx::query_scalar("SELECT COUNT(*) FROM albums WHERE person_name_original IS NULL OR person_name_original = ''").fetch_one(&self.pool).await.unwrap_or(0), total_tags: sqlx::query_scalar("SELECT COUNT(*) FROM album_tags").fetch_one(&self.pool).await.unwrap_or(0), } } // ── 写入 ── pub async fn insert_person(&self, name: &str, display_name: Option<&str>) -> i64 { if let Some(p) = self.get_person_by_name(name).await { return p.id; } sqlx::query("INSERT OR IGNORE INTO persons (name, display_name) VALUES (?, ?)") .bind(name).bind(display_name).execute(&self.pool).await.ok(); 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(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(tx.deref_mut()).await.ok(); if let Ok(Some(r)) = sqlx::query("SELECT id FROM persons WHERE name = ?") .bind(name).fetch_optional(tx.deref_mut()).await { return r.get::("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(tx.deref_mut()).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(tx.deref_mut()).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(); } /// 删除某相册的所有人物关联,返回被影响的旧 person_id 列表 pub async fn delete_album_person_albums(&self, album_id: &str) -> Vec { let old_ids: Vec = sqlx::query_scalar("SELECT person_id FROM person_albums WHERE album_id = ?") .bind(album_id).fetch_all(&self.pool).await.unwrap_or_default(); sqlx::query("DELETE FROM person_albums WHERE album_id = ?").bind(album_id).execute(&self.pool).await.ok(); old_ids } /// 删除某相册的所有人物关联(事务版),返回被影响的旧 person_id 列表 pub async fn delete_album_person_albums_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, album_id: &str) -> Vec { let old_ids: Vec = sqlx::query_scalar("SELECT person_id FROM person_albums WHERE album_id = ?") .bind(album_id).fetch_all(tx.deref_mut()).await.unwrap_or_default(); sqlx::query("DELETE FROM person_albums WHERE album_id = ?").bind(album_id).execute(tx.deref_mut()).await.ok(); old_ids } /// 删除某人物下所有别名(写入新别名前清旧别名,配合 UNIQUE INDEX 防止重复) pub async fn delete_person_aliases(&self, person_id: i64) { sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(person_id).execute(&self.pool).await.ok(); } pub async fn delete_person_aliases_with_tx(&self, tx: &mut Transaction<'_, Sqlite>, person_id: i64) { sqlx::query("DELETE FROM person_aliases WHERE person_id = ?").bind(person_id).execute(tx.deref_mut()).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(); } /// 把 src_id 的所有相册关联、别名移动到 dst_id,然后删除 src_id pub async fn merge_person_into(&self, src_id: i64, dst_id: i64) { let _guard = self.write_lock.lock().await; // 整个合并过程是一个原子操作,用事务包裹避免逐条 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(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(tx.deref_mut()).await.ok(); // 3. 把 src 的原名也作为别名加入 dst(使用事务连接,避免额外占用连接池) if let Some(r) = sqlx::query("SELECT id, name, display_name FROM persons WHERE id = ?") .bind(src_id).fetch_optional(tx.deref_mut()).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 的关联和别名 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(tx.deref_mut()).await.ok(); // 6. 更新 dst 的统计数据 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 { let r = sqlx::query("SELECT * FROM persons WHERE id = ?").bind(id).fetch_optional(&self.pool).await.ok()??; Some(Person { id: r.get("id"), name: r.get("name"), display_name: r.get("display_name"), total_albums: r.get("total_albums"), total_images: r.get("total_images"), original_names: r.get("original_names"), created_at: r.get("created_at"), updated_at: r.get("updated_at"), }) } pub async fn update_album_person(&self, 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(&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(tx.deref_mut()).await.ok(); } /// 清空该人物所有关联相册的 person 字段(当该人物被确认为非人物后调用) pub async fn update_album_person_by_person_id(&self, person_id: i64) { sqlx::query( "UPDATE albums SET person_name_original = NULL, person_name_cleaned = NULL, \ processed_at = datetime('now') WHERE id IN (SELECT album_id FROM person_albums WHERE person_id = ?)" ).bind(person_id).execute(&self.pool).await.ok(); } pub async fn update_album_tags(&self, 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(&self.pool).await.ok(); // 用一个事务批量替换 album_tags,避免逐条 autocommit 导致巨慢 if let Ok(tags) = serde_json::from_str::>(tags_json) { let mut tx = match self.pool.begin().await { Ok(t) => t, Err(_) => return, }; // 单条 DELETE + 一条多值 INSERT sqlx::query("DELETE FROM album_tags WHERE album_id = ?") .bind(album_id) .execute(tx.deref_mut()) .await.ok(); // 构建多值 INSERT, 避免逐条执行 if !tags.is_empty() { let placeholders: Vec = 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(tx.deref_mut()).await.ok(); } tx.commit().await.ok(); } } /// 与 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(tx.deref_mut()).await.ok(); sqlx::query("DELETE FROM album_tags WHERE album_id = ?") .bind(album_id) .execute(tx.deref_mut()) .await.ok(); if let Ok(tags) = serde_json::from_str::>(tags_json) { if !tags.is_empty() { let placeholders: Vec = 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(tx.deref_mut()).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 = ?), 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(&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(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() } }