feat(snot): add search history and purge command

download_imageset.py

@@ -7,9 +7,11 @@ import dataclasses
 import logging
 import os
 import re
+import shlex
 import shutil
 import struct
 import subprocess
+import tempfile
 import urllib.parse
 import zipfile
 from functools import partial
@@ -201,6 +203,62 @@ def _has_images(d: Path) -> bool:
     return any(f.suffix.lower() in IMAGE_EXTS for f in d.rglob("*") if f.is_file())
 
 
+_TRICKLEBAR_APP = Path("/Applications/TrickleBar.app")
+
+
+def download_url(url: str, destination: Path) -> None:
+    destination = destination.resolve()
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    if _TRICKLEBAR_APP.is_dir():
+        logging.info(f"downloading with TrickleBar → {destination}")
+        download_url_with_tricklebar(url=url, destination=destination)
+    else:
+        logging.info(f"downloading with aria2 → {destination}")
+        download_url_with_aria2(url=url, destination=destination)
+
+
+def download_url_with_aria2(url: str, destination: Path) -> None:
+    run_command(args=[
+        "aria2c", "--continue=true", f"--dir={destination.parent}",
+        f"--out={destination.name}", url,
+    ])
+
+
+def download_url_with_tricklebar(url: str, destination: Path) -> None:
+    with tempfile.TemporaryDirectory(prefix="imageset-") as temporary_directory:
+        directory = Path(temporary_directory)
+        completion_pipe = directory / "completion"
+        callback = directory / "on_done.sh"
+        os.mkfifo(path=completion_pipe, mode=0o600)
+        callback.write_text(data=(
+            "#!/bin/sh\n"
+            f"[ -p {shlex.quote(str(completion_pipe))} ] || exit 0\n"
+            f"printf '%s\\0' \"$1\" \"$2\" > {shlex.quote(str(completion_pipe))}\n"
+        ))
+        callback.chmod(mode=0o700)
+        query = urllib.parse.urlencode(query={
+            "on_done": str(callback), "url": url,
+            "dir": str(destination.parent), "out": destination.name, "continue": "true",
+        }, quote_via=urllib.parse.quote)
+        logging.info("waiting for TrickleBar; press Ctrl+C to stop waiting")
+        run_command(args=["open", f"tricklebar://add-download?{query}"])
+        with completion_pipe.open(mode="rb") as completion:
+            status, downloaded_path, _ = completion.read().split(b"\0", maxsplit=2)
+        if status != b"complete":
+            raise click.ClickException(
+                f"TrickleBar download failed: {os.fsdecode(status)}"
+            )
+        if not downloaded_path or not Path(os.fsdecode(downloaded_path)).is_file():
+            raise click.ClickException("TrickleBar completed without a downloaded file")
+        if Path(os.fsdecode(downloaded_path)).resolve() != destination.resolve():
+            raise click.ClickException(
+                f"TrickleBar downloaded to {os.fsdecode(downloaded_path)} instead of {destination}; "
+                "update TrickleBar to a version supporting dir/out URL options"
+            )
+        if destination.with_name(f"{destination.name}.aria2").exists():
+            raise click.ClickException(f"TrickleBar download is unfinished: {destination}")
+
+
 def _needs_resize(d: Path) -> bool:
     images = [f for f in d.rglob("*") if f.is_file() and f.suffix.lower() in IMAGE_EXTS and not f.is_relative_to(d / "resized")]
     if not images:
@@ -284,6 +342,9 @@ def process_imageset(path: Path, processor: ImageProcessor) -> None:
         return
 
     for archive in archives:
+        if archive.with_name(f"{archive.name}.aria2").exists():
+            logging.warning(f"skipping {archive}: download is unfinished")
+            return
         logging.info(f"extracting {archive} → {output_dir}")
         if archive.suffix.lower() == ".zip":
             try:
@@ -342,7 +403,7 @@ def remove_empty_dirs(root_dir: Path) -> None:
     for d in removed:
         d.rmdir()
     if removed:
-        print(f"removed {len(removed)} empty dir(s) in {root_dir}")
+        logging.info(f"removed {len(removed)} empty dir(s) in {root_dir}")
 
 
 def move_to_trash(file_paths: list[Path]) -> None:
@@ -423,9 +484,9 @@ def gallery_cmd(urls: tuple[str, ...], name: str | None, cwd: Path | None) -> No
     output_dir = (cwd or Path(".")) / resolved_name
 
     if output_dir.exists() and _has_images(output_dir):
