Merge pull request #591 from Nezreka/feature/quarantine-management
Feature/quarantine management
This commit is contained in:
BoulderBadgeDad
GitHub
commit
4226be761a
9 changed files with 903 additions and 14 deletions
|
|
@ -29,8 +29,22 @@ def _get_config_manager():
|
|||
return config_manager
|
||||
|
||||
|
||||
def move_to_quarantine(file_path: str, context: dict, reason: str, automation_engine=None) -> str:
|
||||
"""Move a file to the quarantine folder and write a metadata sidecar."""
|
||||
def move_to_quarantine(file_path: str, context: dict, reason: str, automation_engine=None, *, trigger: str = "unknown") -> str:
|
||||
"""Move a file to the quarantine folder and write a metadata sidecar.
|
||||
|
||||
`trigger` identifies which check fired (`integrity` / `acoustid` /
|
||||
`bit_depth` / `unknown`) and is persisted in the sidecar so
|
||||
one-click Approve can set the matching `_skip_quarantine_check`
|
||||
bypass when re-running the pipeline.
|
||||
|
||||
Sidecar also persists a JSON-safe snapshot of the full `context`
|
||||
dict via `serialize_quarantine_context`, enabling in-place approve
|
||||
without losing the matched-track metadata. Legacy sidecars (written
|
||||
before this expansion) lack the `context` field — Approve falls
|
||||
back to `recover_to_staging` for those.
|
||||
"""
|
||||
from core.imports.quarantine import serialize_quarantine_context
|
||||
|
||||
download_dir = _get_config_manager().get("soulseek.download_path", "./downloads")
|
||||
quarantine_dir = Path(download_dir) / "ss_quarantine"
|
||||
quarantine_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
|
@ -56,6 +70,8 @@ def move_to_quarantine(file_path: str, context: dict, reason: str, automation_en
|
|||
"expected_track": get_import_clean_title(context, default=original_search.get("title", "Unknown")),
|
||||
"expected_artist": get_import_clean_artist(context, default=(artist_context.get("name", "") if isinstance(artist_context, dict) else "Unknown")),
|
||||
"context_key": context.get("context_key", "unknown"),
|
||||
"trigger": trigger,
|
||||
"context": serialize_quarantine_context(context),
|
||||
}
|
||||
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -163,15 +163,23 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
_duration_tolerance_override = resolve_duration_tolerance(
|
||||
config_manager.get('post_processing.duration_tolerance_seconds', 0)
|
||||
)
|
||||
try:
|
||||
integrity = check_audio_integrity(
|
||||
file_path,
|
||||
_expected_duration_ms,
|
||||
length_tolerance_s=_duration_tolerance_override,
|
||||
)
|
||||
except Exception as integrity_error:
|
||||
logger.error(f"[Integrity] Check raised unexpectedly (continuing): {integrity_error}")
|
||||
# Per-check quarantine bypass — set by `approve_quarantine_entry`
|
||||
# when the user explicitly approves a previously-quarantined
|
||||
# file. Skips ONLY the named check; other gates still run.
|
||||
_bypass_check = context.get('_skip_quarantine_check')
|
||||
if _bypass_check == 'integrity':
|
||||
logger.info(f"[Integrity] Skipped (user approval) for {_basename}")
|
||||
integrity = None
|
||||
else:
|
||||
try:
|
||||
integrity = check_audio_integrity(
|
||||
file_path,
|
||||
_expected_duration_ms,
|
||||
length_tolerance_s=_duration_tolerance_override,
|
||||
)
|
||||
except Exception as integrity_error:
|
||||
logger.error(f"[Integrity] Check raised unexpectedly (continuing): {integrity_error}")
|
||||
integrity = None
|
||||
|
||||
if integrity is not None and not integrity.ok:
|
||||
logger.error(f"[Integrity] Rejected {_basename}: {integrity.reason}")
|
||||
|
|
@ -183,6 +191,7 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
context,
|
||||
f"Integrity check failed: {integrity.reason}",
|
||||
automation_engine,
|
||||
trigger='integrity',
|
||||
)
|
||||
logger.error(f"File quarantined due to integrity failure: {quarantine_path}")
|
||||
except Exception as quarantine_error:
|
||||
|
|
@ -218,7 +227,9 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
f"drift={integrity.checks.get('length_drift_s', 'n/a')})"
|
||||
)
|
||||
|
||||
_skip_acoustid = False
|
||||
_skip_acoustid = context.get('_skip_quarantine_check') == 'acoustid'
|
||||
if _skip_acoustid:
|
||||
logger.info(f"[AcoustID] Skipped (user approval) for {_basename}")
|
||||
try:
|
||||
from core.acoustid_verification import AcoustIDVerification, VerificationResult
|
||||
|
||||
|
|
@ -260,6 +271,7 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
context,
|
||||
verification_msg,
|
||||
automation_engine,
|
||||
trigger='acoustid',
|
||||
)
|
||||
logger.error(f"File quarantined due to verification failure: {quarantine_path}")
|
||||
except Exception as quarantine_error:
|
||||
|
|
@ -432,7 +444,9 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
if context['_audio_quality']:
|
||||
logger.info(f"Audio quality detected: {context['_audio_quality']}")
|
||||
|
||||
rejection_reason = check_flac_bit_depth(file_path, context)
|
||||
rejection_reason = None if context.get('_skip_quarantine_check') == 'bit_depth' else check_flac_bit_depth(file_path, context)
|
||||
if context.get('_skip_quarantine_check') == 'bit_depth':
|
||||
logger.info(f"[BitDepth] Skipped (user approval) for {_basename}")
|
||||
if rejection_reason:
|
||||
try:
|
||||
quarantine_path = move_to_quarantine(
|
||||
|
|
@ -440,6 +454,7 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
context,
|
||||
rejection_reason,
|
||||
automation_engine,
|
||||
trigger='bit_depth',
|
||||
)
|
||||
logger.info(f"File quarantined due to bit depth filter: {quarantine_path}")
|
||||
except Exception as quarantine_error:
|
||||
|
|
@ -560,7 +575,9 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
if context['_audio_quality']:
|
||||
logger.info(f"Audio quality detected: {context['_audio_quality']}")
|
||||
|
||||
rejection_reason = check_flac_bit_depth(file_path, context)
|
||||
rejection_reason = None if context.get('_skip_quarantine_check') == 'bit_depth' else check_flac_bit_depth(file_path, context)
|
||||
if context.get('_skip_quarantine_check') == 'bit_depth':
|
||||
logger.info(f"[BitDepth] Skipped (user approval) for {_basename}")
|
||||
if rejection_reason:
|
||||
try:
|
||||
quarantine_path = move_to_quarantine(
|
||||
|
|
@ -568,6 +585,7 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
|
|||
context,
|
||||
rejection_reason,
|
||||
automation_engine,
|
||||
trigger='bit_depth',
|
||||
)
|
||||
logger.info(f"File quarantined due to bit depth filter: {quarantine_path}")
|
||||
except Exception as quarantine_error:
|
||||
|
|
|
|||
329
core/imports/quarantine.py
Normal file
329
core/imports/quarantine.py
Normal file
|
|
@ -0,0 +1,329 @@
|
|||
"""Quarantine entry management — pure helpers for list/delete/approve/recover.
|
||||
|
||||
Quarantined files live in `<download_path>/ss_quarantine/` as
|
||||
`<timestamp>_<original>.<ext>.quarantined` paired with a JSON sidecar
|
||||
`<timestamp>_<original>.json` written by `core.imports.guards.move_to_quarantine`.
|
||||
|
||||
This module provides the read/write/restore primitives. Web routes are
|
||||
thin glue around these. Pipeline re-run on approval is the caller's
|
||||
job (we hand back `(file_path, context, bypass_check)`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("imports.quarantine")
|
||||
|
||||
|
||||
_QUARANTINE_SUFFIX = ".quarantined"
|
||||
|
||||
|
||||
# JSON-serializable scalar predicate. dict / list values get walked
|
||||
# recursively; anything else is dropped during sidecar serialization.
|
||||
_SAFE_SCALARS = (str, int, float, bool, type(None))
|
||||
|
||||
|
||||
def serialize_quarantine_context(context: Any) -> Dict[str, Any]:
|
||||
"""Walk a context dict and emit a JSON-safe copy.
|
||||
|
||||
Drops non-serializable values (sets, custom objects, callables,
|
||||
open file handles, etc) silently — sidecar must round-trip through
|
||||
`json.dump` / `json.load` without raising. Lists are walked element
|
||||
by element; dicts are walked recursively. Anything that isn't a
|
||||
scalar / dict / list is converted to a string fallback so caller
|
||||
still sees *something* (rather than a silent drop) but won't break
|
||||
the JSON write.
|
||||
"""
|
||||
if not isinstance(context, dict):
|
||||
return {}
|
||||
return _coerce_dict(context)
|
||||
|
||||
|
||||
def _coerce_value(value: Any) -> Any:
|
||||
if isinstance(value, _SAFE_SCALARS):
|
||||
return value
|
||||
if isinstance(value, dict):
|
||||
return _coerce_dict(value)
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_coerce_value(v) for v in value]
|
||||
if isinstance(value, set):
|
||||
return [_coerce_value(v) for v in value]
|
||||
# Fallback — preserve via str() so caller sees the value's shape
|
||||
# without breaking JSON serialization.
|
||||
try:
|
||||
return str(value)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _coerce_dict(d: Dict[str, Any]) -> Dict[str, Any]:
|
||||
out: Dict[str, Any] = {}
|
||||
for key, value in d.items():
|
||||
if not isinstance(key, str):
|
||||
try:
|
||||
key = str(key)
|
||||
except Exception:
|
||||
continue
|
||||
out[key] = _coerce_value(value)
|
||||
return out
|
||||
|
||||
|
||||
def _entry_id_from_filename(quarantined_filename: str) -> str:
|
||||
"""Derive a stable entry id from the quarantined filename.
|
||||
|
||||
Strip the `.quarantined` suffix; strip the original file extension;
|
||||
return the bare `<timestamp>_<original>` stem. Sidecar uses the
|
||||
same stem with a `.json` extension, so the id pairs both sides.
|
||||
"""
|
||||
base = quarantined_filename
|
||||
if base.endswith(_QUARANTINE_SUFFIX):
|
||||
base = base[: -len(_QUARANTINE_SUFFIX)]
|
||||
return Path(base).stem
|
||||
|
||||
|
||||
def list_quarantine_entries(quarantine_dir: str) -> List[Dict[str, Any]]:
|
||||
"""Enumerate quarantined files paired with their sidecars.
|
||||
|
||||
Returns one dict per `.quarantined` file with: id, filename,
|
||||
original_filename (from sidecar), reason, expected_track,
|
||||
expected_artist, timestamp, size_bytes, has_full_context (True
|
||||
when the sidecar carries a `context` field — required for one-click
|
||||
Approve), trigger (which check fired: integrity / acoustid /
|
||||
bit_depth / unknown).
|
||||
|
||||
Orphaned `.quarantined` files (no sidecar) still surface — caller
|
||||
can delete them. Orphaned sidecars (no file) are skipped silently.
|
||||
Sorted newest-first by timestamp prefix.
|
||||
"""
|
||||
entries: List[Dict[str, Any]] = []
|
||||
if not os.path.isdir(quarantine_dir):
|
||||
return entries
|
||||
|
||||
for name in os.listdir(quarantine_dir):
|
||||
if not name.endswith(_QUARANTINE_SUFFIX):
|
||||
continue
|
||||
full_path = os.path.join(quarantine_dir, name)
|
||||
if not os.path.isfile(full_path):
|
||||
continue
|
||||
|
||||
entry_id = _entry_id_from_filename(name)
|
||||
sidecar_path = os.path.join(quarantine_dir, f"{entry_id}.json")
|
||||
sidecar: Dict[str, Any] = {}
|
||||
if os.path.isfile(sidecar_path):
|
||||
try:
|
||||
with open(sidecar_path, encoding="utf-8") as f:
|
||||
loaded = json.load(f)
|
||||
if isinstance(loaded, dict):
|
||||
sidecar = loaded
|
||||
except Exception as exc:
|
||||
logger.debug("sidecar read failed for %s: %s", entry_id, exc)
|
||||
|
||||
try:
|
||||
size_bytes = os.path.getsize(full_path)
|
||||
except OSError:
|
||||
size_bytes = 0
|
||||
|
||||
entries.append(
|
||||
{
|
||||
"id": entry_id,
|
||||
"filename": name,
|
||||
"original_filename": sidecar.get("original_filename", name),
|
||||
"reason": sidecar.get("quarantine_reason", "Unknown reason"),
|
||||
"expected_track": sidecar.get("expected_track", ""),
|
||||
"expected_artist": sidecar.get("expected_artist", ""),
|
||||
"timestamp": sidecar.get("timestamp", ""),
|
||||
"size_bytes": size_bytes,
|
||||
"has_full_context": isinstance(sidecar.get("context"), dict),
|
||||
"trigger": sidecar.get("trigger", "unknown"),
|
||||
}
|
||||
)
|
||||
|
||||
entries.sort(key=lambda e: e["id"], reverse=True)
|
||||
return entries
|
||||
|
||||
|
||||
def _resolve_entry_paths(quarantine_dir: str, entry_id: str) -> Tuple[Optional[str], Optional[str]]:
|
||||
"""Locate the `.quarantined` file + JSON sidecar for an entry id.
|
||||
|
||||
Returns (file_path, sidecar_path), either may be None if missing.
|
||||
"""
|
||||
if not os.path.isdir(quarantine_dir) or not entry_id:
|
||||
return None, None
|
||||
file_path: Optional[str] = None
|
||||
for name in os.listdir(quarantine_dir):
|
||||
if not name.endswith(_QUARANTINE_SUFFIX):
|
||||
continue
|
||||
if _entry_id_from_filename(name) == entry_id:
|
||||
file_path = os.path.join(quarantine_dir, name)
|
||||
break
|
||||
sidecar_path = os.path.join(quarantine_dir, f"{entry_id}.json")
|
||||
if not os.path.isfile(sidecar_path):
|
||||
sidecar_path = None
|
||||
return file_path, sidecar_path
|
||||
|
||||
|
||||
def delete_quarantine_entry(quarantine_dir: str, entry_id: str) -> bool:
|
||||
"""Delete the quarantined file + sidecar for the given entry id.
|
||||
|
||||
Returns True if at least one of the two was removed. False when
|
||||
neither existed (entry already gone).
|
||||
"""
|
||||
file_path, sidecar_path = _resolve_entry_paths(quarantine_dir, entry_id)
|
||||
removed = False
|
||||
if file_path and os.path.isfile(file_path):
|
||||
try:
|
||||
os.remove(file_path)
|
||||
removed = True
|
||||
except OSError as exc:
|
||||
logger.error("Failed to delete quarantine file %s: %s", file_path, exc)
|
||||
if sidecar_path and os.path.isfile(sidecar_path):
|
||||
try:
|
||||
os.remove(sidecar_path)
|
||||
removed = True
|
||||
except OSError as exc:
|
||||
logger.error("Failed to delete quarantine sidecar %s: %s", sidecar_path, exc)
|
||||
return removed
|
||||
|
||||
|
||||
def _restore_filename(quarantined_filename: str, sidecar_original: Optional[str] = None) -> str:
|
||||
"""Resolve the filename to restore.
|
||||
|
||||
Sidecar's `original_filename` wins when provided — it's the
|
||||
canonical record of what the file was named before quarantine.
|
||||
Otherwise parse the `<YYYYMMDD_HHMMSS>_<original>.<ext>.quarantined`
|
||||
convention written by `move_to_quarantine`, dropping the timestamp
|
||||
prefix and `.quarantined` suffix. Final fallback returns the
|
||||
quarantined filename minus the suffix unchanged.
|
||||
"""
|
||||
if sidecar_original:
|
||||
return sidecar_original
|
||||
base = quarantined_filename
|
||||
if base.endswith(_QUARANTINE_SUFFIX):
|
||||
base = base[: -len(_QUARANTINE_SUFFIX)]
|
||||
parts = base.split("_", 2)
|
||||
if len(parts) >= 3 and parts[0].isdigit() and parts[1].isdigit():
|
||||
return parts[2]
|
||||
return base
|
||||
|
||||
|
||||
def approve_quarantine_entry(
|
||||
quarantine_dir: str,
|
||||
entry_id: str,
|
||||
restore_dir: str,
|
||||
) -> Optional[Tuple[str, Dict[str, Any], str]]:
|
||||
"""Restore a quarantined file for re-import via the post-process pipeline.
|
||||
|
||||
Reads the sidecar's `context` + `trigger`, moves the file out of
|
||||
quarantine to `restore_dir` (with the original filename + extension),
|
||||
deletes the sidecar.
|
||||
|
||||
Returns `(restored_file_path, context, trigger)` so the caller can
|
||||
set the appropriate `_skip_quarantine_check` bypass flag and
|
||||
dispatch the post-process pipeline.
|
||||
|
||||
Returns None when:
|
||||
- the entry doesn't exist
|
||||
- the sidecar lacks a serialized `context` (legacy thin sidecar
|
||||
— caller should fall back to `recover_to_staging` instead)
|
||||
- the file move fails
|
||||
"""
|
||||
file_path, sidecar_path = _resolve_entry_paths(quarantine_dir, entry_id)
|
||||
if not file_path or not sidecar_path:
|
||||
logger.warning("approve: entry %s missing file or sidecar", entry_id)
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(sidecar_path, encoding="utf-8") as f:
|
||||
sidecar = json.load(f)
|
||||
except Exception as exc:
|
||||
logger.error("approve: sidecar read failed for %s: %s", entry_id, exc)
|
||||
return None
|
||||
|
||||
context = sidecar.get("context")
|
||||
if not isinstance(context, dict):
|
||||
logger.info("approve: entry %s has thin sidecar (no context) — caller should recover-to-staging", entry_id)
|
||||
return None
|
||||
|
||||
trigger = str(sidecar.get("trigger", "unknown"))
|
||||
|
||||
original_name = sidecar.get("original_filename") or _restore_filename(os.path.basename(file_path))
|
||||
os.makedirs(restore_dir, exist_ok=True)
|
||||
restored_path = os.path.join(restore_dir, original_name)
|
||||
restored_path = _ensure_unique_path(restored_path)
|
||||
|
||||
try:
|
||||
shutil.move(file_path, restored_path)
|
||||
except OSError as exc:
|
||||
logger.error("approve: failed to restore %s -> %s: %s", file_path, restored_path, exc)
|
||||
return None
|
||||
|
||||
try:
|
||||
os.remove(sidecar_path)
|
||||
except OSError as exc:
|
||||
logger.warning("approve: failed to remove sidecar %s: %s", sidecar_path, exc)
|
||||
|
||||
return restored_path, context, trigger
|
||||
|
||||
|
||||
def recover_to_staging(
|
||||
quarantine_dir: str,
|
||||
staging_dir: str,
|
||||
entry_id: str,
|
||||
) -> Optional[str]:
|
||||
"""Move a quarantined file into Staging for manual import.
|
||||
|
||||
Strips the timestamp prefix + `.quarantined` suffix, drops the file
|
||||
into `staging_dir` so the user can finish via the existing Import
|
||||
flow. Sidecar is removed. Used as the fallback path for legacy thin
|
||||
sidecars (no embedded `context`) where one-click Approve is
|
||||
impossible.
|
||||
"""
|
||||
file_path, sidecar_path = _resolve_entry_paths(quarantine_dir, entry_id)
|
||||
if not file_path:
|
||||
return None
|
||||
|
||||
sidecar_original = None
|
||||
if sidecar_path:
|
||||
try:
|
||||
with open(sidecar_path, encoding="utf-8") as f:
|
||||
sidecar_original = json.load(f).get("original_filename")
|
||||
except Exception as exc:
|
||||
logger.debug("recover: sidecar read failed for %s: %s", entry_id, exc)
|
||||
|
||||
restored_name = _restore_filename(os.path.basename(file_path), sidecar_original)
|
||||
os.makedirs(staging_dir, exist_ok=True)
|
||||
target = _ensure_unique_path(os.path.join(staging_dir, restored_name))
|
||||
|
||||
try:
|
||||
shutil.move(file_path, target)
|
||||
except OSError as exc:
|
||||
logger.error("recover: failed to move %s -> %s: %s", file_path, target, exc)
|
||||
return None
|
||||
|
||||
if sidecar_path and os.path.isfile(sidecar_path):
|
||||
try:
|
||||
os.remove(sidecar_path)
|
||||
except OSError as exc:
|
||||
logger.warning("recover: failed to remove sidecar %s: %s", sidecar_path, exc)
|
||||
|
||||
return target
|
||||
|
||||
|
||||
def _ensure_unique_path(target: str) -> str:
|
||||
"""Append `_(2)`, `_(3)`, ... before the extension when target exists."""
|
||||
if not os.path.exists(target):
|
||||
return target
|
||||
base, ext = os.path.splitext(target)
|
||||
counter = 2
|
||||
while True:
|
||||
candidate = f"{base}_({counter}){ext}"
|
||||
if not os.path.exists(candidate):
|
||||
return candidate
|
||||
counter += 1
|
||||
273
tests/imports/test_quarantine_management.py
Normal file
273
tests/imports/test_quarantine_management.py
Normal file
|
|
@ -0,0 +1,273 @@
|
|||
import json
|
||||
import os
|
||||
|
||||
from core.imports.quarantine import (
|
||||
approve_quarantine_entry,
|
||||
delete_quarantine_entry,
|
||||
list_quarantine_entries,
|
||||
recover_to_staging,
|
||||
serialize_quarantine_context,
|
||||
)
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# serialize_quarantine_context — JSON-safe coercion
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_serialize_passes_scalar_dict_unchanged():
|
||||
ctx = {"title": "DNA.", "track_number": 2, "active": True, "missing": None, "duration_ms": 185000}
|
||||
out = serialize_quarantine_context(ctx)
|
||||
assert out == ctx
|
||||
|
||||
|
||||
def test_serialize_walks_nested_dicts():
|
||||
ctx = {"track_info": {"name": "DNA.", "artists": [{"name": "Kendrick"}, {"name": "Rihanna"}]}}
|
||||
out = serialize_quarantine_context(ctx)
|
||||
assert out == ctx
|
||||
|
||||
|
||||
def test_serialize_coerces_set_to_list():
|
||||
ctx = {"sources": {"spotify", "deezer"}}
|
||||
out = serialize_quarantine_context(ctx)
|
||||
assert sorted(out["sources"]) == ["deezer", "spotify"]
|
||||
|
||||
|
||||
def test_serialize_coerces_tuple_to_list():
|
||||
ctx = {"pair": (1, 2, 3)}
|
||||
out = serialize_quarantine_context(ctx)
|
||||
assert out == {"pair": [1, 2, 3]}
|
||||
|
||||
|
||||
def test_serialize_stringifies_unknown_objects():
|
||||
class Custom:
|
||||
def __str__(self):
|
||||
return "<custom obj>"
|
||||
out = serialize_quarantine_context({"obj": Custom()})
|
||||
assert out["obj"] == "<custom obj>"
|
||||
|
||||
|
||||
def test_serialize_non_dict_returns_empty_dict():
|
||||
assert serialize_quarantine_context(None) == {}
|
||||
assert serialize_quarantine_context("string") == {}
|
||||
assert serialize_quarantine_context([1, 2, 3]) == {}
|
||||
|
||||
|
||||
def test_serialize_round_trips_through_json():
|
||||
ctx = {
|
||||
"track_info": {"name": "X", "artists": [{"name": "A"}, {"name": "B"}]},
|
||||
"spotify_artist": {"name": "A", "id": "abc"},
|
||||
"duration_ms": 180000,
|
||||
"sources": {"spotify"},
|
||||
}
|
||||
serialized = serialize_quarantine_context(ctx)
|
||||
json.dumps(serialized) # must not raise
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# list_quarantine_entries
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def _write_entry(quarantine_dir, entry_id, original_name, *, with_context=False, trigger="integrity", reason="boom", file_bytes=b"X" * 100):
|
||||
qfile = quarantine_dir / f"{entry_id}_{original_name}.quarantined"
|
||||
qfile.write_bytes(file_bytes)
|
||||
sidecar = {
|
||||
"original_filename": original_name,
|
||||
"quarantine_reason": reason,
|
||||
"expected_track": "Track",
|
||||
"expected_artist": "Artist",
|
||||
"timestamp": "2026-05-14T12:00:00",
|
||||
"trigger": trigger,
|
||||
}
|
||||
if with_context:
|
||||
sidecar["context"] = {"track_info": {"name": "Track"}, "context_key": entry_id}
|
||||
sidecar_path = quarantine_dir / f"{entry_id}_{os.path.splitext(original_name)[0]}.json"
|
||||
sidecar_path.write_text(json.dumps(sidecar))
|
||||
return qfile, sidecar_path
|
||||
|
||||
|
||||
def test_list_returns_empty_for_missing_dir(tmp_path):
|
||||
assert list_quarantine_entries(str(tmp_path / "nope")) == []
|
||||
|
||||
|
||||
def test_list_returns_empty_for_empty_dir(tmp_path):
|
||||
assert list_quarantine_entries(str(tmp_path)) == []
|
||||
|
||||
|
||||
def test_list_returns_entry_with_sidecar_fields(tmp_path):
|
||||
_write_entry(tmp_path, "20260514_120000", "song.flac", reason="Duration mismatch")
|
||||
entries = list_quarantine_entries(str(tmp_path))
|
||||
assert len(entries) == 1
|
||||
e = entries[0]
|
||||
assert e["original_filename"] == "song.flac"
|
||||
assert e["reason"] == "Duration mismatch"
|
||||
assert e["expected_track"] == "Track"
|
||||
assert e["expected_artist"] == "Artist"
|
||||
assert e["has_full_context"] is False
|
||||
assert e["trigger"] == "integrity"
|
||||
assert e["size_bytes"] == 100
|
||||
|
||||
|
||||
def test_list_flags_full_context_entries(tmp_path):
|
||||
_write_entry(tmp_path, "20260514_120000", "song.flac", with_context=True)
|
||||
entries = list_quarantine_entries(str(tmp_path))
|
||||
assert entries[0]["has_full_context"] is True
|
||||
|
||||
|
||||
def test_list_handles_orphan_quarantined_file_without_sidecar(tmp_path):
|
||||
qfile = tmp_path / "20260514_120000_orphan.flac.quarantined"
|
||||
qfile.write_bytes(b"X")
|
||||
entries = list_quarantine_entries(str(tmp_path))
|
||||
assert len(entries) == 1
|
||||
assert entries[0]["reason"] == "Unknown reason"
|
||||
assert entries[0]["has_full_context"] is False
|
||||
|
||||
|
||||
def test_list_skips_orphan_sidecars_without_file(tmp_path):
|
||||
sidecar = tmp_path / "20260514_120000_only.json"
|
||||
sidecar.write_text(json.dumps({"original_filename": "only.flac", "quarantine_reason": "x"}))
|
||||
assert list_quarantine_entries(str(tmp_path)) == []
|
||||
|
||||
|
||||
def test_list_sorts_newest_first(tmp_path):
|
||||
_write_entry(tmp_path, "20260101_120000", "old.flac")
|
||||
_write_entry(tmp_path, "20260514_120000", "new.flac")
|
||||
entries = list_quarantine_entries(str(tmp_path))
|
||||
assert entries[0]["original_filename"] == "new.flac"
|
||||
assert entries[1]["original_filename"] == "old.flac"
|
||||
|
||||
|
||||
def test_list_swallows_corrupt_sidecar_gracefully(tmp_path):
|
||||
qfile = tmp_path / "20260514_120000_song.flac.quarantined"
|
||||
qfile.write_bytes(b"X")
|
||||
sidecar = tmp_path / "20260514_120000_song.json"
|
||||
sidecar.write_text("{ this is not valid json")
|
||||
entries = list_quarantine_entries(str(tmp_path))
|
||||
assert len(entries) == 1
|
||||
assert entries[0]["reason"] == "Unknown reason"
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# delete_quarantine_entry
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_delete_removes_both_file_and_sidecar(tmp_path):
|
||||
_write_entry(tmp_path, "20260514_120000", "song.flac")
|
||||
assert delete_quarantine_entry(str(tmp_path), "20260514_120000_song") is True
|
||||
assert not (tmp_path / "20260514_120000_song.flac.quarantined").exists()
|
||||
assert not (tmp_path / "20260514_120000_song.json").exists()
|
||||
|
||||
|
||||
def test_delete_returns_false_when_entry_missing(tmp_path):
|
||||
assert delete_quarantine_entry(str(tmp_path), "nonexistent") is False
|
||||
|
||||
|
||||
def test_delete_handles_orphan_file_without_sidecar(tmp_path):
|
||||
qfile = tmp_path / "20260514_120000_orphan.flac.quarantined"
|
||||
qfile.write_bytes(b"X")
|
||||
assert delete_quarantine_entry(str(tmp_path), "20260514_120000_orphan") is True
|
||||
assert not qfile.exists()
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# approve_quarantine_entry — full-context path
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_approve_restores_file_and_returns_context_and_trigger(tmp_path):
|
||||
quarantine = tmp_path / "ss_quarantine"
|
||||
quarantine.mkdir()
|
||||
restore = tmp_path / "restore"
|
||||
|
||||
_write_entry(quarantine, "20260514_120000", "song.flac", with_context=True, trigger="integrity")
|
||||
|
||||
result = approve_quarantine_entry(str(quarantine), "20260514_120000_song", str(restore))
|
||||
assert result is not None
|
||||
restored_path, context, trigger = result
|
||||
assert os.path.basename(restored_path) == "song.flac"
|
||||
assert os.path.isfile(restored_path)
|
||||
assert context["track_info"]["name"] == "Track"
|
||||
assert trigger == "integrity"
|
||||
# Sidecar removed after approve
|
||||
assert not (quarantine / "20260514_120000_song.json").exists()
|
||||
|
||||
|
||||
def test_approve_returns_none_for_thin_sidecar_without_context(tmp_path):
|
||||
_write_entry(tmp_path, "20260514_120000", "song.flac", with_context=False)
|
||||
result = approve_quarantine_entry(str(tmp_path), "20260514_120000_song", str(tmp_path / "restore"))
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_approve_returns_none_for_missing_entry(tmp_path):
|
||||
assert approve_quarantine_entry(str(tmp_path), "nope", str(tmp_path)) is None
|
||||
|
||||
|
||||
def test_approve_avoids_filename_collision(tmp_path):
|
||||
quarantine = tmp_path / "q"
|
||||
quarantine.mkdir()
|
||||
restore = tmp_path / "r"
|
||||
restore.mkdir()
|
||||
(restore / "song.flac").write_bytes(b"existing")
|
||||
_write_entry(quarantine, "20260514_120000", "song.flac", with_context=True)
|
||||
result = approve_quarantine_entry(str(quarantine), "20260514_120000_song", str(restore))
|
||||
assert result is not None
|
||||
restored_path = result[0]
|
||||
assert os.path.basename(restored_path) == "song_(2).flac"
|
||||
assert (restore / "song.flac").read_bytes() == b"existing"
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
# recover_to_staging — fallback for thin sidecars
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_recover_strips_prefix_and_suffix(tmp_path):
|
||||
quarantine = tmp_path / "q"
|
||||
quarantine.mkdir()
|
||||
staging = tmp_path / "s"
|
||||
|
||||
qfile, _ = _write_entry(quarantine, "20260514_120000", "song.flac")
|
||||
|
||||
target = recover_to_staging(str(quarantine), str(staging), "20260514_120000_song")
|
||||
assert target is not None
|
||||
assert os.path.basename(target) == "song.flac"
|
||||
assert os.path.isfile(target)
|
||||
assert not qfile.exists()
|
||||
|
||||
|
||||
def test_recover_uses_sidecar_original_filename_when_available(tmp_path):
|
||||
quarantine = tmp_path / "q"
|
||||
quarantine.mkdir()
|
||||
staging = tmp_path / "s"
|
||||
qfile = quarantine / "20260514_120000_munged_name.flac.quarantined"
|
||||
qfile.write_bytes(b"X")
|
||||
sidecar = quarantine / "20260514_120000_munged_name.json"
|
||||
sidecar.write_text(json.dumps({"original_filename": "Pretty Track Name.flac"}))
|
||||
|
||||
target = recover_to_staging(str(quarantine), str(staging), "20260514_120000_munged_name")
|
||||
assert target is not None
|
||||
assert os.path.basename(target) == "Pretty Track Name.flac"
|
||||
|
||||
|
||||
def test_recover_returns_none_for_missing_entry(tmp_path):
|
||||
assert recover_to_staging(str(tmp_path / "q"), str(tmp_path / "s"), "nope") is None
|
||||
|
||||
|
||||
def test_recover_avoids_filename_collision(tmp_path):
|
||||
quarantine = tmp_path / "q"
|
||||
quarantine.mkdir()
|
||||
staging = tmp_path / "s"
|
||||
staging.mkdir()
|
||||
(staging / "song.flac").write_bytes(b"existing")
|
||||
_write_entry(quarantine, "20260514_120000", "song.flac")
|
||||
|
||||
target = recover_to_staging(str(quarantine), str(staging), "20260514_120000_song")
|
||||
assert target is not None
|
||||
assert os.path.basename(target) == "song_(2).flac"
|
||||
|
||||
|
||||
def test_recover_removes_sidecar_after_move(tmp_path):
|
||||
quarantine = tmp_path / "q"
|
||||
quarantine.mkdir()
|
||||
staging = tmp_path / "s"
|
||||
_, sidecar = _write_entry(quarantine, "20260514_120000", "song.flac")
|
||||
|
||||
recover_to_staging(str(quarantine), str(staging), "20260514_120000_song")
|
||||
assert not sidecar.exists()
|
||||
|
|
@ -8194,6 +8194,93 @@ def clear_quarantine():
|
|||
logger.error(f"[Quarantine] Error clearing quarantine: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
def _get_quarantine_dir():
|
||||
return os.path.join(
|
||||
docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')),
|
||||
'ss_quarantine',
|
||||
)
|
||||
|
||||
|
||||
@app.route('/api/quarantine/list', methods=['GET'])
|
||||
def list_quarantine():
|
||||
"""Return all quarantined files with sidecar metadata."""
|
||||
try:
|
||||
from core.imports.quarantine import list_quarantine_entries
|
||||
entries = list_quarantine_entries(_get_quarantine_dir())
|
||||
return jsonify({"success": True, "entries": entries})
|
||||
except Exception as e:
|
||||
logger.error(f"[Quarantine] Error listing entries: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/quarantine/<entry_id>', methods=['DELETE'])
|
||||
def delete_quarantine_item(entry_id):
|
||||
"""Delete a single quarantined file + sidecar."""
|
||||
try:
|
||||
from core.imports.quarantine import delete_quarantine_entry
|
||||
ok = delete_quarantine_entry(_get_quarantine_dir(), entry_id)
|
||||
if not ok:
|
||||
return jsonify({"success": False, "error": "Entry not found"}), 404
|
||||
return jsonify({"success": True})
|
||||
except Exception as e:
|
||||
logger.error(f"[Quarantine] Error deleting {entry_id}: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/quarantine/<entry_id>/approve', methods=['POST'])
|
||||
def approve_quarantine_item(entry_id):
|
||||
"""One-click approve: restore the file and re-run post-process with the
|
||||
matching per-check bypass flag set so the original quarantine trigger
|
||||
is skipped. Other checks still run."""
|
||||
try:
|
||||
from core.imports.quarantine import approve_quarantine_entry
|
||||
# Restore inside the soulseek download dir so existing path-resolution
|
||||
# logic finds it. Unique subdir keeps it from re-mingling with active
|
||||
# transfers.
|
||||
restore_dir = os.path.join(
|
||||
docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')),
|
||||
'Transfer',
|
||||
)
|
||||
result = approve_quarantine_entry(_get_quarantine_dir(), entry_id, restore_dir)
|
||||
if result is None:
|
||||
return jsonify({
|
||||
"success": False,
|
||||
"error": "Cannot one-click approve — entry has thin sidecar (no embedded context). Use 'Recover to Staging' instead.",
|
||||
}), 400
|
||||
restored_path, context, trigger = result
|
||||
# Mark the bypass so the pipeline skips the trigger that fired.
|
||||
context['_skip_quarantine_check'] = trigger
|
||||
# Re-dispatch through the same pipeline. Run async so the HTTP
|
||||
# request returns quickly — UI polls /list to see the entry vanish.
|
||||
context_key = f"approve_{entry_id}_{int(time.time())}"
|
||||
threading.Thread(
|
||||
target=lambda: _post_process_matched_download(context_key, context, restored_path),
|
||||
daemon=True,
|
||||
).start()
|
||||
logger.info(f"[Quarantine] Approved {entry_id} (bypass={trigger}) → re-running pipeline")
|
||||
return jsonify({"success": True, "trigger_bypassed": trigger})
|
||||
except Exception as e:
|
||||
logger.error(f"[Quarantine] Error approving {entry_id}: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/quarantine/<entry_id>/recover', methods=['POST'])
|
||||
def recover_quarantine_item(entry_id):
|
||||
"""Fallback for legacy thin sidecars: move file into Staging so the user
|
||||
can manually finish via the existing Import flow."""
|
||||
try:
|
||||
from core.imports.quarantine import recover_to_staging
|
||||
from core.imports.staging import get_staging_path
|
||||
target = recover_to_staging(_get_quarantine_dir(), get_staging_path(), entry_id)
|
||||
if not target:
|
||||
return jsonify({"success": False, "error": "Entry not found"}), 404
|
||||
return jsonify({"success": True, "staged_path": target})
|
||||
except Exception as e:
|
||||
logger.error(f"[Quarantine] Error recovering {entry_id}: {e}")
|
||||
return jsonify({"success": False, "error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/scan/request', methods=['POST'])
|
||||
def request_media_scan():
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -2200,6 +2200,7 @@
|
|||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
|
@ -7571,6 +7572,9 @@
|
|||
<button class="library-history-tab" data-tab="import" onclick="switchHistoryTab('import')">
|
||||
Server Imports <span class="library-history-tab-count" id="history-import-count">0</span>
|
||||
</button>
|
||||
<button class="library-history-tab" data-tab="quarantine" onclick="switchHistoryTab('quarantine')">
|
||||
Quarantine <span class="library-history-tab-count" id="history-quarantine-count">0</span>
|
||||
</button>
|
||||
</div>
|
||||
<div class="history-source-bar" id="history-source-bar" style="display:none"></div>
|
||||
<div class="library-history-list" id="library-history-list"></div>
|
||||
|
|
|
|||
|
|
@ -3382,7 +3382,18 @@ function processModalStatusUpdate(playlistId, data) {
|
|||
case 'post_processing': statusText = '⌛ Processing...'; break;
|
||||
case 'completed': statusText = '✅ Completed'; completedCount++; break;
|
||||
case 'not_found': statusText = '🔇 Not Found'; notFoundCount++; break;
|
||||
case 'failed': statusText = '❌ Failed'; failedOrCancelledCount++; break;
|
||||
case 'failed': {
|
||||
// Distinguish quarantine outcomes from generic
|
||||
// failures — the file is recoverable, not lost.
|
||||
const _em = (task.error_message || '').toLowerCase();
|
||||
if (_em.includes('integrity check failed') || _em.includes('bit depth filter') || _em.includes('verification failed') || _em.includes('quarantin')) {
|
||||
statusText = '🛡️ Quarantined';
|
||||
} else {
|
||||
statusText = '❌ Failed';
|
||||
}
|
||||
failedOrCancelledCount++;
|
||||
break;
|
||||
}
|
||||
case 'cancelled': statusText = '🚫 Cancelled'; failedOrCancelledCount++; break;
|
||||
default: statusText = `⚪ ${task.status}`; break;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3416,6 +3416,7 @@ const WHATS_NEW = {
|
|||
'2.5.2': [
|
||||
// --- May 13, 2026 — 2.5.2 release ---
|
||||
{ date: 'May 13, 2026 — 2.5.2 release' },
|
||||
{ title: 'Quarantine Management — See, Approve, Delete Files Without Touching The Filesystem', desc: 'github issue #584: quarantined files used to just sit in `ss_quarantine/` with a thin sidecar — no UI, no recovery, no way to see what got dropped or why. new **Quarantine** tab on the existing Library History modal (downloads page → Download History button) lists every quarantined file with the same row chrome as the Downloads + Server Imports tabs: thumb placeholder, expected track + artist, original filename, trigger badge (Duration / AcoustID / Bit Depth), relative time, expandable details panel showing the full failure reason. three per-row actions: **Approve** (restores the file, re-runs post-processing with ONLY the failing check skipped, lands in your library with full tags + lyrics + scan), **Recover** (legacy fallback for entries quarantined before this PR with thin sidecars — moves to Staging so you finish via Import flow), **Delete** (permanent removal of file + sidecar). all three use the themed soulsync confirm modal + toast feedback (no native browser alert / confirm). per-check bypass means approving a duration-mismatch file still runs AcoustID; approving an AcoustID failure still runs bit-depth — other quality gates stay live so you can only override one trigger at a time. files that fail a different check after approval get re-quarantined with the new trigger label so you can decide again. sidecar now persists the full json-safe context so approve has everything the pipeline needs to re-process. download modal status differentiates "🛡️ Quarantined" from "❌ Failed" so recoverable files are visible at a glance. logic lifted to pure helpers in `core/imports/quarantine.py` (list / delete / approve / recover_to_staging / serialize_quarantine_context) with 27 boundary tests covering orphan files / orphan sidecars / corrupt sidecars / collision-safe filename restoration / full-context vs thin-sidecar dispatch / json round-trip safety. four new endpoints. pipeline change is per-check conditionals at the existing quarantine sites — no blanket skip-all flag.', page: 'downloads' },
|
||||
{ title: 'Configurable Duration Tolerance For Quarantined Tracks', desc: 'discord question: tracks were quarantining when their actual length drifted by a few seconds from what spotify/musicbrainz reported (3s tolerance hardcoded, 5s for tracks >10min). live recordings, alternate masterings, and some legitimate uploads routinely drift more than that. new setting on settings → metadata → post-processing: "duration tolerance (seconds)". `0 = auto` (preserves the existing 3s/5s defaults). raise it to 10 / 15 / 20 if your library has a lot of drift-prone material. capped at 60s — past that the check is effectively off. applies to ALL matched downloads (soulseek / tidal / qobuz / hifi / youtube / deezer-direct) since they all flow through the same post-process integrity check. logic lifted to a pure helper `core/imports/file_integrity.py:resolve_duration_tolerance` that coerces the config value (none / empty / 0 / negative / unparseable / above-cap) to either a float override or `None` for the auto-scaled default. 12 tests pin every input shape.', page: 'settings' },
|
||||
{ title: 'Soulseek Downloads: Multi-Artist Tags Now Get Written Properly', desc: 'discord report: tracks downloaded via soulseek were getting tagged with primary artist only (no collab artists), while the same track downloaded via deezer tagged everyone correctly. trace: the soulseek matched-download context constructed `original_search_result` with `artist` (singular string) but no `artists` (list), even though the full multi-artist list lived on `track_info` (the matched spotify track object). `core/metadata/source.py:extract_source_metadata` only read `original_search.artists`, so soulseek path always fell through to the single-artist branch. fix: lifted artist resolution into a pure helper `core/metadata/artist_resolution.py:resolve_track_artists` that walks `original_search.artists` → `track_info.artists` → `artist_dict.name` fallback chain. handles all three list-item shapes (spotify-style dicts, bare strings, anything else stringified). 13 tests pin the resolution order, fallback chain, mixed-shape normalization, whitespace stripping, empty/none handling. composes with the existing deezer per-track upgrade (still fires when single-artist + track_id available) and feat_in_title / artist_separator settings (still drive the joined ARTIST string downstream).', page: 'downloads' },
|
||||
{ title: 'Download Missing Modal: Tracklist Got A Polish Pass', desc: 'visual tune-up only — column layout untouched. hairline row dividers, accent gradient + edge bar on hover, monospace track numbers (glow accent on row hover), monospace tabular duration. status text in both library-match + download-status columns picks up a leading colored dot with a soft halo (green found / amber missing / blue checking / orange downloading / red failed) and pulses while in-flight. artist column centered. soft scrollbar.', page: 'downloads' },
|
||||
|
|
|
|||
|
|
@ -3094,6 +3094,150 @@ function openLibraryHistoryModal() {
|
|||
}
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
// Quarantine tab — rendered inside the Library History modal as a third
|
||||
// tab next to Downloads + Server Imports. Reuses the existing list +
|
||||
// pagination chrome; provides per-row Approve / Recover / Delete actions.
|
||||
// ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
async function loadQuarantineList() {
|
||||
const list = document.getElementById('library-history-list');
|
||||
const pagination = document.getElementById('library-history-pagination');
|
||||
const sourceBar = document.getElementById('history-source-bar');
|
||||
if (!list) return;
|
||||
list.innerHTML = '<div class="library-history-loading">Loading...</div>';
|
||||
if (pagination) pagination.innerHTML = '';
|
||||
if (sourceBar) sourceBar.style.display = 'none';
|
||||
|
||||
try {
|
||||
const resp = await fetch('/api/quarantine/list');
|
||||
const data = await resp.json();
|
||||
const entries = data.entries || [];
|
||||
const countEl = document.getElementById('history-quarantine-count');
|
||||
if (countEl) countEl.textContent = entries.length;
|
||||
|
||||
if (!data.success) {
|
||||
list.innerHTML = `<div class="library-history-empty">Error: ${escapeHtml(data.error || 'Failed to load')}</div>`;
|
||||
return;
|
||||
}
|
||||
if (entries.length === 0) {
|
||||
list.innerHTML = '<div class="library-history-empty">🛡️<br><br>No quarantined files. Nice and clean.</div>';
|
||||
return;
|
||||
}
|
||||
list.innerHTML = entries.map(renderQuarantineEntry).join('');
|
||||
} catch (err) {
|
||||
console.error('Error loading quarantine entries:', err);
|
||||
list.innerHTML = '<div class="library-history-empty">Error loading quarantine</div>';
|
||||
}
|
||||
}
|
||||
|
||||
function renderQuarantineEntry(entry) {
|
||||
const triggerLabels = { integrity: 'Duration / Integrity', acoustid: 'AcoustID Mismatch', bit_depth: 'Bit Depth Filter', unknown: 'Unknown' };
|
||||
const triggerColors = { integrity: '#facc15', acoustid: '#ef5350', bit_depth: '#fb923c', unknown: '#888' };
|
||||
const triggerLabel = triggerLabels[entry.trigger] || entry.trigger || 'Unknown';
|
||||
const triggerColor = triggerColors[entry.trigger] || '#888';
|
||||
|
||||
const id = escapeHtml(entry.id);
|
||||
const approveLabel = entry.has_full_context ? 'Approve' : 'Recover';
|
||||
const approveTitle = entry.has_full_context
|
||||
? 'Re-run post-processing with only the failing check skipped'
|
||||
: 'Legacy entry — move to Staging, finish via Import flow';
|
||||
const approveCall = entry.has_full_context
|
||||
? `approveQuarantineEntry('${id}')`
|
||||
: `recoverQuarantineEntry('${id}')`;
|
||||
|
||||
const meta = [entry.expected_artist, entry.original_filename].filter(Boolean).join(' — ');
|
||||
const triggerBadge = `<span class="library-history-badge" style="border-color:${triggerColor};color:${triggerColor}">${escapeHtml(triggerLabel)}</span>`;
|
||||
const reasonDetail = `<div class="library-history-entry-source"><span class="lh-prov-label">Reason:</span> ${escapeHtml(entry.reason || 'Unknown')}</div>`;
|
||||
|
||||
return `<div class="library-history-entry lh-expandable" onclick="this.classList.toggle('lh-expanded')">
|
||||
<div class="library-history-thumb-placeholder">🛡️</div>
|
||||
<div class="library-history-entry-content">
|
||||
<div class="library-history-entry-row1">
|
||||
<div class="library-history-entry-text">
|
||||
<div class="library-history-entry-title">${escapeHtml(entry.expected_track || entry.original_filename || 'Unknown')}</div>
|
||||
<div class="library-history-entry-meta">${escapeHtml(meta)}</div>
|
||||
</div>
|
||||
<div class="library-history-entry-badges">${triggerBadge}</div>
|
||||
<div class="library-history-entry-time">${formatHistoryTime(entry.timestamp)}</div>
|
||||
<button class="lh-audit-btn" title="${approveTitle}" onclick="event.stopPropagation();${approveCall}">${approveLabel}</button>
|
||||
<button class="lh-audit-btn" title="Delete permanently" style="border-color:rgba(248,113,113,0.4);color:#f87171" onclick="event.stopPropagation();deleteQuarantineEntry('${id}')">Delete</button>
|
||||
<span class="lh-expand-btn">▾</span>
|
||||
</div>
|
||||
<div class="library-history-entry-details">
|
||||
${reasonDetail}
|
||||
</div>
|
||||
</div>
|
||||
</div>`;
|
||||
}
|
||||
|
||||
async function approveQuarantineEntry(entryId) {
|
||||
const ok = await showConfirmDialog({
|
||||
title: 'Approve Quarantined File',
|
||||
message: 'Re-run post-processing for this file with only the failing check skipped. The file will be tagged, lyrics generated, and moved into your library. Other quality gates (AcoustID + bit-depth) still run.',
|
||||
confirmText: 'Approve & Import',
|
||||
cancelText: 'Cancel',
|
||||
});
|
||||
if (!ok) return;
|
||||
try {
|
||||
const r = await fetch(`/api/quarantine/${encodeURIComponent(entryId)}/approve`, { method: 'POST' });
|
||||
const data = await r.json();
|
||||
if (!data.success) {
|
||||
showToast(`Approve failed: ${data.error}`, 'error');
|
||||
} else {
|
||||
showToast(`Approved — skipped ${data.trigger_bypassed} check, re-running pipeline.`, 'success');
|
||||
}
|
||||
} catch (err) {
|
||||
showToast(`Approve failed: ${err.message}`, 'error');
|
||||
}
|
||||
loadQuarantineList();
|
||||
}
|
||||
|
||||
async function recoverQuarantineEntry(entryId) {
|
||||
const ok = await showConfirmDialog({
|
||||
title: 'Recover To Staging',
|
||||
message: 'Legacy entry — no embedded context. The file will be moved to your Staging folder so you can finish via the Import page (manual match).',
|
||||
confirmText: 'Move To Staging',
|
||||
cancelText: 'Cancel',
|
||||
});
|
||||
if (!ok) return;
|
||||
try {
|
||||
const r = await fetch(`/api/quarantine/${encodeURIComponent(entryId)}/recover`, { method: 'POST' });
|
||||
const data = await r.json();
|
||||
if (!data.success) {
|
||||
showToast(`Recover failed: ${data.error}`, 'error');
|
||||
} else {
|
||||
showToast('Moved to Staging — finish via the Import page.', 'success');
|
||||
}
|
||||
} catch (err) {
|
||||
showToast(`Recover failed: ${err.message}`, 'error');
|
||||
}
|
||||
loadQuarantineList();
|
||||
}
|
||||
|
||||
async function deleteQuarantineEntry(entryId) {
|
||||
const ok = await showConfirmDialog({
|
||||
title: 'Delete Quarantined File',
|
||||
message: 'This permanently removes the file and its metadata sidecar. Cannot be undone.',
|
||||
confirmText: 'Delete',
|
||||
cancelText: 'Cancel',
|
||||
destructive: true,
|
||||
});
|
||||
if (!ok) return;
|
||||
try {
|
||||
const r = await fetch(`/api/quarantine/${encodeURIComponent(entryId)}`, { method: 'DELETE' });
|
||||
const data = await r.json();
|
||||
if (!data.success) {
|
||||
showToast(`Delete failed: ${data.error}`, 'error');
|
||||
} else {
|
||||
showToast('Quarantined file deleted.', 'success');
|
||||
}
|
||||
} catch (err) {
|
||||
showToast(`Delete failed: ${err.message}`, 'error');
|
||||
}
|
||||
loadQuarantineList();
|
||||
}
|
||||
|
||||
function closeLibraryHistoryModal() {
|
||||
const overlay = document.getElementById('library-history-overlay');
|
||||
if (overlay) overlay.classList.add('hidden');
|
||||
|
|
@ -3110,6 +3254,12 @@ function switchHistoryTab(tab) {
|
|||
|
||||
async function loadLibraryHistory() {
|
||||
const { tab, page, limit } = _libraryHistoryState;
|
||||
if (tab === 'quarantine') {
|
||||
// Refresh the count for the other two tabs in the background so
|
||||
// the badge stays accurate when the user switches over.
|
||||
loadQuarantineList();
|
||||
return;
|
||||
}
|
||||
const list = document.getElementById('library-history-list');
|
||||
const pagination = document.getElementById('library-history-pagination');
|
||||
if (!list) return;
|
||||
|
|
|
|||
Loading…
Reference in a new issue