|
|
import os |
|
|
import tempfile |
|
|
import subprocess |
|
|
import json |
|
|
import re |
|
|
from typing import List, Dict, Optional, Tuple, Generator |
|
|
import gradio as gr |
|
|
|
|
|
try: |
|
|
import spaces |
|
|
HAS_SPACES = True |
|
|
except ImportError: |
|
|
HAS_SPACES = False |
|
|
|
|
|
import torch |
|
|
import numpy as np |
|
|
|
|
|
|
|
|
MODEL_PATH = "microsoft/VibeVoice-ASR" |
|
|
model = None |
|
|
processor = None |
|
|
|
|
|
|
|
|
def get_model(): |
|
|
global model, processor |
|
|
if model is None: |
|
|
from vibevoice.modular.modeling_vibevoice_asr import VibeVoiceASRForConditionalGeneration |
|
|
from vibevoice.processor.vibevoice_asr_processor import VibeVoiceASRProcessor |
|
|
|
|
|
processor = VibeVoiceASRProcessor.from_pretrained(MODEL_PATH) |
|
|
model = VibeVoiceASRForConditionalGeneration.from_pretrained( |
|
|
MODEL_PATH, |
|
|
dtype=torch.bfloat16, |
|
|
device_map="auto", |
|
|
trust_remote_code=True |
|
|
) |
|
|
model.eval() |
|
|
return model, processor |
|
|
|
|
|
|
|
|
def transcribe_audio_inner(audio_path: str) -> List[Dict]: |
|
|
model, processor = get_model() |
|
|
device = next(model.parameters()).device |
|
|
|
|
|
inputs = processor( |
|
|
audio=audio_path, |
|
|
sampling_rate=16000, |
|
|
return_tensors="pt", |
|
|
add_generation_prompt=True, |
|
|
) |
|
|
|
|
|
inputs = {k: v.to(device) if isinstance(v, torch.Tensor) else v for k, v in inputs.items()} |
|
|
|
|
|
with torch.no_grad(): |
|
|
output_ids = model.generate( |
|
|
**inputs, |
|
|
max_new_tokens=8192, |
|
|
temperature=None, |
|
|
do_sample=False, |
|
|
num_beams=1, |
|
|
pad_token_id=processor.pad_id, |
|
|
eos_token_id=processor.tokenizer.eos_token_id, |
|
|
) |
|
|
|
|
|
generated_ids = output_ids[0, inputs['input_ids'].shape[1]:] |
|
|
generated_text = processor.decode(generated_ids, skip_special_tokens=True) |
|
|
|
|
|
try: |
|
|
segments = processor.post_process_transcription(generated_text) |
|
|
except Exception: |
|
|
segments = parse_raw_transcript(generated_text) |
|
|
|
|
|
return segments |
|
|
|
|
|
|
|
|
def parse_raw_transcript(text: str) -> List[Dict]: |
|
|
segments = [] |
|
|
pattern = r'\[(\d+\.?\d*)\s*-\s*(\d+\.?\d*)\]\s*(?:\[([^\]]*)\])?\s*(.+?)(?=\[\d+\.?\d*\s*-|\Z)' |
|
|
matches = re.findall(pattern, text, re.DOTALL) |
|
|
|
|
|
for match in matches: |
|
|
start, end, speaker, content = match |
|
|
segments.append({ |
|
|
'start': float(start), |
|
|
'end': float(end), |
|
|
'speaker': speaker.strip() if speaker else 'Speaker', |
|
|
'text': content.strip() |
|
|
}) |
|
|
|
|
|
if not segments and text.strip(): |
|
|
sentences = re.split(r'(?<=[.!?])\s+', text.strip()) |
|
|
duration_per_sentence = 3.0 |
|
|
for i, sentence in enumerate(sentences): |
|
|
if sentence.strip(): |
|
|
segments.append({ |
|
|
'start': i * duration_per_sentence, |
|
|
'end': (i + 1) * duration_per_sentence, |
|
|
'speaker': 'Speaker', |
|
|
'text': sentence.strip() |
|
|
}) |
|
|
|
|
|
return segments |
|
|
|
|
|
|
|
|
if HAS_SPACES: |
|
|
@spaces.GPU(duration=120) |
|
|
def transcribe_with_gpu(audio_path: str) -> List[Dict]: |
|
|
return transcribe_audio_inner(audio_path) |
|
|
else: |
|
|
def transcribe_with_gpu(audio_path: str) -> List[Dict]: |
|
|
return transcribe_audio_inner(audio_path) |
|
|
|
|
|
|
|
|
def extract_audio(video_path: str) -> str: |
|
|
audio_path = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name |
|
|
cmd = [ |
|
|
"ffmpeg", "-y", "-i", video_path, |
|
|
"-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", |
|
|
audio_path |
|
|
] |
|
|
subprocess.run(cmd, capture_output=True, check=True) |
|
|
return audio_path |
|
|
|
|
|
|
|
|
def get_video_duration(video_path: str) -> float: |
|
|
cmd = [ |
|
|
"ffprobe", "-v", "error", |
|
|
"-show_entries", "format=duration", |
|
|
"-of", "json", video_path |
|
|
] |
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True) |
|
|
data = json.loads(result.stdout) |
|
|
return float(data["format"]["duration"]) |
|
|
|
|
|
|
|
|
def segments_to_transcript(segments: List[Dict]) -> str: |
|
|
lines = [] |
|
|
for seg in segments: |
|
|
start = seg['start'] |
|
|
end = seg['end'] |
|
|
text = seg['text'] |
|
|
lines.append(f"[{start:.2f}-{end:.2f}] {text}") |
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
def parse_transcript_to_segments(transcript: str) -> List[Dict]: |
|
|
segments = [] |
|
|
pattern = r'\[(\d+\.?\d*)-(\d+\.?\d*)\]\s*(.+)' |
|
|
|
|
|
for line in transcript.strip().split("\n"): |
|
|
line = line.strip() |
|
|
if not line: |
|
|
continue |
|
|
match = re.match(pattern, line) |
|
|
if match: |
|
|
start, end, text = match.groups() |
|
|
segments.append({ |
|
|
'start': float(start), |
|
|
'end': float(end), |
|
|
'text': text.strip() |
|
|
}) |
|
|
|
|
|
return segments |
|
|
|
|
|
|
|
|
def find_current_segment_index(segments: List[Dict], current_time: float) -> int: |
|
|
for i, seg in enumerate(segments): |
|
|
if seg['start'] <= current_time < seg['end']: |
|
|
return i |
|
|
return -1 |
|
|
|
|
|
|
|
|
def format_transcript_with_highlight(segments: List[Dict], current_index: int) -> str: |
|
|
lines = [] |
|
|
for i, seg in enumerate(segments): |
|
|
start = seg['start'] |
|
|
end = seg['end'] |
|
|
text = seg['text'] |
|
|
line = f"[{start:.2f}-{end:.2f}] {text}" |
|
|
if i == current_index: |
|
|
line = line.upper() |
|
|
lines.append(line) |
|
|
return "\n".join(lines) |
|
|
|
|
|
|
|
|
def cut_video_segments(video_path: str, segments_to_keep: List[Dict]) -> Optional[str]: |
|
|
if not segments_to_keep: |
|
|
return None |
|
|
|
|
|
segments_to_keep = sorted(segments_to_keep, key=lambda x: x['start']) |
|
|
|
|
|
temp_dir = tempfile.mkdtemp() |
|
|
clip_files = [] |
|
|
|
|
|
for i, seg in enumerate(segments_to_keep): |
|
|
clip_path = os.path.join(temp_dir, f"clip_{i:04d}.mp4") |
|
|
cmd = [ |
|
|
"ffmpeg", "-y", "-i", video_path, |
|
|
"-ss", str(seg['start']), |
|
|
"-to", str(seg['end']), |
|
|
"-c:v", "libx264", "-c:a", "aac", |
|
|
"-avoid_negative_ts", "make_zero", |
|
|
clip_path |
|
|
] |
|
|
subprocess.run(cmd, capture_output=True, check=True) |
|
|
clip_files.append(clip_path) |
|
|
|
|
|
list_file = os.path.join(temp_dir, "list.txt") |
|
|
with open(list_file, "w") as f: |
|
|
for clip in clip_files: |
|
|
f.write(f"file '{clip}'\n") |
|
|
|
|
|
output_path = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name |
|
|
cmd = [ |
|
|
"ffmpeg", "-y", "-f", "concat", "-safe", "0", |
|
|
"-i", list_file, |
|
|
"-c", "copy", |
|
|
output_path |
|
|
] |
|
|
subprocess.run(cmd, capture_output=True, check=True) |
|
|
|
|
|
for clip in clip_files: |
|
|
os.remove(clip) |
|
|
os.remove(list_file) |
|
|
os.rmdir(temp_dir) |
|
|
|
|
|
return output_path |
|
|
|
|
|
|
|
|
def process_upload(video_file): |
|
|
if video_file is None: |
|
|
return None, "", [], "Please upload a video file." |
|
|
|
|
|
video_path = video_file |
|
|
return video_path, "", [], "Video uploaded. Click 'Transcribe' to start transcription." |
|
|
|
|
|
|
|
|
def run_transcription(video_path, progress=gr.Progress()): |
|
|
if video_path is None: |
|
|
return "", [], "No video uploaded." |
|
|
|
|
|
progress(0.1, desc="Extracting audio...") |
|
|
|
|
|
try: |
|
|
audio_path = extract_audio(video_path) |
|
|
except Exception as e: |
|
|
return "", [], f"Error extracting audio: {str(e)}" |
|
|
|
|
|
progress(0.3, desc="Running transcription (this may take a while)...") |
|
|
|
|
|
try: |
|
|
segments = transcribe_with_gpu(audio_path) |
|
|
except Exception as e: |
|
|
return "", [], f"Error during transcription: {str(e)}" |
|
|
finally: |
|
|
if os.path.exists(audio_path): |
|
|
os.remove(audio_path) |
|
|
|
|
|
progress(0.9, desc="Formatting transcript...") |
|
|
|
|
|
transcript = segments_to_transcript(segments) |
|
|
|
|
|
progress(1.0, desc="Done!") |
|
|
|
|
|
return transcript, segments, f"Transcription complete! {len(segments)} segments found." |
|
|
|
|
|
|
|
|
def update_highlight(video_path, original_segments, current_time): |
|
|
if not original_segments: |
|
|
return "" |
|
|
|
|
|
current_index = find_current_segment_index(original_segments, current_time) |
|
|
return format_transcript_with_highlight(original_segments, current_index) |
|
|
|
|
|
|
|
|
def apply_cuts(video_path, edited_transcript, original_segments): |
|
|
if video_path is None: |
|
|
return None, "No video to process." |
|
|
|
|
|
if not original_segments: |
|
|
return None, "No transcript available. Please transcribe first." |
|
|
|
|
|
edited_segments = parse_transcript_to_segments(edited_transcript) |
|
|
|
|
|
original_texts = {seg['text'].strip().lower() for seg in original_segments} |
|
|
edited_texts = {seg['text'].strip().lower() for seg in edited_segments} |
|
|
|
|
|
segments_to_keep = [] |
|
|
for seg in original_segments: |
|
|
if seg['text'].strip().lower() in edited_texts: |
|
|
segments_to_keep.append(seg) |
|
|
|
|
|
if not segments_to_keep: |
|
|
return None, "All segments were removed. Cannot create empty video." |
|
|
|
|
|
deleted_count = len(original_segments) - len(segments_to_keep) |
|
|
|
|
|
if deleted_count == 0: |
|
|
return video_path, "No changes detected. Original video returned." |
|
|
|
|
|
try: |
|
|
output_path = cut_video_segments(video_path, segments_to_keep) |
|
|
if output_path: |
|
|
return output_path, f"Video edited! Removed {deleted_count} segment(s)." |
|
|
else: |
|
|
return None, "Error creating edited video." |
|
|
except Exception as e: |
|
|
return None, f"Error cutting video: {str(e)}" |
|
|
|
|
|
|
|
|
JS_CODE = """ |
|
|
<script> |
|
|
(function() { |
|
|
let lastUpdate = 0; |
|
|
const updateInterval = 500; |
|
|
|
|
|
function findVideoElement() { |
|
|
const videos = document.querySelectorAll('video'); |
|
|
for (const video of videos) { |
|
|
if (video.src && !video.src.includes('blob:')) { |
|
|
return video; |
|
|
} |
|
|
} |
|
|
return videos[0]; |
|
|
} |
|
|
|
|
|
function setupVideoListener() { |
|
|
const video = findVideoElement(); |
|
|
if (!video) { |
|
|
setTimeout(setupVideoListener, 1000); |
|
|
return; |
|
|
} |
|
|
|
|
|
video.addEventListener('timeupdate', function() { |
|
|
const now = Date.now(); |
|
|
if (now - lastUpdate < updateInterval) return; |
|
|
lastUpdate = now; |
|
|
|
|
|
const timeInput = document.querySelector('#current-time-input input'); |
|
|
if (timeInput) { |
|
|
timeInput.value = video.currentTime.toFixed(2); |
|
|
timeInput.dispatchEvent(new Event('input', { bubbles: true })); |
|
|
} |
|
|
}); |
|
|
} |
|
|
|
|
|
if (document.readyState === 'loading') { |
|
|
document.addEventListener('DOMContentLoaded', setupVideoListener); |
|
|
} else { |
|
|
setupVideoListener(); |
|
|
} |
|
|
|
|
|
const observer = new MutationObserver(function(mutations) { |
|
|
setupVideoListener(); |
|
|
}); |
|
|
observer.observe(document.body, { childList: true, subtree: true }); |
|
|
})(); |
|
|
</script> |
|
|
""" |
|
|
|
|
|
|
|
|
with gr.Blocks(title="TextCut - Edit Videos by Editing Transcripts") as demo: |
|
|
gr.Markdown("# TextCut") |
|
|
gr.Markdown("Edit videos by simply editing their transcript. Upload a video, transcribe it, then delete lines to cut those parts from the video.") |
|
|
gr.HTML(JS_CODE) |
|
|
|
|
|
original_segments = gr.State([]) |
|
|
|
|
|
with gr.Row(): |
|
|
with gr.Column(scale=1): |
|
|
gr.Markdown("### Transcript") |
|
|
transcript_box = gr.Textbox( |
|
|
label="Transcript (delete lines to cut those parts)", |
|
|
lines=15, |
|
|
interactive=True, |
|
|
placeholder="Transcript will appear here after transcription..." |
|
|
) |
|
|
|
|
|
current_time = gr.Number( |
|
|
label="Current Video Time (seconds)", |
|
|
value=0, |
|
|
visible=True, |
|
|
elem_id="current-time-input" |
|
|
) |
|
|
|
|
|
highlight_btn = gr.Button("Update Highlight", size="sm") |
|
|
|
|
|
with gr.Column(scale=1): |
|
|
gr.Markdown("### Video") |
|
|
video_input = gr.Video( |
|
|
label="Upload Video", |
|
|
sources=["upload"], |
|
|
interactive=True |
|
|
) |
|
|
|
|
|
with gr.Row(): |
|
|
transcribe_btn = gr.Button("Transcribe", variant="primary") |
|
|
cut_btn = gr.Button("Apply Cuts", variant="secondary") |
|
|
|
|
|
status_text = gr.Textbox(label="Status", interactive=False, lines=2) |
|
|
|
|
|
gr.Markdown("### Edited Video Output") |
|
|
video_output = gr.Video(label="Edited Video") |
|
|
|
|
|
video_input.change( |
|
|
fn=process_upload, |
|
|
inputs=[video_input], |
|
|
outputs=[video_input, transcript_box, original_segments, status_text] |
|
|
) |
|
|
|
|
|
transcribe_btn.click( |
|
|
fn=run_transcription, |
|
|
inputs=[video_input], |
|
|
outputs=[transcript_box, original_segments, status_text] |
|
|
) |
|
|
|
|
|
highlight_btn.click( |
|
|
fn=update_highlight, |
|
|
inputs=[video_input, original_segments, current_time], |
|
|
outputs=[transcript_box] |
|
|
) |
|
|
|
|
|
current_time.change( |
|
|
fn=update_highlight, |
|
|
inputs=[video_input, original_segments, current_time], |
|
|
outputs=[transcript_box] |
|
|
) |
|
|
|
|
|
cut_btn.click( |
|
|
fn=apply_cuts, |
|
|
inputs=[video_input, transcript_box, original_segments], |
|
|
outputs=[video_output, status_text] |
|
|
) |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
demo.launch() |
|
|
|
