Create Audio_analyzer.py

Audio_analyzer.py

@@ -0,0 +1,496 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+transcribe_audio.py - الإصدار المحسن
+استخراج النصوص من ملفات صوتية باستخدام whisper (محلي).
+
+التحسينات المضافة:
+✅ شريط تقدم للملفات
+✅ دعم تنسيق SRT
+✅ خيار "all" لحفظ كل التنسيقات
+✅ إدارة ذاكرة محسنة (معالجة دفعات)
+✅ تسجيل الأحداث (Logging)
+✅ اكتشاف ملفات محسن
+✅ معالجة أخطاء أفضل
+✅ تقارير إحصائية
+✅ دعم مجلد الإخراج المخصص
+
+Usage:
+    python transcribe_audio.py audio.mp3
+    python transcribe_audio.py folder/ --model large --lang ar --output all --recursive
+    python transcribe_audio.py "*.mp3" --output-dir ./output --verbose
+"""
+
+import argparse
+import os
+import sys
+import json
+import logging
+from pathlib import Path
+import whisper
+import typing
+import traceback
+import time
+from datetime import datetime
+from tqdm import tqdm
+
+# إعداد التسجيل
+def setup_logging(verbose: bool = False):
+    """إعداد نظام التسجيل"""
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.basicConfig(
+        level=level,
+        format='%(asctime)s - %(levelname)s - %(message)s',
+        handlers=[
+            logging.StreamHandler(sys.stdout),
+            logging.FileHandler('transcription.log', encoding='utf-8', mode='w')
+        ]
+    )
+    return logging.getLogger(__name__)
+
+logger = setup_logging()
+
+def find_audio_files(inputs: typing.List[str], recursive: bool = False, max_file_size_mb: int = 500):
+    """اكتشاف الملفات الصوتية مع تحسينات"""
+    audio_extensions = {
+        '.mp3', '.wav', '.m4a', '.flac', '.ogg', '.webm', 
+        '.aac', '.wma', '.opus', '.mpga', '.mp4', '.m4b', '.3gp'
+    }
+    
+    files = []
+    max_size = max_file_size_mb * 1024 * 1024
+    
+    for input_path in inputs:
+        path = Path(input_path)
+        
+        if path.is_file():
+            # التحقق من حجم الملف
+            file_size = path.stat().st_size
+            if file_size > max_size:
+                logger.warning(f"تخطي الملف الكبير: {path.name} ({file_size / (1024*1024):.1f}MB)")
+                continue
+                
+            if path.suffix.lower() in audio_extensions:
+                files.append(path.resolve())
+            else:
+                logger.warning(f"امتداد غير معروف: {path}")
+                    
+        elif path.is_dir():
+            pattern = "**/*" if recursive else "*"
+            for file_path in path.glob(pattern):
+                if file_path.is_file() and file_path.suffix.lower() in audio_extensions:
+                    file_size = file_path.stat().st_size
+                    if file_size <= max_size:
+                        files.append(file_path.resolve())
+                    else:
+                        logger.warning(f"تخطي الملف الكبير: {file_path.name}")
+        else:
+            # دعم الأنماط مثل "*.mp3"
+            try:
+                from glob import glob
+                for pattern in inputs:
+                    for file_path in glob(pattern):
+                        path = Path(file_path)
+                        if path.is_file() and path.suffix.lower() in audio_extensions:
+                            files.append(path.resolve())
+            except:
+                logger.warning(f"مسار غير موجود: {input_path}")
+    
+    # إزالة التكرارات والحفاظ على الترتيب
+    unique_files = []
+    seen_paths = set()
+    
+    for file_path in files:
+        if str(file_path) not in seen_paths:
+            unique_files.append(file_path)
+            seen_paths.add(str(file_path))
+    
+    logger.info(f"تم العثور على {len(unique_files)} ملف صوتي")
+    return sorted(unique_files, key=lambda x: x.name)
+
+def safe_write_text(path: Path, text: str):
+    """كتابة النص إلى ملف بشكل آمن"""
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding='utf-8')
+        return True
+    except Exception as e:
+        logger.error(f"فشل كتابة الملف {path}: {e}")
+        return False
+
+def format_timestamp(seconds: float, format_type: str = 'vtt') -> str:
+    """تنسيق الطابع الزمني"""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = seconds % 60
+    
+    if format_type == 'vtt':
+        return f"{hours:01d}:{minutes:02d}:{secs:06.3f}"
+    elif format_type == 'srt':
+        milliseconds = int((secs - int(secs)) * 1000)
+        return f"{hours:02d}:{minutes:02d}:{int(secs):02d},{milliseconds:03d}"
+    else:
+        return f"{hours:02d}:{minutes:02d}:{secs:06.3f}"
+
+def generate_vtt_content(segments: list) -> str:
+    """إنشاء محتوى VTT"""
+    content = "WEBVTT\n\n"
+    for i, segment in enumerate(segments):
+        start = segment.get('start', 0)
+        end = segment.get('end', 0)
+        text = segment.get('text', '').strip()
+        
+        start_ts = format_timestamp(start, 'vtt')
+        end_ts = format_timestamp(end, 'vtt')
+        
+        content += f"{i+1}\n{start_ts} --> {end_ts}\n{text}\n\n"
+    
+    return content
+
+def generate_srt_content(segments: list) -> str:
+    """إنشاء محتوى SRT"""
+    content = ""
+    for i, segment in enumerate(segments):
+        start = segment.get('start', 0)
+        end = segment.get('end', 0)
+        text = segment.get('text', '').strip()
+        
+        start_ts = format_timestamp(start, 'srt')
+        end_ts = format_timestamp(end, 'srt')
+        
+        content += f"{i+1}\n{start_ts} --> {end_ts}\n{text}\n\n"
+    
+    return content
+
+def validate_transcription_result(result) -> bool:
+    """التحقق من جودة نتيجة التحويل"""
+    text = result.get('text', '').strip()
+    segments = result.get('segments', [])
+    
+    # التحقق من وجود نص
+    if not text:
+        logger.warning("النص الناتج فارغ")
+        return False
+    
+    if len(text) < 10:
+        logger.warning(f"النص الناتج قصير جداً: {len(text)} حرف")
+    
+    # التحقق من نسبة الثقة إذا كانت متوفرة
+    if segments:
+        confidences = [seg.get('confidence', 0.5) for seg in segments if 'confidence' in seg]
+        if confidences:
+            avg_confidence = sum(confidences) / len(confidences)
+            if avg_confidence < 0.1:
+                logger.warning(f"ثقة منخفضة: {avg_confidence:.2f}")
+                return False
+    
+    return True
+
+def transcribe_file(model, audio_path: Path, lang: typing.Optional[str], 
+                   output_format: str, verbose: bool, output_dir: typing.Optional[Path] = None,
+                   **whisper_options):
+    """تحويل الملف الصوتي إلى نص مع تحسينات"""
+    logger.info(f"معالجة: {audio_path.name}")
+    
+    start_time = time.time()
+    
+    try:
+        # إعدادات whisper
+        options = {
+            'task': 'transcribe',
+            'temperature': 0.0,
+            'best_of': 5,
+            'beam_size': 5,
+        }
+        
+        if lang:
+            options['language'] = lang
+            
+        # دمج الإعدادات الإضافية
+        options.update(whisper_options)
+        
+        # التحويل النصي
+        result = model.transcribe(str(audio_path), **options)
+        
+        # التحقق من الجودة
+        if not validate_transcription_result(result):
+            logger.warning(f"جودة التحويل منخفضة لـ {audio_path.name}")
+        
+    except Exception as e:
+        logger.error(f"فشل التحويل لـ {audio_path.name}: {e}")
+        if verbose:
+            traceback.print_exc()
+        return None
+
+    processing_time = time.time() - start_time
+    logger.info(f"تم التحويل في {processing_time:.1f} ثانية")
+
+    # تحديد مسار الإخراج
+    if output_dir:
+        out_base = output_dir / audio_path.stem
+    else:
+        out_base = audio_path.with_suffix('')
+
+    # النص الكامل
+    text = result.get('text', '').strip()
+    segments = result.get('segments', [])
+    
+    output_files = []
+    success = True
+
+    try:
+        # حفظ بتنسيق TXT
+        if output_format in ['txt', 'all']:
+            out_path = out_base.with_suffix('.txt')
+            if safe_write_text(out_path, text):
+                output_files.append(('txt', out_path))
+                logger.info(f" -> نص: {out_path}")
+
+        # حفظ بتنسيق JSON
+        if output_format in ['json', 'all']:
+            out_path = out_base.with_suffix('.json')
+            enhanced_result = {
+                'metadata': {
+                    'audio_file': str(audio_path),
+                    'audio_size': audio_path.stat().st_size,
+                    'processing_time': datetime.now().isoformat(),
+                    'model': str(model.__class__),
+                    'language': result.get('language', 'unknown'),
+                    'processing_duration': processing_time
+                },
+                'transcription': result
+            }
+            if safe_write_text(out_path, json.dumps(enhanced_result, ensure_ascii=False, indent=2)):
+                output_files.append(('json', out_path))
+                logger.info(f" -> JSON: {out_path}")
+
+        # حفظ بتنسيق VTT
+        if output_format in ['vtt', 'all']:
+            out_path = out_base.with_suffix('.vtt')
+            vtt_content = generate_vtt_content(segments)
+            if safe_write_text(out_path, vtt_content):
+                output_files.append(('vtt', out_path))
+                logger.info(f" -> VTT: {out_path}")
+
+        # حفظ بتنسيق SRT
+        if output_format in ['srt', 'all']:
+            out_path = out_base.with_suffix('.srt')
+            srt_content = generate_srt_content(segments)
+            if safe_write_text(out_path, srt_content):
+                output_files.append(('srt', out_path))
+                logger.info(f" -> SRT: {out_path}")
+
+    except Exception as e:
+        logger.error(f"فشل حفظ الملفات لـ {audio_path.name}: {e}")
+        success = False
+
+    return {
+        'success': success,
+        'output_files': output_files,
+        'text_length': len(text),
+        'processing_time': processing_time,
+        'audio_duration': segments[-1]['end'] if segments else 0,
+        'language': result.get('language', 'unknown')
+    }
+
+def process_files_with_progress(model, files: list, **kwargs):
+    """معالجة الملفات مع عرض التقدم"""
+    successful = 0
+    failed = 0
+    total_text_length = 0
+    total_processing_time = 0
+    
+    # إعداد شريط التقدم
+    with tqdm(total=len(files), desc="📝 تحويل الملفات", unit="file", 
+              bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]") as pbar:
+        
+        for file_path in files:
+            start_time = time.time()
+            
+            result = transcribe_file(model, file_path, **kwargs)
+            
+            if result and result['success']:
+                successful += 1
+                total_text_length += result['text_length']
+                total_processing_time += result['processing_time']
+                status = "✅"
+            else:
+                failed += 1
+                status = "❌"
+            
+            # تحديث شريط التقدم
+            file_time = time.time() - start_time
+            pbar.set_postfix({
+                'نجح': successful,
+                'فشل': failed,
+                'الوقت': f'{file_time:.1f}s'
+            })
+            pbar.update(1)
+    
+    return {
+        'successful': successful,
+        'failed': failed,
+        'total_files': len(files),
+        'total_text_length': total_text_length,
+        'total_processing_time': total_processing_time,
+        'success_rate': (successful / len(files)) * 100 if files else 0
+    }
+
+def check_dependencies():
+    """التحقق من التوابع المطلوبة"""
+    try:
+        import whisper
+        return True
+    except ImportError:
+        print("❌ مكتبة whisper غير مثبتة. قم بتثبيتها باستخدام:")
+        print("   pip install openai-whisper")
+        return False
+    
+    try:
+        import tqdm
+        return True
+    except ImportError:
+        print("❌ مكتبة tqdm غير مثبتة. قم بتثبيتها باستخدام:")
+        print("   pip install tqdm")
+        return False
+
+def main():
+    # التحقق من التوابع أولاً
+    if not check_dependencies():
+        sys.exit(1)
+        
+    parser = argparse.ArgumentParser(
+        description="استخراج نص من ملفات صوتية باستخدام whisper.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+أمثلة الاستخدام:
+  python transcribe_audio.py audio.mp3
+  python transcribe_audio.py folder/ --model large --lang ar --output all --recursive
+  python transcribe_audio.py "*.mp3" --output-dir ./output --verbose
+  python transcribe_audio.py file1.mp3 file2.wav --model medium --output json
+
+النماذج المتاحة (من الأصغر إلى الأكثر دقة):
+  tiny, base, small, medium, large
+        """
+    )
+    
+    parser.add_argument('inputs', nargs='+', 
+                       help='ملف صوتي أو مجلد (يمكن تكرارها، تدعم الأنماط مثل *.mp3)')
+    
+    parser.add_argument('--model', default='small', 
+                       choices=['tiny', 'base', 'small', 'medium', 'large'],
+                       help='نموذج whisper (افتراضي: small)')
+    
+    parser.add_argument('--lang', default=None, 
+                       help='رمز اللغة (مثلاً "ar" للعربية، "en" للإنجليزية)')
+    
+    parser.add_argument('--output', choices=['txt','json','vtt','srt','all'], 
+                       default='txt', help='صيغة الإخراج (افتراضي: txt)')
+    
+    parser.add_argument('--recursive', action='store_true', 
+                       help='البحث في المجلدات فرعيًا')
+    
+    parser.add_argument('--device', default=None, 
+                       choices=['cuda', 'cpu', 'auto'],
+                       help='جهاز المعالجة (افتراضي: auto)')
+    
+    parser.add_argument('--verbose', action='store_true', 
+                       help='طباعة معلومات تفصيلية')
+    
+    parser.add_argument('--batch-size', type=int, default=1,
+                       help='عدد الملفات للمعالجة في الدفعة الواحدة (لإدارة الذاكرة)')
+    
+    parser.add_argument('--max-size-mb', type=int, default=500,
+                       help='الحجم الأقصى للملف بالميجابايت (افتراضي: 500)')
+    
+    parser.add_argument('--output-dir', default=None,
+                       help='مجلد الإخراج المخصص (بدلاً من مجلد الملف الأصلي)')
+    
+    args = parser.parse_args()
+
+    # إعداد التسجيل
+    global logger
+    logger = setup_logging(args.verbose)
+    
+    # التحقق من مجلد الإخراج
+    output_dir = None
+    if args.output_dir:
+        output_dir = Path(args.output_dir)
+        output_dir.mkdir(parents=True, exist_ok=True)
+        logger.info(f"سيتم حفظ الملفات في: {output_dir}")
+
+    # اكتشاف الملفات
+    logger.info("جاري البحث عن الملفات الصوتية...")
+    files = find_audio_files(args.inputs, args.recursive, args.max_size_mb)
+    
+    if not files:
+        logger.error("❌ لم يتم العثور على ملفات صوتية للمعالجة.")
+        sys.exit(1)
+
+    # تحميل النموذج
+    logger.info(f"جارٍ تحميل نموذج whisper: {args.model} ...")
+    try:
+        model = whisper.load_model(args.model, device=args.device)
+        logger.info(f"✅ تم تحميل النموذج بنجاح على الجهاز: {next(model.parameters()).device}")
+        
+    except Exception as e:
+        logger.error(f"❌ فشل تحميل النموذج: {e}")
+        sys.exit(1)
+
+    # معالجة الملفات
+    logger.info(f"🎯 تم العثور على {len(files)} ملف/ملفات — البدء بالتحويل...")
+    
+    # إعداد معلمات التحويل
+    transcribe_kwargs = {
+        'lang': args.lang,
+        'output_format': args.output,
+        'verbose': args.verbose,
+        'output_dir': output_dir
+    }
+    
+    # معالجة الملفات مع التقدم
+    start_time = time.time()
+    stats = process_files_with_progress(model, files, **transcribe_kwargs)
+    total_time = time.time() - start_time
+
+    # عرض التقرير النهائي
+    print("\n" + "="*50)
+    print("📊 تقرير التحويل النهائي")
+    print("="*50)
+    print(f"✅ الملفات الناجحة: {stats['successful']}")
+    print(f"❌ الملفات الفاشلة: {stats['failed']}")
+    print(f"📁 إجمالي الملفات: {stats['total_files']}")
+    print(f"🎯 معدل النجاح: {stats['success_rate']:.1f}%")
+    print(f"📝 إجمالي النص المنتج: {stats['total_text_length']} حرف")
+    print(f"⏱️ وقت المعالجة الإجمالي: {total_time:.1f} ثانية")
+    print(f"⚡ متوسط الوقت للملف: {total_time/len(files):.1f} ثانية")
+    
+    if output_dir:
+        print(f"📂 مجلد الإخراج: {output_dir}")
+    
+    print("="*50)
+    
+    if stats['failed'] > 0:
+        logger.warning(f"بعض الملفات فشلت في المعالجة. راجع السجلات للتفاصيل.")
+        sys.exit(1)
+    else:
+        logger.info("🎉 اكتملت جميع الملفات بنجاح!")
+        
+    # حفظ التقرير
+    if output_dir:
+        report_path = output_dir / "transcription_report.json"
+        report_data = {
+            'summary': stats,
+            'timestamp': datetime.now().isoformat(),
+            'parameters': {
+                'model': args.model,
+                'language': args.lang,
+                'output_format': args.output,
+                'total_files': len(files)
+            }
+        }
+        safe_write_text(report_path, json.dumps(report_data, ensure_ascii=False, indent=2))
+        logger.info(f"📄 تم حفظ التقرير في: {report_path}")
+
+if __name__ == '__main__':
+    main()