Spaces:
Running
Running
File size: 3,471 Bytes
41979e6 7f442a1 41979e6 16f9715 41979e6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
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)
@router.post("/transcribe")
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)}")
|