melbar/aitrailer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update cutter report

Browse Source
This commit is contained in:
Melbar
2026-05-18 08:48:26 +02:00
parent 68ec775916
commit fa40821319
4 changed files with 172 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli.py
+165
View File
@@ -384,6 +384,7 @@ def _normalize_cached_results(beats: list, results: list, cfg) -> list:
fps = _scene_fps_light(scene, cfg)
adjusted_in_s = result.in_point_s
phase_changed = False
scene_changed = int(scene["scene_id"]) != result.scene_id
starts_before_scene = result.in_point_s < float(scene["start_s"])
if scene_changed or starts_before_scene or result.duration_s <= 0.12:
@@ -392,6 +393,25 @@ def _normalize_cached_results(beats: list, results: list, cfg) -> list:
scene = _scene_for_time_light(scenes, adjusted_in_s, cfg) or scene
fps = _scene_fps_light(scene, cfg)
should_phase_probe = (
scene_changed
or starts_before_scene
or not result.is_confirmed
or result.match_score < cfg.cv.deep_scan.match_threshold
)
phase_score = result.match_score
if should_phase_probe:
probe = _phase_probe_segment_in_scene(beat, scene, adjusted_in_s, cfg)
if probe is not None:
probed_in_s, probed_score = probe
max_shift_s = max(0.12, min(0.75, beat.duration_s * 0.35))
if abs(probed_in_s - adjusted_in_s) <= max_shift_s:
adjusted_in_s = probed_in_s
phase_changed = True
phase_score = max(float(result.match_score), float(probed_score))
scene = _scene_for_time_light(scenes, adjusted_in_s, cfg) or scene
fps = _scene_fps_light(scene, cfg)
matchable_duration_s = beat.duration_s
try:
from src.cv.global_scan import estimate_matchable_reference_duration
@@ -414,6 +434,7 @@ def _normalize_cached_results(beats: list, results: list, cfg) -> list:
if (
scene_changed
or starts_before_scene
or phase_changed
or result.duration_s <= 0.12
or result.out_point_s > adjusted_in_s + max_duration_s + (1.0 / fps)
):
@@ -423,6 +444,8 @@ def _normalize_cached_results(beats: list, results: list, cfg) -> list:
in_point_s=adjusted_in_s,
out_point_s=adjusted_in_s + max_duration_s,
in_point_frame=int(adjusted_in_s * fps),
match_score=phase_score,
is_confirmed=phase_score >= cfg.cv.deep_scan.match_threshold,
)
coverage = (
@@ -2404,6 +2427,141 @@ def cmd_run(args: argparse.Namespace, cfg) -> None:
cmd_export(args, cfg)
def cmd_preview(args: argparse.Namespace, cfg) -> None:
"""Assemble a rough preview video from cached source matches, with original audio."""
import subprocess
log = logging.getLogger(__name__)
results_path = _results_cache_path(cfg)
if not results_path.exists():
log.error("No match_results.json — run 'match' first.")
return
data = sorted(
json.loads(results_path.read_text(encoding="utf-8")),
key=lambda r: r["beat_id"],
)
beats_path = cfg.paths.cache_dir / "trailer_beats.json"
beats_by_id: dict = {}
if beats_path.exists():
for b in json.loads(beats_path.read_text(encoding="utf-8")):
beats_by_id[int(b["beat_id"])] = b
clip_width = 1280
fps = 25
out_dir = cfg.paths.output_dir / "preview_clips"
out_dir.mkdir(parents=True, exist_ok=True)
preview_out = cfg.paths.output_dir / "preview.mp4"
def _run(cmd: list, timeout: int = 120) -> bool:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
if r.returncode != 0:
log.debug("ffmpeg stderr: %s", r.stderr[-600:])
return r.returncode == 0
def extract_with_audio(src: Path, start_s: float, duration_s: float, out: Path) -> bool:
preroll = 2.0 if start_s >= 2.0 else 0.0
input_seek = max(0.0, start_s - preroll)
accurate_seek = start_s - input_seek
return _run([
"ffmpeg", "-y", "-loglevel", "error",
"-ss", f"{input_seek:.3f}", "-i", str(src),
"-ss", f"{accurate_seek:.3f}", "-t", f"{max(0.04, duration_s):.3f}",
"-map", "0:v:0", "-map", "0:a:0",
"-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
"-vf", f"fps={fps},scale={clip_width}:-2,setsar=1,setpts=PTS-STARTPTS",
"-c:a", "aac", "-ar", "48000", "-ac", "2",
"-pix_fmt", "yuv420p", "-movflags", "+faststart", str(out),
])
def black_silence(duration_s: float, out: Path) -> bool:
return _run([
"ffmpeg", "-y", "-loglevel", "error",
"-f", "lavfi", "-i", f"color=black:s={clip_width}x720:r={fps}",
"-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo",
"-t", f"{max(0.5, duration_s):.3f}",
"-c:v", "libx264", "-preset", "veryfast", "-crf", "23",
"-c:a", "aac", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(out),
])
def concat_clips(parts: list[Path], out: Path) -> bool:
lst = out.with_suffix(".txt")
lst.write_text(
"\n".join(f"file '{p.resolve().as_posix()}'" for p in parts),
encoding="utf-8",
)
ok = _run([
"ffmpeg", "-y", "-loglevel", "error",
"-f", "concat", "-safe", "0", "-i", str(lst),
"-c", "copy", str(out),
], timeout=300)
lst.unlink(missing_ok=True)
return ok
beat_clips: list[Path] = []
for rec in data:
bid = int(rec["beat_id"])
segs = rec.get("segments", [])
src = Path(rec["source_path"]) if rec.get("source_path") else None
clip_out = out_dir / f"beat_{bid:02d}.mp4"
if src is None or not src.exists():
beat = beats_by_id.get(bid, {})
dur = max(0.5, float(beat.get("end_s", 1)) - float(beat.get("start_s", 0)))
log.info("Beat %02d: NO MATCH — black/silence %.2fs", bid, dur)
if black_silence(dur, clip_out):
beat_clips.append(clip_out)
continue
if len(segs) >= 2:
parts: list[Path] = []
for idx, seg in enumerate(segs):
in_s = float(seg["in_point_s"])
dur = max(0.04, float(seg["out_point_s"]) - in_s)
seg_src = Path(seg["source_path"]) if seg.get("source_path") else src
part = out_dir / f"beat_{bid:02d}_seg{idx:02d}.mp4"
log.info("Beat %02d seg%d: scene=%s %.2fs–%.2fs", bid, idx, seg.get("scene_id"), in_s, in_s + dur)
if extract_with_audio(seg_src, in_s, dur, part):
parts.append(part)
if not parts:
log.warning("Beat %02d: no segments extracted", bid)
continue
if len(parts) == 1:
parts[0].rename(clip_out)
beat_clips.append(clip_out)
else:
if concat_clips(parts, clip_out):
beat_clips.append(clip_out)
for p in parts:
p.unlink(missing_ok=True)
else:
in_s = float(rec["in_point_s"])
beat = beats_by_id.get(bid, {})
beat_dur = float(beat["end_s"]) - float(beat["start_s"]) if beat else 0.0
source_dur = float(rec["out_point_s"]) - in_s
dur = max(0.04, beat_dur if beat_dur > 0.04 else source_dur)
log.info("Beat %02d: scene=%s %.2fs+%.2fs (trailer=%.2fs src=%.2fs)", bid, rec.get("scene_id"), in_s, dur, beat_dur, source_dur)
if extract_with_audio(src, in_s, dur, clip_out):
beat_clips.append(clip_out)
else:
log.warning("Beat %02d: extraction failed", bid)
if not beat_clips:
log.error("No clips extracted — aborting.")
return
log.info("Concatenating %d beat clips → %s", len(beat_clips), preview_out)
if concat_clips(beat_clips, preview_out):
size_mb = preview_out.stat().st_size / 1_048_576
log.info("Preview ready: %s (%.1f MB)", preview_out, size_mb)
print(f"\n Preview → {preview_out} ({size_mb:.1f} MB)")
else:
log.error("Final concat failed — per-beat clips are in %s", out_dir)
# ---------------------------------------------------------------------------
# Argument parser
# ---------------------------------------------------------------------------
@@ -2474,6 +2632,12 @@ def _build_parser() -> argparse.ArgumentParser:
p_run.add_argument("--beat", type=int,
help="Run match/report/export for only one cached beat")
# preview
sub.add_parser(
"preview",
help="Build output/preview.mp4 from cached matches — source clips with audio in beat order",
)
return parser
@@ -2498,6 +2662,7 @@ def main() -> None:
"report": cmd_report,
"export": cmd_export,
"run": cmd_run,
"preview": cmd_preview,
}
handler = dispatch[args.command]