Upload 4 files

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+# 시스템 패키지 설치
+RUN apt-get update && apt-get install -y \
+    ffmpeg \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+# Python 패키지 설치
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 앱 복사
+COPY app.py .
+
+# 포트 노출
+EXPOSE 7860
+
+# 실행
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,11 +1,145 @@
 ---
-title: SpeechlibProject
-emoji: 👁
-colorFrom: purple
-colorTo: green
+title: Speechlib API
+emoji: 🎤
+colorFrom: blue
+colorTo: purple
 sdk: docker
+app_file: app.py
 pinned: false
-license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Speechlib REST API (ECAPA-TDNN)
+
+화자 분리(Speaker Diarization) + 화자 식별(Speaker Identification) + 음성 인식(STT) REST API
+
+## Features
+
+- **화자 분리**: pyannote/speaker-diarization-3.1로 여러 화자 구분
+- **화자 식별**: speechbrain ECAPA-TDNN으로 등록된 화자 식별 (고정밀)
+- **음성 인식**: faster-whisper (large-v3-turbo)를 사용한 STT
+
+## API Endpoints
+
+### GET /
+API 상태 확인
+
+### GET /health
+헬스 체크
+
+### POST /transcribe
+단순 STT + 화자 분리 (화자 식별 없음)
+
+**Parameters (multipart/form-data):**
+- `audio`: 오디오 파일 (필수)
+- `language`: 언어 코드 (기본값: ko)
+- `hf_token`: HuggingFace 토큰 (필수)
+
+### POST /process
+전체 기능: 화자 분리 + 화자 식별 + STT
+
+**Parameters (multipart/form-data):**
+- `audio`: 분석할 오디오 파일 (필수)
+- `voice_sample`: 화자 샘플 파일 (선택)
+- `speaker_name`: 식별할 화자 이름 (기본값: speaker)
+- `language`: 언어 코드 (기본값: ko)
+- `hf_token`: HuggingFace 토큰 (필수)
+
+## Usage Example
+
+### cURL
+
+```bash
+# 단순 STT
+curl -X POST "https://YOUR_SPACE.hf.space/transcribe" \
+  -F "audio=@audio.wav" \
+  -F "language=ko" \
+  -F "hf_token=hf_YOUR_TOKEN"
+
+# 화자 식별 포함
+curl -X POST "https://YOUR_SPACE.hf.space/process" \
+  -F "audio=@conversation.wav" \
+  -F "voice_sample=@speaker_sample.wav" \
+  -F "speaker_name=홍길동" \
+  -F "language=ko" \
+  -F "hf_token=hf_YOUR_TOKEN"
+```
+
+### Python
+
+```python
+import requests
+
+# 단순 STT
+response = requests.post(
+    "https://YOUR_SPACE.hf.space/transcribe",
+    files={"audio": open("audio.wav", "rb")},
+    data={"language": "ko", "hf_token": "hf_YOUR_TOKEN"}
+)
+print(response.json())
+
+# 화자 식별 포함
+response = requests.post(
+    "https://YOUR_SPACE.hf.space/process",
+    files={
+        "audio": open("conversation.wav", "rb"),
+        "voice_sample": open("speaker_sample.wav", "rb")
+    },
+    data={
+        "speaker_name": "홍길동",
+        "language": "ko",
+        "hf_token": "hf_YOUR_TOKEN"
+    }
+)
+print(response.json())
+```
+
+### JavaScript/Node.js
+
+```javascript
+const FormData = require('form-data');
+const fs = require('fs');
+const axios = require('axios');
+
+const form = new FormData();
+form.append('audio', fs.createReadStream('audio.wav'));
+form.append('language', 'ko');
+form.append('hf_token', 'hf_YOUR_TOKEN');
+
+const response = await axios.post(
+  'https://YOUR_SPACE.hf.space/transcribe',
+  form,
+  { headers: form.getHeaders() }
+);
+console.log(response.data);
+```
+
+## Response Format
+
+```json
+{
+  "success": true,
+  "segments": [
+    {
+      "start": 0.0,
+      "end": 2.5,
+      "text": "안녕하세요",
+      "speaker": "홍길동",
+      "similarity": 85.3
+    }
+  ],
+  "speaker_stats": {
+    "홍길동": {
+      "count": 10,
+      "duration": 45.5
+    }
+  },
+  "total_segments": 20
+}
+```
+
+## Notes
+
+- ECAPA-TDNN은 유사도 임계값 25% 이상일 때 화자 매칭
+- GPU 사용 가능 시 자동으로 GPU 활용
+- 지원 오디오 포맷: wav, mp3, m4a, ogg, flac, aac
+- API 문서: https://YOUR_SPACE.hf.space/docs

