video-ai/react-flask-app/server/processing.py

import os
import tempfile
from moviepy import VideoFileClip
from openai import OpenAI
import math
from pydub import AudioSegment
from docx import Document
from docx.shared import Pt, RGBColor
from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
import pypandoc
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configuration
MAX_FILE_SIZE = 26214400  # 25MB in bytes
CHUNK_SIZE = MAX_FILE_SIZE // 2  # Split into ~12MB chunks
OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'outputs')
os.makedirs(OUTPUT_DIR, exist_ok=True)

def setup_api_keys():
    """Setup API keys and configurations."""
    api_key = os.getenv('OPENAI_API_KEY')
    if not api_key:
        raise ValueError("OpenAI API key not found in environment variables")
    
    client = OpenAI(api_key=api_key)
    return client

def extract_audio(video_path):
    """Extract audio from video file and return path to audio file."""
    print("Extracting audio from video file...")
    temp_audio_path = os.path.join(OUTPUT_DIR, "temp_audio.mp3")
    video = VideoFileClip(video_path)
    video.audio.write_audiofile(temp_audio_path)
    video.close()
    return temp_audio_path

def split_audio_by_duration(file_path, duration_seconds=600):
    """Split an audio file into chunks of a specified duration (default 10 minutes)."""
    chunks_dir = os.path.join(OUTPUT_DIR, 'chunks')
    os.makedirs(chunks_dir, exist_ok=True)
    
    try:
        audio = AudioSegment.from_file(file_path)
    except Exception as e:
        print(f"Error loading audio file: {e}")
        return []
    
    chunks = []
    chunk_length_ms = duration_seconds * 1000  # Convert seconds to milliseconds
    
    for i, chunk_start in enumerate(range(0, len(audio), chunk_length_ms)):
        chunk_end = chunk_start + chunk_length_ms
        chunk = audio[chunk_start:chunk_end]
        
        chunk_filename = os.path.join(chunks_dir, f"chunk_{i}.mp3")
        try:
            chunk.export(chunk_filename, format="mp3")
            chunks.append(chunk_filename)
            print(f"Created chunk: {chunk_filename}")
        except Exception as e:
            print(f"Error exporting chunk: {e}")
    
    return chunks

def transcribe_audio(client, audio_chunks):
    """Transcribe audio chunks using OpenAI's Whisper API."""
    transcripts = []
    for chunk in audio_chunks:
        print(f"Transcribing chunk: {chunk}")
        audio_file = open(chunk, "rb")
        try:
            transcript = client.audio.transcriptions.create(
                file=audio_file,
                model="whisper-1",
                response_format="text",
                language="it",
                prompt="L'audio è il pezzo di una riunione registrata; potrebbe essere la continuazione di un precedente pezzo di audio"
            )
            transcripts.append(transcript)
        except Exception as e:
            print(f"Error in transcription: {e}")
    
    return ' '.join(transcripts)

def summarize_transcription(client, transcript):
    """Use GPT-4 to summarize the transcription."""
    system_message = """You will find next a transcript of a Teams Meeting.
    [PAY ATTENTION: The text may not be perfect as it was transcribed from video]
    
    Your task is to create detailed notes from this meeting transcript,
    including as much information as possible. At the end, include a summary
    of open points and decisions made during the meeting.
    
    The result should be in business Italian language.
    """
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": transcript}
        ],
        temperature=0.1,
        max_tokens=16000
    )
    
    return response.choices[0].message.content

def clean_summary(client, summary):
    """Clean and format the summary using GPT."""
    system_message = """Following is an automated generated text from a video transcript.
    Due to audio quality and lack of knowledge about the context, the text may not be complete or accurate.
    
    [ORIGINAL TEXT]
    {0}
    
    Please clean this text, fixing any inconsistencies, formatting issues, or errors in transcription.
    The result should be well-structured and in business Italian language.
    """.format(summary)
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": "Please clean and format this text."}
        ],
        temperature=0.5,
        max_tokens=16000
    )
    
    return response.choices[0].message.content

