dillon_stuff/TalkEdit
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

UI improvements, moved file name and moved buttons left

Browse Source
This commit is contained in:
dillonj
2026-04-15 19:54:39 -06:00
parent 7479acd3ee
commit b7a795f986
9 changed files with 500 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
backend/services/video_editor.py
View File

@ -13,6 +13,78 @@ from typing import List
logger = logging.getLogger(__name__)
def _clamp_speed(speed: float) -> float:
return max(0.25, min(4.0, float(speed)))
def _build_atempo_chain(speed: float) -> str:
"""Build an FFmpeg atempo chain since each atempo node only supports 0.5..2.0."""
s = _clamp_speed(speed)
filters = []
while s > 2.0:
filters.append("atempo=2.0")
s /= 2.0
while s < 0.5:
filters.append("atempo=0.5")
s /= 0.5
filters.append(f"atempo={s:.6f}")
return ",".join(filters)
def _split_keep_segments_by_speed(
keep_segments: List[dict],
speed_ranges: List[dict] = None,
) -> List[dict]:
"""Split keep segments by speed ranges, attaching speed multiplier per piece."""
if not keep_segments:
return []
normalized_ranges = []
for r in speed_ranges or []:
start = float(r.get("start", 0.0))
end = float(r.get("end", 0.0))
if end <= start:
continue
normalized_ranges.append({
"start": start,
"end": end,
"speed": _clamp_speed(float(r.get("speed", 1.0))),
})
normalized_ranges.sort(key=lambda x: x["start"])
result = []
for keep in keep_segments:
k_start = float(keep["start"])
k_end = float(keep["end"])
if k_end <= k_start:
continue
cuts = {k_start, k_end}
for sr in normalized_ranges:
overlap_start = max(k_start, sr["start"])
overlap_end = min(k_end, sr["end"])
if overlap_end > overlap_start:
cuts.add(overlap_start)
cuts.add(overlap_end)
points = sorted(cuts)
for i in range(len(points) - 1):
seg_start = points[i]
seg_end = points[i + 1]
if seg_end - seg_start < 1e-6:
continue
speed = 1.0
for sr in normalized_ranges:
if seg_start >= sr["start"] and seg_end <= sr["end"]:
speed = sr["speed"]
break
result.append({"start": seg_start, "end": seg_end, "speed": speed})
return result
def _find_ffmpeg() -> str:
"""Locate ffmpeg binary."""
for cmd in ["ffmpeg", "ffmpeg.exe"]:
@ -113,6 +185,7 @@ def export_reencode(
format_hint: str = "mp4",
mute_ranges: List[dict] = None,
gain_ranges: List[dict] = None,
speed_ranges: List[dict] = None,
global_gain_db: float = 0.0,
) -> str:
"""
@ -150,8 +223,11 @@ def export_reencode(
has_audio_filters = bool(mute_ranges) or bool(gain_ranges) or abs(float(global_gain_db)) > 1e-6
# Handle filtered full-timeline audio case (mute/gain/global gain)
if has_audio_filters:
speed_segments = _split_keep_segments_by_speed(keep_segments, speed_ranges)
has_speed = any(abs(seg.get("speed", 1.0) - 1.0) > 1e-6 for seg in speed_segments)
# Handle filtered full-timeline audio case (mute/gain/global gain) when no speed warping is needed
if has_audio_filters and not has_speed:
audio_filter = build_audio_filter()
# Video filter - just scaling if needed
@ -189,18 +265,25 @@ def export_reencode(
resolution,
)
else:
# Original cutting logic
# Cutting logic with optional per-segment speed changes
if not keep_segments:
raise ValueError("No segments to export")
filter_parts = []
for i, seg in enumerate(keep_segments):
filter_parts.append(
f"[0:v]trim=start={seg['start']}:end={seg['end']},setpts=PTS-STARTPTS[v{i}];"
f"[0:a]atrim=start={seg['start']}:end={seg['end']},asetpts=PTS-STARTPTS[a{i}];"
)
segments_for_concat = speed_segments if speed_segments else _split_keep_segments_by_speed(keep_segments, None)
if not segments_for_concat:
raise ValueError("No segments to export")
n = len(keep_segments)
filter_parts = []
for i, seg in enumerate(segments_for_concat):
speed = _clamp_speed(seg.get("speed", 1.0))
v_chain = f"trim=start={seg['start']}:end={seg['end']},setpts=PTS-STARTPTS"
a_chain = f"atrim=start={seg['start']}:end={seg['end']},asetpts=PTS-STARTPTS"
if abs(speed - 1.0) > 1e-6:
v_chain += f",setpts=PTS/{speed:.6f}"
a_chain += f",{_build_atempo_chain(speed)}"
filter_parts.append(f"[0:v]{v_chain}[v{i}];[0:a]{a_chain}[a{i}];")
n = len(segments_for_concat)
concat_inputs = "".join(f"[v{i}][a{i}]" for i in range(n))
filter_parts.append(f"{concat_inputs}concat=n={n}:v=1:a=1[outv][outa]")
@ -228,7 +311,13 @@ def export_reencode(
output_path,
]
logger.info(f"Re-encoding {n} segments -> {output_path} ({resolution})")
logger.info(
"Re-encoding %s segments (speed-adjusted=%s) -> %s (%s)",
n,
has_speed,
output_path,
resolution,
)
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
@ -246,6 +335,7 @@ def export_reencode_with_subs(
format_hint: str = "mp4",
mute_ranges: List[dict] = None,
gain_ranges: List[dict] = None,
speed_ranges: List[dict] = None,
global_gain_db: float = 0.0,
) -> str:
"""
@ -284,8 +374,11 @@ def export_reencode_with_subs(
has_audio_filters = bool(mute_ranges) or bool(gain_ranges) or abs(float(global_gain_db)) > 1e-6
# Handle filtered full-timeline audio case (mute/gain/global gain)
if has_audio_filters:
speed_segments = _split_keep_segments_by_speed(keep_segments, speed_ranges)
has_speed = any(abs(seg.get("speed", 1.0) - 1.0) > 1e-6 for seg in speed_segments)
# Handle filtered full-timeline audio case (mute/gain/global gain) when no speed warping is needed
if has_audio_filters and not has_speed:
audio_filter = build_audio_filter()
# Video filter with subtitles
@ -322,18 +415,25 @@ def export_reencode_with_subs(
resolution,
)
else:
# Original cutting logic with subtitles
# Cutting logic with subtitles and optional speed changes
if not keep_segments:
raise ValueError("No segments to export")
filter_parts = []
for i, seg in enumerate(keep_segments):
filter_parts.append(
f"[0:v]trim=start={seg['start']}:end={seg['end']},setpts=PTS-STARTPTS[v{i}];"
f"[0:a]atrim=start={seg['start']}:end={seg['end']},asetpts=PTS-STARTPTS[a{i}];"
)
segments_for_concat = speed_segments if speed_segments else _split_keep_segments_by_speed(keep_segments, None)
if not segments_for_concat:
raise ValueError("No segments to export")
n = len(keep_segments)
filter_parts = []
for i, seg in enumerate(segments_for_concat):
speed = _clamp_speed(seg.get("speed", 1.0))
v_chain = f"trim=start={seg['start']}:end={seg['end']},setpts=PTS-STARTPTS"
a_chain = f"atrim=start={seg['start']}:end={seg['end']},asetpts=PTS-STARTPTS"
if abs(speed - 1.0) > 1e-6:
v_chain += f",setpts=PTS/{speed:.6f}"
a_chain += f",{_build_atempo_chain(speed)}"
filter_parts.append(f"[0:v]{v_chain}[v{i}];[0:a]{a_chain}[a{i}];")
n = len(segments_for_concat)
concat_inputs = "".join(f"[v{i}][a{i}]" for i in range(n))
filter_parts.append(f"{concat_inputs}concat=n={n}:v=1:a=1[outv][outa]")
@ -364,7 +464,13 @@ def export_reencode_with_subs(
output_path,
]
logger.info(f"Re-encoding {n} segments with subtitles -> {output_path} ({resolution})")
logger.info(
"Re-encoding %s segments with subtitles (speed-adjusted=%s) -> %s (%s)",
n,
has_speed,
output_path,
resolution,
)
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0: