video_processing.py

import os
import random
import glob
import shutil
import tempfile
from concurrent.futures import ThreadPoolExecutor
from moviepy.editor import (
    AudioFileClip,
    CompositeVideoClip,
    ImageClip,
    VideoFileClip,
    concatenate_videoclips,
    vfx,
)
from moviepy.video.tools.subtitles import SubtitlesClip
import tqdm

from sentence_transformers import SentenceTransformer, util

# Khởi tạo model sentence transformer
model = SentenceTransformer('all-MiniLM-L6-v2')

# Tăng số lượng ảnh lên 30
NUM_IMAGES = 30

def add_transitions(clips, transition_duration=1):
    """
    Thêm hiệu ứng chuyển cảnh giữa các clip.
    """
    final_clips = []
    for i, clip in enumerate(clips):
        start_time = i * (clip.duration - transition_duration)
        end_time = start_time + clip.duration

        if i > 0:
            # Tạo hiệu ứng fade in
            fade_in = clip.fx(vfx.fadeout, duration=transition_duration)
            fade_in = fade_in.set_start(start_time)
            final_clips.append(fade_in)

        if i < len(clips) - 1:
            # Tạo hiệu ứng fade out
            fade_out = clip.fx(vfx.fadein, duration=transition_duration)
            fade_out = fade_out.set_end(end_time)
            final_clips.append(fade_out)

        # Thêm clip gốc
        final_clips.append(clip.set_start(start_time).set_end(end_time))

    return CompositeVideoClip(final_clips)

def create_video(sentences, audio_files, video_files, output_path="output_video.mp4"):
    """
    Tạo video từ các câu, file âm thanh và file video.
    """
    clips = []
    for sentence, audio_path, video_path in tqdm.tqdm(zip(sentences, audio_files, video_files), desc="Tạo video"):
        audio = AudioFileClip(audio_path)
        video = VideoFileClip(video_path).set_duration(audio.duration)
        video = video.set_audio(audio)
        clips.append(video)

    final_video = concatenate_videoclips(clips, method="compose")
    final_video.write_videofile(output_path, fps=24)
    print(f"Đã tạo video: {output_path}")
    return output_path

def process_images_parallel(image_patch, clip_duration):
    """
    Xử lý song song các hình ảnh.
    """
    with ThreadPoolExecutor() as executor:
        futures = []
        for content, image_path in image_patch:
            if image_path:
                future = executor.submit(ImageClip, image_path)
                futures.append((future, clip_duration))

        clips = []
        for future, duration in futures:
            clip = future.result().set_duration(duration)
            clips.append(clip)

    return clips

# Định nghĩa hàm extract_key_contents
def extract_key_contents(script: str) -> list[str]:
    """
    Hàm này dùng để trích xuất các ý chính từ một đoạn script.

    Tham số:
    - script (str): Đoạn văn bản cần xử lý để trích xuất các ý chính.

    Trả về:
    - list[str]: Danh sách các câu được tách ra từ đoạn script.

    Logic xử lý:
    - Đầu tiên, đoạn script được tách thành các câu dựa trên dấu chấm ('.').
    - Mỗi câu được xem như một ý chính và được thêm vào danh sách kết quả.
    """
    # Kiểm tra nếu script là chuỗi rỗng
    if not script:
        return []

    # Tách đoạn script thành các câu dựa trên dấu chấm
    sentences = script.split('.')

    # Loại bỏ các khoảng trắng thừa và các câu rỗng
    sentences = [sentence.strip() for sentence in sentences if sentence.strip()]

    # Trả về danh sách các câu
    return sentences


def process_script_for_video(script, dataset_path, use_dataset):
    """
    Xử lý script để tạo video.
    """
    sentences = extract_key_contents(script)
    return sentences

def create_video_func(script, audio_path, dataset_path, use_dataset):
    """
    Hàm chính để tạo video.
    """
    try:
        sentences = process_script_for_video(script, dataset_path, use_dataset)

        # Tạo thư mục tạm thời để lưu các file âm thanh tách biệt
        temp_dir = tempfile.mkdtemp()

        # Tách file âm thanh thành các đoạn nhỏ
        audio_clips = split_audio(audio_path, len(sentences), temp_dir)

        # Lấy đường dẫn của các video từ dataset
        video_files = glob.glob(os.path.join(dataset_path, "*.mp4")) if use_dataset else []

        # Đảm bảo số lượng câu, âm thanh và video là bằng nhau
        min_length = min(len(sentences), len(audio_clips), len(video_files))
        sentences = sentences[:min_length]
        audio_clips = audio_clips[:min_length]
        video_files = video_files[:min_length]

        output_path = "output_video.mp4"
        create_video(sentences, audio_clips, video_files, output_path)

        return output_path
    except Exception as e:
        print(f"Lỗi khi tạo video: {e}")
        return None
    finally:
        # Xóa thư mục tạm thời
        shutil.rmtree(temp_dir)

def split_audio(audio_path, num_segments, output_dir):
    """
    Chia file âm thanh thành các đoạn nhỏ.
    """
    audio = AudioFileClip(audio_path)
    duration = audio.duration
    segment_duration = duration / num_segments

    audio_clips = []
    for i in range(num_segments):
        start = i * segment_duration
        end = (i + 1) * segment_duration
        segment = audio.subclip(start, end)
        output_path = os.path.join(output_dir, f"segment_{i}.mp3")
        segment.write_audiofile(output_path)
        audio_clips.append(output_path)

    return audio_clips

def find_matching_image(prompt, dataset_path, threshold=0.5):
    """
    Tìm kiếm hình ảnh phù hợp với prompt trong dataset.
    """
    prompt_embedding = model.encode(prompt, convert_to_tensor=True)
    best_match = None
    best_score = -1

    for filename in os.listdir(dataset_path):
        if filename.lower().endswith(('.png', '.jpg', '.jpeg')):
            image_path = os.path.join(dataset_path, filename)
            image_name = os.path.splitext(filename)[0].replace('_', ' ')
            image_embedding = model.encode(image_name, convert_to_tensor=True)
            cosine_score = util.pytorch_cos_sim(prompt_embedding, image_embedding).item()
            if cosine_score > best_score and cosine_score >= threshold:
                best_score = cosine_score
                best_match = image_path
    return best_match