def generate_verbale(client, summary_clean):
    """Generate a formal meeting report (verbale) from the cleaned summary."""
    system_message = """You are tasked with creating a formal meeting report (verbale) from the following meeting summary.
    
    [SUMMARY]
    {0}
    
    Please structure the report in a professional format typical of consulting firms (like Deloitte).
    Focus particularly on decisions made and action items for future meetings.
    
    The report should be in business Italian language.
    """.format(summary_clean)
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_message},
            {"role": "user", "content": "Generate a formal meeting report (verbale)."}
        ],
        temperature=0.5,
        max_tokens=16000
    )
    
    return response.choices[0].message.content

def convert_markdown_to_word(markdown_text, output_file):
    """Convert markdown text to MS Word document format using pypandoc."""
    try:
        # Write markdown to a temporary file
        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.md', encoding='utf-8') as temp_file:
            temp_file.write(markdown_text)
            temp_md_file = temp_file.name
        
        # Ensure output directory exists
        os.makedirs(os.path.dirname(output_file), exist_ok=True)
        
        # Convert to docx using pypandoc
        pypandoc.convert_file(temp_md_file, 'docx', outputfile=output_file)
        
        # Clean up temp file
        os.unlink(temp_md_file)
        
        print(f"Successfully converted to Word: {output_file}")
        return True
    except Exception as e:
        print(f"Error converting to Word: {e}")
        return False

def process_video(file_path):
    """Process a video file and generate transcription, summaries, and documents."""
    base_filename = os.path.splitext(os.path.basename(file_path))[0]
    client = setup_api_keys()
    
    # Output file paths
    transcription_path = os.path.join(OUTPUT_DIR, f"{base_filename}_transcription.txt")
    summary_path = os.path.join(OUTPUT_DIR, f"{base_filename}_summary.md")
    summary_clean_path = os.path.join(OUTPUT_DIR, f"{base_filename}_summary_clean.md")
    verbale_path = os.path.join(OUTPUT_DIR, f"{base_filename}_verbale.md")
    docx_path = os.path.join(OUTPUT_DIR, f"{base_filename}_summary.docx")
    docx_verbale_path = os.path.join(OUTPUT_DIR, f"{base_filename}_verbale.docx")
    
    try:
        # Step 1: Extract audio from video
        audio_path = extract_audio(file_path)
        
        # Step 2: Split audio into chunks
        audio_chunks = split_audio_by_duration(audio_path)
        
        # Step 3: Transcribe audio
        transcription = transcribe_audio(client, audio_chunks)
        with open(transcription_path, "w", encoding='utf-8') as f:
            f.write(transcription)
        print(f"Saved transcription to: {transcription_path}")
        
        # Step 4: Generate summary
        summary = summarize_transcription(client, transcription)
        with open(summary_path, "w", encoding='utf-8') as f:
            f.write(summary)
        print(f"Saved summary to: {summary_path}")
        
        # Step 5: Clean and format summary
        summary_clean = clean_summary(client, summary)
        with open(summary_clean_path, "w", encoding='utf-8') as f:
            f.write(summary_clean)
        print(f"Saved cleaned summary to: {summary_clean_path}")
        
        # Step 6: Generate formal report (verbale)
        verbale = generate_verbale(client, summary_clean)
        with open(verbale_path, "w", encoding='utf-8') as f:
            f.write(verbale)
        print(f"Saved verbale to: {verbale_path}")
        
        # Step 7: Convert markdown to Word documents
        convert_markdown_to_word(summary_clean, docx_path)
        convert_markdown_to_word(verbale, docx_verbale_path)
        
        # Step 8: Clean up temporary files
        for chunk in audio_chunks:
            if os.path.exists(chunk):
                os.remove(chunk)
        if os.path.exists(audio_path):
            os.remove(audio_path)
        
        # Return paths to all generated files
        return {
            'transcription_path': transcription_path,
            'summary_path': summary_path,
            'summary_clean_path': summary_clean_path,
            'verbale_path': verbale_path,
            'docx_path': docx_path,
            'docx_verbale_path': docx_verbale_path
        }
    
    except Exception as e:
        print(f"Error processing video: {e}")
        raise