-        print(f"skipping gallery-dl: {output_dir} already has images")
+        logging.info(f"skipping gallery-dl: {output_dir} already has images")
     else:
-        print(f"gallery-dl → {output_dir}")
+        logging.info(f"gallery-dl → {output_dir}")
         args = ["uvx", "gallery-dl"]
         if name or cwd:
             args += ["-D", str(output_dir)]
@@ -445,24 +506,36 @@ def filehost_cmd(urls: tuple[str, ...], name: str | None, cwd: Path, token: str)
     output_dir.mkdir(parents=True, exist_ok=True)
 
     for url in urls:
-        print(f"unrestricting {url}")
+        logging.info(f"unrestricting {url}")
         dl_url = resolve_download_url_with_alldebrid(token, url)
         meta = _resolve_url(dl_url)
         if meta.suggested_extension in _VIDEO_EXTS:
             video_path = (cwd or Path(".")) / (resolved_name + meta.suggested_extension)
-            print(f"video detected → {video_path}")
+            logging.info(f"video detected → {video_path}")
-            if video_path.exists() and video_path.stat().st_size > 5 * 1024 * 1024:
+            if video_path.with_name(f"{video_path.name}.aria2").exists():
-                print(f"skipping: {video_path.name} already downloaded")
+                logging.warning(f"unfinished download; resuming {video_path}")
+            if (
+                video_path.exists()
+                and not video_path.with_name(f"{video_path.name}.aria2").exists()
+                and video_path.stat().st_size > 5 * 1024 * 1024
+            ):
+                logging.info(f"skipping: {video_path.name} already downloaded")
                 continue
-            run_command(["aria2c", "-o", str(video_path), dl_url])
+            download_url(url=dl_url, destination=video_path)
             continue
 
         dest = output_dir / meta.suggested_filename
-        if dest.exists() and dest.stat().st_size > 5 * 1024 * 1024:
+        if dest.with_name(f"{dest.name}.aria2").exists():
-            print(f"skipping: {dest.name} already downloaded")
+            logging.warning(f"unfinished download; resuming {dest}")
+        if (
+            dest.exists()
+            and not dest.with_name(f"{dest.name}.aria2").exists()
+            and dest.stat().st_size > 5 * 1024 * 1024
+        ):
+            logging.info(f"skipping: {dest.name} already downloaded")
             continue
-        print(f"downloading {meta.suggested_filename} → {output_dir}")
+        logging.info(f"downloading {meta.suggested_filename} → {output_dir}")
-        run_command(["aria2c", f"--dir={output_dir}", dl_url])
+        download_url(url=dl_url, destination=dest)
 
     process_imageset(path=output_dir, processor=detect_resizer())
 

download_imageset_test.py

@@ -6,7 +6,10 @@
 import shutil
 import subprocess
 import sys
+import threading
+import urllib.parse
 import zipfile
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from functools import partial
 from unittest.mock import Mock
