feat: export clips, protect bookmarks and clean session storage

scripts/validate/check-clip.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""Validate a persisted clip before any source cleanup (ffmpeg/ffprobe required)."""
+import json
+import pathlib
+import subprocess
+import sys
+
+saved_path = pathlib.Path(sys.argv[1])
+saved = json.loads(saved_path.read_text())
+clip = saved_path.with_suffix('.mp4')
+session_path = saved_path.parent.parent / 'sessions' / str(saved['source_session']) / 'session.json'
+session = json.loads(session_path.read_text())
+seconds = lambda d: d['secs'] + d['nanos'] / 1e9
+start, end = seconds(saved['start']), seconds(saved['end'])
+assert end > start
+pieces = [s for s in session['segments'] if s['available'] and seconds(s['start']) < end and seconds(s['end']) > start]
+assert len(pieces) >= 2, 'Export must cross a segment boundary'
+
+def probe(path):
+    return json.loads(subprocess.check_output(['ffprobe', '-v', 'error', '-show_streams', '-show_packets', '-of', 'json', str(path)]))
+
+source = probe(session_path.parent / pieces[0]['path'])
+output = probe(clip)
+video = next(s for s in output['streams'] if s['codec_type'] == 'video')
+source_video = next(s for s in source['streams'] if s['codec_type'] == 'video')
+assert video['codec_name'] == source_video['codec_name']
+pts = sorted(float(p['pts_time']) for p in output['packets'] if p['stream_index'] == video['index'])
+fps = pieces[0]['frames_per_second']
+for a, b in zip(pts, pts[1:]):
+    assert abs(b-a-1/fps) < 2e-6, ('Video gap/duplicate', a, b, fps)
+assert abs(pts[0]) < 1/fps
+assert abs(pts[-1]+1/fps-(end-start)) < 1/fps+0.002, (pts[-1], start, end)
+# Metadata records independent AAC epochs, even if earlier files are cleaned up later.
+runs={}
+for segment in session['segments']:
+    if not segment['complete'] or not segment['available']:
+        continue
+    run=str(pathlib.Path(segment['path']).parent)
+    facts=probe(session_path.parent/segment['path'])
+    audios=[s for s in facts['streams'] if s['codec_type']=='audio']
+    totals=runs.setdefault(run,[0]*len(audios))
+    assert segment['audio_start_samples']==totals,(segment['path'],segment['audio_start_samples'],totals)
+    for i,stream in enumerate(audios):
+        totals[i]+=1024*len([p for p in facts['packets'] if p['stream_index']==stream['index']])
+
+for ordinal,stream in enumerate([s for s in output['streams'] if s['codec_type']=='audio']):
+    if stream['codec_type'] == 'audio':
+        timestamps = [float(p['pts_time']) for p in output['packets'] if p['stream_index'] == stream['index']]
+        step = 1024 / int(stream['sample_rate'])
+        assert timestamps
+        import math
+        source_ordinal=ordinal if saved['separate_audio'] or len([s for s in source['streams'] if s['codec_type']=='audio'])!=3 else 2
+        first=pieces[0]
+        run_base=seconds(first['start'])-first['frame_start']/fps
+        phase=run_base+first['audio_start_samples'][source_ordinal]/int(stream['sample_rate'])-start
+        expected_first=phase+max(0,math.ceil(-phase/step))*step
+        assert abs(timestamps[0]-expected_first)<1/int(stream['sample_rate'])+2e-6,('A/V phase',timestamps[0],expected_first)
+        for a, b in zip(timestamps, timestamps[1:]):
+            assert abs(b-a-step) < 2e-6, ('Audio gap/duplicate', a, b)
+source_audio = [s for s in source['streams'] if s['codec_type'] == 'audio']
+expected = 2 if saved['separate_audio'] and len(source_audio) == 3 else 1
+assert len([s for s in output['streams'] if s['codec_type'] == 'audio']) == expected
+result = subprocess.run(['ffmpeg', '-v', 'error', '-xerror', '-i', str(clip), '-f', 'null', '-'], capture_output=True)
+assert result.returncode == 0 and not result.stderr, result.stderr.decode()
+assert saved['clip']['size_bytes'] == clip.stat().st_size
+print(f'PASS clip {clip.name}: {video["codec_name"]}, {len(pts)} CFR frames, continuous audio, zero decode errors, {end-start:.3f}s')