"""HTTP-сервис распознавания: приём файла, очередь, выдача результата.""" import logging import os import shutil 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.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 run() -> None: import uvicorn _setup_console() install_log_buffer() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") 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") # Windows-консоль не разбирает ANSI-последовательности и печатает их как мусор uvicorn.run(app, host=settings.host, port=settings.port, log_level="info", use_colors=(os.name != "nt")) if __name__ == "__main__": run()