@@ -297,5 +300,177 @@ def test_resume_resizing(
     trash.assert_called_once_with(file_paths=[image_path])
 
 
+@pytest.mark.parametrize("extension", [".zip", ".mp4"])
+@pytest.mark.parametrize("unfinished", [False, True])
+def test_filehost_resumes_unfinished_downloads(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+    extension: str,
+    unfinished: bool,
+) -> None:
+    directory = tmp_path / "album"
+    directory.mkdir()
+    destination = (
+        directory / "download.zip" if extension == ".zip" else tmp_path / "album.mp4"
+    )
+    with destination.open(mode="wb") as output:
+        output.truncate(6 * 1024 * 1024)
+    control_file = destination.with_name(f"{destination.name}.aria2")
+    if unfinished:
+        control_file.touch()
+    url = "https://example.com/download"
+    monkeypatch.setattr(
+        download_imageset, "resolve_download_url_with_alldebrid", Mock(return_value=url)
+    )
+    monkeypatch.setattr(
+        download_imageset,
+        "_resolve_url",
+        Mock(return_value=download_imageset.UrlMeta(
+            suggested_filename=f"download{extension}",
+            suggested_extension=extension,
+            size_bytes=destination.stat().st_size,
+        )),
+    )
+    run_command = Mock()
+    monkeypatch.setattr(download_imageset, "_TRICKLEBAR_APP", tmp_path / "missing.app")
+    monkeypatch.setattr(download_imageset, "run_command", run_command)
+    monkeypatch.setattr(download_imageset, "process_imageset", Mock())
+
+    result = CliRunner().invoke(
+        cli=download_imageset.cli,
+        args=["filehost", url, "--name", "album", "--cwd", str(tmp_path), "--token", "test"],
+    )
+
+    assert result.exit_code == 0, result.output
+    if unfinished:
+        assert f"unfinished download; resuming {destination}" in caplog.text
+        run_command.assert_called_once_with(args=[
+            "aria2c", "--continue=true", f"--dir={destination.parent}",
+            f"--out={destination.name}", url,
+        ])
+    else:
+        assert "unfinished download; resuming" not in caplog.text
+        run_command.assert_not_called()
+
+
+def test_aria2_download_failure_propagates(monkeypatch: pytest.MonkeyPatch) -> None:
+    failure = subprocess.CalledProcessError(returncode=1, cmd="aria2c")
+    monkeypatch.setattr(download_imageset, "run_command", Mock(side_effect=failure))
+
+    with pytest.raises(subprocess.CalledProcessError):
+        download_imageset.download_url_with_aria2(
+            url="https://example.com/archive.zip", destination=Path("archive.zip")
+        )
+
+
+@pytest.mark.parametrize(
+    ("status", "file_state", "error"),
+    [
+        ("complete", "valid", None),
+        ("error", "valid", "download failed"),
+        ("complete", "missing", "without a downloaded file"),
+        ("complete", "wrong_path", "instead of"),
+        ("complete", "unfinished", "download is unfinished"),
+    ],
+)
+def test_tricklebar_waits_for_callback(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+    status: str, file_state: str, error: str | None,
+) -> None:
+    destination = tmp_path / "download & photos.zip"
+    downloaded_path = tmp_path / "wrong.zip" if file_state == "wrong_path" else destination
+    if file_state != "missing":
+        downloaded_path.write_bytes(data=b"downloaded archive")
+    if file_state == "unfinished":
+        destination.with_name(f"{destination.name}.aria2").touch()
+    url = "https://example.com/photos.zip?token=a+b&other=value"
+    opened = threading.Event()
+    callback_paths: list[Path] = []
+
+    def open_download(args: list[str]) -> None:
+        assert args[0] == "open"
+        parsed = urllib.parse.urlparse(url=args[1])
+        assert "+" not in parsed.query
+        assert parsed.scheme == "tricklebar"
+        assert parsed.netloc == "add-download"
+        query = urllib.parse.parse_qs(qs=parsed.query)
+        assert query["url"] == [url]
+        assert query["dir"] == [str(destination.parent)]
+        assert query["out"] == [destination.name]
+        assert query["continue"] == ["true"]
+        callback_paths.append(Path(query["on_done"][0]))
+        opened.set()
+
+    monkeypatch.setattr(download_imageset, "run_command", open_download)
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        download = executor.submit(
+            download_imageset.download_url_with_tricklebar, url=url, destination=destination
+        )
+        assert opened.wait(timeout=5)
+        assert not download.done()
+        callback = callback_paths[0]
+        subprocess.run(args=[str(callback), status, str(downloaded_path), "gid", url], check=True)
+        if error is None:
+            assert download.result(timeout=5) is None
+        else:
+            with pytest.raises(download_imageset.click.ClickException, match=error):
+                download.result(timeout=5)
+
+    assert not callback.parent.exists()
+
+
+@pytest.mark.parametrize("installed", [False, True])
+def test_download_selects_backend(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, installed: bool
+) -> None:
+    application = tmp_path / "TrickleBar.app"
+    if installed:
+        application.mkdir()
+    monkeypatch.setattr(download_imageset, "_TRICKLEBAR_APP", application)
+    aria2 = Mock()
+    tricklebar = Mock()
+    monkeypatch.setattr(download_imageset, "download_url_with_aria2", aria2)
+    monkeypatch.setattr(download_imageset, "download_url_with_tricklebar", tricklebar)
+    destination = tmp_path / "photos" / "album.zip"
+    url = "https://example.com/album.zip"
+
+    download_imageset.download_url(url=url, destination=destination)
+
+    assert destination.parent.is_dir()
+    selected = tricklebar if installed else aria2
+    unused = aria2 if installed else tricklebar
+    selected.assert_called_once_with(url=url, destination=destination.resolve())
+    unused.assert_not_called()
+
+
+@pytest.mark.parametrize("process_directory", [False, True])
+def test_process_skips_unfinished_archive(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+    process_directory: bool,
+) -> None:
+    archive_path = tmp_path / "gallery.zip"
+    with zipfile.ZipFile(file=archive_path, mode="w") as archive:
+        archive.writestr(zinfo_or_arcname="photo.png", data=b"image data")
+    archive_path.with_name(f"{archive_path.name}.aria2").touch()
+    trash = Mock()
+    monkeypatch.setattr(download_imageset, "move_to_trash", trash)
+
+    result = CliRunner().invoke(
+        cli=download_imageset.cli,
+        args=["process", str(tmp_path if process_directory else archive_path)],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert "download is unfinished" in caplog.text
+    assert not (tmp_path / "photo.png").exists()
+    assert not (tmp_path / "gallery").exists()
+    assert archive_path.is_file()
+    trash.assert_not_called()
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))

