Vladimir BryzgalovandClaude Opus 5
6812220d8f
Uvicorn заводил свои обработчики с propagate=False, поэтому его строки шли мимо нашего формата и рвались посреди слова. log_config=None отдаёт всё общему обработчику. Занятый порт сервис теперь замечает до запуска и объясняет, что делать, вместо строки uvicorn про Errno 10048. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
452 lines
22 KiB
Python
452 lines
22 KiB
Python
"""HTTP-сервис распознавания: приём файла, очередь, выдача результата."""
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from contextlib import asynccontextmanager
|
||
from pathlib import Path
|
||
|
||
from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile
|
||
from fastapi.openapi.docs import get_swagger_ui_html
|
||
from fastapi.openapi.utils import get_openapi
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.security import HTTPBearer
|
||
|
||
from app.config import ConfigError, Settings, load_settings
|
||
from app.console import setup_logging
|
||
from app.logbuffer import get as log_buffer
|
||
from app.logbuffer import install as install_log_buffer
|
||
from app.pipeline import ModelsMissing, Pipeline, to_wav16k
|
||
from app.security import check_token, client_address, ip_allowed, parse_allowlist
|
||
from app.store import JobStatus, JobStore
|
||
from app.webhook import deliver_async
|
||
from app.version import __version__
|
||
|
||
log = logging.getLogger("talkscore-asr")
|
||
|
||
try:
|
||
settings: Settings = load_settings()
|
||
except ConfigError as exc:
|
||
# Печатаем понятную причину и выходим: со сломанным конфигом
|
||
# сервис всё равно ничего осмысленного не сделает.
|
||
print(f"\n {exc}\n")
|
||
raise SystemExit(1) from None
|
||
allowlist = parse_allowlist(settings.allow_ips)
|
||
trusted_proxies = parse_allowlist(settings.trust_proxy)
|
||
|
||
# Хранилище создаётся не здесь, а при старте сервиса. На Windows процессы
|
||
# поднимаются через spawn и заново импортируют этот модуль: если бы очередь
|
||
# открывалась на уровне модуля, каждый новый процесс при старте помечал бы
|
||
# чужие выполняющиеся задачи как сорванные.
|
||
store: JobStore = None # type: ignore[assignment]
|
||
def make_pipeline() -> Pipeline:
|
||
"""Своя копия моделей на каждого воркера: они не рассчитаны на общий доступ."""
|
||
return Pipeline(
|
||
models_dir=settings.models_dir,
|
||
threads=settings.effective_threads(),
|
||
replacements_path=settings.replacements_path,
|
||
base_dir=settings.base_dir,
|
||
)
|
||
|
||
|
||
pipeline = make_pipeline()
|
||
|
||
_worker_stop = threading.Event()
|
||
_state: dict = {"ready": False, "error": None}
|
||
|
||
|
||
def _process(job_id: str, worker: Pipeline, pool) -> None:
|
||
"""Задача уже помечена в работе тем, кто её забрал."""
|
||
job = store.get(job_id)
|
||
if job is None:
|
||
return
|
||
upload = settings.data_dir / "uploads" / job_id
|
||
try:
|
||
speakers = int(job["options"].get("speakers", settings.speakers))
|
||
if pool is not None:
|
||
# Считаем в отдельном процессе: библиотеки держат GIL, и в потоках
|
||
# задачи выстраиваются в очередь вместо параллельной работы.
|
||
from app.worker import run_job
|
||
|
||
result = pool.submit(run_job, str(upload), speakers, worker.ffmpeg,
|
||
settings.normalize).result()
|
||
else:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
wav = Path(tmp) / "audio.wav"
|
||
to_wav16k(upload, wav, worker.ffmpeg, settings.normalize)
|
||
result = worker.transcribe(wav, num_speakers=speakers)
|
||
result["filename"] = job["filename"]
|
||
store.mark_done(job_id, result)
|
||
log.info("задача %s готова: %.1f с аудио, x%s", job_id,
|
||
result["duration_sec"], result["timing"]["realtime_factor"])
|
||
_notify(job, {"job_id": job_id, "status": JobStatus.DONE, **result})
|
||
except Exception as exc: # noqa: BLE001 - в статус задачи должна попасть любая причина
|
||
log.exception("задача %s провалилась", job_id)
|
||
error = f"{type(exc).__name__}: {exc}"
|
||
store.mark_failed(job_id, error=error)
|
||
_notify(job, {"job_id": job_id, "status": JobStatus.FAILED,
|
||
"filename": job["filename"], "error": error})
|
||
finally:
|
||
upload.unlink(missing_ok=True)
|
||
|
||
|
||
def _notify(job: dict, payload: dict) -> None:
|
||
"""Сообщает о результате, если для задачи задан адрес."""
|
||
url = job["options"].get("webhook") or settings.webhook_url
|
||
if url:
|
||
deliver_async(url, payload, settings.webhook_secret)
|
||
|
||
|
||
def _worker_loop(index: int, worker: Pipeline, pool=None) -> None:
|
||
"""Разбирает очередь. Задача захватывается атомарно, поэтому воркеров может быть много."""
|
||
last_cleanup = 0.0
|
||
while not _worker_stop.is_set():
|
||
if _state["ready"]:
|
||
job_id = store.claim_next()
|
||
if job_id:
|
||
log.info("воркер %d взял задачу %s", index, job_id)
|
||
_process(job_id, worker, pool)
|
||
continue
|
||
# Уборкой занимается только первый воркер, чтобы не делать её хором.
|
||
if index == 0 and time.time() - last_cleanup > 3600:
|
||
removed = store.cleanup(settings.keep_results_hours)
|
||
if removed:
|
||
log.info("удалено старых задач: %d", removed)
|
||
last_cleanup = time.time()
|
||
_worker_stop.wait(1.0)
|
||
|
||
|
||
def _make_pool(count: int):
|
||
"""Пул процессов для расчётов. При неудаче работаем в одном процессе.
|
||
|
||
На Windows процессы поднимаются через spawn, и сбои тут возможны, поэтому
|
||
отказ не должен ронять сервис - он просто станет однопроцессным.
|
||
"""
|
||
from concurrent.futures import ProcessPoolExecutor
|
||
|
||
from app.worker import init_worker
|
||
|
||
try:
|
||
pool = ProcessPoolExecutor(
|
||
max_workers=count,
|
||
initializer=init_worker,
|
||
initargs=(str(settings.models_dir), settings.effective_threads(),
|
||
str(settings.replacements_path), str(settings.base_dir)),
|
||
)
|
||
# Пустая задача проверяет, что процессы действительно поднялись
|
||
pool.submit(str, "ok").result(timeout=300)
|
||
return pool
|
||
except Exception as exc: # noqa: BLE001 - причин может быть много, важен откат
|
||
log.warning("не удалось запустить процессы (%s), работаю в одном", exc)
|
||
return None
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
global store
|
||
|
||
(settings.data_dir / "uploads").mkdir(parents=True, exist_ok=True)
|
||
store = JobStore(settings.data_dir / "jobs.db")
|
||
try:
|
||
pipeline.warmup()
|
||
_state["ready"] = True
|
||
except ModelsMissing as exc:
|
||
_state["error"] = str(exc)
|
||
log.error("сервис запущен без моделей: %s", exc)
|
||
count = settings.effective_workers()
|
||
pool = None
|
||
if count > 1 and _state["ready"]:
|
||
pool = _make_pool(count)
|
||
|
||
workers = []
|
||
for i in range(count):
|
||
worker = pipeline if i == 0 else make_pipeline()
|
||
thread = threading.Thread(target=_worker_loop, args=(i, worker, pool),
|
||
name=f"asr-worker-{i}", daemon=True)
|
||
thread.start()
|
||
workers.append(thread)
|
||
|
||
log.info("сервис слушает %s:%s, воркеров %d по %d потоков, режим %s",
|
||
settings.host, settings.port, count, settings.effective_threads(),
|
||
"процессы" if pool else "один процесс")
|
||
yield
|
||
_worker_stop.set()
|
||
for thread in workers:
|
||
thread.join(timeout=5)
|
||
if pool is not None:
|
||
pool.shutdown(wait=False, cancel_futures=True)
|
||
|
||
|
||
# Штатные /docs и /openapi.json отключены: они не требуют токена. Вместо них
|
||
# ниже свои маршруты, закрытые тем же списком адресов, что и остальной сервис.
|
||
app = FastAPI(
|
||
title="talkscore-asr",
|
||
version=__version__,
|
||
description=(
|
||
"Локальное распознавание русской речи с разделением по говорящим.\n\n"
|
||
"Все методы, кроме `/health`, требуют заголовок `Authorization: Bearer <токен>`. "
|
||
"Нажмите **Authorize** и вставьте токен из `config.toml`."
|
||
),
|
||
lifespan=lifespan,
|
||
docs_url=None,
|
||
redoc_url=None,
|
||
openapi_url=None,
|
||
)
|
||
bearer_scheme = HTTPBearer(description="Токен из config.toml, раздел [security]")
|
||
|
||
|
||
def peer_address(request: Request) -> str | None:
|
||
"""Адрес клиента с поправкой на обратный прокси."""
|
||
peer = request.client.host if request.client else None
|
||
return client_address(peer, request.headers.get("x-forwarded-for"), trusted_proxies)
|
||
|
||
|
||
def ip_guard(request: Request) -> None:
|
||
"""Только проверка адреса: нужна для страницы с описанием методов."""
|
||
client_ip = peer_address(request)
|
||
if not ip_allowed(client_ip, allowlist):
|
||
raise HTTPException(status_code=403, detail="адрес не в списке разрешённых")
|
||
|
||
|
||
def guard(request: Request) -> None:
|
||
"""Проверяет адрес и токен. Порядок важен: сначала сеть, потом секрет."""
|
||
client_ip = peer_address(request)
|
||
if not ip_allowed(client_ip, allowlist):
|
||
log.warning("отказано по адресу: %s", client_ip)
|
||
raise HTTPException(status_code=403, detail="адрес не в списке разрешённых")
|
||
if not check_token(request.headers.get("authorization"), settings.token):
|
||
raise HTTPException(status_code=401, detail="неверный или отсутствующий токен")
|
||
|
||
|
||
@app.get("/docs", include_in_schema=False)
|
||
def docs_page(request: Request):
|
||
"""Описание методов для браузера. Открывается только с разрешённых адресов."""
|
||
ip_guard(request)
|
||
return get_swagger_ui_html(openapi_url="openapi.json", title="talkscore-asr")
|
||
|
||
|
||
@app.get("/openapi.json", include_in_schema=False)
|
||
def openapi_schema(request: Request) -> JSONResponse:
|
||
ip_guard(request)
|
||
schema = get_openapi(title=app.title, version=app.version,
|
||
description=app.description, routes=app.routes)
|
||
schema["components"] = schema.get("components", {})
|
||
schema["components"]["securitySchemes"] = {
|
||
"bearer": {"type": "http", "scheme": "bearer"}}
|
||
schema["security"] = [{"bearer": []}]
|
||
return JSONResponse(schema)
|
||
|
||
|
||
@app.get("/v1/logs", dependencies=[Depends(guard)], summary="Журнал сервиса",
|
||
description="Последние записи журнала: удобно посмотреть причину сбоя, "
|
||
"не заходя на машину. Уровень можно отфильтровать параметром level.")
|
||
def logs(limit: int = Query(100, ge=1, le=500),
|
||
level: str | None = Query(None, description="INFO, WARNING или ERROR")) -> dict:
|
||
rows = log_buffer().tail(limit=limit, level=level)
|
||
return {
|
||
"count": len(rows),
|
||
"records": [
|
||
{"time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(r["time"])),
|
||
"level": r["level"], "logger": r["logger"], "message": r["message"]}
|
||
for r in rows
|
||
],
|
||
}
|
||
|
||
|
||
@app.get("/health", summary="Состояние сервиса",
|
||
description="Единственный метод без токена - годится для мониторинга. "
|
||
"Показывает версию, очередь, число потоков и то, каким сервис "
|
||
"видит ваш адрес.")
|
||
def health(request: Request) -> JSONResponse:
|
||
"""Проверка живости - без токена, чтобы годилась для мониторинга.
|
||
|
||
Показывает адрес обратившегося и вердикт по списку доступа: без этого
|
||
отладка отказов превращается в гадание, а сам адрес клиенту и так известен.
|
||
Список разрешённых адресов при этом не раскрывается.
|
||
"""
|
||
client_ip = peer_address(request)
|
||
return JSONResponse({
|
||
"status": "ok" if _state["ready"] else "no_models",
|
||
"version": __version__,
|
||
"error": _state["error"],
|
||
"queue": store.stats(),
|
||
"threads": settings.effective_threads(),
|
||
"workers": settings.effective_workers(),
|
||
"your_ip": client_ip,
|
||
"your_ip_allowed": ip_allowed(client_ip, allowlist),
|
||
"ip_filter_active": bool(allowlist),
|
||
})
|
||
|
||
|
||
@app.post("/v1/jobs", dependencies=[Depends(guard)], summary="Отправить запись",
|
||
description="Принимает аудиофайл любого формата, который читает ffmpeg: "
|
||
"mp3, wav, m4a, ogg, opus, wma. Возвращает идентификатор задачи - "
|
||
"результат забирается отдельным запросом, потому что обработка "
|
||
"занимает минуты.")
|
||
async def create_job(
|
||
file: UploadFile = File(...),
|
||
speakers: int | None = Query(None, ge=0, le=10,
|
||
description="число говорящих, 0 = определить автоматически"),
|
||
webhook: str | None = Query(None,
|
||
description="куда сообщить о готовности; заменяет адрес из настроек"),
|
||
) -> dict:
|
||
if not _state["ready"]:
|
||
raise HTTPException(status_code=503, detail=_state["error"] or "сервис ещё не готов")
|
||
|
||
job_id = store.create(filename=file.filename or "audio",
|
||
duration_sec=0.0,
|
||
options={"speakers": settings.speakers if speakers is None else speakers,
|
||
"webhook": webhook or ""})
|
||
target = settings.data_dir / "uploads" / job_id
|
||
size = 0
|
||
try:
|
||
with target.open("wb") as out:
|
||
while chunk := await file.read(1 << 20):
|
||
size += len(chunk)
|
||
if size > settings.max_upload_bytes:
|
||
raise HTTPException(status_code=413,
|
||
detail=f"файл больше {settings.max_upload_mb} МБ")
|
||
out.write(chunk)
|
||
except HTTPException:
|
||
target.unlink(missing_ok=True)
|
||
store.mark_failed(job_id, error="файл слишком большой")
|
||
raise
|
||
if size == 0:
|
||
target.unlink(missing_ok=True)
|
||
store.mark_failed(job_id, error="пустой файл")
|
||
raise HTTPException(status_code=400, detail="пустой файл")
|
||
|
||
return {"job_id": job_id, "status": JobStatus.QUEUED,
|
||
"queue_position": store.queue_position(job_id)}
|
||
|
||
|
||
@app.post("/v1/benchmark", dependencies=[Depends(guard)], summary="Подобрать число потоков",
|
||
description="Прогоняет минуту из присланной записи на разном числе потоков "
|
||
"и показывает, где быстрее. Занимает пару минут и на это время "
|
||
"нагружает процессор.")
|
||
async def benchmark(file: UploadFile = File(...)) -> dict:
|
||
from app.bench import run_benchmark
|
||
from app.pipeline import read_wav
|
||
|
||
if not _state["ready"]:
|
||
raise HTTPException(status_code=503, detail=_state["error"] or "сервис ещё не готов")
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
src = Path(tmp) / (file.filename or "audio")
|
||
with src.open("wb") as out:
|
||
while chunk := await file.read(1 << 20):
|
||
out.write(chunk)
|
||
wav = Path(tmp) / "audio.wav"
|
||
to_wav16k(src, wav, pipeline.ffmpeg, settings.normalize)
|
||
samples = read_wav(wav)
|
||
|
||
return run_benchmark(samples, models_dir=settings.models_dir,
|
||
base_dir=settings.base_dir,
|
||
replacements_path=settings.replacements_path)
|
||
|
||
|
||
@app.get("/v1/jobs/{job_id}", dependencies=[Depends(guard)], summary="Забрать результат",
|
||
description="Пока задача не готова, возвращает статус и место в очереди. "
|
||
"Готовый результат содержит реплики с говорящими и тайм-кодами, "
|
||
"статистику разговора и время обработки.")
|
||
def get_job(job_id: str) -> dict:
|
||
job = store.get(job_id)
|
||
if job is None:
|
||
raise HTTPException(status_code=404, detail="задача не найдена")
|
||
|
||
body = {"job_id": job_id, "status": job["status"], "filename": job["filename"]}
|
||
if job["status"] == JobStatus.QUEUED:
|
||
body["queue_position"] = store.queue_position(job_id)
|
||
if job["status"] == JobStatus.DONE:
|
||
body.update(job["result"])
|
||
if job["status"] == JobStatus.FAILED:
|
||
body["error"] = job["error"]
|
||
return body
|
||
|
||
|
||
@app.delete("/v1/jobs/{job_id}", dependencies=[Depends(guard)], summary="Удалить задачу",
|
||
description="Убирает задачу и её результат. Завершённые задачи и так "
|
||
"удаляются сами через срок из keep_results_hours.")
|
||
def delete_job(job_id: str) -> dict:
|
||
if store.get(job_id) is None:
|
||
raise HTTPException(status_code=404, detail="задача не найдена")
|
||
store.delete(job_id)
|
||
return {"deleted": job_id}
|
||
|
||
|
||
def _setup_console() -> None:
|
||
"""Windows-консоль по умолчанию не в UTF-8, иначе русский текст в логах рассыпается."""
|
||
for stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
stream.reconfigure(encoding="utf-8", errors="replace")
|
||
except (AttributeError, ValueError):
|
||
pass
|
||
|
||
|
||
def _port_busy(host: str, port: int) -> bool:
|
||
"""Проверяет, слушает ли кто-то уже этот порт.
|
||
|
||
Проверяем подключением, а не пробной привязкой: привязка оставляет сокет
|
||
в TIME_WAIT и может сама помешать запуску через секунду.
|
||
"""
|
||
target = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host
|
||
with socket.socket() as probe:
|
||
probe.settimeout(0.5)
|
||
try:
|
||
return probe.connect_ex((target, port)) == 0
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def run() -> None:
|
||
import uvicorn
|
||
|
||
_setup_console()
|
||
install_log_buffer()
|
||
setup_logging(logging.INFO)
|
||
|
||
if _port_busy(settings.host, settings.port):
|
||
# Uvicorn напишет про это «[Errno 10048] ... bind on address», из чего
|
||
# неочевидно, что виноват уже работающий экземпляр сервиса.
|
||
print(f"\n Порт {settings.port} уже занят - похоже, сервис уже запущен."
|
||
"\n Проверьте свёрнутые окна talkscore-asr либо найдите процесс:"
|
||
f"\n netstat -ano | findstr :{settings.port}"
|
||
"\n taskkill /PID <номер из последней колонки> /F\n")
|
||
raise SystemExit(1)
|
||
|
||
if not settings.token:
|
||
print("\n В config.toml пустой токен - сервис никого не пустит."
|
||
"\n Впишите значение в [security] token и перезапустите.\n")
|
||
|
||
# Модели проверяются здесь, а не по _state: lifespan отработает уже внутри
|
||
# uvicorn.run, и к тому моменту сообщение выводить поздно.
|
||
if not (settings.models_dir / "gigaam" / "config.json").is_file():
|
||
print("\n Модели не найдены. Сначала запустите download_models.bat"
|
||
"\n Сервис поднимется, но принимать записи не сможет.\n")
|
||
|
||
nets = len(allowlist)
|
||
access = f"только с {nets} адресов" if nets else "со ВСЕХ адресов"
|
||
print(f"\n Настройки: {settings.base_dir / 'config.toml'}"
|
||
f"\n Токен: {'задан' if settings.token else 'НЕ ЗАДАН, сервис никого не пустит'}"
|
||
f"\n Доступ: {access}"
|
||
f"\n Обработка: {settings.effective_workers()} задач одновременно"
|
||
f" по {settings.effective_threads()} потока")
|
||
print(f"\n talkscore-asr слушает http://{settings.host}:{settings.port}"
|
||
f"\n Потоков: {settings.effective_threads()}"
|
||
f"\n Проверка: curl http://localhost:{settings.port}/health"
|
||
"\n Остановить: Ctrl+C\n")
|
||
# log_config=None отключает собственную настройку логов uvicorn. Иначе он
|
||
# заводит свои обработчики с propagate=False, его строки идут мимо нашего
|
||
# формата и рвутся посреди слова, а наши выглядят иначе - в одном окне два
|
||
# разных формата.
|
||
uvicorn.run(app, host=settings.host, port=settings.port, log_level="info",
|
||
log_config=None)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|