feat(api): sync test API (serve) + opt-in LLM correction + cloudflared tunnel

- api/: FastAPI app, X-API-Key 인증(미설정 시 임시키), 엔진 load-once 풀
  (+transcribe lock), POST /v1/transcribe(multipart, 동기), /health, /v1/system,
  /v1/models. 업로드 임시파일 finally 삭제(프라이버시).
- postprocess/: llm.correct(scripts/llm_correct.py 승격; opt-in·allowlist·감사로그·재시도)
  + rules.normalize(EmbeddingGemma 등 정규화).
- results/formats.py: txt/srt/vtt. connectivity/tunnel.py: cloudflared quick tunnel(Colab).
- cli serve: uvicorn 단일워커 + --tunnel cloudflare; config llm_* 필드;
  pyproject api/queue extra 분리(+python-multipart, dev httpx).

검증: 22 단위테스트(API TestClient·formats·postprocess) + 실서버 e2e
(/health·auth 401·실제 전사(JFK)·SRT·임시파일 삭제). KO 품질은 turbo/large-v3 필요(tiny는 한국어 degenerate).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-08 23:20:01 +09:00
parent 480a36edfe
commit 8f6f8969fd
22 changed files with 744 additions and 28 deletions
+86
View File
@@ -0,0 +1,86 @@
"""API — FastAPI TestClient. 엔진은 monkeypatch(가짜)로 모델 로드 회피."""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
import luke_scribe.api.routes.transcribe as route
from luke_scribe.api.app import create_app
from luke_scribe.config import settings
class _FakeSeg:
def __init__(self, start: float, end: float, text: str) -> None:
self.start = start
self.end = end
self.text = text
class _FakeEngine:
def transcribe(self, _audio, **_kw):
return [_FakeSeg(0.0, 1.0, "안녕 vLLM"), _FakeSeg(1.0, 2.0, "두번째")], SimpleNamespace(
language="ko"
)
@pytest.fixture
def client(monkeypatch):
monkeypatch.setattr(route, "get_engine", lambda *a, **k: _FakeEngine())
monkeypatch.setattr(
route, "probe_media", lambda p: SimpleNamespace(path=p, duration_s=2.0, size_bytes=1234)
)
monkeypatch.setattr(settings, "api_keys", ["testkey"])
return TestClient(create_app())
def _files():
return {"file": ("a.wav", b"RIFF0000WAVE", "audio/wav")}
def test_health(client):
assert client.get("/health").json() == {"status": "ok"}
def test_requires_key(client):
assert client.post("/v1/transcribe", files=_files()).status_code == 401
def test_transcribe_ok(client):
r = client.post(
"/v1/transcribe", files=_files(), headers={"X-API-Key": "testkey"}, data={"language": "ko"}
)
assert r.status_code == 200
body = r.json()
assert body["segments"][0]["text"] == "안녕 vLLM"
assert body["model_used"]
assert body["corrected"] is False
def test_413(client, monkeypatch):
monkeypatch.setattr(
route, "probe_media", lambda p: SimpleNamespace(path=p, duration_s=999999, size_bytes=1)
)
r = client.post("/v1/transcribe", files=_files(), headers={"X-API-Key": "testkey"})
assert r.status_code == 413
def test_srt_format(client):
r = client.post(
"/v1/transcribe",
files=_files(),
headers={"X-API-Key": "testkey"},
data={"response_format": "srt"},
)
assert r.status_code == 200
assert "00:00:00,000 --> 00:00:01,000" in r.text
def test_correct_path(client, monkeypatch):
monkeypatch.setattr(route.llm_correct, "correct", lambda text, **k: text + " [보정]")
r = client.post(
"/v1/transcribe", files=_files(), headers={"X-API-Key": "testkey"}, data={"correct": "true"}
)
assert r.status_code == 200
assert r.json()["corrected"] is True
+25
View File
@@ -0,0 +1,25 @@
"""results.formats — txt/srt/vtt."""
from __future__ import annotations
from luke_scribe.results import formats
SEGS = [
{"start": 0.0, "end": 1.5, "text": "안녕 world"},
{"start": 1.5, "end": 3.0, "text": "두번째"},
]
def test_txt():
assert formats.to_txt(SEGS) == "안녕 world\n두번째"
def test_srt():
out = formats.to_srt(SEGS)
assert "1\n00:00:00,000 --> 00:00:01,500\n안녕 world" in out
assert "2\n00:00:01,500 --> 00:00:03,000\n두번째" in out
def test_vtt():
out = formats.to_vtt(SEGS)
assert out.startswith("WEBVTT")
assert "00:00:00.000 --> 00:00:01.500" in out
+41
View File
@@ -0,0 +1,41 @@
"""postprocess.rules / postprocess.llm (urllib monkeypatch)."""
from __future__ import annotations
import json
import pytest
from luke_scribe.postprocess import llm, rules
def test_rules_normalize():
assert rules.normalize("구글 Embedding Gemma 소개") == "구글 EmbeddingGemma 소개"
assert rules.normalize("그대로") == "그대로"
def test_llm_not_configured():
with pytest.raises(llm.LLMNotConfigured):
llm.correct("x", base_url=None, api_key=None)
class _FakeResp:
def __init__(self, payload: dict) -> None:
self._p = payload
def read(self) -> bytes:
return json.dumps(self._p).encode()
def __enter__(self):
return self
def __exit__(self, *_a):
return False
def test_llm_correct_monkeypatched(monkeypatch):
def fake_urlopen(_req, timeout=90): # noqa: ARG001
return _FakeResp({"choices": [{"message": {"content": "EmbeddingGemma 복원됨"}}]})
monkeypatch.setattr(llm.urllib.request, "urlopen", fake_urlopen)
out = llm.correct("인베딩 점마", base_url="http://x/v1", api_key="k", model="m")
assert out == "EmbeddingGemma 복원됨"