ffmpeg_remove_still.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+# /// script
+# dependencies = []
+# ///
+
+import argparse
+from bisect import bisect_left, bisect_right
+import json
+import logging
+import math
+import re
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Shorten still sections using FFmpeg stream copying, without re-encoding.",
+        epilog="Cuts follow video keyframes and audio packets. Still sections may remain longer than requested. "
+               "Audio during removed time is also removed, including any narration.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument("input", type=Path, help="Input video")
+    parser.add_argument("--output", type=Path, help="Output video (default: INPUT.short with the original extension)")
+    parser.add_argument("--min-still", type=float, default=2.0,
+                        help="Minimum still section duration to detect, in seconds")
+    parser.add_argument("--keep-still", type=float, default=0.5,
+                        help="Target seconds to retain per still section; keyframe alignment may retain more")
+    parser.add_argument("--noise", type=float, default=0.001,
+                        help="Frame difference tolerance, from 0 to 1; higher values ignore more movement")
+    parser.add_argument("--dry-run", action="store_true", help="Report cuts without writing output")
+    parser.add_argument("--overwrite", action="store_true", help="Replace an existing output file")
+    arguments = parser.parse_args()
+    if not math.isfinite(arguments.min_still) or arguments.min_still <= 0:
+        parser.error("--min-still must be positive and finite")
+    if not math.isfinite(arguments.keep_still) or arguments.keep_still <= 0:
+        parser.error("--keep-still must be positive and finite")
+    if not 0 <= arguments.noise <= 1:
+        parser.error("--noise must be between 0 and 1")
+    return arguments
+
+
+def probe_video(path: Path) -> tuple[float, float]:
+    result = subprocess.run(
+        args=["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
+        check=True, capture_output=True, text=True,
+    )
+    metadata = json.loads(result.stdout)
+    video = next((stream for stream in metadata["streams"] if stream["codec_type"] == "video"), None)
+    if video is None:
+        raise ValueError("Input has no video stream")
+    duration = float(video.get("duration") or metadata["format"].get("duration", 0))
+    if not math.isfinite(duration) or duration <= 0:
+        raise ValueError("Cannot determine video duration")
+    start_time = float(video.get("start_time") or metadata["format"].get("start_time", 0))
+    return duration, start_time
+
+
+def probe_keyframes(path: Path, start_time: float) -> list[tuple[float, float]]:
+    result = subprocess.run(
+        args=["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_packets",
+              "-show_entries", "packet=pts_time,dts_time,flags", "-of", "csv=p=0", str(path)],
+        check=True, capture_output=True, text=True,
+    )
+    keyframes = []
+    for line in result.stdout.splitlines():
+        fields = line.split(",")
+        if len(fields) >= 3 and "K" in fields[2] and fields[0] != "N/A":
+            presentation_time = float(fields[0]) - start_time
+            decode_time = float(fields[1]) - start_time if fields[1] != "N/A" else presentation_time
+            keyframes.append((max(0.0, presentation_time), decode_time))
+    if not keyframes:
+        raise ValueError("No video keyframes found; cannot safely trim by stream copying")
+    return sorted(keyframes)
+
+
+def parse_freezes(log: str, duration: float) -> list[tuple[float, float]]:
+    freezes = []
+    start = None
+    for kind, timestamp in re.findall(r"lavfi\.freezedetect\.freeze_(start|end):\s*([-+\d.eE]+)", log):
+        timestamp = min(duration, max(0.0, float(timestamp)))
+        if kind == "start":
+            start = timestamp
+        elif start is not None:
+            if timestamp > start:
+                freezes.append((start, timestamp))
+            start = None
+    if start is not None and duration > start:
+        freezes.append((start, duration))
+    return freezes
+
+
+def detect_freezes(path: Path, duration: float, min_still: float, noise: float) -> list[tuple[float, float]]:
+    logging.info("Detecting still sections...")
+    result = subprocess.run(
+        args=[
+            "ffmpeg", "-hide_banner", "-nostdin", "-nostats", "-i", str(path),
+            "-map", "0:v:0", "-vf", f"setpts=PTS-STARTPTS,freezedetect=noise={noise}:duration={min_still}",
+            "-an", "-f", "null", "-",
+        ],
+        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True,
+    )
+    return parse_freezes(log=result.stderr, duration=duration)
+
+
+def plan_segments(duration: float, freezes: list[tuple[float, float]], keep_still: float,
+                  keyframes: list[float]) -> list[tuple[float, float]]:
+    segments = []
+    cursor = 0.0
+    for start, end in freezes:
+        if end - start <= keep_still:
+            continue
+        first = bisect_left(keyframes, start + keep_still / 2)
+        last = bisect_right(keyframes, end - keep_still / 2) - 1
+        if first >= len(keyframes) or last < 0 or keyframes[first] >= keyframes[last]:
+            logging.info(f"Still section {start:.2f}–{end:.2f}s: no removable span between keyframes")
+            continue
+        cut_start = keyframes[first]
+        cut_end = keyframes[last]
+        logging.info(f"Still section {start:.2f}–{end:.2f}s: retain {end - start - (cut_end - cut_start):.2f}s")
+        if cut_start > cursor:
+            segments.append((cursor, cut_start))
+        cursor = cut_end
+    if cursor < duration:
+        segments.append((cursor, duration))
+    return segments
+
+
+def copy_video(source: Path, output: Path, segments: list[tuple[float, float]], duration: float,
+               start_time: float, keyframes: list[tuple[float, float]], overwrite: bool) -> None:
+    decode_times = dict(keyframes)
+    with tempfile.TemporaryDirectory() as temporary_directory:
+        playlist = Path(temporary_directory) / "segments.ffconcat"
+        local_source = Path(temporary_directory) / f"source{source.suffix}"
+        local_source.symlink_to(target=source)
+        entries = ["ffconcat version 1.0"]
+        for start, end in segments:
+            entries.extend([f"file '{local_source.name}'", f"inpoint {start + start_time:.6f}"])
+            if end < duration:
+                # The concat demuxer's outpoint uses DTS, while inpoint uses PTS.
+                entries.append(f"outpoint {decode_times[end] + start_time:.6f}")
+            entries.append(f"duration {end - start:.6f}")
+        playlist.write_text(data="\n".join(entries) + "\n")
+        command = [
+            "ffmpeg", "-hide_banner", "-nostdin", "-loglevel", "warning", "-stats",
+            "-y" if overwrite else "-n", "-f", "concat", "-safe", "0", "-auto_convert", "0", "-i", str(playlist),
+            "-map", "0:v:0", "-map", "0:a?", "-c", "copy",
+        ]
+        if output.suffix.lower() in {".mp4", ".mov"}:
+            command.extend(["-movflags", "+faststart"])
+        command.append(str(output))
+        subprocess.run(args=command, check=True)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+    arguments = parse_args()
+    try:
+        for executable in ("ffmpeg", "ffprobe"):
+            if shutil.which(cmd=executable) is None:
+                raise ValueError(f"{executable} must be installed and available on PATH")
+        source = arguments.input.resolve(strict=True)
+        output = (arguments.output or source.with_name(f"{source.stem}.short{source.suffix or '.mkv'}")).resolve()
+        if source == output:
+            raise ValueError("Input and output must be different files")
+        if output.exists() and source.samefile(output):
+            raise ValueError("Input and output must be different files")
+        if not arguments.dry_run and output.exists() and not arguments.overwrite:
+            raise ValueError(f"Output exists: {output}; use --overwrite to replace it")
+        duration, start_time = probe_video(path=source)
+        freezes = detect_freezes(path=source, duration=duration, min_still=arguments.min_still, noise=arguments.noise)
+        keyframes = probe_keyframes(path=source, start_time=start_time)
+        segments = plan_segments(duration=duration, freezes=freezes, keep_still=arguments.keep_still,
+                                 keyframes=[time for time, _ in keyframes])
+        retained = sum(end - start for start, end in segments)
+        logging.info(f"{duration:.2f}s → {retained:.2f}s; removing {duration - retained:.2f}s ({(1 - retained / duration):.1%})")
+        if arguments.dry_run:
+            return
+        copy_video(source=source, output=output, segments=segments, duration=duration,
+                   start_time=start_time, keyframes=keyframes, overwrite=arguments.overwrite)
+        logging.info(f"Saved {output}")
+    except (OSError, ValueError, subprocess.CalledProcessError) as error:
+        logging.error(str(error))
+        if isinstance(error, subprocess.CalledProcessError) and error.stderr:
+            logging.error(error.stderr)
+        raise SystemExit(1) from error
+
+
+if __name__ == "__main__":
+    main()

ffmpeg_remove_still_test.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# dependencies = ["pytest"]
+# ///
+
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+import ffmpeg_remove_still as script
+
+
+@pytest.mark.parametrize("log, expected", [
+    ("", []),
+    ("lavfi.freezedetect.freeze_start: 0", [(0.0, 10.0)]),
+    ("lavfi.freezedetect.freeze_start: 2\nlavfi.freezedetect.freeze_end: 5", [(2.0, 5.0)]),
+    ("lavfi.freezedetect.freeze_start: 2\nlavfi.freezedetect.freeze_end: 5\n"
+     "lavfi.freezedetect.freeze_start: 8", [(2.0, 5.0), (8.0, 10.0)]),
+])
+def test_parse_freezes(log: str, expected: list[tuple[float, float]]) -> None:
+    assert script.parse_freezes(log=log, duration=10.0) == expected
+
+
+@pytest.mark.parametrize("freezes, keep_still, keyframes, expected", [
+    ([], 1.0, [0.0], [(0.0, 10.0)]),
+    ([(0.0, 10.0)], 1.0, [0.0, 0.5, 9.5], [(0.0, 0.5), (9.5, 10.0)]),
+    ([(2.0, 6.0)], 2.0, [0.0, 3.0, 5.0], [(0.0, 3.0), (5.0, 10.0)]),
+    ([(2.0, 3.0)], 2.0, [0.0, 2.5], [(0.0, 10.0)]),
+    ([(0.0, 4.0), (6.0, 10.0)], 1.0, [0.0, 0.5, 3.5, 6.5, 9.5],
+     [(0.0, 0.5), (3.5, 6.5), (9.5, 10.0)]),
+    ([(2.0, 6.0)], 1.0, [0.0, 3.0, 5.0, 7.0], [(0.0, 3.0), (5.0, 10.0)]),
+    ([(2.0, 6.0)], 1.0, [0.0, 4.0, 8.0], [(0.0, 10.0)]),
+    ([(2.0, 6.0)], 1.0, [0.0, 8.0], [(0.0, 10.0)]),
+])
+def test_plan_segments(freezes: list[tuple[float, float]], keep_still: float,
+                       keyframes: list[float], expected: list[tuple[float, float]]) -> None:
+    assert script.plan_segments(duration=10.0, freezes=freezes, keep_still=keep_still, keyframes=keyframes) == expected
+
+
+@pytest.mark.skipif(not shutil.which(cmd="ffmpeg") or not shutil.which(cmd="ffprobe"), reason="Requires FFmpeg")
+@pytest.mark.parametrize("has_audio, keep_still, keyframe_interval, expected_duration", [
+    (False, 0.5, 5, 4.0), (True, 0.5, 5, 4.0), (True, 1.5, 5, 6.0),
+    (True, 0.5, 100, 8.0),
+])
+def test_copies_packets_without_encoding(tmp_path: Path, has_audio: bool, keep_still: float,
+                                        keyframe_interval: int, expected_duration: float) -> None:
+    source = tmp_path / "demo's recording.mp4"
+    output = tmp_path / "short.mp4"
+    command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-nostdin"]
+    for pattern in ["testsrc2=size=64x64:rate=10:duration=1", "color=red:size=64x64:rate=10:duration=3",
+                    "testsrc2=size=64x64:rate=10:duration=1", "color=blue:size=64x64:rate=10:duration=3"]:
+        command.extend(["-f", "lavfi", "-i", pattern])
+    if has_audio:
+        command.extend(["-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=8"])
+    command.extend(["-filter_complex", "[0:v][1:v][2:v][3:v]concat=n=4:v=1:a=0[v]", "-map", "[v]"])
+    if has_audio:
+        command.extend(["-map", "4:a", "-c:a", "aac"])
+    command.extend(["-c:v", "libx264", "-g", str(keyframe_interval), "-keyint_min", str(keyframe_interval),
+                    "-sc_threshold", "0", "-bf", "2", "-flags", "+cgop", str(source)])
+    subprocess.run(args=command, check=True, capture_output=True)
+    copied = subprocess.run(
+        args=[sys.executable, str(Path(script.__file__).resolve()), str(source), "--output", str(output),
+              "--min-still", "2", "--keep-still", str(keep_still)],
+        check=True, capture_output=True, text=True,
+    )
+    assert "non-monoton" not in copied.stderr.lower()
+    result = subprocess.run(
+        args=["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(output)],
+        check=True, capture_output=True, text=True,
+    )
+    metadata = json.loads(result.stdout)
+    video = next(stream for stream in metadata["streams"] if stream["codec_type"] == "video")
+    assert float(video["duration"]) == pytest.approx(expected_duration, abs=0.15)
+    audio = [stream for stream in metadata["streams"] if stream["codec_type"] == "audio"]
+    assert bool(audio) == has_audio
+    if has_audio:
+        assert float(audio[0]["duration"]) == pytest.approx(float(video["duration"]), abs=0.1)
+    packet_hashes = []
+    for path in (source, output):
+        packets = subprocess.run(
+            args=["ffprobe", "-v", "error", "-show_packets", "-show_data_hash", "sha256",
+                  "-show_entries", "packet=stream_index,data_hash", "-of", "json", str(path)],
+            check=True, capture_output=True, text=True,
+        )
+        packet_hashes.append({(packet["stream_index"], packet["data_hash"])
+                              for packet in json.loads(packets.stdout)["packets"]})
+    assert packet_hashes[1] <= packet_hashes[0]
+    decoded = subprocess.run(
+        args=["ffmpeg", "-v", "error", "-i", str(output), "-f", "null", "-"],
+        check=True, capture_output=True, text=True,
+    )
+    assert decoded.stderr == ""
+    original = source.read_bytes()
+    refused = subprocess.run(
+        args=[sys.executable, str(Path(script.__file__).resolve()), str(source), "--output", str(source), "--overwrite"],
+        capture_output=True, text=True,
+    )
+    assert refused.returncode == 1
+    assert source.read_bytes() == original
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))

