app.py

import os
import whisper
from flask import Flask, request, render_template, redirect, url_for, flash, jsonify, send_file
from werkzeug.utils import secure_filename
import uuid
import threading
import time
import json
from pathlib import Path
import librosa
import numpy as np

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'
app.config['MAX_CONTENT_LENGTH'] = 500 * 1024 * 1024  # 500MB max file size

# Allowed file extensions for media files
ALLOWED_EXTENSIONS = {'mp3', 'mp4', 'mpeg', 'mpga', 'm4a', 'wav', 'webm', 'mov', 'avi', 'flv', 'mkv'}

# Global dictionary to track processing status
processing_status = {}

# Load whisper model (you can change this to different models)
model = None

def load_whisper_model():
    global model
    if model is None:
        model = whisper.load_model("tiny.en")
    return model

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

def get_file_size_str(size_bytes):
    """Convert bytes to human readable string"""
    if size_bytes == 0:
        return "0B"
    size_names = ["B", "KB", "MB", "GB"]
    i = 0
    while size_bytes >= 1024 and i < len(size_names) - 1:
        size_bytes /= 1024.0
        i += 1
    return f"{size_bytes:.1f}{size_names[i]}"

def process_file_async(file_path, job_id, original_filename):
    """Process file in background thread"""
    try:
        processing_status[job_id] = {
            'status': 'processing', 
            'progress': 10,
            'filename': original_filename
        }
        
        # Validate file exists and has content
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        file_size = os.path.getsize(file_path)
        if file_size == 0:
            raise ValueError("File is empty")
        
        if file_size < 1024:  # Less than 1KB
            raise ValueError("File too small, likely corrupted or not a valid audio file")
        
        processing_status[job_id]['progress'] = 20
        
        # Try to validate audio file by loading a small portion
        try:
            # Load a small sample to validate the file
            audio_data, sample_rate = librosa.load(file_path, duration=1.0, sr=None)
            if len(audio_data) == 0:
                raise ValueError("Audio file contains no audio data")
        except Exception as audio_error:
            raise ValueError(f"Invalid or corrupted audio file: {str(audio_error)}")
        
        # Load model if not already loaded
        processing_status[job_id]['progress'] = 30
        model = load_whisper_model()
        
        # Process the audio file
        processing_status[job_id]['progress'] = 50
        
        # Add additional error handling for Whisper processing
        try:
            result = model.transcribe(file_path)
            if not result or 'text' not in result:
                raise ValueError("Whisper failed to generate transcription")
        except Exception as whisper_error:
            error_msg = str(whisper_error)
            if "reshape tensor" in error_msg.lower():
                raise ValueError("Audio file format not compatible with Whisper model. Try converting to WAV or MP3 format.")
            else:
                raise ValueError(f"Whisper processing failed: {error_msg}")
        
        # Save the transcription to a text file next to the original
        file_dir = os.path.dirname(file_path)
        file_name = os.path.splitext(os.path.basename(file_path))[0]
        txt_filename = os.path.join(file_dir, f"{file_name}_transcription.txt")
        
        with open(txt_filename, 'w', encoding='utf-8') as f:
            f.write(result['text'])
        
        processing_status[job_id] = {
            'status': 'completed', 
            'progress': 100,
            'text': result['text'],
            'txt_file': txt_filename,
            'filename': original_filename
        }
            
    except Exception as e:
        processing_status[job_id] = {
            'status': 'error', 
            'progress': 0,
            'error': str(e),
            'filename': original_filename
        }

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/browse')
def browse_files():
    """Browse local filesystem for media files"""
    path = request.args.get('path', os.path.expanduser('~'))
    
    try:
        # Ensure the path exists and is a directory
        if not os.path.exists(path) or not os.path.isdir(path):
            path = os.path.expanduser('~')
        
        # Get directory contents
        items = []
        
        # Add parent directory link (if not at root)
        parent_path = os.path.dirname(path)
        if parent_path != path:
            items.append({
                'name': '..',
                'path': parent_path,
                'type': 'directory',
                'size': '',
                'is_media': False
            })
        
        # List directory contents
        try:
            for item in sorted(os.listdir(path)):
                item_path = os.path.join(path, item)
                
                if os.path.isdir(item_path):
                    items.append({
                        'name': item,
                        'path': item_path,
                        'type': 'directory',
                        'size': '',
                        'is_media': False
                    })
                elif os.path.isfile(item_path):
                    is_media = allowed_file(item)
                    file_size = os.path.getsize(item_path)
                    items.append({
                        'name': item,
                        'path': item_path,
                        'type': 'file',
                        'size': get_file_size_str(file_size),
                        'is_media': is_media
                    })
        except PermissionError:
            flash(f'Permission denied accessing {path}')
            return redirect(url_for('browse_files', path=os.path.expanduser('~')))
        
        return jsonify({
            'current_path': path,
            'items': items
        })
        
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/process', methods=['POST'])
def process_files():
    """Process selected local files"""
    data = request.get_json()
    file_paths = data.get('files', [])
    
    if not file_paths:
        return jsonify({'error': 'No files selected'}), 400
    
    job_ids = []
    
    for file_path in file_paths:
        # Verify file exists and is a media file
        if os.path.exists(file_path) and os.path.isfile(file_path) and allowed_file(file_path):
            # Generate unique job ID
            job_id = str(uuid.uuid4())
            job_ids.append(job_id)
            
            # Start background processing
            filename = os.path.basename(file_path)
            thread = threading.Thread(target=process_file_async, args=(file_path, job_id, filename))
            thread.daemon = True
            thread.start()
    
    if job_ids:
        return jsonify({'job_ids': job_ids, 'redirect': url_for('processing_page')})
    else:
        return jsonify({'error': 'No valid media files selected'}), 400

@app.route('/processing')
def processing_page():
    job_ids = request.args.getlist('job_ids')
    return render_template('processing.html', job_ids=job_ids)

@app.route('/status/<job_id>')
def get_status(job_id):
    status = processing_status.get(job_id, {'status': 'not_found'})
    return jsonify(status)

@app.route('/result/<job_id>')
def get_result(job_id):
    status = processing_status.get(job_id, {'status': 'not_found'})
    if status.get('status') == 'completed':
        return render_template('result.html', 
                             text=status.get('text', ''), 
                             job_id=job_id,
                             filename=status.get('filename', ''))
    else:
        return redirect(url_for('index'))

@app.route('/download/<job_id>')
def download_result(job_id):
    status = processing_status.get(job_id, {})
    if status.get('status') == 'completed' and 'txt_file' in status:
        txt_file = status['txt_file']
        if os.path.exists(txt_file):
            return send_file(txt_file, as_attachment=True, download_name=os.path.basename(txt_file))
    
    flash('File not found or processing not completed')
    return redirect(url_for('index'))

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)