HoaMP3 / app.py
HoaHuggingFace's picture
Update app.py
b1df6dd verified
Raw
History Blame Contribute Delete
24.3 kB
import importlib
import subprocess
import sys
import zipfile
import os
import time
import gradio as gr
from moviepy.editor import VideoFileClip, AudioFileClip
from pydub import AudioSegment
import shutil
import atexit
def install_package(name, pkg_type, upgrade=False, import_name=None, check_cmd=None):
if pkg_type == "pip":
import_name = import_name or name.replace("-", "_")
try:
importlib.import_module(import_name)
if upgrade:
print(f"⏫ Nâng cấp {name} ...")
subprocess.check_call([sys.executable, "-m", "pip", "install", "--upgrade", name])
else:
print(f"✅ Đã có sẵn: {name}")
except ImportError:
print(f"⏳ Đang cài đặt: {name} ...")
subprocess.check_call([sys.executable, "-m", "pip", "install", name])
install_package("gradio", "pip", upgrade=False)
import os
import time
import gradio as gr
from moviepy.editor import VideoFileClip, AudioFileClip
from pydub import AudioSegment
import shutil
import atexit
# ====== [WEBM] Các định dạng được coi là "audio-only" (xử lý như audio, bỏ qua video track) ======
WEBM_AS_AUDIO_EXTS = {'.webm'}
# Lưu danh sách temp directories để cleanup sau
temp_dirs = []
def cleanup_temp_files():
"""Xóa tất cả temporary files khi thoát"""
for temp_dir in temp_dirs:
try:
if os.path.exists(temp_dir):
shutil.rmtree(temp_dir)
except Exception as e:
print(f"Không thể xóa {temp_dir}: {e}")
atexit.register(cleanup_temp_files)
def webm_to_mp3(webm_path, output_dir=None):
"""
[WEBM] Convert file .webm sang .mp3 bằng FFmpeg (extract audio track).
Trả về đường dẫn file mp3 đã tạo.
"""
if output_dir is None:
output_dir = os.path.dirname(webm_path)
os.makedirs(output_dir, exist_ok=True)
file_basename = os.path.splitext(os.path.basename(webm_path))[0]
mp3_path = os.path.join(output_dir, f"{file_basename}.mp3")
cmd = [
'ffmpeg',
'-i', webm_path,
'-vn',
'-acodec', 'libmp3lame',
'-q:a', '2',
mp3_path,
'-y'
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
raise RuntimeError(
f"FFmpeg không thể convert WebM sang MP3:\n{result.stderr.decode(errors='replace')}"
)
return mp3_path
def get_duration(file_path):
"""Lấy duration từ file"""
if not file_path:
return 100
clip = None
try:
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext in WEBM_AS_AUDIO_EXTS:
result = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1', file_path],
capture_output=True, text=True
)
duration_str = result.stdout.strip()
return float(duration_str) if duration_str else 100
if file_ext in ['.mp4', '.mkv', '.avi', '.mov', '.flv']:
clip = VideoFileClip(file_path)
duration = clip.duration
clip.close()
else:
audio = AudioSegment.from_file(file_path)
duration = len(audio) / 1000.0
return duration
except:
return 100
finally:
if clip:
clip.close()
def process_upload(file_path):
"""Xử lý file upload"""
if not file_path:
return None, None, "", 999999, 0
clip = None
try:
file_ext = os.path.splitext(file_path)[1].lower()
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path) / (1024 * 1024)
yield None, None, "Đang xử lý file upload...", 999999, 0
if file_ext in WEBM_AS_AUDIO_EXTS:
temp_dir = os.path.join(os.getcwd(), "temp_webm_upload")
os.makedirs(temp_dir, exist_ok=True)
temp_dirs.append(temp_dir)
mp3_path = webm_to_mp3(file_path, output_dir=temp_dir)
duration = get_duration(file_path)
file_type = "Audio (WebM)"
video_file = None
info_text = f"""📁 **Tên file:** {file_name}
🎵 **Loại:** {file_type}
💾 **Kích thước:** {file_size:.2f} MB
⏱️ **Thời lượng:** {int(duration//60)}:{int(duration%60):02d} ({duration:.1f}s)
✅ **Đã extract audio sang MP3**"""
yield (
mp3_path,
video_file,
info_text,
duration,
0
)
return
if file_ext in ['.mp4', '.mkv', '.avi', '.mov', '.flv']:
clip = VideoFileClip(file_path)
duration = clip.duration
clip.close()
file_type = "Video"
video_file = file_path
elif file_ext in ['.mp3', '.wav', '.m4a', '.ogg', '.flac', '.aac']:
audio = AudioSegment.from_file(file_path)
duration = len(audio) / 1000.0
file_type = "Audio"
video_file = None
else:
yield None, None, f"⚠️ Định dạng file không được hỗ trợ: {file_ext}", 999999, 0
return
info_text = f"""📁 **Tên file:** {file_name}
🎬 **Loại:** {file_type}
💾 **Kích thước:** {file_size:.2f} MB
⏱️ **Thời lượng:** {int(duration//60)}:{int(duration%60):02d} ({duration:.1f}s)"""
yield (
file_path,
video_file,
info_text,
duration,
0
)
except Exception as e:
yield None, None, f"❌ Lỗi: {str(e)}", 999999, 0
finally:
if clip:
clip.close()
def convert_to_mp3(audio_file, video_file, bitrate, sample_rate):
"""
Convert audio/video sang MP3 với bitrate và sample rate tùy chọn (tối ưu cho transcribe).
"""
if not audio_file and not video_file:
yield None, "⚠️ Không có file để convert."
return
try:
input_file = video_file if video_file else audio_file
file_basename = os.path.splitext(os.path.basename(input_file))[0]
temp_dir = os.path.join(os.getcwd(), "temp_converted")
os.makedirs(temp_dir, exist_ok=True)
temp_dirs.append(temp_dir)
mp3_file = os.path.join(temp_dir, f"{file_basename}_{bitrate}_{sample_rate}hz.mp3")
yield None, "⏳ Đang convert sang MP3..."
cmd = [
'ffmpeg',
'-i', input_file,
'-vn',
'-acodec', 'libmp3lame',
'-b:a', bitrate,
'-ar', str(sample_rate),
mp3_file,
'-y'
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
yield None, f"❌ Lỗi FFmpeg: {result.stderr.decode(errors='replace')}"
return
file_size_kb = os.path.getsize(mp3_file) / 1024
status_msg = (
f"✅ Convert thành công!\n"
f"📁 File: {os.path.basename(mp3_file)}\n"
f"🎵 Bitrate: {bitrate} | Sample rate: {sample_rate} Hz\n"
f"💾 Kích thước: {file_size_kb:.1f} KB"
)
yield mp3_file, status_msg
except Exception as e:
import traceback
yield None, f"❌ Lỗi: {str(e)}\n{traceback.format_exc()}"
def clear_outputs():
return (
gr.update(value=None, visible=False), # source_mp3_output
gr.update(value=None, visible=False), # converted_mp3_output
gr.update(value=None, visible=False), # cut_zip_output
"", # cut_status
gr.update(visible=False) # clear_output_btn
)
def cut_file_by_duration(audio_file, minutes_per_chunk):
"""Cắt file mp3 thành các đoạn theo số phút"""
if not audio_file:
return [], "⚠️ Không có file audio để cắt."
try:
import datetime
chunk_secs = int(minutes_per_chunk) * 60
total_duration = get_duration(audio_file)
if total_duration <= chunk_secs:
return [], f"⚠️ File ngắn hơn {minutes_per_chunk} phút ({total_duration:.0f}s). Không cần cắt."
temp_dir = os.path.join(os.getcwd(), "temp_cut")
os.makedirs(temp_dir, exist_ok=True)
temp_dirs.append(temp_dir)
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
file_basename = os.path.splitext(os.path.basename(audio_file))[0]
output_files = []
part = 1
start = 0.0
while start < total_duration:
end = min(start + chunk_secs, total_duration)
out_path = os.path.join(temp_dir, f"{file_basename}_part{part:02d}_{ts}.mp3")
cmd = ['ffmpeg', '-ss', str(start), '-i', audio_file,
'-t', str(end - start), '-acodec', 'libmp3lame', '-q:a', '2', out_path, '-y']
result = subprocess.run(cmd, capture_output=True)
if result.returncode == 0:
output_files.append(out_path)
start += chunk_secs
part += 1
status = f"✅ Đã cắt thành {len(output_files)} đoạn × {minutes_per_chunk} phút"
return output_files, status
except Exception as e:
import traceback
return [], f"❌ Lỗi: {str(e)}\n{traceback.format_exc()}"
def cut_file_by_parts(audio_file, num_parts):
"""Cắt file mp3 thành N phần bằng nhau"""
if not audio_file:
return [], "⚠️ Không có file audio để cắt."
try:
import datetime
num_parts = int(num_parts)
total_duration = get_duration(audio_file)
chunk_secs = total_duration / num_parts
temp_dir = os.path.join(os.getcwd(), "temp_cut")
os.makedirs(temp_dir, exist_ok=True)
temp_dirs.append(temp_dir)
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
file_basename = os.path.splitext(os.path.basename(audio_file))[0]
output_files = []
for part in range(1, num_parts + 1):
start = (part - 1) * chunk_secs
out_path = os.path.join(temp_dir, f"{file_basename}_part{part:02d}of{num_parts}_{ts}.mp3")
cmd = ['ffmpeg', '-ss', str(start), '-i', audio_file,
'-t', str(chunk_secs), '-acodec', 'libmp3lame', '-q:a', '2', out_path, '-y']
result = subprocess.run(cmd, capture_output=True)
if result.returncode == 0:
output_files.append(out_path)
status = f"✅ Đã cắt thành {len(output_files)} phần (~{chunk_secs/60:.1f} phút/phần)"
return output_files, status
except Exception as e:
import traceback
return [], f"❌ Lỗi: {str(e)}\n{traceback.format_exc()}"
def do_cut_file(audio_file, cut_mode, minutes_val, parts_val):
"""Dispatcher: cắt theo phút hoặc theo phần, đóng gói zip"""
if not audio_file:
yield gr.update(value=None, visible=False), "⚠️ Không có file audio."
return
yield gr.update(visible=False), "⏳ Đang cắt file..."
if cut_mode == "Theo phút":
files, status = cut_file_by_duration(audio_file, minutes_val)
else:
files, status = cut_file_by_parts(audio_file, parts_val)
if not files:
yield gr.update(value=None, visible=False), status
return
import zipfile, datetime
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
file_basename = os.path.splitext(os.path.basename(audio_file))[0]
zip_dir = os.path.join(os.getcwd(), "temp_cut")
os.makedirs(zip_dir, exist_ok=True)
zip_path = os.path.join(zip_dir, f"{file_basename}_cut_{ts}.zip")
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
for f in files:
zf.write(f, os.path.basename(f))
zip_size_mb = os.path.getsize(zip_path) / (1024 * 1024)
status += f"\n📦 ZIP: {os.path.basename(zip_path)} ({zip_size_mb:.1f} MB)"
yield gr.update(value=zip_path, visible=True), status
# ====== GRADIO INTERFACE ======
css = """
#textbox_id textarea {
color: black !important;
font-size: 16px !important;
font-family: 'IBM Plex Sans', sans-serif !important;
}
#textbox_id placeholder::textarea {
color: black !important;
font-size: 16px !important;
font-family: 'IBM Plex Sans', sans-serif !important;
}
#method_dropdown .menu button {
color: *primary_50 !important;
font-size: 16px !important;
}
"""
theme = gr.themes.Default().set(
block_background_fill='*primary_50',
block_border_color='*button_primary_border_color',
block_label_text_color='*secondary_600',
block_info_text_color='*primary_700',
block_title_text_color='*primary_700',
body_text_size='*text_lg'
)
with gr.Blocks(theme=theme, css=css, title="Media Transcriber with Trimming") as demo:
gr.Markdown("# 🎤 Media Transcriber with Trimming")
gr.Markdown("Download từ YouTube (convert MP3) hoặc upload file, cắt (tùy chọn), và transcribe với faster-whisper")
with gr.Tab("Transcription"):
current_file = gr.State()
video_file_state = gr.State()
with gr.Row():
with gr.Column(scale=2):
upload_file = gr.File(
label="Upload file audio/video từ máy tính (.mp3 .wav .m4a .ogg .flac .aac .webm .mp4 .mkv .avi .mov .flv)",
file_types=["video", "audio", ".webm"],
type="filepath"
)
media_info = gr.Markdown(value="Nhập URL hoặc upload file để xem thông tin...")
with gr.Row():
bitrate_dropdown = gr.Dropdown(
choices=["32k", "64k"],
value="32k",
label="🎚️ Bitrate",
info="32k: nhỏ hơn | 64k: tốt hơn một chút",
scale=1,
)
sample_rate_dropdown = gr.Dropdown(
choices=["8000", "16000"],
value="16000",
label="📊 Sample Rate (Hz)",
info="8kHz: tối thiểu | 16kHz: khuyến nghị cho transcribe",
scale=1,
)
convert_btn = gr.Button("🎵 Convert MP3", variant="secondary", size="lg")
convert_status = gr.Markdown(value="")
transcribe_btn = gr.Button("🎙️ Transcribe", variant="primary", size="lg", interactive=False)
with gr.Accordion("🔧 Phương thức & Timestamps", open=False):
with gr.Row():
method_dropdown = gr.Dropdown(
choices=["model.transcribe", "BatchedInferencePipeline"],
label="Phương thức",
scale=3,
elem_id="method_dropdown",
value="BatchedInferencePipeline",
info="model.transcribe: chất lượng cao | Batched: nhanh hơn"
)
beam_size_dropdown = gr.Dropdown(
choices=["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"],
label="Beam Size",
scale=2,
elem_id="method_dropdown",
value="3",
info="Cao hơn = chính xác hơn nhưng chậm hơn"
)
include_timestamps_checkbox = gr.Checkbox(
label="Bao gồm Timestamps (HH:MM:SS)",
value=True,
info="Bao gồm timestamps trong kết quả transcription"
)
with gr.Accordion("⚙️ Tham số nâng cao (Batch & VAD)", open=False):
with gr.Row():
batch_size_slider = gr.Slider(
minimum=1,
maximum=64,
value=16,
step=1,
label="Batch Size (BatchedInferencePipeline)",
info="Số lượng đoạn âm thanh xử lý cùng lúc. Ảnh hưởng đến tốc độ và bộ nhớ GPU.",
visible=True
)
with gr.Row():
min_silence_duration_ms_slider = gr.Slider(
minimum=0,
maximum=2000,
value=500,
step=50,
label="VAD: Min Silence Duration (ms)",
info="Thời lượng im lặng tối thiểu để tách phân đoạn. Ảnh hưởng đến việc phát hiện câu/từ."
)
speech_pad_ms_slider = gr.Slider(
minimum=0,
maximum=1000,
value=400,
step=50,
label="VAD: Speech Pad (ms)",
info="Thêm thời gian vào đầu/cuối mỗi phân đoạn giọng nói. Giúp giữ lại bối cảnh."
)
with gr.Row():
no_speech_threshold_slider = gr.Slider(
minimum=0.0,
maximum=1.0,
value=0.55,
step=0.05,
label="VAD: No Speech Threshold",
info="Ngưỡng xác định khi nào không có lời nói."
)
condition_on_previous_text_checkbox = gr.Checkbox(
label="Condition on Previous Text",
value=False,
info="Sử dụng văn bản trước đó làm điều kiện để cải thiện tính nhất quán."
)
with gr.Row():
minimum_speech_duration_slider = gr.Slider(
minimum=0.0,
maximum=5.0,
value=0.1,
step=0.05,
label="VAD: Minimum Speech Duration (s)",
info="Thời lượng tối thiểu của một đoạn giọng nói. Giúp lọc các âm thanh ngắn, nhiễu."
)
with gr.Accordion("✂️ Cắt file MP3", open=False):
cut_mode_radio = gr.Radio(
choices=["Theo phút", "Theo phần"],
value="Theo phút",
label="Chế độ cắt",
info="Chỉ chọn một chế độ"
)
with gr.Row():
cut_minutes_dropdown = gr.Dropdown(
choices=["3", "5", "10", "15"],
value="5",
label="⏱️ Phút mỗi đoạn",
info="Áp dụng khi chọn Theo phút",
interactive=True,
scale=1,
)
cut_parts_dropdown = gr.Dropdown(
choices=["2", "3", "5", "10"],
value="2",
label="🔢 Số phần",
info="Áp dụng khi chọn Theo phần",
interactive=False,
scale=1,
)
cut_btn = gr.Button("✂️ Cut File", variant="secondary", size="lg")
cut_status = gr.Markdown(value="")
with gr.Column(scale=3):
placeholder_text = (
"📖 Hướng dẫn: Nhập Youtube video URL và bấm ENTER \n\n"
"📖 Hoặc: Upload file từ máy tính \n\n"
)
transcript_output = gr.Textbox(
label="Transcript",
lines=20,
max_lines=50,
show_copy_button=True,
placeholder=placeholder_text,
autoscroll=True,
elem_id="textbox_id",
)
statistics_output = gr.Textbox(
label="Thông tin transcription",
lines=5,
max_lines=20,
show_copy_button=True,
visible=True,
)
clear_output_btn = gr.Button("🗑️ Clear Output", variant="stop", size="sm", visible=False)
with gr.Row():
source_mp3_output = gr.File(label="📥 Source MP3", visible=False)
converted_mp3_output = gr.File(label="📥 Converted MP3", visible=False)
with gr.Row():
file_output = gr.File(label="📄 TXT", visible=False)
srt_output = gr.File(label="🎬 SRT", visible=False)
vtt_output = gr.File(label="🌐 VTT", visible=False)
with gr.Row():
trimmed_mp3_output = gr.File(label="✂️ Trimmed MP3", visible=False)
trimmed_mp4_output = gr.File(label="✂️ Trimmed Video", visible=False)
cut_zip_output = gr.File(label="📦 Cut ZIP", visible=False)
# ====== EVENT HANDLERS ======
# File upload
upload_file.change(
fn=process_upload,
inputs=[upload_file],
outputs=[current_file, video_file_state, media_info, gr.State(999999), gr.State(0)]
)
# Show source MP3 after file is set
current_file.change(
fn=lambda f: gr.update(value=f, visible=True) if f else gr.update(visible=False),
inputs=[current_file],
outputs=[source_mp3_output]
)
# ====== NÚT CONVERT MP3 ======
convert_btn.click(
fn=lambda: (gr.update(interactive=False), gr.update(value="⏳ Đang convert...")),
inputs=None,
outputs=[convert_btn, convert_status]
).then(
fn=convert_to_mp3,
inputs=[current_file, video_file_state, bitrate_dropdown, sample_rate_dropdown],
outputs=[converted_mp3_output, convert_status]
).then(
fn=lambda f: (gr.update(interactive=True), gr.update(visible=True) if f else gr.update(visible=False)),
inputs=[converted_mp3_output],
outputs=[convert_btn, converted_mp3_output]
)
# ====== CUT FILE ======
cut_mode_radio.change(
fn=lambda mode: (
gr.update(interactive=(mode == "Theo phút")),
gr.update(interactive=(mode == "Theo phần"))
),
inputs=[cut_mode_radio],
outputs=[cut_minutes_dropdown, cut_parts_dropdown]
)
cut_btn.click(
fn=lambda: gr.update(interactive=False),
inputs=None,
outputs=[cut_btn]
).then(
fn=do_cut_file,
inputs=[current_file, cut_mode_radio, cut_minutes_dropdown, cut_parts_dropdown],
outputs=[cut_zip_output, cut_status]
).then(
fn=lambda: gr.update(interactive=True),
inputs=None,
outputs=[cut_btn]
)
# Clear Output button handler
clear_output_btn.click(
fn=clear_outputs,
inputs=None,
outputs=[
source_mp3_output,
converted_mp3_output,
cut_zip_output,
cut_status,
clear_output_btn
]
)
# ====== QUEUE + LAUNCH ======
demo.queue(default_concurrency_limit=2)
demo.launch()