Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| خادم API لمعالج النصوص المتكامل باستخدام Whisper | |
| """ | |
| from fastapi import FastAPI, File, UploadFile, Form, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from pydantic import BaseModel | |
| from typing import List, Optional | |
| import tempfile | |
| import os | |
| import logging | |
| import json | |
| from whisper_text_processor import WhisperTextProcessor | |
| # إعداد التسجيل | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # إنشاء تطبيق FastAPI | |
| app = FastAPI( | |
| title="Whisper Text Processor API", | |
| description="API متكامل لمعالجة النصوص باستخدام Whisper مع التصحيح التلقائي ومقارنة النصوص", | |
| version="1.0.0", | |
| docs_url="/docs", | |
| redoc_url="/redoc" | |
| ) | |
| # إعداد CORS للسماح بالوصول من أي مصدر | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # متغير عام لمعالج النصوص | |
| processor = None | |
| # نماذج البيانات | |
| class TextCorrectionRequest(BaseModel): | |
| text: str | |
| reference_texts: List[str] | |
| max_distance: Optional[int] = 3 | |
| threshold_freq: Optional[int] = 1 | |
| class TextComparisonRequest(BaseModel): | |
| reference_text: str | |
| transcribed_text: str | |
| class ProcessingResponse(BaseModel): | |
| success: bool | |
| message: str | |
| data: Optional[dict] = None | |
| async def startup_event(): | |
| """تهيئة المعالج عند بدء التشغيل""" | |
| global processor | |
| try: | |
| logger.info("تهيئة معالج النصوص...") | |
| processor = WhisperTextProcessor() | |
| logger.info("تم تهيئة معالج النصوص بنجاح") | |
| except Exception as e: | |
| logger.error(f"فشل في تهيئة معالج النصوص: {str(e)}") | |
| # استخدام معالج مبسط في حالة فشل تحميل Whisper | |
| from test_processor_simple import SimpleTextProcessor | |
| processor = SimpleTextProcessor() | |
| logger.info("تم تهيئة معالج النصوص المبسط") | |
| async def root(): | |
| """الصفحة الرئيسية""" | |
| html_content = """ | |
| <!DOCTYPE html> | |
| <html dir="rtl" lang="ar"> | |
| <head> | |
| <meta charset="UTF-8"> | |
| <meta name="viewport" content="width=device-width, initial-scale=1.0"> | |
| <title>Whisper Text Processor API</title> | |
| <style> | |
| body { font-family: 'Arial', sans-serif; margin: 40px; background: #f5f5f5; } | |
| .container { max-width: 800px; margin: 0 auto; background: white; padding: 30px; border-radius: 10px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); } | |
| .header { text-align: center; color: #2c3e50; margin-bottom: 30px; } | |
| .endpoint { background: #ecf0f1; padding: 15px; margin: 15px 0; border-radius: 5px; } | |
| .method { display: inline-block; padding: 5px 10px; border-radius: 3px; color: white; font-weight: bold; } | |
| .get { background: #27ae60; } | |
| .post { background: #3498db; } | |
| .code { background: #2c3e50; color: white; padding: 10px; border-radius: 5px; font-family: monospace; } | |
| </style> | |
| </head> | |
| <body> | |
| <div class="container"> | |
| <div class="header"> | |
| <h1>🎤 Whisper Text Processor API</h1> | |
| <p>API متكامل لمعالجة النصوص باستخدام Whisper مع التصحيح التلقائي</p> | |
| </div> | |
| <h2>📋 نقاط النهاية المتاحة:</h2> | |
| <div class="endpoint"> | |
| <span class="method get">GET</span> | |
| <strong>/health</strong> - فحص حالة الخادم | |
| </div> | |
| <div class="endpoint"> | |
| <span class="method post">POST</span> | |
| <strong>/correct-text</strong> - تصحيح النص باستخدام القاموس المرجعي | |
| </div> | |
| <div class="endpoint"> | |
| <span class="method post">POST</span> | |
| <strong>/compare-texts</strong> - مقارنة نصين وحساب معدلات الخطأ | |
| </div> | |
| <div class="endpoint"> | |
| <span class="method post">POST</span> | |
| <strong>/process-audio</strong> - معالجة شاملة للملف الصوتي مع التصحيح | |
| </div> | |
| <div class="endpoint"> | |
| <span class="method get">GET</span> | |
| <strong>/docs</strong> - وثائق API التفاعلية (Swagger) | |
| </div> | |
| <h2>🚀 مثال على الاستخدام:</h2> | |
| <div class="code"> | |
| curl -X POST "http://localhost:8000/correct-text" \\ | |
| -H "Content-Type: application/json" \\ | |
| -d '{ | |
| "text": "يحتاز غشاء الطبل", | |
| "reference_texts": ["يهتز غشاء الطبل تنتقل عظيمات السمع"] | |
| }' | |
| </div> | |
| </div> | |
| </body> | |
| </html> | |
| """ | |
| return html_content | |
| async def health_check(): | |
| """فحص حالة الخادم""" | |
| return { | |
| "status": "healthy", | |
| "message": "خادم معالج النصوص يعمل بشكل طبيعي", | |
| "processor_loaded": processor is not None | |
| } | |
| async def correct_text(request: TextCorrectionRequest): | |
| """تصحيح النص باستخدام القاموس المرجعي""" | |
| try: | |
| if not processor: | |
| raise HTTPException(status_code=500, detail="معالج النصوص غير متاح") | |
| # إنشاء القاموس من النصوص المرجعية | |
| processor.create_dictionary(request.reference_texts) | |
| # تصحيح النص | |
| corrected_text, corrections = processor.correct_text( | |
| request.text, | |
| max_distance=request.max_distance, | |
| threshold_freq=request.threshold_freq | |
| ) | |
| return ProcessingResponse( | |
| success=True, | |
| message="تم تصحيح النص بنجاح", | |
| data={ | |
| "original_text": request.text, | |
| "corrected_text": corrected_text, | |
| "corrections": corrections, | |
| "corrections_count": len(corrections) | |
| } | |
| ) | |
| except Exception as e: | |
| logger.error(f"خطأ في تصحيح النص: {str(e)}") | |
| raise HTTPException(status_code=500, detail=f"خطأ في تصحيح النص: {str(e)}") | |
| async def compare_texts(request: TextComparisonRequest): | |
| """مقارنة نصين وحساب معدلات الخطأ""" | |
| try: | |
| if not processor: | |
| raise HTTPException(status_code=500, detail="معالج النصوص غير متاح") | |
| # مقارنة النصوص | |
| comparison_result = processor.compare_texts( | |
| request.reference_text, | |
| request.transcribed_text | |
| ) | |
| return ProcessingResponse( | |
| success=True, | |
| message="تم مقارنة النصوص بنجاح", | |
| data=comparison_result | |
| ) | |
| except Exception as e: | |
| logger.error(f"خطأ في مقارنة النصوص: {str(e)}") | |
| raise HTTPException(status_code=500, detail=f"خطأ في مقارنة النصوص: {str(e)}") | |
| async def process_audio( | |
| audio_file: UploadFile = File(...), | |
| reference_text: str = Form(...), | |
| reference_texts: Optional[str] = Form(None), | |
| max_distance: Optional[int] = Form(3), | |
| threshold_freq: Optional[int] = Form(1) | |
| ): | |
| """معالجة شاملة للملف الصوتي مع التصحيح والمقارنة""" | |
| try: | |
| if not processor: | |
| raise HTTPException(status_code=500, detail="معالج النصوص غير متاح") | |
| # التحقق من نوع الملف | |
| if not audio_file.content_type.startswith('audio/'): | |
| raise HTTPException(status_code=400, detail="يجب أن يكون الملف من نوع صوتي") | |
| # حفظ الملف الصوتي مؤقتاً | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file: | |
| content = await audio_file.read() | |
| temp_file.write(content) | |
| temp_audio_path = temp_file.name | |
| try: | |
| # تحضير النصوص المرجعية | |
| ref_texts = [reference_text] | |
| if reference_texts: | |
| try: | |
| additional_texts = json.loads(reference_texts) | |
| if isinstance(additional_texts, list): | |
| ref_texts.extend(additional_texts) | |
| except json.JSONDecodeError: | |
| # إذا لم يكن JSON صالح، نعتبره نص واحد | |
| ref_texts.append(reference_texts) | |
| # معالجة شاملة للصوت | |
| if hasattr(processor, 'process_audio_with_correction'): | |
| results = processor.process_audio_with_correction( | |
| temp_audio_path, | |
| reference_text, | |
| ref_texts | |
| ) | |
| else: | |
| # استخدام معالج مبسط | |
| results = { | |
| "message": "معالج مبسط - يتطلب ملف صوتي حقيقي لاختبار Whisper", | |
| "reference_text": reference_text, | |
| "reference_texts": ref_texts | |
| } | |
| return ProcessingResponse( | |
| success=True, | |
| message="تم معالجة الملف الصوتي بنجاح", | |
| data=results | |
| ) | |
| finally: | |
| # حذف الملف المؤقت | |
| if os.path.exists(temp_audio_path): | |
| os.unlink(temp_audio_path) | |
| except Exception as e: | |
| logger.error(f"خطأ في معالجة الملف الصوتي: {str(e)}") | |
| raise HTTPException(status_code=500, detail=f"خطأ في معالجة الملف الصوتي: {str(e)}") | |
| async def generate_report( | |
| audio_file: UploadFile = File(...), | |
| reference_text: str = Form(...), | |
| reference_texts: Optional[str] = Form(None) | |
| ): | |
| """إنشاء تقرير HTML شامل للمعالجة""" | |
| try: | |
| if not processor: | |
| raise HTTPException(status_code=500, detail="معالج النصوص غير متاح") | |
| # معالجة الملف الصوتي أولاً | |
| processing_result = await process_audio( | |
| audio_file, reference_text, reference_texts | |
| ) | |
| if not processing_result.data: | |
| raise HTTPException(status_code=500, detail="فشل في معالجة الملف الصوتي") | |
| # إنشاء تقرير HTML | |
| if hasattr(processor, 'generate_html_report'): | |
| html_report = processor.generate_html_report(processing_result.data) | |
| return HTMLResponse(content=html_report) | |
| else: | |
| # تقرير مبسط | |
| simple_report = f""" | |
| <!DOCTYPE html> | |
| <html dir="rtl" lang="ar"> | |
| <head> | |
| <meta charset="UTF-8"> | |
| <title>تقرير معالجة النصوص</title> | |
| </head> | |
| <body> | |
| <h1>تقرير معالجة النصوص</h1> | |
| <h2>النص المرجعي:</h2> | |
| <p>{reference_text}</p> | |
| <h2>نتائج المعالجة:</h2> | |
| <pre>{json.dumps(processing_result.data, ensure_ascii=False, indent=2)}</pre> | |
| </body> | |
| </html> | |
| """ | |
| return HTMLResponse(content=simple_report) | |
| except Exception as e: | |
| logger.error(f"خطأ في إنشاء التقرير: {str(e)}") | |
| raise HTTPException(status_code=500, detail=f"خطأ في إنشاء التقرير: {str(e)}") | |
| if __name__ == "__main__": | |
| import uvicorn | |
| # تشغيل الخادم | |
| uvicorn.run( | |
| "api_server:app", | |
| host="0.0.0.0", | |
| port=8000, | |
| reload=True, | |
| log_level="info" | |
| ) | |