snot/snot.html

@@ -162,7 +162,12 @@
 <div x-data="app()" class="gapped" @keydown.window="onKeydown">
     <form @submit.prevent="search" class="search-form">
         <div class="row">
-            <input class="input" type="text" x-ref="query" x-model="form.query" placeholder="Search videos" style="width: 300px" />
+            <input class="input" type="text" x-ref="query" x-model="form.query" list="search-history" placeholder="Search videos" style="width: 300px" />
+            <datalist id="search-history">
+                <template x-for="query in searchHistory" :key="query">
+                    <option :value="query"></option>
+                </template>
+            </datalist>
             <input class="button" type="submit" />
         </div>
     </form>
@@ -225,8 +230,38 @@
 
 <script defer src="./alpine@3.x.x.cdn.min.js"></script>
 <script>
+    const searchHistory = {
+        /** @returns {string[]} The most recent searches. */
+        get history() {
+            try {
+                const history = JSON.parse(localStorage.getItem("snot.searchHistory") || "[]");
+                return Array.isArray(history) ? history.filter((query) => typeof query === "string" && query.trim()).slice(0, 40) : [];
+            } catch (error) {
+                console.warn("Could not load search history", error);
+                return [];
+            }
+        },
+        /**
+         * @param {string} value
+         * @returns {string[]} The updated searches.
+         */
+        add(value) {
+            const query = value.trim();
+            if (!query) return this.history;
+            const history = [query, ...this.history.filter((item) => item !== query)].slice(0, 40);
+            try {
+                localStorage.setItem("snot.searchHistory", JSON.stringify(history));
+            } catch (error) {
+                console.warn("Could not save search history", error);
+            }
+            return history;
+        },
+    };
+
+    /** @returns {object} The Alpine application state and methods. */
     const app = () => ({
         results: [],
+        searchHistory: [],
         errorMessage: "",
         form: {
             query: "",
@@ -236,9 +271,17 @@
             minSizeMB: 0,
         },
         sort: { by: null, dir: "asc" },
+        /**
+         * @param {string} url
+         * @returns {string} The Alfred trigger URL.
+         */
         buildAlfredURL(url) {
             return `alfred://runtrigger/dev.abdus.alfred.media/video/?argument=${encodeURIComponent(url)}`;
         },
+        /**
+         * @param {string} field The result field to sort by.
+         * @returns {void}
+         */
         toggleSort(field) {
             if (this.sort.by === field) {
                 this.sort.dir = this.sort.dir === "asc" ? "desc" : "asc";
@@ -247,6 +290,10 @@
                 this.sort.dir = "asc";
             }
         },
+        /**
+         * @param {KeyboardEvent} e
+         * @returns {Promise<void>}
+         */
         async onKeydown(e) {
             if (e.target.tagName === "INPUT" || e.target.tagName === "TEXTAREA") {
                 return;
@@ -256,6 +303,7 @@
                 this.$refs.filterInput.focus();
             }
         },
+        /** @returns {object[]} The filtered and sorted results. */
         get visibleResults() {
             const isEmpty = !this.filter.query && this.filter.minSizeMB === 0;
             let rows = isEmpty
@@ -284,6 +332,10 @@
             { key: "file_path", value: (it) => it.file_path, label: "File Path" },
             { key: "created_at", value: (it) => (it.created_at ? it.created_at.slice(0, 10) : ""), label: "Created At", sortable: true },
         ],
+        /**
+         * @param {string} actor
+         * @returns {string} The actor filter token.
+         */
         actorToken(actor) {
             const escaped = actor.replaceAll('"', '\\"');
             if (escaped.includes(" ")) {
@@ -291,11 +343,20 @@
             }
             return `actor:${escaped}`;
         },
+        /**
+         * @param {string} actor
+         * @returns {boolean} Whether the actor is already in the query.
+         */
         hasActorFilter(actor) {
             const escaped = actor.replace(/[.*+?^${}()|[\]\\]/g, "\\$&").replaceAll('"', '\\"');
             const pattern = new RegExp(`(^|\\s|\\|)actor:(?:"${escaped}"|${escaped})(?=\\s|\\||$)`, "i");
             return pattern.test(this.form.query);
         },
+        /**
+         * @param {string} actor
+         * @param {MouseEvent} event
+         * @returns {void}
+         */
         addActorFilter(actor, event) {
             if (!actor) {
                 return;
@@ -315,12 +376,15 @@
             this.form.query = this.form.query ? `${this.form.query} ${token}` : token;
             this.search();
         },
+        /** @returns {Promise<void>} */
         async init() {
+            this.searchHistory = searchHistory.history;
             if (window.ENV?.query) {
                 this.form.query = window.ENV?.query ?? "";
                 await this.search();
             }
         },
+        /** @returns {string[] | null} Playable URLs, if available. */
         get playableURLs() {
             const playable = this.visibleResults.some((item) => item.file_path.startsWith("/mnt/box/files"));
             if (!playable) {
@@ -332,6 +396,7 @@
                 .sort(() => Math.random() - 0.5);
             return urls;
         },
+        /** @returns {string | null} The playlist trigger URL, if available. */
         get playlistLink() {
             const urls = this.playableURLs;
             if (!urls) {
@@ -341,6 +406,7 @@
             return `alfred://runtrigger/dev.abdus.alfred.media/mpv/?argument=${playlistArg}`;
         },
 
+        /** @returns {string | null} The grid trigger URL, if available. */
         get gridLink() {
             const urls = this.playableURLs;
             if (!urls) {
@@ -350,8 +416,10 @@
             return `alfred://runtrigger/dev.abdus.alfred.media/gridvid/?argument=${playlistArg}`;
         },
 
+        /** @returns {Promise<void>} */
         async search() {
             this.errorMessage = "";
+            const query = this.form.query.trim();
             const response = await fetch("/search", {
                 method: "POST",
                 headers: { "Content-Type": "application/json" },
@@ -363,6 +431,9 @@
                 return;
             }
             this.results = await response.json();
+            if (query) {
+                this.searchHistory = searchHistory.add(query);
+            }
             window.app.setTitle(`s·m·u·t - ${this.results.length} results for "${this.form.query}"`);
         },
     });

snot/snot.py

@@ -31,9 +31,10 @@ import click
 import httpx
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
+logging.getLogger("httpx").setLevel(logging.WARNING)
 
 HandlerFunc = Callable[["Request"], "Response"]
-client = httpx.Client(base_url="http://klein:18543", timeout=30)
+client = httpx.Client(base_url="http://klein.talpa-carat.ts.net:18543", timeout=30)
 
 
 @dataclasses.dataclass
@@ -747,6 +748,19 @@ def scan(paths: list[Path]):
     scan_videos(video_paths=list(paths))
 
 
+@cli.command()
+def purge() -> None:
+    """Remove box entries not seen in the last 24 hours."""
+    cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=24)
+    response = client.delete(
+        "/videos",
+        params={"category": "eq.box", "last_seen_at": f"lt.{cutoff.isoformat()}"},
+        headers={"Prefer": "return=representation"},
+    )
+    response.raise_for_status()
+    logging.info(f"Purged {len(response.json())} box entries")
+
+
 @cli.command("mark-deletion")
 @click.argument("paths", nargs=-1, type=Path)
 def mark_deletion(paths: list[Path]):