Add automatic sensitivity fallback and default vpipe cleanup

src/videoclean/cli.py

@@ -52,7 +52,7 @@ def _build_parser():
     clean.add_argument("--intent", default=None)
     clean.add_argument("--preview", action="store_true")
     clean.add_argument("--confirm", action="store_true")
-    clean.add_argument("--detect-mode", choices=["fast", "balanced", "sensitive"], default="balanced")
+    clean.add_argument("--detect-mode", choices=["auto", "fast", "balanced", "sensitive"], default="auto")
     clean.add_argument("--ocr", choices=["auto", "off", "rapidocr"], default="auto")
     clean.add_argument("--detector", dest="detector_mode", choices=["dbnet", "hybrid"], default="dbnet")
     clean.add_argument("--inpaint-method", choices=["telea", "ns"], default="telea")

src/videoclean/engine.py

@@ -41,7 +41,7 @@ class WipeEngine:
         self, task: str = "detext", *, verbose: bool = False, gap: int = 10,
         dual: bool = False, detector: Optional[TextDetector] = None,
         model: str = "opencv", model_options: Optional[dict] = None,
-        detect_mode: str = "balanced", ocr: str = "auto", **legacy,
+        detect_mode: str = "auto", ocr: str = "auto", **legacy,
     ):
         if task not in _TASK_CLASSES:
             raise ValueError(f"Unknown task: {task}. Choose from: {list(_TASK_CLASSES)}")
@@ -265,8 +265,9 @@ class WipeEngine:
 
     def _build_fresh_clean_plan(self, video, result, selected_ids, snapshot, directed):
         selection = self._explicit_selection_kwargs(result.candidates, selected_ids, directed)
+        effective_mode = snapshot.get("effective_detect_mode", snapshot["detect_mode"])
         return build_refined_wipe_plan(
-            video, result, compute_source(video), refine=snapshot["detect_mode"] != "fast",
+            video, result, compute_source(video), refine=effective_mode != "fast",
             request=snapshot, progress=lambda done, total: self._emit_progress(ProgressEvent("refine", done, total)),
             check_cancelled=self._check_cancelled, **selection,
         )
@@ -297,7 +298,9 @@ class WipeEngine:
         effective_targets = list(dict.fromkeys(effective_targets))
         normalized = {normalize_target(target) for target in effective_targets}
         detect_text = not requested_regions and not normalized or bool(normalized & {"subtitle", "timestamp", "watermark", "scene_text", "unknown_text"}) or bool(intent and not normalized)
-        params = resolve_detect_params(detect_mode or self._detect_mode, has_subtitle_target="subtitle" in normalized)
+        requested_mode = detect_mode or self._detect_mode
+        if requested_mode not in {"auto", "fast", "balanced", "sensitive"}:
+            raise ValueError("detect_mode must be auto, fast, balanced, or sensitive")
         recognizer = self._build_recognizer(ocr or self._ocr)
         mode = detector_mode or "dbnet"
         if mode not in {"dbnet", "hybrid"}:
@@ -307,18 +310,42 @@ class WipeEngine:
             detector = _default_detector()
             if mode == "hybrid":
                 detector = HybridTextDetector(detector)
-        self._emit_progress(ProgressEvent("detect", 0, params["sample_count"], message=f"Scanning {params['sample_count']} sampled frames"))
-        result = detect_clean_candidates(video, detector=detector, regions=requested_regions, detect_text=detect_text, include_logo="logo" in normalized, include_translucent_watermark="watermark" in normalized, sample_count=params["sample_count"], consistency=params["consistency"], subtitle_fallback=params["subtitle_fallback"], recognizer=recognizer)
-        selected = select_clean_candidates(result.candidates, targets=effective_targets, intent=intent)
-        print(f"[videoclean] found {len(result.candidates)} candidate(s); selected {len(selected)}")
+        def run_pass(mode_name):
+            params = resolve_detect_params(mode_name, has_subtitle_target="subtitle" in normalized)
+            self._emit_progress(ProgressEvent("detect", 0, params["sample_count"], message=f"Scanning {params['sample_count']} sampled frames ({mode_name})"))
+            detected = detect_clean_candidates(
+                video, detector=detector, regions=requested_regions,
+                detect_text=detect_text, include_logo="logo" in normalized,
+                include_translucent_watermark="watermark" in normalized,
+                sample_count=params["sample_count"], consistency=params["consistency"],
+                subtitle_fallback=params["subtitle_fallback"], recognizer=recognizer,
+            )
+            chosen = select_clean_candidates(detected.candidates, targets=effective_targets, intent=intent)
+            return detected, chosen, mode_name
+
+        if requested_mode == "auto":
+            result, selected, effective_mode = run_pass("balanced")
+            if not result.candidates:
+                print("[videoclean] no candidates in balanced mode; retrying sensitive mode")
+                self._emit_progress(ProgressEvent("detect", 0, 1, message="Retrying with sensitive mode"))
+                result, selected, effective_mode = run_pass("sensitive")
+        else:
+            result, selected, effective_mode = run_pass(requested_mode)
+        print(f"[videoclean] {effective_mode} mode found {len(result.candidates)} candidate(s); selected {len(selected)}")
         if not result.candidates:
             print("[videoclean] no text or overlay candidates were detected")
         write_clean_artifacts(result, selected, output)
