Files
furumi-ng/furumi-agent/src/merge.rs
AB-UK 0ba1caaa23
All checks were successful
Publish Metadata Agent Image / build-and-push-image (push) Successful in 1m13s
Publish Web Player Image / build-and-push-image (push) Successful in 1m13s
Publish Server Image / build-and-push-image (push) Successful in 2m28s
Added merge
2026-03-19 01:09:49 +00:00

210 lines
8.1 KiB
Rust

use std::sync::Arc;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::db;
use crate::web::AppState;
use crate::ingest::normalize::call_ollama;
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct MergeProposal {
pub canonical_artist_name: String,
pub winner_artist_id: i64,
pub album_mappings: Vec<AlbumMapping>,
pub notes: String,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct AlbumMapping {
pub source_album_id: i64,
pub canonical_name: String,
pub merge_into_album_id: Option<i64>,
}
pub async fn propose_merge(state: &Arc<AppState>, merge_id: Uuid) -> anyhow::Result<()> {
db::update_merge_status(&state.pool, merge_id, "processing", None).await?;
let merge = db::get_artist_merge(&state.pool, merge_id).await?
.ok_or_else(|| anyhow::anyhow!("Merge not found: {}", merge_id))?;
let source_ids: Vec<i64> = serde_json::from_str(&merge.source_artist_ids)
.map_err(|e| anyhow::anyhow!("Invalid source_artist_ids: {}", e))?;
let artists_data = db::get_artists_full_data(&state.pool, &source_ids).await?;
let user_message = build_merge_message(&artists_data);
let response = call_ollama(
&state.config.ollama_url,
&state.config.ollama_model,
&state.merge_prompt,
&user_message,
state.config.ollama_auth.as_deref(),
).await?;
let proposal = parse_merge_response(&response)?;
let notes = proposal.notes.clone();
let proposal_json = serde_json::to_string(&proposal)?;
db::update_merge_proposal(&state.pool, merge_id, &proposal_json, &notes).await?;
tracing::info!(id = %merge_id, "Merge proposal generated");
Ok(())
}
fn build_merge_message(artists: &[db::ArtistFullData]) -> String {
let mut msg = String::from("## Artists to merge\n\n");
for artist in artists {
msg.push_str(&format!("### Artist ID {}: \"{}\"\n", artist.id, artist.name));
if artist.albums.is_empty() {
msg.push_str(" (no albums)\n");
}
for album in &artist.albums {
let year_str = album.year.map(|y| format!(" ({})", y)).unwrap_or_default();
msg.push_str(&format!(" Album ID {}: \"{}\"{}\n", album.id, album.name, year_str));
for track in &album.tracks {
let num = track.track_number.map(|n| format!("{:02}. ", n)).unwrap_or_default();
msg.push_str(&format!(" - {}\"{}\" [track_id={}]\n", num, track.title, track.id));
}
}
msg.push('\n');
}
msg
}
fn parse_merge_response(response: &str) -> anyhow::Result<MergeProposal> {
let cleaned = response.trim();
let json_str = if cleaned.starts_with("```") {
let start = cleaned.find('{').unwrap_or(0);
let end = cleaned.rfind('}').map(|i| i + 1).unwrap_or(cleaned.len());
&cleaned[start..end]
} else {
cleaned
};
serde_json::from_str(json_str)
.map_err(|e| anyhow::anyhow!("Failed to parse merge LLM response: {} — raw: {}", e, response))
}
pub async fn execute_merge(state: &Arc<AppState>, merge_id: Uuid) -> anyhow::Result<()> {
let merge = db::get_artist_merge(&state.pool, merge_id).await?
.ok_or_else(|| anyhow::anyhow!("Merge not found"))?;
let proposal_str = merge.proposal.ok_or_else(|| anyhow::anyhow!("No proposal to execute"))?;
let proposal: MergeProposal = serde_json::from_str(&proposal_str)?;
let source_ids: Vec<i64> = serde_json::from_str(&merge.source_artist_ids)?;
let loser_ids: Vec<i64> = source_ids.iter().copied()
.filter(|&id| id != proposal.winner_artist_id).collect();
// 1. Rename winner artist to canonical name
db::rename_artist(&state.pool, proposal.winner_artist_id, &proposal.canonical_artist_name).await?;
// 2. Process album mappings
for mapping in &proposal.album_mappings {
if let Some(target_id) = mapping.merge_into_album_id {
merge_albums_into(&state.pool, mapping.source_album_id, target_id).await?;
} else {
// Rename album, then move to winner — merging if winner already has same name
db::rename_album(&state.pool, mapping.source_album_id, &mapping.canonical_name).await?;
if let Some(existing_id) = db::find_album_by_artist_id_and_name(
&state.pool, proposal.winner_artist_id, &mapping.canonical_name,
).await? {
if existing_id != mapping.source_album_id {
merge_albums_into(&state.pool, mapping.source_album_id, existing_id).await?;
}
// else: source_album already belongs to winner with that name — nothing to do
} else {
db::set_album_artist(&state.pool, mapping.source_album_id, proposal.winner_artist_id).await?;
}
}
}
// 3. Move remaining albums from losers to winner, merging name conflicts
for &loser_id in &loser_ids {
let albums = db::get_albums_for_artist(&state.pool, loser_id).await?;
for album in albums {
if let Some(existing_id) = db::find_album_by_artist_id_and_name(
&state.pool, proposal.winner_artist_id, &album.name,
).await? {
merge_albums_into(&state.pool, album.id, existing_id).await?;
} else {
db::set_album_artist(&state.pool, album.id, proposal.winner_artist_id).await?;
}
}
}
// 4. Move track_artists from losers to winner
for &loser_id in &loser_ids {
db::move_track_artists(&state.pool, loser_id, proposal.winner_artist_id).await?;
}
// 5. Move files on disk and update storage paths
let tracks = db::get_tracks_with_albums_for_artist(&state.pool, proposal.winner_artist_id).await?;
for track in &tracks {
let current = std::path::Path::new(&track.storage_path);
let filename = match current.file_name() {
Some(f) => f.to_string_lossy().to_string(),
None => continue,
};
let album_name = track.album_name.as_deref().unwrap_or("Unknown Album");
let new_path = state.config.storage_dir
.join(sanitize(&proposal.canonical_artist_name))
.join(sanitize(album_name))
.join(&filename);
if current != new_path.as_path() {
if current.exists() {
if let Some(parent) = new_path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
let moved = tokio::fs::rename(current, &new_path).await;
if moved.is_err() {
if let Ok(_) = tokio::fs::copy(current, &new_path).await {
let _ = tokio::fs::remove_file(current).await;
}
}
}
db::update_track_storage_path(&state.pool, track.id, &new_path.to_string_lossy()).await?;
}
}
// 6. Delete loser artists
for &loser_id in &loser_ids {
db::delete_artist(&state.pool, loser_id).await?;
}
// 7. Mark approved
db::update_merge_status(&state.pool, merge_id, "approved", None).await?;
tracing::info!(id = %merge_id, "Merge executed successfully");
Ok(())
}
/// Merge source album into target: deduplicate tracks, move the rest, delete source.
async fn merge_albums_into(pool: &sqlx::PgPool, source_id: i64, target_id: i64) -> anyhow::Result<()> {
let dup_ids = db::get_duplicate_track_ids_in_albums(pool, source_id, target_id).await?;
for dup_id in dup_ids {
if let Ok(Some(path)) = db::get_track_storage_path(pool, dup_id).await {
let p = std::path::Path::new(&path);
if p.exists() {
let _ = tokio::fs::remove_file(p).await;
}
}
db::delete_track(pool, dup_id).await?;
}
db::move_tracks_to_album(pool, source_id, target_id).await?;
db::delete_album(pool, source_id).await?;
Ok(())
}
fn sanitize(name: &str) -> String {
name.chars()
.map(|c| match c {
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' | '\0' => '_',
_ => c,
})
.collect::<String>()
.trim()
.trim_matches('.')
.to_owned()
}