Python实现音频去广告和字幕提取<

import glob
import os
from concurrent.futures import ThreadPoolExecutor, as_completed

import numpy as np
import librosa
import torch
import whisper
from pydub import AudioSegment
import soundfile as sf
import torch.nn.functional as F
import shutil

# 已知的广告片段文件
AD_SNIPPET_FILE  = "./testcase/test2.wav"

# 待处理的音频文件目录
audio_dir = "./testcase"
TAIL_SECONDS = 300  # 只截取最后5分钟处理
SIMILARITY_THRESHOLD = 0.8  # 相似度阈值(0~1之间, 需根据实际情况调整)
SUCCESS_DIR = "./success"
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

def load_audio_segment(file_path, sr=16000, tail_seconds=None):
info = sf.info(file_path)
total_duration = info.duration
if tail_seconds is not None and tail_seconds < total_duration:
    start_time = total_duration - tail_seconds
    audio, _ = librosa.load(file_path, sr=sr, mono=True, offset=start_time, duration=tail_seconds)
    return audio, start_time

else:
    audio, _ = librosa.load(file_path, sr=sr, mono=True)
    return audio, 0.0

def find_audio_snippet(main_audio_path, snippet_audio, snippet_norm, sr=16000, tail_seconds=300):
"""
在 main_audio_path 中寻找 snippet_audio 音频片段的出现位置。
snippet_audio 为事先加载好的 numpy 数组，snippet_norm 为 snippet 的二范数，用于相似度计算。
返回 (ad_start_time, ad_end_time, similarity)
若未找到则返回 (None, None, None)
"""
main_audio, main_start = load_audio_segment(main_audio_path, sr=sr, tail_seconds=tail_seconds)
if len(snippet_audio) > len(main_audio):
    return None, None, None
# 转换到 GPU 张量
main_audio_t = torch.from_numpy(main_audio).float().to(device).unsqueeze(0).unsqueeze(0)  # [1,1,M]
snippet_audio_t = torch.from_numpy(snippet_audio).float().to(device).unsqueeze(0).unsqueeze(0)  # [1,1,S]
# 使用 conv1d 来进行类似相似度搜索 (无 snippet 翻转)
correlation = F.conv1d(main_audio_t, snippet_audio_t)
correlation = correlation[0, 0].cpu().numpy()
best_index = np.argmax(correlation)
best_value = correlation[best_index]
# 相似度计算：归一化
similarity = best_value / snippet_norm if snippet_norm > 0 else 0
ad_start_time = main_start + best_index / sr
ad_end_time = ad_start_time + len(snippet_audio) / s

return ad_start_time, ad_end_time, similarity

def process_file(filename, snippet_audio, snippet_norm, sr=16000, tail_seconds=300, similarity_threshold=0.8):
"""
处理单个文件，找到广告并移除。
"""
ad_start, ad_end, similarity = find_audio_snippet(filename, snippet_audio, snippet_norm, sr=sr,
                                                  tail_seconds=tail_seconds)

if ad_start is not None and similarity > similarity_threshold:
    # 去除广告段落
    audio = AudioSegment.from_file(filename)
    part1 = audio[:ad_start * 1000]
    part2 = audio[ad_end * 1000:]
    cleaned = part1 + part2
    cleaned_file = f"output/{os.path.basename(filename)}"
    cleaned.export(cleaned_file, format="mp3")
    shutil.move(filename, SUCCESS_DIR)
    return f"{filename} 已移除广告，生成 {cleaned_file}，相似度：{similarity}"
else:
    return f"{filename} 中未高相似度检测到广告或相似度过低（{similarity}）"


def remove_ads():
sr = 16000
# 预先加载广告片段
snippet_audio, _ = librosa.load(AD_SNIPPET_FILE , sr=sr, mono=True)
# 计算snippet的范数，用于相似度归一化
snippet_norm = np.dot(snippet_audio, snippet_audio)

file_list = [os.path.join(audio_dir, f) for f in os.listdir(audio_dir) if
             f.endswith(".mp3")]

# 使用多线程加速处理
# 线程数可根据您的机器资源调整
max_workers = 20
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = {
        executor.submit(process_file, file, snippet_audio, snippet_norm, sr, TAIL_SECONDS,
                        SIMILARITY_THRESHOLD): file
        for file in file_list
    }

    for future in as_completed(futures):
        file = futures[future]
        try:
            res = future.result()
            print(res)
        except Exception as e:
            print(f"{file} 处理时出错: {e}")

def format_lrc_timestamp(seconds: float) -> str:
"""将秒数转换为 LRC 格式时间戳 [mm:ss.xx]"""
total_seconds = int(seconds)
m = total_seconds // 60
s = total_seconds % 60
# 毫秒取两位小数
ms = (seconds - total_seconds) * 100
return f"[{m:02d}:{s:02d}.{int(ms):02d}]"

def trans_files():
# 请将此路径替换为你的音频目录路径
audio_directory = "./audio_files"
# 可根据需要选择模型大小，如 "small"、"medium"、"large"
trans_text(audio_directory, model_name="medium", language="zh")

def transcribe_to_lrc(audio_path: str, lrc_path: str, model, language: str = "zh"):
"""
使用已加载的 whisper model 对 audio_path 进行转录，
并将结果保存为 lrc_path 文件。
"""
result = model.transcribe(audio_path, language=language)
segments = result.get("segments", [])

with open(lrc_path, "w", encoding="utf-8") as f:
    # 可根据需要添加标签信息，如标题、歌手、专辑
    f.write("[ti:未知标题]\n")
    f.write("[ar:未知作者]\n")
    f.write("[al:未知专辑]\n\n")

    for seg in segments:
        start_time = format_lrc_timestamp(seg['start'])
        text = seg['text'].strip()
        f.write(f"{start_time}{text}\n")


def trans_text(audio_dir: str, model_name: str = "medium", language: str = "zh"):
# 尝试使用 GPU
device = "cuda:0" if torch.cuda.is_available() else "cpu"
print(f"使用设备: {device}")

# 加载模型到指定设备
# 模型大小可根据资源调整，如：tiny, base, small, medium, large
print(f"加载 Whisper 模型 ({model_name})，请稍候...")
model = whisper.load_model(model_name, device=device)
print("模型加载完成。")

# 遍历指定目录下所有 mp3
audio_files = glob.glob(os.path.join(audio_dir, "*.mp3"))
if not audio_files:
    print("指定目录中未找到 MP3 文件。")
    return

for audio_path in audio_files:
    base_name = os.path.splitext(audio_path)[0]
    lrc_path = base_name + ".lrc"

    print(f"处理文件: {audio_path} -> {lrc_path}")
    transcribe_to_lrc(audio_path, lrc_path, model=model, language=language)
    print(f"完成: {lrc_path}")

print("所有文件处理完成！")

def trans_files():
# 请将此路径替换为你的音频目录路径
audio_directory = "./audio_files"
# 可根据需要选择模型大小，如 "small"、"medium"、"large"
trans_text(audio_directory, model_name="medium", language="zh")