Spaces:
Running
Running
| from fastapi import APIRouter, UploadFile, File, Depends, HTTPException, Query | |
| from fastapi.responses import JSONResponse | |
| from preprocessing import clean_audio | |
| from asr import transcribe_file, MedicalASR | |
| from diarization import diarize_segments | |
| from privacy import MedicalPrivacyProcessor | |
| from config import settings | |
| from typing import Optional | |
| import tempfile, os, uuid | |
| from fastapi.concurrency import run_in_threadpool | |
| router = APIRouter() | |
| # Gizlilik işlemcisi | |
| privacy_processor = MedicalPrivacyProcessor() | |
| def get_asr_model(): | |
| """ASR modelini oluşturur ve döndürür""" | |
| config = { | |
| "language": "tr", | |
| "model": settings.ASR_MODEL, | |
| "domain": "medical" | |
| } | |
| return MedicalASR(config) | |
| async def transcribe_audio( | |
| file: UploadFile = File(...), | |
| diarize: bool = True, | |
| enhance_audio: bool = True, | |
| anonymize: Optional[bool] = Query(None, description="Kişisel verileri anonimleştir"), | |
| asr_model: MedicalASR = Depends(get_asr_model) | |
| ): | |
| """Ses dosyasını transkribe eder""" | |
| try: | |
| # Geçici dosya oluştur | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file: | |
| temp_file.write(await file.read()) | |
| temp_file_path = temp_file.name | |
| # CPU/GPU yoğun işlemi thread pool'da çalıştır | |
| result = await run_in_threadpool( | |
| asr_model.transcribe, | |
| temp_file_path, | |
| speaker_diarization=diarize, | |
| enhance_audio=enhance_audio | |
| ) | |
| # Geçici dosyayı temizle | |
| os.unlink(temp_file_path) | |
| # Kişisel veri anonimleştirme | |
| # Eğer açıkça belirtilmediyse uygulama konfigürasyonuna göre davran | |
| should_anonymize = anonymize if anonymize is not None else app.state.asr_config.get("anonymize_data", True) | |
| identified_data = {} | |
| if should_anonymize: | |
| # Ana metni anonimleştir | |
| anonymized_text, main_data = privacy_processor.anonymize_text(result["text"]) | |
| result["text"] = anonymized_text | |
| identified_data.update(main_data) | |
| # Diyarizasyon segmentlerini anonimleştir | |
| if "diarization" in result: | |
| for segment in result["diarization"]: | |
| if segment["text"]: | |
| anonymized_segment, segment_data = privacy_processor.anonymize_text(segment["text"]) | |
| segment["text"] = anonymized_segment | |
| # Her segment için tespit edilen verileri güncelle | |
| for key, values in segment_data.items(): | |
| identified_data.setdefault(key, []).extend(values) | |
| # Tespit edilen verileri güvenli şekilde sakla | |
| # NOT: Gerçek uygulamada bu verileri şifreli bir veritabanında saklayın | |
| session_id = str(uuid.uuid4()) | |
| # Burada verileri güvenli bir şekilde saklama kodunuz olacak | |
| # Yanıtta tespit edilen veri tipleri hakkında bilgi ver | |
| result["anonymized"] = True | |
| result["identified_data_types"] = {k: len(v) for k, v in identified_data.items() if v} | |
| result["session_id"] = session_id | |
| return JSONResponse(content=result) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=f"Transkripsiyon hatası: {str(e)}") | |