Perceptual-hash dedup: web UI + background job (3.7, part 2)

Wires the fingerprint engine into the web Maintenance view.

- web.rs: a DedupState job (one at a time) on its own thread. POST
  /api/maintenance/dedup/scan snapshots the library, mtime-gates against
  stored fingerprints, fingerprints the new/changed videos in parallel
  (progress counter), prunes vanished entries, groups by visual
  similarity, and builds review rows (title/channel/size/files + a
  recommended-keep = largest copy). GET /api/maintenance/dedup/status
  polls progress + results. Deletion reuses /api/maintenance/remove.
- index.html: a "Similar content (perceptual)" section in the Maintenance
  modal — Scan button, live progress bar, grouped results with
  checkboxes (recommended-keep pre-unchecked), and a re-scan that drops
  deleted copies. Poller self-cancels when the modal closes.

Integration test (ffmpeg-gated): generates orig + CRF-38 downscaled
re-encode + an unrelated clip, runs the real scan end-to-end, and asserts
the first two group while the third stays out. 118 tests pass.

Desktop Maintenance UI next.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Luna 2026-06-07 06:17:58 -07:00
parent 6d5c1cae33
commit f30e32707e
3 changed files with 293 additions and 1 deletions

View file

@ -24,7 +24,7 @@
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::path::{Path as StdPath, PathBuf}; use std::path::{Path as StdPath, PathBuf};
use std::process::Stdio; use std::process::Stdio;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use axum::{ use axum::{
@ -133,6 +133,43 @@ pub struct WebState {
/// re-serializing. The Arc lets concurrent responses share one /// re-serializing. The Arc lets concurrent responses share one
/// allocation. Cleared / replaced lazily on the next miss. /// allocation. Cleared / replaced lazily on the next miss.
pub library_body_cache: Mutex<Option<(u64, std::sync::Arc<String>)>>, pub library_body_cache: Mutex<Option<(u64, std::sync::Arc<String>)>>,
/// Background perceptual-dedup job state. The expensive fingerprint pass
/// runs on its own thread; this carries live progress + the last result
/// so `/api/maintenance/dedup/status` can be polled. `Arc` so the worker
/// thread can hold a handle past the request that started it.
pub dedup: std::sync::Arc<DedupState>,
}
/// Live state for the background perceptual-dedup job (see
/// [`post_dedup_scan`]). One job runs at a time, guarded by `running`.
#[derive(Default)]
pub struct DedupState {
running: AtomicBool,
done: AtomicUsize,
total: AtomicUsize,
result: Mutex<Option<Vec<SimilarGroup>>>,
error: Mutex<Option<String>>,
}
/// One video in a perceptual-similarity group, with everything the review UI
/// needs to show it and (optionally) delete its files.
#[derive(serde::Serialize, Clone)]
pub struct SimilarVideo {
video_id: String,
title: String,
channel: String,
location: String,
file_size: Option<u64>,
/// Absolute paths of the video + its sidecars, for the delete action.
files: Vec<String>,
/// The copy the UI recommends keeping (largest file in the group).
recommended_keep: bool,
}
/// A cluster of videos that share visual content across different IDs.
#[derive(serde::Serialize, Clone)]
pub struct SimilarGroup {
videos: Vec<SimilarVideo>,
} }
/// Failed-login tracking entry. After [`LOGIN_LOCKOUT_AFTER`] failures from /// Failed-login tracking entry. After [`LOGIN_LOCKOUT_AFTER`] failures from
@ -1976,6 +2013,132 @@ struct RemoveRequest {
paths: Vec<PathBuf>, paths: Vec<PathBuf>,
} }
/// `POST /api/maintenance/dedup/scan` — start (or no-op if already running)
/// the background perceptual-dedup pass. Fingerprints any new/changed videos
/// (cached by mtime), then groups by visual similarity. Returns immediately;
/// poll `/api/maintenance/dedup/status` for progress + results.
async fn post_dedup_scan(State(state): State<Arc<WebState>>) -> impl IntoResponse {
if state.dedup.running.swap(true, Ordering::SeqCst) {
return Json(serde_json::json!({ "started": false, "running": true }));
}
state.dedup.done.store(0, Ordering::Relaxed);
state.dedup.total.store(0, Ordering::Relaxed);
*state.dedup.error.lock_recover() = None;
// Snapshot what the worker needs from the library (cheap clones of paths +
// display strings) so it doesn't hold the lock during the ffmpeg pass.
let (inputs, by_path, valid_paths) = {
let lib = state.library.lock_recover();
let root = state.library_root.clone();
let mut inputs: Vec<crate::fingerprint::FpInput> = Vec::new();
let mut by_path: HashMap<String, SimilarVideo> = HashMap::new();
let mut valid_paths: HashSet<String> = HashSet::new();
for ch in lib.iter() {
let channel = ch.name.clone();
let videos = ch.videos.iter().chain(ch.playlists.iter().flat_map(|p| p.videos.iter()));
for v in videos {
let Some(vp) = &v.video_path else { continue }; // need a real file
let path_str = vp.display().to_string();
valid_paths.insert(path_str.clone());
inputs.push(crate::fingerprint::FpInput {
path: vp.clone(),
mtime_unix: v.mtime_unix.map(|m| m as i64).unwrap_or(0),
video_id: v.id.clone(),
duration_secs: v.duration_secs.unwrap_or(0.0),
});
let mut files = vec![path_str.clone()];
for p in [v.thumb_path.as_ref(), v.info_path.as_ref(), v.description_path.as_ref()]
.into_iter().flatten() { files.push(p.display().to_string()); }
for s in &v.subtitles { files.push(s.path.display().to_string()); }
let location = vp.parent()
.map(|d| d.strip_prefix(&root).unwrap_or(d).display().to_string())
.unwrap_or_default();
by_path.insert(path_str, SimilarVideo {
video_id: v.id.clone(), title: v.title.clone(), channel: channel.clone(),
location, file_size: v.file_size, files, recommended_keep: false,
});
}
}
(inputs, by_path, valid_paths)
};
let db = state.db.clone();
let dedup = state.dedup.clone();
std::thread::spawn(move || run_dedup(db, dedup, inputs, by_path, valid_paths));
Json(serde_json::json!({ "started": true }))
}
/// `GET /api/maintenance/dedup/status` — progress + results of the dedup job.
async fn get_dedup_status(State(state): State<Arc<WebState>>) -> impl IntoResponse {
let d = &state.dedup;
Json(serde_json::json!({
"running": d.running.load(Ordering::Relaxed),
"done": d.done.load(Ordering::Relaxed),
"total": d.total.load(Ordering::Relaxed),
"groups": d.result.lock_recover().clone(),
"error": d.error.lock_recover().clone(),
}))
}
/// Worker body for the dedup job: mtime-gate, fingerprint the missing videos
/// in parallel, prune vanished entries, then group by visual similarity and
/// stash the result. Always clears `running` on exit.
fn run_dedup(
db: Database,
dedup: std::sync::Arc<DedupState>,
inputs: Vec<crate::fingerprint::FpInput>,
by_path: HashMap<String, SimilarVideo>,
valid_paths: HashSet<String>,
) {
use crate::fingerprint;
let outcome: Result<Vec<SimilarGroup>, String> = (|| {
let known = db.fingerprint_mtimes().map_err(|e| e.to_string())?;
let todo: Vec<fingerprint::FpInput> = inputs
.into_iter()
.filter(|i| known.get(&i.path.display().to_string()) != Some(&i.mtime_unix))
.collect();
dedup.total.store(todo.len(), Ordering::Relaxed);
let workers = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
let computed = fingerprint::compute_batch(todo, workers, &dedup.done);
for c in &computed {
let _ = db.upsert_fingerprint(
&c.input.path.display().to_string(), c.input.mtime_unix,
&c.input.video_id, c.input.duration_secs, &c.hashes,
);
}
let _ = db.prune_fingerprints(&valid_paths);
let stored = db.load_fingerprints().map_err(|e| e.to_string())?;
let records: Vec<fingerprint::FpRecord> = stored.iter().map(|s| fingerprint::FpRecord {
video_id: s.video_id.clone(), duration_secs: s.duration_secs, hashes: s.hashes.clone(),
}).collect();
let groups_idx = fingerprint::group_similar(
&records, fingerprint::DEFAULT_DUR_TOL,
fingerprint::DEFAULT_MAX_HAM, fingerprint::DEFAULT_MIN_MATCH,
);
let mut groups = Vec::new();
for g in groups_idx {
let mut vids: Vec<SimilarVideo> =
g.iter().filter_map(|&i| by_path.get(&stored[i].path).cloned()).collect();
if vids.len() < 2 { continue; } // stale paths dropped out
// Recommend keeping the largest file (best quality) in each group.
let keep = vids.iter().enumerate()
.max_by_key(|(_, v)| v.file_size.unwrap_or(0)).map(|(i, _)| i);
for (i, v) in vids.iter_mut().enumerate() { v.recommended_keep = Some(i) == keep; }
groups.push(SimilarGroup { videos: vids });
}
Ok(groups)
})();
match outcome {
Ok(groups) => *dedup.result.lock_recover() = Some(groups),
Err(e) => *dedup.error.lock_recover() = Some(e),
}
dedup.running.store(false, Ordering::SeqCst);
}
/// `POST /api/maintenance/remove` — delete the listed duplicate files. /// `POST /api/maintenance/remove` — delete the listed duplicate files.
/// Paths outside the library root are refused. /// Paths outside the library root are refused.
async fn post_maintenance_remove( async fn post_maintenance_remove(
@ -2611,6 +2774,7 @@ async fn serve(config: Config, shutdown_rx: std::sync::mpsc::Receiver<()>) {
progress_tx, progress_tx,
login_attempts: Mutex::new(HashMap::new()), login_attempts: Mutex::new(HashMap::new()),
library_body_cache: Mutex::new(None), library_body_cache: Mutex::new(None),
dedup: std::sync::Arc::new(DedupState::default()),
}); });
// Broadcast progress snapshots to WebSocket subscribers. Ticks fast // Broadcast progress snapshots to WebSocket subscribers. Ticks fast
@ -2721,6 +2885,8 @@ async fn serve(config: Config, shutdown_rx: std::sync::mpsc::Receiver<()>) {
.route("/api/plex/generate", post(post_plex_generate)) .route("/api/plex/generate", post(post_plex_generate))
.route("/api/maintenance/scan", get(get_maintenance_scan)) .route("/api/maintenance/scan", get(get_maintenance_scan))
.route("/api/maintenance/remove", post(post_maintenance_remove)) .route("/api/maintenance/remove", post(post_maintenance_remove))
.route("/api/maintenance/dedup/scan", post(post_dedup_scan))
.route("/api/maintenance/dedup/status", get(get_dedup_status))
.route("/api/maintenance/repair/:id", post(post_maintenance_repair)) .route("/api/maintenance/repair/:id", post(post_maintenance_repair))
.route("/api/cookies", get(get_cookies).post(post_cookies).delete(delete_cookies)) .route("/api/cookies", get(get_cookies).post(post_cookies).delete(delete_cookies))
.route("/api/scheduler/run", post(post_scheduler_run)) .route("/api/scheduler/run", post(post_scheduler_run))

View file

@ -2162,7 +2162,86 @@ function renderMaintenance(r){
</div>`; </div>`;
} }
} }
h+=`<h3 style="font-size:13px;margin:16px 0 8px">Similar content (perceptual)</h3>
<div class="settings-hint" style="margin-bottom:8px">Finds the same video re-uploaded under a different ID — mirrors, re-encodes, resolution changes — by comparing sampled frames. The first scan fingerprints your library (a few minutes); after that it's cached, so re-scans are quick.</div>
<div id="dedup-area"></div>`;
body.innerHTML=h; body.innerHTML=h;
initDedupArea();
}
/* ── Perceptual dedup (similar content) ─────────────────────────── */
let dedupTimer=null;
async function initDedupArea(){
if(dedupTimer){clearInterval(dedupTimer);dedupTimer=null}
try{
const s=await(await api('/api/maintenance/dedup/status')).json();
if(s.running){renderDedupProgress(s);startDedupPoll();}
else if(s.groups){renderSimilarGroups(s.groups);}
else dedupIdle(s.error);
}catch(e){const a=document.getElementById('dedup-area');if(a)a.innerHTML=`<div style="color:#f87171;font-size:12px">${esc(e.message)}</div>`}
}
function dedupIdle(err){
const a=document.getElementById('dedup-area');if(!a)return;
a.innerHTML=`${err?`<div style="color:#f87171;font-size:12px;margin-bottom:6px">Last scan error: ${esc(err)}</div>`:''}
<button class="primary" onclick="startDedup(this)">🔍 Scan for similar content</button>`;
}
async function startDedup(btn){
if(btn)btn.disabled=true;
try{await api('/api/maintenance/dedup/scan',{method:'POST'});startDedupPoll();}
catch(e){setStatus('Error: '+e.message);if(btn)btn.disabled=false}
}
function startDedupPoll(){
if(dedupTimer)clearInterval(dedupTimer);
dedupTimer=setInterval(async()=>{
if(!document.getElementById('dedup-area')){clearInterval(dedupTimer);dedupTimer=null;return} // modal closed
let s;try{s=await(await api('/api/maintenance/dedup/status')).json()}catch{return}
if(s.running){renderDedupProgress(s);return}
clearInterval(dedupTimer);dedupTimer=null;
if(s.error)dedupIdle(s.error); else renderSimilarGroups(s.groups||[]);
},800);
}
function renderDedupProgress(s){
const a=document.getElementById('dedup-area');if(!a)return;
const pct=s.total?Math.round(100*s.done/s.total):0;
a.innerHTML=`<div style="font-size:12px;color:var(--muted);margin-bottom:6px">Fingerprinting ${s.done} / ${s.total||'…'} new videos…</div>
<div style="height:6px;background:var(--bg);border-radius:3px;overflow:hidden"><div style="height:100%;width:${pct}%;background:var(--accent);transition:width .3s"></div></div>`;
}
function renderSimilarGroups(groups){
const a=document.getElementById('dedup-area');if(!a)return;
let h=`<div style="display:flex;gap:8px;align-items:center;margin-bottom:8px">
<span style="font-size:12px;color:var(--muted)">${groups.length} similar group${groups.length===1?'':'s'}</span>
<button onclick="startDedup(this)">⟳ Re-scan</button></div>`;
if(!groups.length){h+='<div style="color:var(--muted);font-size:12px">No visually-similar duplicates found across different IDs. 🎉</div>';a.innerHTML=h;return}
groups.forEach((g,gi)=>{
h+=`<div style="border:1px solid var(--border);border-radius:6px;padding:8px;margin-bottom:8px">
<div style="font-weight:600;font-size:12px;margin-bottom:6px">Group ${gi+1} — ${g.videos.length} copies</div>`;
g.videos.forEach(v=>{
const tag=v.recommended_keep?'<span style="color:#4ade80">keep</span>':'<span style="color:#f87171">remove</span>';
const size=v.file_size?fmtSize(v.file_size):'no video';
h+=`<label style="display:flex;align-items:center;gap:8px;font-size:12px;padding:3px 0">
<input type="checkbox" class="sim-chk" data-files='${esc(JSON.stringify(v.files))}' ${v.recommended_keep?'':'checked'}>
<span style="flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap" title="${esc(v.location)}">${esc(v.title)} <span style="color:var(--muted)">· ${esc(v.channel)} · ${size} · [${esc(v.video_id)}]</span></span>
${tag}
</label>`;
});
h+=`</div>`;
});
h+=`<button class="primary" onclick="removeSimilar(this)">🗑 Remove checked copies</button>`;
a.innerHTML=h;
}
async function removeSimilar(btn){
const chks=[...document.querySelectorAll('.sim-chk:checked')];
let paths=[];
for(const c of chks){try{paths=paths.concat(JSON.parse(c.dataset.files))}catch{}}
if(!paths.length){setStatus('Nothing selected.');return}
if(!confirm(`Delete ${paths.length} file(s) across the checked copies? This cannot be undone.`))return;
btn.disabled=true;
try{
const r=await(await api('/api/maintenance/remove',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({paths})})).json();
setStatus(`Removed ${r.removed} file(s)`+(r.errors&&r.errors.length?`, ${r.errors.length} error(s)`:''));
await loadLibrary();
startDedup(); // re-scan (fingerprints cached) so deleted copies drop out
}catch(e){setStatus('Error: '+e.message);btn.disabled=false}
} }
async function removeDuplicates(btn){ async function removeDuplicates(btn){

View file

@ -25,6 +25,13 @@ fn have_curl() -> bool {
.status().map(|s| s.success()).unwrap_or(false) .status().map(|s| s.success()).unwrap_or(false)
} }
/// True if `ffmpeg` is usable — the perceptual-dedup test needs it to both
/// generate fixtures and fingerprint them; it skips otherwise.
fn have_ffmpeg() -> bool {
Command::new("ffmpeg").arg("-version").stdout(Stdio::null()).stderr(Stdio::null())
.status().map(|s| s.success()).unwrap_or(false)
}
/// A running `yt-offline --web` child against a scratch dir. Killed and /// A running `yt-offline --web` child against a scratch dir. Killed and
/// its dir removed on drop. /// its dir removed on drop.
struct Server { struct Server {
@ -351,3 +358,43 @@ fn full_text_search_indexes_titles_and_descriptions() {
assert!(!s.get("/api/search?q=quantumchromodynamics").1.contains("vid123"), assert!(!s.get("/api/search?q=quantumchromodynamics").1.contains("vid123"),
"unrelated query must not hit"); "unrelated query must not hit");
} }
#[test]
fn perceptual_dedup_groups_reencodes() {
if !have_curl() { eprintln!("skip: no curl"); return; }
if !have_ffmpeg() { eprintln!("skip: no ffmpeg"); return; }
let s = Server::start();
let chan = s.dir.join("ch/channels/Demo");
std::fs::create_dir_all(&chan).unwrap();
let gen = |args: &[&str]| {
let ok = Command::new("ffmpeg").arg("-nostdin").arg("-y").arg("-loglevel").arg("error")
.args(args).status().map(|st| st.success()).unwrap_or(false);
assert!(ok, "ffmpeg gen failed: {args:?}");
};
let orig = chan.join("orig [aaa].mp4");
let reenc = chan.join("reenc [bbb].mp4");
let diff = chan.join("diff [ccc].mp4");
gen(&["-f","lavfi","-i","testsrc=duration=20:size=640x480:rate=10", orig.to_str().unwrap()]);
gen(&["-i", orig.to_str().unwrap(), "-vf","scale=320:240","-r","15","-c:v","libx264","-crf","38", reenc.to_str().unwrap()]);
gen(&["-f","lavfi","-i","testsrc2=duration=20:size=640x480:rate=10", diff.to_str().unwrap()]);
for stem in ["orig [aaa]", "reenc [bbb]", "diff [ccc]"] {
std::fs::write(chan.join(format!("{stem}.info.json")), r#"{"duration":20.0}"#).unwrap();
}
assert_eq!(s.post("/api/rescan", "").0, 200);
assert_eq!(s.post("/api/maintenance/dedup/scan", "").0, 200);
// Poll until the background job finishes (fingerprinting 3 short clips).
let mut body = String::new();
for _ in 0..160 {
let (_, b) = s.get("/api/maintenance/dedup/status");
if field(&b, "running") == Some("false") { body = b; break; }
std::thread::sleep(std::time::Duration::from_millis(250));
}
assert!(field(&body, "running") == Some("false"), "dedup job never finished: {body}");
// orig + its re-encode group together; the unrelated clip stays out.
assert!(body.contains("aaa"), "group should contain orig: {body}");
assert!(body.contains("bbb"), "group should contain the re-encode: {body}");
assert!(!body.contains("ccc"), "unrelated video must not be grouped: {body}");
}