+
         if confirm:
             selected = self._confirm_candidates(result.candidates, selected)
             write_clean_artifacts(result, selected, output)
         self._emit_progress(ProgressEvent("detect", 1, 1))
-        return result, {candidate.id for candidate in selected}, {"intent": intent, "targets": effective_targets, "regions": requested_regions, "detect_mode": detect_mode or self._detect_mode, "ocr": ocr or self._ocr, "detector_mode": mode}, bool(targets or intent or regions or confirm)
+        snapshot = {
+            "intent": intent, "targets": effective_targets, "regions": requested_regions,
+            "detect_mode": requested_mode, "effective_detect_mode": effective_mode,
+            "ocr": ocr or self._ocr, "detector_mode": mode,
+        }
+        return result, {candidate.id for candidate in selected}, snapshot, bool(targets or intent or regions or confirm)
 
     @staticmethod
     def _confirm_candidates(candidates, selected):

tests/test_auto_detect.py

@@ -0,0 +1,58 @@
+from types import SimpleNamespace
+
+from videoclean.cli import _build_parser
+from videoclean.engine import WipeEngine
+
+
+def test_cli_defaults_to_auto_detection():
+    args = _build_parser().parse_args(["clean", "input.mp4"])
+    assert args.detect_mode == "auto"
+
+
+def test_explicit_modes_remain_available():
+    for mode in ("fast", "balanced", "sensitive"):
+        args = _build_parser().parse_args(["clean", "input.mp4", "--detect-mode", mode])
+        assert args.detect_mode == mode
+
+
+def test_auto_mode_runs_balanced_then_sensitive(monkeypatch, tmp_path):
+    empty = SimpleNamespace(candidates=[])
+    found = SimpleNamespace(candidates=[])
+    calls = []
+
+    def fake_detect(*args, **kwargs):
+        calls.append(kwargs["sample_count"])
+        return empty if len(calls) == 1 else found
+
+    monkeypatch.setattr("videoclean.detect.detect_clean_candidates", fake_detect)
+    monkeypatch.setattr("videoclean.detect._default_detector", lambda: object())
+    monkeypatch.setattr("videoclean.detect.write_clean_artifacts", lambda *args: None)
+    monkeypatch.setattr("videoclean.engine.WipeEngine._build_recognizer", staticmethod(lambda mode: None))
+    engine = WipeEngine(task="clean", detect_mode="auto", ocr="off")
+    result, selected, snapshot, directed = engine._detect_clean(
+        "input.mp4", object(), [], None, [], None, "off", str(tmp_path), False, "dbnet"
+    )
+    assert calls == [50, 80]
+    assert snapshot["detect_mode"] == "auto"
+    assert snapshot["effective_detect_mode"] == "sensitive"
+    assert selected == set()
+    assert directed is False
+
+
+def test_explicit_fast_does_not_retry(monkeypatch, tmp_path):
+    calls = []
+
+    def fake_detect(*args, **kwargs):
+        calls.append(kwargs["sample_count"])
+        return SimpleNamespace(candidates=[])
+
+    monkeypatch.setattr("videoclean.detect.detect_clean_candidates", fake_detect)
+    monkeypatch.setattr("videoclean.detect._default_detector", lambda: object())
+    monkeypatch.setattr("videoclean.detect.write_clean_artifacts", lambda *args: None)
+    monkeypatch.setattr("videoclean.engine.WipeEngine._build_recognizer", staticmethod(lambda mode: None))
+    engine = WipeEngine(task="clean", detect_mode="fast", ocr="off")
+    _, _, snapshot, _ = engine._detect_clean(
+        "input.mp4", object(), [], None, [], None, "off", str(tmp_path), False, "dbnet"
+    )
+    assert calls == [24]
+    assert snapshot["effective_detect_mode"] == "fast"

vpipe

@@ -15,9 +15,10 @@ def main() -> int:
     parser = argparse.ArgumentParser(description="Batch-clean videos with videoclean")
     parser.add_argument("source", help="Video file or directory")
     parser.add_argument("-o", "--output", help="Output directory")
-    parser.add_argument("--clean", action="store_true", help="Remove detected overlays")
+    parser.add_argument("--clean", action="store_true", help="Compatibility flag; cleaning is the default")
+    parser.add_argument("--no-clean", "--copy", action="store_true", help="Copy files without cleaning")
     parser.add_argument("-y", "--yes", action="store_true", help="Run without confirmation")
-    parser.add_argument("--detect-mode", choices=["fast", "balanced", "sensitive"], default="balanced")
+    parser.add_argument("--detect-mode", choices=["auto", "fast", "balanced", "sensitive"], default="auto")
     parser.add_argument("--inpaint-method", choices=["telea", "ns"], default="telea")
     parser.add_argument("--inpaint-radius", type=float, default=3.0)
     parser.add_argument("--inpaint-dilate", type=int, default=0)
@@ -41,7 +42,7 @@ def main() -> int:
 
     for video in videos:
         destination = output / f"{video.stem}.mp4"
-        if not args.clean:
+        if args.no_clean:
             shutil.copy2(video, destination)
             continue
         command = [