app.py

@@ -0,0 +1,528 @@
+"""
+Speechlib REST API - HuggingFace Spaces (ECAPA-TDNN 버전)
+화자 분리 + 화자 식별 + STT
+"""
+import os
+import tempfile
+import json
+import numpy as np
+import shutil
+from typing import List, Dict, Optional
+from contextlib import asynccontextmanager
+
+# 환경 설정
+os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"
+os.environ["HF_HUB_DISABLE_SYMLINKS"] = "1"
+
+import torch
+
+# PyTorch 호환성 패치 (버전에 따라 분기)
+if hasattr(torch.serialization, 'add_safe_globals'):
+    torch.serialization.add_safe_globals([torch.torch_version.TorchVersion])
+    from pyannote.audio.core import task as pyannote_task
+    from pyannote.audio.core.io import Audio
+    torch.serialization.add_safe_globals([
+        pyannote_task.Specifications,
+        pyannote_task.Problem,
+        pyannote_task.Resolution,
+        Audio
+    ])
+
+# weights_only=False 패치
+original_load = torch.load
+def patched_load(*args, **kwargs):
+    if 'weights_only' not in kwargs:
+        kwargs['weights_only'] = False
+    return original_load(*args, **kwargs)
+torch.load = patched_load
+
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.responses import JSONResponse
+import uvicorn
+import torchaudio
+from pydub import AudioSegment
+
+
+class SpeakerPipelineECAPA:
+    """
+    ECAPA-TDNN 임베딩을 사용한 화자 식별 파이프라인
+    """
+
+    def __init__(
+        self,
+        hf_token: str,
+        whisper_model: str = "large-v3-turbo",
+        similarity_threshold: float = 0.25,
+        device: str = None
+    ):
+        self.hf_token = hf_token
+        self.whisper_model_size = whisper_model
+        self.similarity_threshold = similarity_threshold
+
+        # GPU 사용 가능하면 GPU, 아니면 CPU
+        if device is None:
+            self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        else:
+            self.device = device
+
+        self.registered_speakers: Dict[str, np.ndarray] = {}
+
+        # 모델들 (lazy loading)
+        self._diarization_pipeline = None
+        self._ecapa_model = None
+        self._whisper_model = None
+
+        print(f"[SpeakerPipeline ECAPA-TDNN] 초기화")
+        print(f"  - Device: {self.device}")
+        print(f"  - 임계값: {similarity_threshold}")
+
+    @property
+    def diarization_pipeline(self):
+        if self._diarization_pipeline is None:
+            print("[로딩] pyannote/speaker-diarization-3.1...")
+            from pyannote.audio import Pipeline
+            self._diarization_pipeline = Pipeline.from_pretrained(
+                "pyannote/speaker-diarization-3.1",
+                use_auth_token=self.hf_token
+            )
+            self._diarization_pipeline.to(torch.device(self.device))
+        return self._diarization_pipeline
+
+    @property
+    def ecapa_model(self):
+        if self._ecapa_model is None:
+            print("[로딩] speechbrain ECAPA-TDNN...")
+            from speechbrain.inference.speaker import EncoderClassifier
+            self._ecapa_model = EncoderClassifier.from_hparams(
+                source="speechbrain/spkrec-ecapa-voxceleb",
+                savedir="pretrained_models/spkrec-ecapa-voxceleb",
+                run_opts={"device": self.device}
+            )
+        return self._ecapa_model
+
+    @property
+    def whisper_model(self):
+        if self._whisper_model is None:
+            print(f"[로딩] faster-whisper {self.whisper_model_size}...")
+            from faster_whisper import WhisperModel
+            compute_type = "float16" if self.device == "cuda" else "int8"
+            self._whisper_model = WhisperModel(
+                self.whisper_model_size,
+                device=self.device,
+                compute_type=compute_type
+            )
+        return self._whisper_model
+
+    def _load_audio(self, audio_path: str) -> tuple:
+        """오디오 로드 및 전처리"""
+        ext = os.path.splitext(audio_path)[1].lower()
+
+        if ext in ['.m4a', '.mp4', '.aac', '.ogg', '.flac', '.mp3']:
+            audio = AudioSegment.from_file(audio_path)
+            audio = audio.set_channels(1).set_frame_rate(16000)
+            with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp:
+                tmp_path = tmp.name
+            audio.export(tmp_path, format='wav')
+            waveform, sample_rate = torchaudio.load(tmp_path)
+            os.unlink(tmp_path)
+        else:
+            waveform, sample_rate = torchaudio.load(audio_path)
+
+        if waveform.shape[0] > 1:
+            waveform = waveform.mean(dim=0, keepdim=True)
+
+        if sample_rate != 16000:
+            resampler = torchaudio.transforms.Resample(sample_rate, 16000)
+            waveform = resampler(waveform)
+            sample_rate = 16000
+
+        return waveform, sample_rate
+
+    def get_embedding_ecapa(self, waveform: torch.Tensor) -> np.ndarray:
+        """ECAPA-TDNN으로 임베딩 추출"""
+        if waveform.dim() == 2:
+            waveform = waveform.squeeze(0)
+
+        waveform = waveform.to(self.device)
+
+        with torch.no_grad():
+            embedding = self.ecapa_model.encode_batch(waveform.unsqueeze(0))
+
+        return embedding.squeeze().cpu().numpy()
+
+    def register_speaker(self, name: str, audio_paths: List[str]) -> None:
+        """화자 등록"""
+        print(f"\n[화자 등록] {name} ({len(audio_paths)}개 샘플)")
+        embeddings = []
+
+        for path in audio_paths:
+            if not os.path.exists(path):
+                continue
+
+            try:
+                waveform, sr = self._load_audio(path)
+                emb = self.get_embedding_ecapa(waveform)
+                emb = emb / np.linalg.norm(emb)
+                embeddings.append(emb)
+                print(f"  ✓ {os.path.basename(path)}")
+            except Exception as e:
+                print(f"  ✗ 에러({os.path.basename(path)}): {e}")
+
+        if not embeddings:
+            print(f"  [경고] 유효한 샘플이 없습니다!")
+            return
+
+        avg_embedding = np.mean(embeddings, axis=0)
+        avg_embedding = avg_embedding / np.linalg.norm(avg_embedding)
+        self.registered_speakers[name] = avg_embedding
+        print(f"[화자 등록] {name} 완료!")
+
+    def process(self, audio_path: str, language: str = "ko") -> List[Dict]:
+        """메인 처리 함수"""
+        print(f"\n[처리 시작] {os.path.basename(audio_path)}")
+
+        waveform, sample_rate = self._load_audio(audio_path)
+        audio_dict = {"waveform": waveform, "sample_rate": sample_rate}
+
+        # 1. 화자 분리
+        print("[1/3] 화자 분리 중...")
+        raw_diarization = self.diarization_pipeline(audio_dict)
+
+        diarization = None
+        if hasattr(raw_diarization, "itertracks"):
+            diarization = raw_diarization
+        else:
+            for attr in dir(raw_diarization):
+                if attr.startswith("_"): continue
+                try:
+                    val = getattr(raw_diarization, attr)
+                    if hasattr(val, "itertracks"):
+                        diarization = val
+                        break
+                except: pass
+
+        if diarization is None:
+            raise RuntimeError("화자 분리 결과를 파싱할 수 없습니다.")
+
+        segments = []
+        for turn, _, speaker in diarization.itertracks(yield_label=True):
+            segments.append({
+                "start": turn.start,
+                "end": turn.end,
+                "diarization_speaker": speaker
+            })
+        print(f"  → {len(segments)}개 세그먼트 감지")
+
+        # 2. 화자 식별 (ECAPA-TDNN)
+        if self.registered_speakers:
+            print("[2/3] 화자 식별 중 (ECAPA-TDNN)...")
+
+            speaker_embeddings = {}
+            speakers_found = set(seg["diarization_speaker"] for seg in segments)
+
+            for spk in speakers_found:
+                spk_embs = []
+                for seg in segments:
+                    if seg["diarization_speaker"] != spk:
+                        continue
+
+                    duration = seg["end"] - seg["start"]
+                    if duration < 0.5:
+                        continue
+
+                    try:
+                        start_sample = int(seg["start"] * sample_rate)
+                        end_sample = int(seg["end"] * sample_rate)
+                        end_sample = min(end_sample, waveform.shape[1])
+                        seg_waveform = waveform[:, start_sample:end_sample]
+
+                        if seg_waveform.shape[1] < sample_rate * 0.3:
+                            continue
+
+                        emb = self.get_embedding_ecapa(seg_waveform)
+                        emb = emb / np.linalg.norm(emb)
+                        spk_embs.append(emb)
+                    except:
+                        pass
+
+                if spk_embs:
+                    speaker_embeddings[spk] = spk_embs
+
+            # 화자 매핑
+            speaker_mapping = {}
+            speaker_scores = {}
+
+            for spk, embs in speaker_embeddings.items():
+                avg_emb = np.mean(embs, axis=0)
+                avg_emb = avg_emb / np.linalg.norm(avg_emb)
+
+                speaker_scores[spk] = {}
+                for name, ref_emb in self.registered_speakers.items():
+                    sim = np.dot(avg_emb, ref_emb)
+                    speaker_scores[spk][name] = sim
+
+            # 경쟁 매칭
+            for reg_name in self.registered_speakers.keys():
+                best_spk = None
+                best_sim = -1
+
+                for spk in speaker_scores.keys():
+                    if spk in [m[0] for m in speaker_mapping.values() if m[0] != spk]:
+                        continue
+
+                    sim = speaker_scores[spk].get(reg_name, -1)
+                    if sim > best_sim:
+                        best_sim = sim
+                        best_spk = spk
+
+                if best_spk and best_sim >= self.similarity_threshold:
+                    speaker_mapping[best_spk] = (reg_name, best_sim)
+
+            for spk in speaker_scores.keys():
+                if spk not in speaker_mapping:
+                    speaker_mapping[spk] = (spk, 0.0)
+
+            for seg in segments:
+                d_spk = seg["diarization_speaker"]
+                if d_spk in speaker_mapping:
+                    seg["speaker"], seg["similarity"] = speaker_mapping[d_spk]
+                else:
+                    seg["speaker"] = d_spk
+                    seg["similarity"] = 0.0
+        else:
+            for seg in segments:
+                seg["speaker"] = seg["diarization_speaker"]
+                seg["similarity"] = 0.0
+
+        # 3. STT
+        print("[3/3] 음성 인식(STT) 중...")
+        whisper_segs, _ = self.whisper_model.transcribe(
+            audio_path, language=language, beam_size=5, vad_filter=True
+        )
+        whisper_results = [{"start": s.start, "end": s.end, "text": s.text.strip()} for s in whisper_segs]
+
+        # 4. 병합
+        final_results = []
+        for w_seg in whisper_results:
+            best_speaker = "Unknown"
+            best_overlap = 0
+            best_sim = 0.0
+
+            for d_seg in segments:
+                overlap = max(0, min(w_seg["end"], d_seg["end"]) - max(w_seg["start"], d_seg["start"]))
+                if overlap > best_overlap:
+                    best_overlap = overlap
+                    best_speaker = d_seg["speaker"]
+                    best_sim = d_seg.get("similarity", 0.0)
+
+            final_results.append({
+                "start": w_seg["start"],
+                "end": w_seg["end"],
+                "text": w_seg["text"],
+                "speaker": best_speaker,
+                "similarity": round(best_sim * 100, 1)
+            })
+
+        return final_results
+
+
+# 전역 파이프라인 인스턴스
+_pipeline: Optional[SpeakerPipelineECAPA] = None
+
+
+def get_pipeline(hf_token: str) -> SpeakerPipelineECAPA:
+    """파이프라인 싱글톤 인스턴스 반환"""
+    global _pipeline
+    if _pipeline is None:
+        _pipeline = SpeakerPipelineECAPA(hf_token=hf_token)
+    return _pipeline
+
+
+# FastAPI 앱
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # 시작 시
+    print("🚀 Speechlib API 서버 시작")
+    yield
+    # 종료 시
+    print("👋 Speechlib API 서버 종료")
+
+
+app = FastAPI(
+    title="Speechlib API",
+    description="화자 분리 + 화자 식별 + STT REST API (ECAPA-TDNN)",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+
+@app.get("/")
+async def root():
+    """API 상태 확인"""
+    return {
+        "status": "ok",
+        "message": "Speechlib API (ECAPA-TDNN)",
+        "endpoints": {
+            "/transcribe": "POST - 단순 STT + 화자 분리",
+            "/process": "POST - 전체 기능 (화자 식별 포함)"
+        }
+    }
+
+
+@app.get("/health")
+async def health_check():
+    """헬스 체크"""
+    return {"status": "healthy", "device": "cuda" if torch.cuda.is_available() else "cpu"}
+
+
+@app.post("/transcribe")
+async def transcribe(
+    audio: UploadFile = File(..., description="오디오 파일"),
+    language: str = Form(default="ko", description="언어 코드 (ko, en, ja, zh)"),
+    hf_token: str = Form(..., description="HuggingFace 토큰")
+):
+    """
+    단순 STT + 화자 분리 (화자 식별 없음)
+    """
+    temp_dir = None
+    try:
+        # 임시 파일 저장
+        temp_dir = tempfile.mkdtemp()
+        audio_path = os.path.join(temp_dir, audio.filename)
+
+        with open(audio_path, "wb") as f:
+            content = await audio.read()
+            f.write(content)
+
+        # 파이프라인 실행
+        pipeline = get_pipeline(hf_token)
+        pipeline.registered_speakers.clear()  # 화자 식별 없음
+
+        results = pipeline.process(audio_path, language=language)
+
+        # 결과 포맷팅
+        segments = []
+        speaker_stats = {}
+
+        for r in results:
+            segments.append({
+                "start": round(r["start"], 2),
+                "end": round(r["end"], 2),
+                "text": r["text"],
+                "speaker": r["speaker"]
+            })
+
+            speaker = r["speaker"]
+            if speaker not in speaker_stats:
+                speaker_stats[speaker] = {"count": 0, "duration": 0}
+            speaker_stats[speaker]["count"] += 1
+            speaker_stats[speaker]["duration"] += r["end"] - r["start"]
+
+        for speaker in speaker_stats:
+            speaker_stats[speaker]["duration"] = round(speaker_stats[speaker]["duration"], 2)
+
+        return JSONResponse(content={
+            "success": True,
+            "segments": segments,
+            "speaker_stats": speaker_stats,
+            "total_segments": len(segments)
+        })
+
+    except Exception as e:
+        import traceback
+        return JSONResponse(
+            status_code=500,
+            content={
+                "success": False,
+                "error": str(e),
+                "traceback": traceback.format_exc()
+            }
+        )
+    finally:
+        if temp_dir and os.path.exists(temp_dir):
+            shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+@app.post("/process")
+async def process_audio(
+    audio: UploadFile = File(..., description="분석할 오디오 파일"),
+    voice_sample: UploadFile = File(default=None, description="화자 샘플 파일 (선택)"),
+    speaker_name: str = Form(default="speaker", description="식별할 화자 이름"),
+    language: str = Form(default="ko", description="언어 코드 (ko, en, ja, zh)"),
+    hf_token: str = Form(..., description="HuggingFace 토큰")
+):
+    """
+    전체 기능: 화자 분리 + 화자 식별 + STT
+    """
+    temp_dir = None
+    try:
+        # 임시 디렉토리 생성
+        temp_dir = tempfile.mkdtemp()
+
+        # 메인 오디오 저장
+        audio_path = os.path.join(temp_dir, audio.filename)
+        with open(audio_path, "wb") as f:
+            content = await audio.read()
+            f.write(content)
+
+        # 파이프라인 가져오기
+        pipeline = get_pipeline(hf_token)
+        pipeline.registered_speakers.clear()
+
+        # 화자 샘플이 있으면 등록
+        if voice_sample and voice_sample.filename:
+            sample_path = os.path.join(temp_dir, voice_sample.filename)
+            with open(sample_path, "wb") as f:
+                sample_content = await voice_sample.read()
+                f.write(sample_content)
+            pipeline.register_speaker(speaker_name, [sample_path])
+
+        # 처리
+        results = pipeline.process(audio_path, language=language)
+
+        # 결과 포맷팅
+        segments = []
+        speaker_stats = {}
+
+        for r in results:
+            segments.append({
+                "start": round(r["start"], 2),
+                "end": round(r["end"], 2),
+                "text": r["text"],
+                "speaker": r["speaker"],
+                "similarity": r["similarity"]
+            })
+
+            speaker = r["speaker"]
+            if speaker not in speaker_stats:
+                speaker_stats[speaker] = {"count": 0, "duration": 0}
+            speaker_stats[speaker]["count"] += 1
+            speaker_stats[speaker]["duration"] += r["end"] - r["start"]
+
+        for speaker in speaker_stats:
+            speaker_stats[speaker]["duration"] = round(speaker_stats[speaker]["duration"], 2)
+
+        return JSONResponse(content={
+            "success": True,
+            "segments": segments,
+            "speaker_stats": speaker_stats,
+            "total_segments": len(segments)
+        })
+
+    except Exception as e:
+        import traceback
+        return JSONResponse(
+            status_code=500,
+            content={
+                "success": False,
+                "error": str(e),
+                "traceback": traceback.format_exc()
+            }
+        )
+    finally:
+        if temp_dir and os.path.exists(temp_dir):
+            shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -0,0 +1,12 @@
+fastapi>=0.100.0
+uvicorn>=0.23.0
+python-multipart>=0.0.6
+torch==2.4.0
+torchaudio==2.4.0
+pyannote.audio==3.3.2
+speechbrain==1.0.0
+faster-whisper>=1.0.0
+pydub>=0.25.1
+numpy<2.0.0
+ffmpeg-python
+soundfile