#!/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 @app.on_event("startup") 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("تم تهيئة معالج النصوص المبسط") @app.get("/", response_class=HTMLResponse) async def root(): """الصفحة الرئيسية""" html_content = """ Whisper Text Processor API

🎤 Whisper Text Processor API

API متكامل لمعالجة النصوص باستخدام Whisper مع التصحيح التلقائي

📋 نقاط النهاية المتاحة:

GET /health - فحص حالة الخادم
POST /correct-text - تصحيح النص باستخدام القاموس المرجعي
POST /compare-texts - مقارنة نصين وحساب معدلات الخطأ
POST /process-audio - معالجة شاملة للملف الصوتي مع التصحيح
GET /docs - وثائق API التفاعلية (Swagger)

🚀 مثال على الاستخدام:

curl -X POST "http://localhost:8000/correct-text" \\ -H "Content-Type: application/json" \\ -d '{ "text": "يحتاز غشاء الطبل", "reference_texts": ["يهتز غشاء الطبل تنتقل عظيمات السمع"] }'
""" return html_content @app.get("/health") async def health_check(): """فحص حالة الخادم""" return { "status": "healthy", "message": "خادم معالج النصوص يعمل بشكل طبيعي", "processor_loaded": processor is not None } @app.post("/correct-text") 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)}") @app.post("/compare-texts") 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)}") @app.post("/process-audio") 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)}") @app.post("/generate-report") 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""" تقرير معالجة النصوص

تقرير معالجة النصوص

النص المرجعي:

{reference_text}

نتائج المعالجة:

{json.dumps(processing_result.data, ensure_ascii=False, indent=2)}
""" 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" )