From 00e2a95fb72904e0664455c9d47cc42b9ab6a04f Mon Sep 17 00:00:00 2001 From: zikai <1621362626@qq.com> Date: Mon, 6 Jul 2026 06:54:19 +0000 Subject: [PATCH] =?UTF-8?q?Initial=20commit:=20audio2text=20=E5=8F=8C?= =?UTF-8?q?=E8=AF=AD=E5=AD=97=E5=B9=95=E7=94=9F=E6=88=90=E6=9C=8D=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 音频/视频转双语(英/中)SRT 字幕，Docker 容器化，CPU 开发/GPU 生产同一份代码 - faster-whisper ASR(词级时间戳) + 断句时间戳重算 + NLLB 翻译(模型不共驻) - 分片上传(断点续传) + SQLite 持久化 + 主页/历史/日志页面 - 历史页文件名搜索；缓存定时清理(默认保留7天，可配置) - 双 Dockerfile(cpu/gpu) + setup/start/stop 脚本 --- .dockerignore | 10 + .gitignore | 17 + Dockerfile | 63 +++ README.md | 681 ++++++++++++++++++++++++++++++ app/__init__.py | 0 app/config.py | 136 ++++++ app/controllers/__init__.py | 7 + app/controllers/log_router.py | 31 ++ app/controllers/task_router.py | 88 ++++ app/controllers/upload_router.py | 69 +++ app/database.py | 68 +++ app/main.py | 160 +++++++ app/models/__init__.py | 0 app/models/task.py | 40 ++ app/models/upload_session.py | 49 +++ app/schemas/__init__.py | 0 app/schemas/task.py | 69 +++ app/security.py | 27 ++ app/services/__init__.py | 0 app/services/asr_service.py | 65 +++ app/services/cache_cleaner.py | 194 +++++++++ app/services/ffmpeg_service.py | 64 +++ app/services/log_buffer.py | 109 +++++ app/services/model_manager.py | 136 ++++++ app/services/pipeline.py | 153 +++++++ app/services/reaper.py | 22 + app/services/segmenter.py | 233 ++++++++++ app/services/srt_writer.py | 68 +++ app/services/translate_service.py | 60 +++ app/services/types.py | 38 ++ app/services/upload_service.py | 249 +++++++++++ app/views/__init__.py | 0 app/views/_shared.py | 370 ++++++++++++++++ app/views/history_html.py | 169 ++++++++ app/views/home_html.py | 142 +++++++ app/views/logs_html.py | 167 ++++++++ config.cpu.yaml | 57 +++ config.example.yaml | 55 +++ config.gpu.yaml | 56 +++ docker-compose.yml | 53 +++ requirements.txt | 25 ++ setup.sh | 41 ++ start.sh | 55 +++ stop.sh | 14 + 44 files changed, 4110 insertions(+) create mode 100644 .dockerignore create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 app/__init__.py create mode 100644 app/config.py create mode 100644 app/controllers/__init__.py create mode 100644 app/controllers/log_router.py create mode 100644 app/controllers/task_router.py create mode 100644 app/controllers/upload_router.py create mode 100644 app/database.py create mode 100644 app/main.py create mode 100644 app/models/__init__.py create mode 100644 app/models/task.py create mode 100644 app/models/upload_session.py create mode 100644 app/schemas/__init__.py create mode 100644 app/schemas/task.py create mode 100644 app/security.py create mode 100644 app/services/__init__.py create mode 100644 app/services/asr_service.py create mode 100644 app/services/cache_cleaner.py create mode 100644 app/services/ffmpeg_service.py create mode 100644 app/services/log_buffer.py create mode 100644 app/services/model_manager.py create mode 100644 app/services/pipeline.py create mode 100644 app/services/reaper.py create mode 100644 app/services/segmenter.py create mode 100644 app/services/srt_writer.py create mode 100644 app/services/translate_service.py create mode 100644 app/services/types.py create mode 100644 app/services/upload_service.py create mode 100644 app/views/__init__.py create mode 100644 app/views/_shared.py create mode 100644 app/views/history_html.py create mode 100644 app/views/home_html.py create mode 100644 app/views/logs_html.py create mode 100644 config.cpu.yaml create mode 100644 config.example.yaml create mode 100644 config.gpu.yaml create mode 100644 docker-compose.yml create mode 100644 requirements.txt create mode 100755 setup.sh create mode 100755 start.sh create mode 100755 stop.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..658500d --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +data/ +/models/ +config.yaml +.venv/ +__pycache__/ +*.pyc +logs/ +*.pid +.git/ +.gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9b98e73 --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# Python +.venv/ +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ + +# 运行时数据（走 docker volume，不入库） +data/ +/models/ +config.yaml +*.pid +logs/ + +# 编辑器 +.vscode/ +.idea/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..989e246 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,63 @@ +# audio2text — 一份 Dockerfile，CPU(dev) / GPU(prod) 双形态。 +# docker build --build-arg VARIANT=cpu -t audio2text:cpu . +# docker build --build-arg VARIANT=gpu -t audio2text:gpu . +# 区别仅在基础镜像与 torch 轮子；Python 依赖列表完全一致。 + +ARG VARIANT=cpu +FROM python:3.12-slim AS base-cpu +# GPU 基础镜像带 CUDA 运行时，torch 可装 CUDA 轮子 +FROM nvidia/cuda:12.1.0-runtime-ubuntu22.04 AS base-gpu +RUN apt-get update -y && apt-get install -y --no-install-recommends \ + python3.12 python3.12-venv python3.12-dev python3-pip \ + && rm -rf /var/lib/apt/lists/* \ + && ln -sf /usr/bin/python3.12 /usr/local/bin/python3 \ + && ln -sf /usr/bin/python3.12 /usr/local/bin/python + +FROM base-${VARIANT} AS final +ARG VARIANT +ENV VARIANT=${VARIANT} \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + HF_HOME=/models/huggingface \ + CT2_CACHE=/models/ctranslate2 + +# ffmpeg 是核心系统依赖，必须装 +RUN apt-get update -y && apt-get install -y --no-install-recommends \ + ffmpeg ca-certificates patchelf \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY requirements.txt /app/requirements.txt + + +# CPU 装 CPU 版 torch；GPU 走默认 index（带 CUDA 的轮子） +RUN if [ "$VARIANT" = "cpu" ]; then \ + pip install --upgrade pip && \ + pip install torch --index-url https://download.pytorch.org/whl/cpu ; \ + else \ + pip install --upgrade pip && \ + pip install torch ; \ + fi +RUN pip install -r /app/requirements.txt + +# ctranslate2 的 .so 带可执行栈标志（PT_GNU_STACK X），在某些内核 + Docker 组合下 +# 会触发 "cannot enable executable stack as shared object requires"。用 patchelf +# 清掉该标志（改为 RW），无需放宽容器安全策略。 +# 注意：库在 ctranslate2.libs/ 隐藏目录（pip wheel 拆分产物），不在 ctranslate2/ 包目录。 +RUN for d in /usr/local/lib/python3.12/site-packages/ctranslate2.libs \ + /usr/local/lib/python3.12/site-packages/ctranslate2; do \ + [ -d "$d" ] && find "$d" -name '*.so*' \ + -exec patchelf --clear-execstack {} \; 2>/dev/null || true; \ + done; \ + python -c "import ctranslate2; print('ctranslate2 stack fix verified', ctranslate2.__version__)" + +COPY app /app/app +COPY config.example.yaml /app/config.example.yaml + +# 运行时数据：上传 / 中间产物 / 输出字幕 / 模型缓存 +# 全部走 volume，镜像本身无状态、无敏感数据 +VOLUME ["/data", "/models"] +ENV CONFIG_PATH=/app/config.yaml + +EXPOSE 8000 +CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..eea9ba5 --- /dev/null +++ b/README.md @@ -0,0 +1,681 @@ +# audio2text + +音频 / 视频转双语字幕服务。上传视频 → ffmpeg 提取音频 → faster-whisper 识别英语 → +断句 + 时间戳重算 → NLLB 翻译为中文 → 输出双语 SRT。全程跑在 Docker 容器里，自带 +网页上传界面，支持大文件分片上传与断点续传。 + +仿照 zikai 的 `server/` 风格分层（`controllers → services`），**CPU 开发 / GPU 生产 +同一份代码**，仅靠 `config.yaml` 的 `device` + `model` + `compute_type` 三项切换。 + +--- + +## 目录 + +- [功能特性](#功能特性) +- [架构](#架构) +- [部署：CPU 开发环境](#部署cpu-开发环境) +- [部署：GPU 生产环境](#部署gpu-生产环境) +- [配置文件说明](#配置文件说明) +- [缓存清理与定时任务](#缓存清理与定时任务) +- [HTTP 接口](#http-接口) +- [断句与时间戳重算原理](#断句与时间戳重算原理) +- [模型不共驻（显存策略）](#模型不共驻显存策略) +- [Docker 说明](#docker-说明) +- [依赖](#依赖) +- [常见问题](#常见问题) + +--- + +## 功能特性 + +- **主页** `/`：上传入口（拖拽 / 选择文件，多文件、4 MiB 分片、断点续传）+ 最近 10 个任务的实时进度卡片，完成的可直接下载字幕。 +- **历史任务页** `/history`：分页查看所有历史任务，可下载完成的字幕。 +- **实时日志页** `/logs`：按级别分层查看——debug=详细子步骤、info=仅阶段转换、error=完整 traceback。 +- **大视频处理**：接收完成后用 ffmpeg 提取 16 kHz 单声道 PCM 音频；是否删原始视频由配置决定。 +- **faster-whisper 转写英语**，带词级时间戳。 +- **断句 + 时间戳重算**：按句末标点（`. ! ? ;`）切句、超长句按逗号拆，时间戳取首末词精确值； + 无词级时间戳时退化为段内匀速估算。 +- **NLLB-200 英译中**；ASR 与翻译模型**不共驻**，翻译时卸载 Whisper 独占显存跑大 batch。 +- **双语合并 SRT** 输出（英文在上、中文在下），亦可单独下载英文 / 中文字幕。 +- **任务状态机**：`queued → extracting → transcribing → segmenting → translating → done`， + 页面自动轮询进度。 +- **定时缓存清理**：任务产物（字幕 / 中间音频 / 保留的原始视频）默认保留 7 天，超期后 + 连同 DB 记录一并删除；容器内后台线程定时执行（启动时跑一次，默认每 24 小时一次）， + 保留期与间隔均可配置。 +- **SQLite 持久化**（自包含，无需外部 DB）。 +- `/docs`（Swagger UI）受 Basic Auth 保护。 + +--- + +## 架构 + +Spring 风格分层，HTTP 边界（controllers）与业务逻辑（services）分离： + +``` +audio2text/ +├── Dockerfile # 一份 Dockerfile，ARG VARIANT=cpu|gpu 出两个镜像 +├── docker-compose.yml # cpu / gpu 两个 profile +├── setup.sh / start.sh / stop.sh # 安装 / 启动 / 停止（包装 docker 命令） +├── requirements.txt +├── config.example.yaml # 配置模板（复制为 config.yaml 后填值） +├── README.md +└── app/ + ├── main.py # FastAPI 应用工厂 + ├── config.py # 从 config.yaml 加载的类型化 Settings（pydantic） + ├── database.py # SQLite 引擎 / Session / Base / get_db + ├── security.py # /docs 的 Basic Auth（常量时间比较） + ├── controllers/ # FastAPI 路由 —— HTTP 边界 + │ ├── upload_router.py # 分片上传（建会话/查状态/传片/complete） + │ └── task_router.py # 任务列表 / 状态 / 下载字幕 + ├── services/ # 业务逻辑 + │ ├── types.py # 共享 DTO（Word/Segment/Subtitle，纯 dataclass） + │ ├── upload_service.py # 分片上传会话 + 拼接 + 创建任务 + │ ├── ffmpeg_service.py # 提取音频 16k mono pcm + │ ├── asr_service.py # faster-whisper 转写 + │ ├── segmenter.py # 断句 + 时间戳重算（纯算法，零模型依赖） + │ ├── translate_service.py # NLLB 翻译 + │ ├── model_manager.py # 模型加载/卸载（不共驻核心） + │ ├── pipeline.py # 编排：提取→识别→断句→翻译→写SRT + │ ├── srt_writer.py # SRT 写入 + 双语合并 + │ ├── log_buffer.py # 内存日志缓冲（供 /logs 页面查询） + │ ├── reaper.py # 清理被放弃的上传会话（短 TTL） + │ └── cache_cleaner.py # 定时清理超期任务产物 + 孤儿目录（长保留期） + ├── models/ # SQLAlchemy ORM + │ ├── task.py # Task（转写任务，状态机） + │ └── upload_session.py # UploadSession（分片会话，含 task_id FK） + ├── schemas/ # pydantic 请求/响应 DTO + │ └── task.py + └── views/ + ├── _shared.py # 共享前端资产（BASE_CSS + SHARED_JS + 上传协议 + 页面骨架） + ├── home_html.py # 主页（上传入口 + 最近任务卡片） + ├── history_html.py # 历史任务分页表格（含搜索） + └── logs_html.py # 实时日志页 +``` + +### 数据流 + +``` +浏览器 /（主页） + │ 分片上传 (4 MiB/片, 可断点续传) + ▼ +upload_router ──► upload_service ──► UploadSession(SQLite) + 分片落盘 + │ complete + ▼ +创建 Task(queued) ──► pipeline 后台线程 + │ + ├─ 1. ffmpeg_service.extract_audio → 16k mono wav + │ （按配置删原始视频） + ├─ 2. model_manager.get_asr → asr_service.transcribe → segments(带词级时间戳) + ├─ 3. segmenter.resegment → 规范字幕条目（精确/估算两路） + ├─ 4. model_manager.unload_asr → get_translator + │ translate_service.translate → 中文译文（独占显存大 batch） + └─ 5. srt_writer → en.srt / zh.srt / bilingual.srt + 更新 Task(done) + 写 output_dir +``` + +--- + +## 部署：CPU 开发环境 + +CPU 模式用于本地开发与流程验证，模型选同系列最小尺寸，2GB 内存开发机即可跑通完整流程。 + +### 前置要求 + +- Docker（用于构建镜像 + 运行容器） +- 约 500 MB 磁盘（模型缓存）+ 上传视频空间 + +CPU 模式**不需要** NVIDIA 驱动，普通 Linux / macOS / WSL 均可。 + +### 步骤 + +```bash +cd /root/zikai/audio2text + +# 1. 构建 CPU 镜像 + 复制 config.cpu.yaml → config.yaml +./setup.sh # 默认 AUDIO2TEXT_VARIANT=cpu + +# 2. 启动容器（默认端口 8000） +./start.sh + +# 3. 停止 / 重启 +./stop.sh +./start.sh +``` + +`setup.sh` 做三件事：检查 docker → 构建 `audio2text:cpu` 镜像 → 把 `config.cpu.yaml` +复制为 `config.yaml`（运行时实际读取的文件）。可重复执行；改完配置后重新 `cp` 并重启即可， +无需重建镜像。 + +首次启动会下载模型（Whisper `tiny.en` ~39M + opus-mt ~300MB）到 `./models` volume， +之后秒起。启动后浏览器打开 `http://127.0.0.1:8000/`，拖入视频或音频文件即可。 + +### CPU 模型选型 + +| 组件 | 模型 | 大小 | 说明 | +|---|---|---|---| +| ASR | `tiny.en` | ~39M | Whisper 同系列最小，英文专用版（比通用 `tiny` 在英语上更准） | +| 翻译 | `Helsinki-NLP/opus-mt-en-zh` | ~300MB | 最轻量英译中。NLLB 同系列最小 `distilled-600M` 需 ~2.4GB，2GB 机 OOM，故回退 | + +> 翻译质量与 GPU 的 NLLB-1.3B 有差异，但**完整流程一致**（提取→识别→断句→翻译→双语 SRT）， +> 足以验证端到端逻辑。如需在 CPU 上验证 NLLB 翻译质量，可把 `translation.model` 改为 +> `nllb-200-distilled-600M`（需 ≥4GB 内存）或 `nllb-200-distilled-1.3B`（需 ~5GB 内存）。 + +### 自定义端口 + +```bash +AUDIO2TEXT_PORT=9000 ./start.sh +``` + +--- + +## 部署：GPU 生产环境 + +GPU 模式用于生产，模型质量优先，NVIDIA 3090 24G 上几 GB 视频几分钟出字幕。 + +### 前置要求 + +- Docker +- **NVIDIA GPU 驱动**（宿主机） +- **nvidia container runtime**（让容器能用 GPU；安装 `nvidia-container-toolkit`） +- 约 6 GB 磁盘（模型缓存：large-v3-turbo ~3GB + NLLB-1.3B ~2.5GB） + +验证 GPU 可用： + +```bash +nvidia-smi # 宿主能看到 GPU +docker run --rm --gpus all nvidia/cuda:12.1.0-runtime-ubuntu22.04 nvidia-smi +# 上面容器内也能列出 GPU 即说明 nvidia runtime 已就绪 +``` + +### 步骤 + +```bash +cd /root/zikai/audio2text + +# 1. 构建 GPU 镜像 + 复制 config.gpu.yaml → config.yaml +AUDIO2TEXT_VARIANT=gpu ./setup.sh + +# 2. 启动容器（start.sh 检测到 gpu 镜像 + nvidia-smi 自动加 --gpus all） +./start.sh + +# 3. 停止 / 重启 +./stop.sh +./start.sh +``` + +`start.sh` 的镜像选择逻辑：若本机存在 `audio2text:gpu` 镜像**且**有 `nvidia-smi`，自动用 +GPU 模式（`--gpus all`）；否则回退 CPU 镜像。也可用 docker compose 显式启动： + +```bash +docker compose --profile gpu up -d --build # GPU +docker compose --profile cpu up -d --build # CPU +``` + +### GPU 模型选型 + +| 组件 | 模型 | 显存 | 说明 | +|---|---|---|---| +| ASR | `large-v3-turbo` | ~3GB（FP16） | 8x 速度，质量接近 large-v3 | +| 翻译 | `facebook/nllb-200-distilled-1.3B` | ~2.5GB（FP16） | 质量最好的蒸馏版 | + +ASR 与翻译**不共驻**：翻译阶段先卸载 Whisper 释放显存，独占跑大 batch（`batch_size=16`）， +两者峰值显存互不叠加，远低于 24G 上限。模型缓存（`./models` volume）跨容器复用， +CPU→GPU 切换时 NLLB/Whisper 大模型首次下载、之后秒起。 + +### CPU ↔ GPU 切换 + +同一份代码，仅靠 `AUDIO2TEXT_VARIANT` 切换镜像 + 配置： + +```bash +AUDIO2TEXT_VARIANT=gpu ./setup.sh # 切到 GPU（构建 gpu 镜像 + config.gpu.yaml） +AUDIO2TEXT_VARIANT=cpu ./setup.sh # 切回 CPU（构建 cpu 镜像 + config.cpu.yaml） +./start.sh # 重新启动 +``` + +两套配置的差异仅在 6 项（其余字段完全一致）： + +| 字段 | CPU（`config.cpu.yaml`） | GPU（`config.gpu.yaml`） | +|---|---|---| +| `asr.model` | `tiny.en` | `large-v3-turbo` | +| `asr.device` | `cpu` | `cuda` | +| `asr.compute_type` | `int8` | `float16` | +| `translation.model` | `Helsinki-NLP/opus-mt-en-zh` | `facebook/nllb-200-distilled-1.3B` | +| `translation.device` | `cpu` | `cuda` | +| `translation.batch_size` | `8` | `16` | + +### 启动后的入口 + +两种模式通用： + +| 入口 | 地址 | +|---|---| +| 主页 | `http://127.0.0.1:8000/`（上传入口 + 最近 10 任务进度卡片） | +| 历史任务 | `http://127.0.0.1:8000/history`（分页查看所有任务，可按文件名搜索、下载字幕） | +| 日志页 | `http://127.0.0.1:8000/logs`（按级别分层、自动刷新） | +| API 文档 | `http://127.0.0.1:8000/docs`（Basic Auth，凭据见 config.yaml `docs` 段） | +| 健康检查 | `http://127.0.0.1:8000/health` | +| 任务列表 | `http://127.0.0.1:8000/api/tasks` | + +--- + +## 配置文件说明 + +项目预置两份配置文件，`setup.sh` 按 `AUDIO2TEXT_VARIANT` 自动复制对应文件为 +`config.yaml`（运行时实际读取的文件，不入库）： + +| 文件 | 激活方式 | 说明 | +|---|---|---| +| `config.cpu.yaml` | `./setup.sh`（默认） | CPU 开发，最小模型 | +| `config.gpu.yaml` | `AUDIO2TEXT_VARIANT=gpu ./setup.sh` | GPU 生产，质量优先 | +| `config.example.yaml` | — | 带完整注释的字段参考模板 | + +也可手动切换：`cp config.gpu.yaml config.yaml` 后重启容器即可，无需重建镜像（镜像不含配置）。 +运行时通过环境变量 `CONFIG_PATH` 指定路径（容器内默认 `/app/config.yaml`）。所有路径相对 +容器内文件系统。`config.py` 用 pydantic 做类型校验，缺字段时回退默认值。 + +### CPU / GPU 两份配置的差异 + +其余字段（存储、断句、日志、docs）两份配置完全一致，仅以下 6 项不同： + +| 字段 | `config.cpu.yaml` | `config.gpu.yaml` | +|---|---|---| +| `asr.model` | `tiny.en` | `large-v3-turbo` | +| `asr.device` | `cpu` | `cuda` | +| `asr.compute_type` | `int8` | `float16` | +| `translation.model` | `Helsinki-NLP/opus-mt-en-zh` | `facebook/nllb-200-distilled-1.3B` | +| `translation.device` | `cpu` | `cuda` | +| `translation.batch_size` | `8` | `16` | + +### 完整字段 + +#### `server` — 服务监听 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `host` | str | `0.0.0.0` | 容器内监听地址（由 `docker -p` 映射到宿主） | +| `port` | int | `8000` | 容器内监听端口 | +| `workers` | int | `1` | uvicorn worker 数。ML 推理为重，固定单 worker 避免显存重复占用 | + +#### `storage` — 文件存储 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `upload_dir` | str | `/data/uploads` | 上传视频落盘根目录（按 `yyyy/mm` 分子目录） | +| `work_dir` | str | `/data/.work` | 分片会话暂存 + 中间音频 + SQLite 数据库 | +| `output_dir` | str | `/data/outputs` | 生成的 SRT 字幕输出目录 | +| `chunk_bytes` | int | `1048576` | 流式分片大小（1 MiB）。注意：前端上传页固定 4 MiB，此项影响服务端缓冲 | +| `chunk_session_ttl_seconds` | int | `300` | 被放弃的分片会话存活秒数，超时后后台 reaper 清理（短 TTL，与下方缓存清理不同） | +| `cache_retention_days` | int | `7` | 任务产物（字幕 / 中间音频 / 保留的原始视频）保留天数；超期任务连同 DB 记录一并删除。`0` = 禁用清理 | +| `cache_cleanup_interval_hours` | int | `24` | 定时清理间隔（小时）。容器启动时跑一次，之后按此间隔循环 | + +#### `processing` — 处理流程 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `delete_original_after_extract` | bool | `true` | 提取音频成功后删除原始视频，省空间。`false` 则保留视频 | +| `keep_audio` | bool | `false` | 任务完成后是否保留中间 wav。`false` 则只留字幕、删 wav | + +#### `asr` — 语音识别（faster-whisper） + +| 字段 | 类型 | 默认（CPU） | 说明 | +|---|---|---|---| +| `model` | str | `tiny.en` | Whisper 模型名。CPU dev 用 `tiny.en`（39M，英文专用，同系列最小）；GPU prod 用 `large-v3-turbo`（8x 速度，质量接近 large-v3） | +| `device` | str | `cpu` | `cpu` 或 `cuda` | +| `compute_type` | str | `int8` | CPU 用 `int8`；GPU 用 `float16` | +| `language` | str | `en` | 识别语言，仅英语 | +| `word_timestamps` | bool | `true` | 词级时间戳：让断句精确（取首末词时间戳）而非纯匀速估算。建议开 | +| `vad_filter` | bool | `true` | 过滤静音段，提升识别质量与速度 | + +#### `translation` — 翻译（NLLB-200） + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `model` | str | `facebook/nllb-200-distilled-1.3B` | HuggingFace 模型名。GPU 生产用 1.3B（质量最好）；CPU dev 用 `Helsinki-NLP/opus-mt-en-zh`（~300MB，2GB 机可跑）。NLLB 同系列最小为 `distilled-600M`（~1.2GB，需 ≥4GB 内存） | +| `device` | str | `cpu` | `cpu` 或 `cuda` | +| `src_lang` | str | `eng_Latn` | NLLB 语言码：英语 | +| `tgt_lang` | str | `zho_Hans` | NLLB 语言码：简体中文 | +| `batch_size` | int | `16` | 翻译批量大小。不与 ASR 共驻时显存独占，可用大 batch | +| `max_length` | int | `256` | 单条翻译最大 token 数 | + +#### `segmentation` — 断句与字幕规范化 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `max_words_per_line` | int | `14` | 单行最多词数，超出按逗号拆分 | +| `max_duration_seconds` | float | `7.0` | 单条字幕最长 7 秒 | +| `min_duration_seconds` | float | `1.0` | 单条字幕最短 1 秒（太短则与下条合并） | +| `max_chars_per_line` | int | `42` | SRT 规范：每行 ≤42 字符，超出按词折行（≤2 行） | + +#### `logging` — 日志 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `level` | str | `info` | 控制台输出的最低级别：`debug` / `info` / `warning` / `error`。不影响 `/logs` 页面（页面可自由切换级别查看） | +| `buffer_size` | int | `2000` | `/logs` 页面内存缓冲条数（有界 deque，旧记录自动淘汰） | + +日志分层语义： + +| 级别 | 内容 | 示例 | +|---|---|---| +| **debug**（详细） | 子步骤：ffmpeg 命令、模型加载/卸载、转写逐段、翻译逐批进度 | `加载 ASR 模型 model=tiny.en device=cpu` / `ffmpeg 命令：ffmpeg -y ...` | +| **info**（简略） | 仅任务阶段转换，看当前进行到哪一步 | `任务 1 [transcribing 55%] 识别出 3 段` | +| **error**（详细） | 完整 traceback（文件名+行号+调用链），可点击展开 | `任务 1 失败：ffmpeg 失败 (code=183)...` + traceback | + +> **注意**：`logging.level` 只控制控制台输出级别。`/logs` 页面始终全量缓冲（DEBUG 起）， +> 页面上的级别按钮是查询过滤，不受此配置限制——所以控制台设 `info` 保持简略，而 `/logs` +> 页面切到 DEBUG 仍能看到所有详细子步骤。 + +#### `docs` — API 文档保护 + +| 字段 | 类型 | 默认 | 说明 | +|---|---|---|---| +| `enabled` | bool | `true` | 是否开启 `/docs` `/redoc` `/openapi.json` | +| `username` | str | `admin` | Basic Auth 用户名 | +| `password` | str | `CHANGE_ME` | Basic Auth 明文密码（常量时间比较）。**部署前务必修改** | +| `realm` | str | `audio2text docs` | WWW-Authenticate realm | + +### 配置示例 + +```yaml +server: + host: 0.0.0.0 + port: 8000 + workers: 1 + +storage: + upload_dir: /data/uploads + work_dir: /data/.work + output_dir: /data/outputs + chunk_bytes: 1048576 + chunk_session_ttl_seconds: 300 + cache_retention_days: 7 # 任务产物保留天数，超期清理（0=禁用） + cache_cleanup_interval_hours: 24 # 定时清理间隔（启动时跑一次，之后循环） + +processing: + delete_original_after_extract: true + keep_audio: false + +asr: + model: tiny.en # GPU: large-v3-turbo + device: cpu # GPU: cuda + compute_type: int8 # GPU: float16 + language: en + word_timestamps: true + vad_filter: true + +translation: + model: facebook/nllb-200-distilled-1.3B + device: cpu # GPU: cuda + src_lang: eng_Latn + tgt_lang: zho_Hans + batch_size: 16 + max_length: 256 + +segmentation: + max_words_per_line: 14 + max_duration_seconds: 7.0 + min_duration_seconds: 1.0 + max_chars_per_line: 42 + +logging: + level: info # debug | info | warning | error（控制台输出最低级别） + buffer_size: 2000 + +docs: + enabled: true + username: admin + password: "CHANGE_ME" + realm: "audio2text docs" +``` + +--- + +## 缓存清理与定时任务 + +每个任务落盘的产物（字幕、中间音频、保留的原始视频）会持续占用磁盘。容器内置定时 +清理（`app/services/cache_cleaner.py`），无需外部 cron： + +### 清理什么 + +| 产物 | 路径 | 何时产生 | +|---|---|---| +| 字幕输出 | `/task_/` | 任务完成 | +| 中间音频 | `/task_.wav` | `keep_audio=true` 且管线未删时残留 | +| 保留的原始视频 | `/yyyy/mm/.` | `delete_original_after_extract=false` 时 | +| 孤儿目录 | 上述目录中无对应 Task 的残留 | 进程崩溃 / 异常退出留下 | + +### 清理策略 + +1. **超期任务**：`Task.created_at` 早于 `now - cache_retention_days`（默认 7 天）的任务， + 删除其全部产物，并删除对应的 `Task` 与 `UploadSession` 行——避免历史页出现指向已删 + 文件的死链接。 +2. **孤儿扫描**：`output_dir` / `work_dir` 下名为 `task_` 但 DB 中已无该 Task 的目录 + （崩溃残留），按目录 `mtime` 判超期后删除。 +3. **DB 一致性**：删任务时先删关联的 `UploadSession`（FK），再删 `Task`，保持引用完整。 + +### 触发时机 + +- **启动时跑一次**：容器启动 lifespan 中立即执行（`purge_expired_cache`），清掉停机期间 + 超期的产物。 +- **后台定时循环**：守护线程 `cache-cleaner` 按 `cache_cleanup_interval_hours`（默认 24h） + 循环执行，随进程退出而终止。 +- **手动触发**（调试用）：进容器跑 `python -m app.services.cache_cleaner`，打印清理统计 JSON。 + +### 相关配置（`storage` 段） + +| 字段 | 默认 | 说明 | +|---|---|---| +| `cache_retention_days` | `7` | 保留天数。`0` = 禁用清理（产物永久保留） | +| `cache_cleanup_interval_hours` | `24` | 定时循环间隔（小时） | + +### 与上传会话 reaper 的区别 + +| 机制 | 清理对象 | 判定 | 触发 | +|---|---|---|---| +| **reaper**（`reaper.py`） | 被放弃的**分片上传会话**（未 complete 的） | `status=pending` 且 `updated_at` 超 `chunk_session_ttl_seconds`（300s） | 仅启动时一次 | +| **cache_cleaner**（本节） | 已完成/失败**任务的产物** + 崩溃孤儿 | `created_at` 超 `cache_retention_days`（7d）/ 孤儿 mtime 超期 | 启动一次 + 定时循环 | + +> 后台清理线程与请求线程并发写同一 SQLite 库，`database.py` 已设 `busy_timeout=30s`， +> 拿锁时阻塞等待而非立即报 `database is locked`。单 worker 部署下无并发写入压力。 + +--- + +## HTTP 接口 + +| 方法 | 路径 | 认证 | 说明 | +|---|---|---|---| +| GET | `/` | 无 | 主页（上传入口 + 最近 10 任务进度卡片） | +| GET | `/health` | 无 | 存活探针 | +| GET | `/history` | 无 | 历史任务页（分页表格，可按文件名搜索、下载字幕） | +| GET | `/logs` | 无 | 实时日志页（按级别过滤、自动刷新、可展开 traceback） | +| GET | `/docs` `/redoc` | Basic Auth | API 文档 | +| POST | `/api/tasks/chunk-uploads` | 无 | 创建分片上传会话 | +| GET | `/api/tasks/chunk-uploads/{id}/status` | 无 | 查已传分片（断点续传） | +| POST | `/api/tasks/chunk-uploads/{id}/chunks/{index}` | 无 | 上传单个分片（原始二进制 body） | +| POST | `/api/tasks/chunk-uploads/{id}/complete` | 无 | 拼接 + 创建转写任务 | +| GET | `/api/tasks` | 无 | 任务列表（`limit` / `offset` 分页，`q` 按文件名模糊搜索） | +| GET | `/api/tasks/{id}` | 无 | 任务状态（status / progress / error） | +| GET | `/api/tasks/{id}/subtitle?type=bilingual\|en\|zh` | 无 | 下载字幕 | +| GET | `/api/logs?level=debug\|info\|warning\|error&tail=N` | 无 | 查询日志（按级别过滤，最近 N 条） | +| DELETE | `/api/logs` | 无 | 清空日志缓冲 | + +### 分片上传协议（与 server 一致） + +1. **建会话** `POST /api/tasks/chunk-uploads`，body 含 `filename` / `size_bytes` / + `chunk_size` / `total_chunks`，返回 `upload_id`。 +2. **查状态** `GET .../status`，返回 `uploaded_chunks`（已传分片下标列表）。 + 断点续传时先查此接口，只补传缺失分片。 +3. **传分片** `POST .../chunks/{index}`，body 为原始二进制。分片可乱序、可重传覆盖。 +4. **完成** `POST .../complete`，服务端按 index 顺序拼接为正式视频文件，创建转写 Task + 并入队。complete 幂等：重复调用返回同一 `task_id`。 + +### 请求/响应示例 + +创建会话： + +```bash +curl -X POST http://127.0.0.1:8000/api/tasks/chunk-uploads \ + -H 'Content-Type: application/json' \ + -d '{"filename":"demo.mp4","size_bytes":10485760,"chunk_size":4194304,"total_chunks":3}' +# → {"upload_id":"a1b2...","filename":"demo.mp4","size_bytes":10485760,"chunk_size":4194304,"total_chunks":3} +``` + +查任务状态： + +```bash +curl http://127.0.0.1:8000/api/tasks/1 +# → {"id":1,"filename":"demo.mp4","status":"done","progress":100.0,"error":null,"has_subtitle":true,...} +``` + +下载字幕： + +```bash +curl -OJ http://127.0.0.1:8000/api/tasks/1/subtitle?type=bilingual +``` + +--- + +## 断句与时间戳重算原理 + +Whisper 原始 segment 的断句通常很混乱：每段不是完整句子，时间戳也不对齐句界。 +`segmenter.py` 基于词级时间戳重组，两路策略： + +### 精确路（`word_timestamps=true`，默认） + +1. 汇集所有词的 `(text, start, end)`。 +2. 按**句末标点**（`. ! ? ;`）切句。 +3. 超长句（> `max_words_per_line` 或 > `max_duration_seconds`）按**逗号**（`, : —`）再拆； + 无逗号则按词数等分。 +4. 每条字幕的时间戳：`start = 首词.start`，`end = 末词.end`，**精确无误**。 + +### 匀速估算路（无词级时间戳时 fallback） + +段内按字符数比例分配时间 —— 即「短时匀速」假设，零模型开销： + +``` +句start = 段start + (前缀字符数 / 段总字符数) × 段时长 +``` + +### SRT 规范化 + +最后统一处理：单条 1–7 秒（过短合并）、≤2 行、每行 ≤42 字符（按词折行）。 + +--- + +## 模型不共驻（显存策略） + +ASR 与翻译模型**不会同时驻留 GPU**。`model_manager.py` 单例跟踪当前加载的模型类型： + +- `get_translator()`：若 ASR 在内存 → 先 `del WhisperModel` + `gc.collect()` + + `torch.cuda.empty_cache()` 释放显存 → 再加载 NLLB。 +- `get_asr()`：若翻译器在内存 → 先卸载 → 再加载 Whisper。 + +翻译阶段独占显存，因此可用大 `batch_size`。24G 3090 上：Whisper large-v3-turbo FP16 +~3GB / NLLB-1.3B FP16 ~2.5GB，互不叠加，远低于显存上限。 + +--- + +## Docker 说明 + +### 一份 Dockerfile，两个镜像 + +`ARG VARIANT=cpu|gpu` 控制基础镜像与 torch 轮子： + +| VARIANT | 基础镜像 | torch | +|---|---|---| +| `cpu`（默认） | `python:3.12-slim` | CPU 版（`--index-url .../whl/cpu`） | +| `gpu` | `nvidia/cuda:12.1.0-runtime-ubuntu22.04` | CUDA 版 | + +两个镜像的 Python 依赖列表（`requirements.txt`）完全一致，仅 torch 不同。镜像内 apt 装 +`ffmpeg` + `patchelf`。 + +### Volume 挂载 + +| 容器路径 | 宿主路径 | 用途 | +|---|---|---| +| `/data` | `./data` | 上传视频、中间音频、输出字幕、SQLite 数据库 | +| `/models` | `./models` | 模型缓存（HF + ctranslate2），跨容器复用避免重下 | +| `/app/config.yaml` | `./config.yaml` | 配置文件（只读挂载） | + +镜像本身无状态、无敏感数据。 + +### docker-compose + +`docker-compose.yml` 提供 `audio2text-cpu` / `audio2text-gpu` 两个 profile： + +```bash +docker compose --profile cpu up -d # CPU +docker compose --profile gpu up -d # GPU（需 nvidia runtime） +``` + +### ctranslate2 可执行栈修复 + +ctranslate2 的 `.so`（在 `ctranslate2.libs/` 隐藏目录）带 PT_GNU_STACK 可执行栈标志， +在某些内核 + Docker 组合下会报 `cannot enable executable stack as shared object requires`。 +Dockerfile 在构建时用 `patchelf --clear-execstack` 清掉该标志，无需放宽容器安全策略。 +构建末尾有 `python -c "import ctranslate2"` 验证。 + +--- + +## 依赖 + +### Python（`requirements.txt`） + +| 包 | 用途 | +|---|---| +| `fastapi` + `uvicorn[standard]` + `python-multipart` | Web 服务 | +| `pydantic` + `pydantic-settings` | 配置类型校验 | +| `PyYAML` | 读 config.yaml | +| `SQLAlchemy` | SQLite ORM | +| `faster-whisper` + `ctranslate2` | 语音识别 | +| `transformers` + `sentencepiece` + `accelerate` | NLLB 翻译 | +| `psutil` | 进程信息 | + +torch 单独安装（CPU / CUDA 轮子不同），不在 requirements.txt 中。 + +### 系统 + +- `ffmpeg`（镜像内 apt 装）—— 提取音频 +- `patchelf`（镜像内 apt 装）—— 修复 ctranslate2 可执行栈 +- GPU 镜像额外需要宿主 NVIDIA 驱动 + nvidia container runtime + +--- + +## 常见问题 + +### Q: CPU 开发机能跑 NLLB 吗？ + +`config.cpu.yaml` 默认用 opus-mt-en-zh（~300MB），2GB 内存开发机即可跑通完整流程。 +若想在 CPU 上验证 NLLB 翻译质量，可手动改 `translation.model`： +- `facebook/nllb-200-distilled-600M`（~1.2GB，同系列最小）——需 ≥4GB 内存，2GB 机会 OOM。 +- `facebook/nllb-200-distilled-1.3B`（~2.5GB，GPU 生产同款）——需 ~5GB 内存。 + +生产环境（3090 24G）用 NLLB-1.3B 质量最好。 + +### Q: 模型下载到哪里？每次启动都重下吗？ + +模型缓存到 `/models` volume（`HF_HOME=/models/huggingface`、`CT2_CACHE=/models/ctranslate2`）。 +首次启动下载，之后跨容器复用秒起。删除 `./models` 目录会强制重下。 + +### Q: 上传大视频中断了怎么办？ + +分片上传支持断点续传。重新上传同一文件时，前端先调 `status` 接口查已传分片，只补传缺失的。 +分片可乱序、可重传覆盖。 + +### Q: 怎么保留原始视频不删？ + +把 `config.yaml` 的 `processing.delete_original_after_extract` 改为 `false`。 +注意：保留的视频仍受缓存清理策略约束——任务超期（默认 7 天）后会被 `cache_cleaner` +连同字幕一起删除。想永久保留请把 `storage.cache_retention_days` 设为 `0`（禁用清理）。 + +### Q: 字幕 / 任务记录多久会被自动清理？能禁用吗？ + +默认保留 7 天（`storage.cache_retention_days`）。超期任务的字幕、中间音频、保留的原始 +视频连同 DB 记录一并删除，启动时跑一次 + 每 `cache_cleanup_interval_hours`（默认 24h） +循环一次。设 `cache_retention_days: 0` 可禁用自动清理（产物永久保留，需自行管理磁盘）。 +手动触发：`docker exec audio2text python -m app.services.cache_cleaner`。 + +### Q: GPU 镜像构建好了但 start.sh 还是用 CPU？ + +`start.sh` 检测到 `audio2text:gpu` 镜像**且**本机有 `nvidia-smi` 才用 GPU。确认宿主装了 +NVIDIA 驱动 + nvidia container runtime。也可用 `docker compose --profile gpu up -d` 显式启动。 diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..36cef0a --- /dev/null +++ b/app/config.py @@ -0,0 +1,136 @@ +"""运行时配置：所有参数从 config.yaml 读取，对齐 server/config.py 的风格。 + +CPU dev / GPU prod 仅靠 device / model / compute_type 三项切换，代码完全不变。 +""" + +from __future__ import annotations + +import os +from functools import lru_cache +from pathlib import Path + +import yaml +from pydantic import BaseModel, Field, field_validator + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_CONFIG_PATH = PROJECT_ROOT / "config.yaml" + + +class ServerConfig(BaseModel): + host: str = "0.0.0.0" + port: int = 8000 + workers: int = 1 + + +class StorageConfig(BaseModel): + upload_dir: str = "/data/uploads" + work_dir: str = "/data/.work" + output_dir: str = "/data/outputs" + chunk_bytes: int = 1024 * 1024 + chunk_session_ttl_seconds: int = 300 + # 缓存保留：超期任务产物（输出字幕 / 中间音频 / 保留的原始视频）连同 DB 记录一并清理 + cache_retention_days: int = 7 + # 定时清理间隔（小时）：启动时跑一次，之后按此间隔循环 + cache_cleanup_interval_hours: int = 24 + + +class ProcessingConfig(BaseModel): + delete_original_after_extract: bool = True + keep_audio: bool = False + + +class AsrConfig(BaseModel): + model: str = "small" + device: str = "cpu" # cpu | cuda + compute_type: str = "int8" # cpu: int8；gpu: float16 + language: str = "en" + word_timestamps: bool = True + vad_filter: bool = True + + +class TranslationConfig(BaseModel): + model: str = "facebook/nllb-200-distilled-1.3B" + device: str = "cpu" # cpu | cuda + src_lang: str = "eng_Latn" + tgt_lang: str = "zho_Hans" + batch_size: int = 16 + max_length: int = 256 + + +class SegmentationConfig(BaseModel): + max_words_per_line: int = 14 + max_duration_seconds: float = 7.0 + min_duration_seconds: float = 1.0 + max_chars_per_line: int = 42 + + +class LoggingConfig(BaseModel): + """日志配置：控制台 + 内存缓冲的最低级别，以及缓冲条数。 + + 分层语义： + - debug：详细（ffmpeg 命令、模型加载/卸载、转写逐段、翻译逐批进度） + - info：简略（仅任务阶段转换，如 "任务 N [transcribing 55%]"） + - error：详细错误（完整 traceback，由 logger.exception 自带） + """ + + level: str = "info" # debug | info | warning | error + buffer_size: int = 2000 # /logs 页面内存缓冲条数 + + @field_validator("level", mode="before") + @classmethod + def _normalize_level(cls, v): + return str(v).lower() if v else "info" + + +class DocsConfig(BaseModel): + """/docs Basic Auth 凭据（明文，对齐 server）。""" + + enabled: bool = True + username: str = "admin" + password: str = "" + realm: str = "audio2text docs" + + @field_validator("password", mode="before") + @classmethod + def _coerce(cls, v): + return "" if v is None else str(v) + + +class Settings(BaseModel): + server: ServerConfig = ServerConfig() + storage: StorageConfig = StorageConfig() + processing: ProcessingConfig = ProcessingConfig() + asr: AsrConfig = AsrConfig() + translation: TranslationConfig = TranslationConfig() + segmentation: SegmentationConfig = SegmentationConfig() + logging: LoggingConfig = LoggingConfig() + docs: DocsConfig = DocsConfig() + + def upload_dir(self) -> Path: + return Path(self.storage.upload_dir) + + def work_dir(self) -> Path: + return Path(self.storage.work_dir) + + def output_dir(self) -> Path: + return Path(self.storage.output_dir) + + +def _load_yaml(path: Path) -> dict: + if not path.exists(): + raise FileNotFoundError( + f"未找到配置文件 {path}；请先 cp config.example.yaml config.yaml 并填值。" + ) + return yaml.safe_load(path.read_text(encoding="utf-8")) or {} + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + path = Path(os.getenv("CONFIG_PATH", str(DEFAULT_CONFIG_PATH))) + return Settings.model_validate(_load_yaml(path)) + + +def reload_settings() -> Settings: + """清缓存并重新读取，供脚本与测试使用。""" + get_settings.cache_clear() + return get_settings() diff --git a/app/controllers/__init__.py b/app/controllers/__init__.py new file mode 100644 index 0000000..884ebcc --- /dev/null +++ b/app/controllers/__init__.py @@ -0,0 +1,7 @@ +"""路由聚合：导出各 controller 的 router，供 main.py include。""" + +from .log_router import router as log_router +from .task_router import router as task_router +from .upload_router import router as upload_router + +__all__ = ["log_router", "task_router", "upload_router"] diff --git a/app/controllers/log_router.py b/app/controllers/log_router.py new file mode 100644 index 0000000..399c2b2 --- /dev/null +++ b/app/controllers/log_router.py @@ -0,0 +1,31 @@ +"""日志路由：查询内存日志缓冲，供 /logs 页面消费。 + +分层语义： +- debug：详细（ffmpeg 命令、模型加载/卸载、转写逐段、翻译逐批进度） +- info：简略（仅任务阶段转换，如 "任务 N [transcribing 55%]"） +- warning / error：更详细，error 含完整 traceback +""" + +from __future__ import annotations + +from fastapi import APIRouter, Query + +from ..services.log_buffer import get_log_buffer + +router = APIRouter(prefix="/api/logs", tags=["logs"]) + + +@router.get("", summary="查询日志（按级别过滤）") +def get_logs( + level: str = Query("info", pattern="^(debug|info|warning|error)$", + description="最小级别：debug=详细, info=简略, error=仅错误"), + tail: int = Query(200, ge=1, le=2000, description="最多返回条数"), +) -> dict: + records = get_log_buffer().get_records(level=level, tail=tail) + return {"level": level, "tail": tail, "count": len(records), "logs": records} + + +@router.delete("", summary="清空日志缓冲") +def clear_logs() -> dict: + get_log_buffer().clear() + return {"status": "cleared"} diff --git a/app/controllers/task_router.py b/app/controllers/task_router.py new file mode 100644 index 0000000..eaec1bc --- /dev/null +++ b/app/controllers/task_router.py @@ -0,0 +1,88 @@ +"""任务路由：列表 / 状态 / 下载字幕。""" + +from __future__ import annotations + +from pathlib import Path + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import FileResponse +from sqlalchemy.orm import Session + +from ..config import get_settings +from ..database import get_db +from ..models.task import Task +from ..schemas.task import TaskListResponse, TaskResponse + +router = APIRouter(prefix="/api/tasks", tags=["task"]) + + +def _to_resp(task: Task) -> TaskResponse: + return TaskResponse( + id=task.id, + filename=task.filename, + status=task.status, + progress=task.progress, + error=task.error, + created_at=task.created_at, + updated_at=task.updated_at, + ) + + +@router.get("", response_model=TaskListResponse, summary="任务列表") +def list_tasks( + limit: int = Query(100, ge=1, le=500), + offset: int = Query(0, ge=0), + q: str = Query("", description="按文件名模糊搜索（大小写不敏感，匹配子串）"), + db: Session = Depends(get_db), +) -> TaskListResponse: + q_base = db.query(Task).order_by(Task.id.desc()) + if q.strip(): + # SQLite 的 LIKE 默认大小写不敏感（ASCII），ilike 等价于 LIKE + like = f"%{q.strip()}%" + q_base = q_base.filter(Task.filename.ilike(like)) + total = q_base.count() + tasks = q_base.offset(offset).limit(limit).all() + return TaskListResponse(tasks=[_to_resp(t) for t in tasks], total=total) + + +@router.get("/{task_id}", response_model=TaskResponse, summary="任务状态") +def get_task(task_id: int, db: Session = Depends(get_db)) -> TaskResponse: + task = db.get(Task, task_id) + if task is None: + raise HTTPException(404, f"任务不存在：{task_id}") + return _to_resp(task) + + +@router.get("/{task_id}/subtitle", summary="下载字幕") +def download_subtitle( + task_id: int, + type: str = Query("bilingual", pattern="^(bilingual|en|zh)$"), + db: Session = Depends(get_db), +) -> FileResponse: + task = db.get(Task, task_id) + if task is None: + raise HTTPException(404, f"任务不存在：{task_id}") + if task.status != "done": + raise HTTPException(409, f"任务尚未完成（当前状态：{task.status}）") + + rel = { + "bilingual": task.bilingual_srt_path, + "en": task.en_srt_path, + "zh": task.zh_srt_path, + }[type] + if not rel: + raise HTTPException(404, f"该类型字幕不存在：{type}") + + s = get_settings() + path = s.output_dir() / rel + if not path.is_file(): + raise HTTPException(404, f"字幕文件丢失：{path}") + + base = Path(task.filename).stem + suffix = "" if type == "bilingual" else f".{type}" + download_name = f"{base}{suffix}.srt" + return FileResponse( + path=path, + media_type="application/x-subrip", + filename=download_name, + ) diff --git a/app/controllers/upload_router.py b/app/controllers/upload_router.py new file mode 100644 index 0000000..35116d4 --- /dev/null +++ b/app/controllers/upload_router.py @@ -0,0 +1,69 @@ +"""分片上传路由：建会话 / 查状态 / 传分片 / complete。 + +协议与 server 完全一致，区别仅在 complete 后创建的是转写 Task 而非 UploadedFile。 +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, Request +from sqlalchemy.orm import Session + +from ..database import get_db +from ..schemas.task import ( + ChunkUploadResponse, + CompleteResponse, + CreateSessionRequest, + CreateSessionResponse, + SessionStatusResponse, +) +from ..services.upload_service import UploadService + +router = APIRouter(prefix="/api/tasks/chunk-uploads", tags=["upload"]) + + +def _service(db: Session = Depends(get_db)) -> UploadService: + return UploadService(db) + + +@router.post("", response_model=CreateSessionResponse, summary="创建分片上传会话") +def create_session( + body: CreateSessionRequest, + service: UploadService = Depends(_service), +) -> CreateSessionResponse: + return service.create_session(body) + + +@router.get("/{upload_id}/status", response_model=SessionStatusResponse, summary="查询会话状态（断点续传）") +def session_status( + upload_id: str, + service: UploadService = Depends(_service), +) -> SessionStatusResponse: + return service.get_status(upload_id) + + +@router.post("/{upload_id}/chunks/{index}", response_model=ChunkUploadResponse, summary="上传单个分片") +async def upload_chunk( + upload_id: str, + index: int, + request: Request, + service: UploadService = Depends(_service), +) -> ChunkUploadResponse: + uploaded = service.write_chunk(upload_id, index, await request.body()) + return ChunkUploadResponse(upload_id=upload_id, index=index, uploaded_chunks=uploaded) + + +@router.post("/{upload_id}/complete", response_model=CompleteResponse, summary="完成拼接并创建转写任务") +def complete_session( + upload_id: str, + service: UploadService = Depends(_service), +) -> CompleteResponse: + """拼接分片 + 创建转写任务 + 入队管线。 + + controller 负责编排：service.complete 只管存储（拼接 + 建 Task）， + 管线触发由 controller 调用，service 不依赖 pipeline（避免循环依赖）。 + """ + resp = service.complete(upload_id) + # 仅新建任务时入队（幂等 complete 返回的也是同一 task_id，enqueue 幂等无副作用） + from ..services.pipeline import enqueue_task + enqueue_task(resp.task_id) + return resp diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..3b74a97 --- /dev/null +++ b/app/database.py @@ -0,0 +1,68 @@ +"""SQLite 引擎、Session、Base、get_db 依赖。 + +自包含：无需外部 MySQL，容器内单文件 SQLite 即可。对齐 server/database.py 的接口形态， +但用 SQLite（本项目独立运行、无并发写入压力）。 +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Generator + +from sqlalchemy import create_engine +from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker + +from .config import get_settings + +_engine = None +_SessionLocal = None + + +def _db_path() -> Path: + """SQLite 文件落在 work_dir 下，跟数据一起走 volume。""" + s = get_settings() + p = s.work_dir() / "audio2text.db" + p.parent.mkdir(parents=True, exist_ok=True) + return p + + +def get_engine(): + global _engine, _SessionLocal + if _engine is None: + url = f"sqlite:///{_db_path()}" + _engine = create_engine( + url, + # 后台 reaper / 缓存清理线程与请求线程并发写同一库；busy_timeout 让等待方 + # 在拿锁时阻塞 5s 而非立即报 database is locked。 + connect_args={"check_same_thread": False, "timeout": 30}, + future=True, + ) + _SessionLocal = sessionmaker(bind=_engine, autoflush=False, expire_on_commit=False) + return _engine + + +def get_session_local(): + get_engine() + return _SessionLocal + + +class Base(DeclarativeBase): + pass + + +def init_db_schema() -> None: + """建表（幂等）。""" + from .models.task import Task # noqa: F401 + from .models.upload_session import UploadSession # noqa: F401 + + get_engine() + Base.metadata.create_all(get_engine()) + + +def get_db() -> Generator[Session, None, None]: + """FastAPI 依赖：每请求一个 Session，结束自动关闭。""" + db = get_session_local()() + try: + yield db + finally: + db.close() diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..f8f8586 --- /dev/null +++ b/app/main.py @@ -0,0 +1,160 @@ +"""FastAPI 应用工厂（对齐 server/main.py 风格）。 + +路由概览： + GET / -> 版本号 + GET /health -> 存活探针 + GET /logs -> 实时日志页（公开） + GET /docs -> Swagger UI（Basic Auth） + POST /api/tasks/chunk-uploads/... -> 分片上传（建会话/查状态/传片/complete） + GET /api/tasks -> 任务列表 + GET /api/tasks/{id} -> 任务状态 + GET /api/tasks/{id}/subtitle -> 下载字幕 + GET /api/logs -> 查询日志（按级别过滤） +""" + +from __future__ import annotations + +import asyncio +import logging +import threading +from contextlib import asynccontextmanager + +from fastapi import Depends, FastAPI +from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html +from fastapi.responses import HTMLResponse, JSONResponse +from sqlalchemy.orm import Session + +from .config import get_settings +from .controllers import log_router, task_router, upload_router +from .database import get_db, init_db_schema +from .security import require_docs_auth +from .services.cache_cleaner import purge_expired_cache, run_forever as run_cache_cleaner +from .services.log_buffer import init_log_buffer +from .services.reaper import reap_stale_sessions +from .views.history_html import render as render_history_html +from .views.home_html import render as render_home_html +from .views.logs_html import render as render_logs_html + +# 日志分层： +# - audio2text logger 始终设 DEBUG，确保所有记录（含子步骤）都能产生。 +# - 控制台 handler 用配置的 level：config=info 时控制台只显示简略。 +# - 内存 handler 始终 DEBUG（全收），/api/logs?level= 查询时按参数过滤可见级别。 +# 这样 /logs 页面切到 DEBUG 能看到详细，而控制台仍按 config.level 简略输出。 +_LOG_LEVEL_MAP = {"debug": logging.DEBUG, "info": logging.INFO, + "warning": logging.WARNING, "error": logging.ERROR} +_console_level = _LOG_LEVEL_MAP.get(get_settings().logging.level, logging.INFO) +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +# 控制台 handler（basicConfig 安装在 root）按配置级别过滤 +for _h in logging.getLogger().handlers: + _h.setLevel(_console_level) +logger = logging.getLogger("audio2text") +logger.setLevel(logging.DEBUG) + +# 内存日志缓冲：捕获所有 audio2text.* 日志，供 /logs 页面查询。 +# handler 级别设为 DEBUG（最低），由 /api/logs 查询时按 level 参数过滤。 +_memory_handler = init_log_buffer(get_settings().logging.buffer_size) +_memory_handler.setLevel(logging.DEBUG) +logger.addHandler(_memory_handler) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + try: + init_db_schema() + logger.info("SQLite 表已就绪。") + except Exception as exc: # pragma: no cover + logger.error("初始化数据库失败：%s", exc) + logger.info( + "audio2text 启动：ASR=%s/%s，翻译=%s", + get_settings().asr.model, get_settings().asr.device, get_settings().translation.device, + ) + # 启动时清理一次被放弃的分片会话 + try: + await asyncio.to_thread(reap_stale_sessions) + except Exception as exc: # pragma: no cover + logger.warning("启动 reaper 失败：%s", exc) + # 缓存清理：启动时跑一次 + 后台定时循环（守护线程，随进程退出） + s = get_settings() + try: + await asyncio.to_thread(purge_expired_cache) + except Exception as exc: # pragma: no cover + logger.warning("启动缓存清理失败：%s", exc) + try: + t = threading.Thread( + target=run_cache_cleaner, + args=(s.storage.cache_cleanup_interval_hours,), + name="cache-cleaner", + daemon=True, + ) + t.start() + logger.info( + "缓存清理调度已启动：保留期 %d 天，间隔 %d 小时。", + s.storage.cache_retention_days, s.storage.cache_cleanup_interval_hours, + ) + except Exception as exc: # pragma: no cover + logger.warning("启动缓存清理调度失败：%s", exc) + yield + + +def create_app() -> FastAPI: + s = get_settings() + app = FastAPI( + title="audio2text", + description=( + "音频/视频转双语字幕服务。\n\n" + "- 上传视频 → ffmpeg 提取音频 → faster-whisper 识别英语 → 断句+时间戳重算\n" + " → NLLB 翻译为中文 → 双语 SRT\n\n" + "模型不共驻：ASR 与翻译分阶段加载，翻译时独占显存跑大 batch。\n" + "/docs 需 Basic Auth，凭据见 config.yaml 的 docs 段。" + ), + version="1.0.0", + docs_url=None, + redoc_url=None, + openapi_url=None, + lifespan=lifespan, + ) + + app.include_router(upload_router) + app.include_router(task_router) + app.include_router(log_router) + + # 受 Basic Auth 保护的文档接口 + @app.get("/openapi.json") + def protected_openapi(_: str = Depends(require_docs_auth)) -> JSONResponse: + return JSONResponse(app.openapi()) + + @app.get("/docs") + def protected_docs(_: str = Depends(require_docs_auth)): + return get_swagger_ui_html( + openapi_url="/openapi.json", title="audio2text docs", swagger_favicon_url="" + ) + + @app.get("/redoc") + def protected_redoc(_: str = Depends(require_docs_auth)): + return get_redoc_html( + openapi_url="/openapi.json", title="audio2text docs", redoc_favicon_url="" + ) + + @app.get("/", response_class=HTMLResponse) + def home_page() -> HTMLResponse: + return HTMLResponse(render_home_html()) + + @app.get("/health") + def health() -> dict: + return {"status": "ok"} + + @app.get("/history", response_class=HTMLResponse) + def history_page() -> HTMLResponse: + return HTMLResponse(render_history_html()) + + @app.get("/logs", response_class=HTMLResponse) + def logs_page() -> HTMLResponse: + return HTMLResponse(render_logs_html()) + + return app + + +app = create_app() diff --git a/app/models/__init__.py b/app/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/models/task.py b/app/models/task.py new file mode 100644 index 0000000..17b8140 --- /dev/null +++ b/app/models/task.py @@ -0,0 +1,40 @@ +"""Task ORM：一个上传完成的视频对应一个转写任务，承载状态机。""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import DateTime, Float, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from ..database import Base + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class Task(Base): + __tablename__ = "task" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + # 原始视频文件名（用户上传时的名字） + filename: Mapped[str] = mapped_column(String(512), nullable=False) + # 视频在 upload_dir 下的相对路径（提取音频前后可能被删） + source_path: Mapped[str] = mapped_column(String(1024), nullable=False) + # 状态机：queued → extracting → transcribing → segmenting → translating → done | failed + status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") + # 0-100 进度 + progress: Mapped[float] = mapped_column(Float, nullable=False, default=0.0) + # 失败原因 + error: Mapped[str | None] = mapped_column(Text, nullable=True) + # 输出字幕路径（双语合并 / 英文 / 中文） + bilingual_srt_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) + en_srt_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) + zh_srt_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) + + created_at: Mapped[datetime] = mapped_column(DateTime, default=_now) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=_now, onupdate=_now) + + def __repr__(self) -> str: + return f"" diff --git a/app/models/upload_session.py b/app/models/upload_session.py new file mode 100644 index 0000000..374bdac --- /dev/null +++ b/app/models/upload_session.py @@ -0,0 +1,49 @@ +"""分片上传会话 ORM：支撑断点续传。对齐 server 的 UploadSession 形态（SQLite 版）。""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, TypeDecorator +from sqlalchemy.orm import Mapped, mapped_column + +from ..database import Base + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class _IntList(TypeDecorator): + """把 list[int] 存成 JSON 字符串。SQLite 没有 ARRAY 类型。""" + + impl = Text + cache_ok = True + + def process_bind_param(self, value: Any, dialect) -> str | None: + return json.dumps(value) if value is not None else None + + def process_result_value(self, value: Any, dialect) -> list[int]: + return json.loads(value) if value else [] + + +class UploadSession(Base): + __tablename__ = "upload_session" + + upload_id: Mapped[str] = mapped_column(String(64), primary_key=True) + filename: Mapped[str] = mapped_column(String(512), nullable=False) + size_bytes: Mapped[int] = mapped_column(Integer, nullable=False) + chunk_size: Mapped[int] = mapped_column(Integer, nullable=False) + total_chunks: Mapped[int] = mapped_column(Integer, nullable=False) + uploaded_chunks: Mapped[list[int]] = mapped_column(_IntList, default=list) + # pending → completed（complete 成功）| abandoned（reaper 清理） + status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending") + # 拼接完成后的视频相对 upload_dir 路径 + final_path: Mapped[str | None] = mapped_column(String(1024), nullable=True) + # complete 后关联的 Task.id（直接引用，避免反向查找 source_path） + task_id: Mapped[int | None] = mapped_column(ForeignKey("task.id"), nullable=True) + + created_at: Mapped[datetime] = mapped_column(DateTime, default=_now) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=_now, onupdate=_now) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/schemas/task.py b/app/schemas/task.py new file mode 100644 index 0000000..dfb1248 --- /dev/null +++ b/app/schemas/task.py @@ -0,0 +1,69 @@ +"""任务与分片上传接口 DTO。""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +# ---------------- 分片上传（对齐 server 的协议） ---------------- + +class CreateSessionRequest(BaseModel): + """创建一个分片上传会话。""" + + filename: str = Field(..., description="客户端原始文件名") + size_bytes: int = Field(..., ge=0, description="文件总字节数") + chunk_size: int = Field(..., gt=0, description="分片大小（字节）") + total_chunks: int = Field(..., gt=0, description="分片总数") + + +class CreateSessionResponse(BaseModel): + upload_id: str + filename: str + size_bytes: int + chunk_size: int + total_chunks: int + + +class SessionStatusResponse(BaseModel): + upload_id: str + filename: str + size_bytes: int + chunk_size: int + total_chunks: int + uploaded_chunks: list[int] + completed: bool + task_id: int | None = Field(None, description="completed=true 时指向 Task.id") + + +class ChunkUploadResponse(BaseModel): + upload_id: str + index: int + uploaded_chunks: list[int] + + +class CompleteResponse(BaseModel): + """complete 的回执：创建出转写任务，返回 task_id。""" + + task_id: int + filename: str + size_bytes: int + status: str + + +# ---------------- 任务查询 ---------------- + +class TaskResponse(BaseModel): + id: int + filename: str + status: str + progress: float + error: str | None + created_at: datetime + updated_at: datetime + + +class TaskListResponse(BaseModel): + tasks: list[TaskResponse] + total: int diff --git a/app/security.py b/app/security.py new file mode 100644 index 0000000..c76c150 --- /dev/null +++ b/app/security.py @@ -0,0 +1,27 @@ +"""/docs Basic Auth：对齐 server/security.py。明文密码，常量时间比较。""" + +from __future__ import annotations + +import secrets + +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPBasic, HTTPBasicCredentials + +from .config import get_settings + +security = HTTPBasic() + + +def require_docs_auth(credentials: HTTPBasicCredentials = Depends(security)) -> str: + s = get_settings() + if not s.docs.enabled or not s.docs.password: + raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="docs disabled") + ok_user = secrets.compare_digest(credentials.username.encode(), s.docs.username.encode()) + ok_pass = secrets.compare_digest(credentials.password.encode(), s.docs.password.encode()) + if not (ok_user and ok_pass): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid credentials", + headers={"WWW-Authenticate": f'Basic realm="{s.docs.realm}"'}, + ) + return credentials.username diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/asr_service.py b/app/services/asr_service.py new file mode 100644 index 0000000..ffca502 --- /dev/null +++ b/app/services/asr_service.py @@ -0,0 +1,65 @@ +"""语音识别服务：faster-whisper，输出带词级时间戳的 segments。 + +CPU dev: tiny.en + int8；GPU prod: large-v3-turbo + float16。同一份代码。 +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +from ..config import get_settings +from .model_manager import get_model_manager +from .types import Segment, Word + +logger = logging.getLogger("audio2text.asr") + + +def transcribe(wav_path: Path) -> list[Segment]: + """转写 wav，返回 segments（含词级时间戳）。 + + Args: + wav_path: 16kHz mono PCM wav + + Returns: + list[Segment]，每个 Segment 带词级 words（若 word_timestamps 启用）。 + """ + s = get_settings().asr + if not wav_path.is_file(): + raise FileNotFoundError(f"音频不存在：{wav_path}") + + model = get_model_manager().get_asr() + logger.debug("开始转写 %s（model=%s language=%s）", wav_path.name, s.model, s.language) + + segments_gen, info = model.transcribe( + str(wav_path), + language=s.language, + word_timestamps=s.word_timestamps, + vad_filter=s.vad_filter, + beam_size=5, + ) + logger.debug( + "音频时长 %.1fs，检测语言=%s（置信度 %.2f）", + info.duration, info.language, info.language_probability, + ) + + segments: list[Segment] = [] + for seg in segments_gen: + words: list[Word] = [] + if s.word_timestamps and getattr(seg, "words", None): + for w in seg.words: + words.append(Word( + text=w.word.strip(), + start=float(w.start), + end=float(w.end), + probability=float(getattr(w, "probability", 1.0)), + )) + segments.append(Segment( + text=seg.text.strip(), + start=float(seg.start), + end=float(seg.end), + words=words, + )) + logger.debug("转写完成：%d 段，%d 词。", + len(segments), sum(len(s.words) for s in segments)) + return segments diff --git a/app/services/cache_cleaner.py b/app/services/cache_cleaner.py new file mode 100644 index 0000000..1637168 --- /dev/null +++ b/app/services/cache_cleaner.py @@ -0,0 +1,194 @@ +"""缓存清理：删除超期任务产物 + 孤儿目录，并删除对应 DB 记录。 + +「缓存」指每个任务落盘的产物： + /task_/ 双语 / 英文 / 中文字幕 + /task_.wav 中间音频（若 keep_audio=true 未被管线删掉） + /yyyy/mm/. 保留的原始视频（若 delete_original_after_extract=false） + +超期 = `created_at` 早于 `now - cache_retention_days`。超期任务的全部产物连同 +Task / UploadSession 行一起删除，避免历史页出现指向已删文件的死链接。 + +另做一次孤儿扫描：output_dir / work_dir 下存在但无对应 Task 的目录（进程崩溃残留）， +按目录 mtime 判超期后删除，让磁盘不被异常退出留下的碎片占满。 +""" + +from __future__ import annotations + +import logging +import shutil +import time +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from sqlalchemy.orm import Session + +from ..config import get_settings +from ..database import get_session_local +from ..models.task import Task +from ..models.upload_session import UploadSession + +logger = logging.getLogger("audio2text.cache") + + +def _now_utc() -> datetime: + return datetime.now(timezone.utc) + + +def _aware(dt: datetime | None) -> datetime | None: + """SQLite 存 naive datetime，统一补 UTC。""" + if dt is None: + return None + return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc) + + +def _rm_tree(path: Path) -> bool: + """删目录或文件，失败仅警告不抛。返回是否实际删除。""" + try: + if path.is_dir(): + shutil.rmtree(path, ignore_errors=False) + return True + if path.is_file(): + path.unlink(missing_ok=True) + return True + except Exception as exc: # pragma: no cover + logger.warning("清理 %s 失败：%s", path, exc) + return False + + +def purge_expired_cache(db: Session | None = None) -> dict: + """执行一次清理，返回统计 dict。 + + 可传入已有 Session（如复用请求会话），不传则自建一个并关闭。 + """ + own_db = db is None + if own_db: + db = get_session_local()() + try: + return _purge(db) + finally: + if own_db: + db.close() + + +def _purge(db: Session) -> dict: + s = get_settings() + retention = max(0, s.storage.cache_retention_days) + cutoff = _now_utc() - timedelta(days=retention) + upload_root = s.upload_dir() + work_root = s.work_dir() + output_root = s.output_dir() + + stats = {"tasks": 0, "outputs": 0, "audio": 0, "videos": 0, + "orphans": 0, "retention_days": retention} + + if retention <= 0: + logger.info("缓存清理已禁用（cache_retention_days=%s）。", retention) + return stats + + # ---------- 1. 超期任务 ---------- + tasks = db.query(Task).all() + for task in tasks: + created = _aware(task.created_at) + if created is None or created >= cutoff: + continue + # 输出目录 + out_dir = output_root / f"task_{task.id}" + if out_dir.is_dir() and _rm_tree(out_dir): + stats["outputs"] += 1 + # 中间音频 + wav = work_root / f"task_{task.id}.wav" + if wav.is_file() and _rm_tree(wav): + stats["audio"] += 1 + # 保留的原始视频（若未在提取后删除） + if task.source_path: + src = upload_root / task.source_path + if src.is_file() and _rm_tree(src): + stats["videos"] += 1 + # 删 DB 记录（先删关联的 UploadSession，再删 Task） + db.query(UploadSession).filter(UploadSession.task_id == task.id).delete() + db.delete(task) + stats["tasks"] += 1 + logger.info( + "清理超期任务 id=%s file=%s created=%s", + task.id, task.filename, created.isoformat(), + ) + if stats["tasks"]: + db.commit() + + # ---------- 2. 孤儿目录扫描 ---------- + existing_ids = {row[0] for row in db.query(Task.id).all()} + stats["orphans"] += _purge_orphans(output_root, "task_", cutoff, existing_ids) + stats["orphans"] += _purge_orphans(work_root, "task_", cutoff, existing_ids, suffix=".wav") + + logger.info( + "缓存清理完成：任务 %d（输出 %d / 音频 %d / 视频 %d）+ 孤儿 %d，保留期 %d 天", + stats["tasks"], stats["outputs"], stats["audio"], stats["videos"], + stats["orphans"], retention, + ) + return stats + + +def _purge_orphans(root: Path, prefix: str, cutoff: datetime, + existing_ids: set[int], suffix: str | None = None) -> int: + """删 root 下名为 `prefix` 但无对应 Task 的孤儿条目。 + + output_dir 下是目录（task_）；work_dir 下可能是 task_.wav 文件。 + 用 mtime 判超期，避免误删刚产生的中间产物。 + """ + if not root.is_dir(): + return 0 + n = 0 + for entry in root.iterdir(): + name = entry.name + if not name.startswith(prefix): + continue + rest = name[len(prefix):] + if suffix: + if not rest.endswith(suffix): + continue + rest = rest[: -len(suffix)] + try: + tid = int(rest) + except ValueError: + continue # 不是 task_ 命名，跳过 + if tid in existing_ids: + continue + # 孤儿：按 mtime 判超期 + try: + mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=timezone.utc) + except OSError: + continue + if mtime >= cutoff: + continue + if _rm_tree(entry): + n += 1 + logger.info("清理孤儿缓存 %s（mtime=%s）", entry, mtime.isoformat()) + return n + + +def run_forever(interval_hours: int) -> None: + """后台线程入口：先跑一次，再按间隔循环。供 main.py lifespan 拉起。""" + interval = max(1, interval_hours) * 3600 + logger.info("缓存清理调度启动：间隔 %d 小时，保留期 %d 天。", + interval_hours, get_settings().storage.cache_retention_days) + # 启动时先跑一次 + try: + purge_expired_cache() + except Exception as exc: # pragma: no cover + logger.warning("启动缓存清理失败：%s", exc) + while True: + time.sleep(interval) + try: + purge_expired_cache() + except Exception as exc: # pragma: no cover + logger.warning("定时缓存清理失败：%s", exc) + + +if __name__ == "__main__": # pragma: no cover + # 容器内手动触发：python -m app.services.cache_cleaner + import json + + logging.basicConfig(level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s") + result = purge_expired_cache() + print(json.dumps(result, ensure_ascii=False)) diff --git a/app/services/ffmpeg_service.py b/app/services/ffmpeg_service.py new file mode 100644 index 0000000..124e39e --- /dev/null +++ b/app/services/ffmpeg_service.py @@ -0,0 +1,64 @@ +"""ffmpeg 服务：从视频/音频文件提取 16kHz 单声道 PCM wav。 + +16kHz mono PCM 正是 Whisper 的标准输入，省去模型内重采样。 +""" + +from __future__ import annotations + +import logging +import shutil +import subprocess +from pathlib import Path + +logger = logging.getLogger("audio2text.ffmpeg") + + +class FFmpegError(RuntimeError): + pass + + +def extract_audio( + src: Path, + out_wav: Path, + sample_rate: int = 16000, +) -> Path: + """提取音频为 16kHz 单声道 PCM wav。 + + Args: + src: 输入视频/音频文件 + out_wav: 输出 wav 路径 + sample_rate: 采样率，默认 16000（Whisper 标准） + + Returns: + out_wav 路径 + + Raises: + FFmpegError: ffmpeg 不可用或提取失败 + """ + if shutil.which("ffmpeg") is None: + raise FFmpegError("ffmpeg 未安装；容器内应通过 apt 装好。") + if not src.is_file(): + raise FFmpegError(f"源文件不存在：{src}") + + out_wav.parent.mkdir(parents=True, exist_ok=True) + # -vn 去视频；-ac 1 单声道；-ar 16k 采样率；-c:a pcm_s16le 16bit PCM + cmd = [ + "ffmpeg", "-y", "-loglevel", "error", + "-i", str(src), + "-vn", "-ac", "1", "-ar", str(sample_rate), + "-c:a", "pcm_s16le", + str(out_wav), + ] + logger.debug("提取音频：%s -> %s", src.name, out_wav.name) + logger.debug("ffmpeg 命令：%s", " ".join(cmd)) + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) + except subprocess.TimeoutExpired as exc: + raise FFmpegError(f"ffmpeg 超时（>1h）：{src}") from exc + if result.returncode != 0: + raise FFmpegError( + f"ffmpeg 失败 (code={result.returncode}): {result.stderr.strip()[:500]}" + ) + if not out_wav.is_file(): + raise FFmpegError(f"ffmpeg 未生成输出文件：{out_wav}") + return out_wav diff --git a/app/services/log_buffer.py b/app/services/log_buffer.py new file mode 100644 index 0000000..7ce16d2 --- /dev/null +++ b/app/services/log_buffer.py @@ -0,0 +1,109 @@ +"""内存日志缓冲：捕获所有 audio2text.* 日志到有界 deque，供 /logs 页面查询。 + +设计： +- MemoryLogHandler 挂到 `audio2text` logger，子 logger 的记录经传播自动汇入。 +- deque(maxlen=N) 有界，旧记录自动淘汰，避免内存无限增长。 +- emit() 在 logging 内部锁下执行，deque.append 线程安全。 +- 查询时按级别过滤（≥指定级别）、取最近 N 条，返回结构化 dict。 +""" + +from __future__ import annotations + +import logging +import threading +from collections import deque +from datetime import datetime, timezone + +# 级别名 → 数值，用于过滤 +_LEVELS = { + "debug": logging.DEBUG, + "info": logging.INFO, + "warning": logging.WARNING, + "error": logging.ERROR, +} + + +class _BoundedLogRecord: + """从 LogRecord 提取显示所需字段，避免持有完整对象引用。""" + + __slots__ = ("ts", "level_no", "level_name", "logger_name", "message", "traceback") + + def __init__(self, record: logging.LogRecord) -> None: + self.ts: float = record.created + self.level_no: int = record.levelno + self.level_name: str = record.levelname + self.logger_name: str = record.name + # getMessage() 应用 %-style 懒格式化 + self.message: str = record.getMessage() + # exc_info 存的是 (type, value, tb) 三元组，格式化为字符串 + self.traceback: str | None = None + if record.exc_info: + import traceback as _tb + self.traceback = "".join(_tb.format_exception(*record.exc_info)) + + +class MemoryLogHandler(logging.Handler): + """把日志记录存入有界 deque，供 /api/logs 查询。""" + + def __init__(self, buffer_size: int = 2000) -> None: + super().__init__() + self._buffer: deque[_BoundedLogRecord] = deque(maxlen=buffer_size) + self._lock = threading.Lock() + + def emit(self, record: logging.LogRecord) -> None: + try: + entry = _BoundedLogRecord(record) + with self._lock: + self._buffer.append(entry) + except Exception: # pragma: no cover —— logging 不应抛 + self.handleError(record) + + def get_records( + self, level: str = "info", tail: int = 200, + ) -> list[dict]: + """返回最近 tail 条、≥指定级别的日志（JSON 友好的 dict 列表）。 + + Args: + level: debug | info | warning | error（最小级别） + tail: 最多返回条数 + """ + min_level = _LEVELS.get(level.lower(), logging.INFO) + with self._lock: + snapshot = list(self._buffer) + # 过滤 + 取最近 tail 条 + filtered = [r for r in snapshot if r.level_no >= min_level] + result = [] + for r in filtered[-tail:]: + ts_str = datetime.fromtimestamp(r.ts, tz=timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S" + ) + result.append({ + "ts": ts_str, + "level": r.level_name, + "logger": r.logger_name, + "msg": r.message, + "traceback": r.traceback, + }) + return result + + def clear(self) -> None: + with self._lock: + self._buffer.clear() + + +# 进程级单例 +_buffer: MemoryLogHandler | None = None + + +def get_log_buffer() -> MemoryLogHandler: + global _buffer + if _buffer is None: + _buffer = MemoryLogHandler() + return _buffer + + +def init_log_buffer(buffer_size: int) -> MemoryLogHandler: + """（重新）创建缓冲并返回，供 main.py 启动时按配置初始化。""" + global _buffer + _buffer = MemoryLogHandler(buffer_size=buffer_size) + return _buffer diff --git a/app/services/model_manager.py b/app/services/model_manager.py new file mode 100644 index 0000000..38bff3d --- /dev/null +++ b/app/services/model_manager.py @@ -0,0 +1,136 @@ +"""模型管理器：ASR 与翻译模型不共驻，任一时刻 GPU 上只有一个模型。 + +设计： +- 单例 ModelManager 跟踪当前加载的模型类型（none / asr / translator） +- get_asr(): 若翻译器在内存 → 先卸载（del + gc + empty_cache）→ 加载 faster-whisper +- get_translator(): 若 ASR 在内存 → 先卸载 → 加载 NLLB +- 翻译阶段独占显存，可用大 batch_size；ASR 阶段同理 + +这样在 24G 3090 上无需担心显存叠加，CPU dev 时也省内存。 +""" + +from __future__ import annotations + +import gc +import logging +import threading +from typing import Any + +from ..config import get_settings + +logger = logging.getLogger("audio2text.models") + +# 全局单例 + 锁：模型加载/卸载必须串行 +_lock = threading.Lock() + + +def _free_memory() -> None: + """释放 Python 对象与 GPU 缓存。""" + gc.collect() + try: + import torch + if torch.cuda.is_available(): + torch.cuda.empty_cache() + torch.cuda.synchronize() + except Exception: # pragma: no cover + pass + + +class ModelManager: + """ASR / 翻译模型的不共驻管理器。""" + + def __init__(self) -> None: + self._asr: Any = None # faster_whisper.WhisperModel + self._translator: Any = None # transformers pipeline + self._current: str = "none" # none | asr | translator + + # ---------------- ASR ---------------- + + def get_asr(self) -> Any: + """返回已加载的 faster-whisper 模型；必要时先卸载翻译器。""" + with _lock: + if self._asr is not None: + return self._asr + if self._translator is not None: + self._unload_translator_locked() + s = get_settings().asr + logger.debug("加载 ASR 模型 model=%s device=%s compute_type=%s", + s.model, s.device, s.compute_type) + from faster_whisper import WhisperModel + # device/compute_type 组合：cpu+int8 / cuda+float16 + self._asr = WhisperModel( + s.model, device=s.device, compute_type=s.compute_type, + ) + self._current = "asr" + logger.debug("ASR 模型已就绪。") + return self._asr + + def unload_asr(self) -> None: + with _lock: + self._unload_asr_locked() + + def _unload_asr_locked(self) -> None: + if self._asr is None: + return + logger.debug("卸载 ASR 模型（释放显存供翻译器独占）。") + # faster-whisper 模型无显式 close，del 即可 + del self._asr + self._asr = None + self._current = "none" if self._translator is None else "translator" + _free_memory() + + # ---------------- 翻译器 ---------------- + + def get_translator(self) -> Any: + """返回已加载的 NLLB 翻译 pipeline；必要时先卸载 ASR。""" + with _lock: + if self._translator is not None: + return self._translator + if self._asr is not None: + self._unload_asr_locked() + s = get_settings().translation + logger.debug("加载翻译模型 model=%s device=%s", s.model, s.device) + from transformers import pipeline + self._translator = pipeline( + "translation", + model=s.model, + device=s.device, + src_lang=s.src_lang, + tgt_lang=s.tgt_lang, + ) + self._current = "translator" + logger.debug("翻译模型已就绪（独占显存，可用大 batch）。") + return self._translator + + def unload_translator(self) -> None: + with _lock: + self._unload_translator_locked() + + def _unload_translator_locked(self) -> None: + if self._translator is None: + return + logger.debug("卸载翻译模型。") + # 释放 pipeline 持有的 model + tokenizer + mdl = getattr(self._translator, "model", None) + tok = getattr(self._translator, "tokenizer", None) + del self._translator, mdl, tok + self._translator = None + self._current = "none" if self._asr is None else "asr" + _free_memory() + + # ---------------- 状态 ---------------- + + @property + def current(self) -> str: + return self._current + + +# 进程级单例 +_manager: ModelManager | None = None + + +def get_model_manager() -> ModelManager: + global _manager + if _manager is None: + _manager = ModelManager() + return _manager diff --git a/app/services/pipeline.py b/app/services/pipeline.py new file mode 100644 index 0000000..fc6f920 --- /dev/null +++ b/app/services/pipeline.py @@ -0,0 +1,153 @@ +"""转写管线编排：提取音频 → ASR → 断句 → 翻译 → 写 SRT。 + +任务状态机： + queued → extracting → transcribing → segmenting → translating → done + 任一步失败 → failed + +模型不共驻：ASR 与翻译分阶段加载，翻译时先卸载 Whisper 释放显存跑大 batch。 +管线在后台线程跑（每个任务一个线程），通过 DB 更新状态与进度。 +""" + +from __future__ import annotations + +import logging +import threading +import traceback +from datetime import datetime, timezone +from pathlib import Path + +from ..config import get_settings +from ..database import get_session_local +from ..models.task import Task +from . import ffmpeg_service, asr_service, segmenter, translate_service, srt_writer +from .types import Subtitle + +logger = logging.getLogger("audio2text.pipeline") + +# 进度锚点（各阶段在 0-100 中的占比） +P_EXTRACT = 5.0 +P_TRANSCRIBE_START = 5.0 +P_TRANSCRIBE_END = 55.0 +P_SEGMENT_START = 55.0 +P_SEGMENT_END = 60.0 +P_TRANSLATE_START = 60.0 +P_TRANSLATE_END = 98.0 +P_DONE = 100.0 + + +def enqueue_task(task_id: int) -> None: + """把任务交给后台线程处理（非阻塞，供 upload_service.complete 调用）。""" + t = threading.Thread(target=_run_task, args=(task_id,), daemon=True) + t.start() + logger.info("任务 %d 已入队（后台线程 %s）。", task_id, t.name) + + +def _run_task(task_id: int) -> None: + """后台执行完整管线。所有异常都被捕获并写入 task.error。""" + db = get_session_local()() + try: + task = db.get(Task, task_id) + if task is None: + logger.error("任务 %d 不存在。", task_id) + return + _pipeline(db, task) + except Exception as exc: + logger.exception("任务 %d 失败：%s", task_id, exc) + _mark_failed(db, task_id, str(exc)) + finally: + db.close() + + +def _pipeline(db, task: Task) -> None: + s = get_settings() + src = s.upload_dir() / task.source_path + + # ---------- 1. 提取音频 ---------- + _set_status(db, task, "extracting", P_EXTRACT) + wav = s.work_dir() / f"task_{task.id}.wav" + ffmpeg_service.extract_audio(src, wav) + + # 按配置决定是否删原始视频（提取成功后） + if s.processing.delete_original_after_extract and src.is_file(): + try: + src.unlink() + logger.info("已删除原始视频 %s（delete_original_after_extract=true）。", src.name) + except OSError as exc: + logger.warning("删除原始视频失败 %s: %s", src, exc) + + # ---------- 2. 语音识别 ---------- + _set_status(db, task, "transcribing", P_TRANSCRIBE_START) + segments = asr_service.transcribe(wav) + _set_status(db, task, "transcribing", P_TRANSCRIBE_END, + note=f"识别出 {len(segments)} 段") + + # ---------- 3. 断句 + 时间戳重算 ---------- + _set_status(db, task, "segmenting", P_SEGMENT_START) + subs = segmenter.resegment(segments) + _set_status(db, task, "segmenting", P_SEGMENT_END, + note=f"重组为 {len(subs)} 条字幕") + + # ---------- 4. 翻译 ---------- + _set_status(db, task, "translating", P_TRANSLATE_START) + # 翻译阶段：model_manager 会自动卸载 ASR、加载翻译器（独占显存） + zh_texts = translate_service.translate(subs) + _set_status(db, task, "translating", P_TRANSLATE_END, + note=f"翻译 {len(zh_texts)} 条") + + # ---------- 5. 写 SRT ---------- + out_dir = s.output_dir() + stem = Path(task.filename).stem + en_path = out_dir / f"task_{task.id}/{stem}.en.srt" + zh_path = out_dir / f"task_{task.id}/{stem}.zh.srt" + bi_path = out_dir / f"task_{task.id}/{stem}.srt" + + srt_writer.write_srt(subs, en_path) + # 中文 SRT（用译文 + 同时间戳） + zh_subs = [Subtitle(text=zh, start=sub.start, end=sub.end) + for zh, sub in zip(zh_texts, subs)] + srt_writer.write_srt(zh_subs, zh_path) + srt_writer.write_bilingual_srt(subs, zh_texts, bi_path) + + # 记录相对路径 + task.en_srt_path = str(en_path.relative_to(out_dir)) + task.zh_srt_path = str(zh_path.relative_to(out_dir)) + task.bilingual_srt_path = str(bi_path.relative_to(out_dir)) + task.status = "done" + task.progress = P_DONE + task.updated_at = datetime.now(timezone.utc) + db.commit() + + # 清理中间音频 + if not s.processing.keep_audio and wav.is_file(): + try: + wav.unlink() + except OSError: + pass + + logger.info("任务 %d 完成：%s", task.id, bi_path.name) + + +# ---------------- DB 状态更新 ---------------- + +def _set_status(db, task: Task, status: str, progress: float, note: str = "") -> None: + task.status = status + task.progress = progress + task.updated_at = datetime.now(timezone.utc) + db.commit() + if note: + logger.info("任务 %d [%s %.0f%%] %s", task.id, status, progress, note) + else: + logger.info("任务 %d [%s %.0f%%]", task.id, status, progress) + + +def _mark_failed(db, task_id: int, error: str) -> None: + try: + task = db.get(Task, task_id) + if task is None: + return + task.status = "failed" + task.error = error[:2000] + task.updated_at = datetime.now(timezone.utc) + db.commit() + except Exception: # pragma: no cover + logger.error("写入失败状态时又失败：\n%s", traceback.format_exc()) diff --git a/app/services/reaper.py b/app/services/reaper.py new file mode 100644 index 0000000..7c6e0d7 --- /dev/null +++ b/app/services/reaper.py @@ -0,0 +1,22 @@ +"""后台 reaper：清理被放弃的分片上传会话与临时文件。供 main.py 启动时调用。""" + +from __future__ import annotations + +import logging + +from ..database import get_session_local +from .upload_service import UploadService + +logger = logging.getLogger("audio2text.reaper") + + +def reap_stale_sessions() -> int: + """执行一次过期会话清理。""" + db = get_session_local()() + try: + return UploadService(db).reap_stale_sessions() + except Exception as exc: # pragma: no cover + logger.warning("reaper 执行失败：%s", exc) + return 0 + finally: + db.close() diff --git a/app/services/segmenter.py b/app/services/segmenter.py new file mode 100644 index 0000000..4198277 --- /dev/null +++ b/app/services/segmenter.py @@ -0,0 +1,233 @@ +"""断句 + 时间戳重算（纯算法，零模型开销）。 + +Whisper 原始 segment 的断句通常很混乱：每段不一定是完整句子，时间戳也不对齐句界。 +本模块基于词级时间戳重新断句，得到规范的字幕条目。 + +两路策略： +1. 精确路（有 word_timestamps）：按句末标点（. ! ? ;）切句，超长句再按逗号拆， + 每条字幕的时间戳直接取首词.start ~ 末词.end，精确无误。 +2. 匀速估算路（无 word_timestamps）：段内按字符数比例分配时间—— + 句start = 段start + (前缀字符数 / 段总字符数) × 段时长。 + 即「短时匀速」假设，无需大模型。 + +最后做 SRT 规范化：单条 1–7 秒、≤2 行、每行 ≤42 字符。 +""" + +from __future__ import annotations + +import logging +import re + +from ..config import get_settings +from .types import Segment, Subtitle, Word + +logger = logging.getLogger("audio2text.segmenter") + +# 句末标点：句号、感叹号、问号、分号 +_SENTENCE_END = re.compile(r"[.!?;]+") +# 句内停顿：逗号、冒号、破折号 +_CLAUSE_BREAK = re.compile(r"[,:\-—]+") + + +def resegment(segments: list[Segment]) -> list[Subtitle]: + """把 ASR segments 重组为规范字幕条目。 + + Args: + segments: asr_service.transcribe 的输出（可能含词级时间戳） + + Returns: + list[Subtitle]，已做时长/字数规范化。 + """ + cfg = get_settings().segmentation + max_words = cfg.max_words_per_line + max_dur = cfg.max_duration_seconds + min_dur = cfg.min_duration_seconds + max_chars = cfg.max_chars_per_line + + # 第一步：把所有词串成大列表（精确路）或退化为段级（估算路） + all_words: list[Word] = [] + has_word_ts = True + for seg in segments: + if not seg.words: + has_word_ts = False + break + all_words.extend(seg.words) + + if has_word_ts and all_words: + subs = _resegment_by_words(all_words, max_words, max_dur) + else: + logger.warning("无词级时间戳，退化为匀速估算路。") + subs = _resegment_by_estimate(segments, max_words, max_dur) + + # 规范化：合并过短条目、拆分行宽 + subs = _normalize(subs, min_dur, max_chars) + logger.debug("断句完成：%d 条字幕。", len(subs)) + return subs + + +# ---------------- 精确路：按词级时间戳 ---------------- + +def _resegment_by_words( + words: list[Word], max_words: int, max_dur: float, +) -> list[Subtitle]: + """按句末标点切句，超长句按逗号拆，时间戳取首末词。""" + subs: list[Subtitle] = [] + # 当前句的词 + current: list[Word] = [] + + def flush(wlist: list[Word]) -> None: + if not wlist: + return + text = " ".join(w.text for w in wlist).strip() + if not text: + return + subs.append(Subtitle( + text=text, + start=wlist[0].start, + end=wlist[-1].end, + )) + + for w in words: + current.append(w) + # 句末标点 → 收尾 + if _SENTENCE_END.search(w.text): + _maybe_split_and_flush(current, max_words, max_dur, flush) + current = [] + continue + # 超长（词数或时长）→ 优先在最近的逗号处断 + cur_dur = (current[-1].end - current[0].start) if len(current) > 1 else 0 + if len(current) >= max_words or cur_dur >= max_dur: + _maybe_split_and_flush(current, max_words, max_dur, flush) + current = [] + + flush(current) + return subs + + +def _maybe_split_and_flush( + wlist: list[Word], max_words: int, max_dur: float, flush, +) -> None: + """若 wlist 过长，在逗号处再拆；否则整条 flush。""" + if len(wlist) <= max_words and (len(wlist) <= 1 or + wlist[-1].end - wlist[0].start < max_dur): + flush(wlist) + return + # 找逗号断点 + parts: list[list[Word]] = [] + cur: list[Word] = [] + for w in wlist: + cur.append(w) + if _CLAUSE_BREAK.search(w.text) and len(cur) >= max_words // 2: + parts.append(cur) + cur = [] + if cur: + parts.append(cur) + # 若逗号拆不开（无逗号），强制按 max_words 等分 + if len(parts) == 1 and len(parts[0]) > max_words: + parts = [parts[0][i:i + max_words] for i in range(0, len(parts[0]), max_words)] + for p in parts: + flush(p) + + +# ---------------- 匀速估算路：段内按字符比例 ---------------- + +def _resegment_by_estimate( + segments: list[Segment], max_words: int, max_dur: float, +) -> list[Subtitle]: + """无词级时间戳时：先按文本断句，再按字符数比例估算时间戳。""" + subs: list[Subtitle] = [] + for seg in segments: + text = seg.text.strip() + if not text: + continue + dur = seg.end - seg.start + # 按句末标点切 + sentences = _split_sentences(text) + if not sentences: + sentences = [text] + # 段内按字符数比例分配时间 + total_chars = sum(len(s) for s in sentences) or 1 + cursor = seg.start + for sent in sentences: + frac = len(sent) / total_chars + est_end = cursor + dur * frac + # 超长句再按逗号拆（时间按字符比例再分） + if len(sent.split()) > max_words or (est_end - cursor) > max_dur: + for clause in _split_clauses(sent): + cfrac = len(clause) / len(sent) if len(sent) else 1 + c_end = cursor + (est_end - cursor) * cfrac + subs.append(Subtitle(text=clause.strip(), start=cursor, end=c_end)) + cursor = c_end + else: + subs.append(Subtitle(text=sent.strip(), start=cursor, end=est_end)) + cursor = est_end + return subs + + +def _split_sentences(text: str) -> list[str]: + """按句末标点切句，保留标点。""" + parts = _SENTENCE_END.split(text) + marks = _SENTENCE_END.findall(text) + out = [] + for i, p in enumerate(parts): + p = p.strip() + if not p: + continue + out.append(p + (marks[i] if i < len(marks) else "")) + return out + + +def _split_clauses(sentence: str) -> list[str]: + """按逗号/冒号拆子句，保留标点。""" + parts = _CLAUSE_BREAK.split(sentence) + marks = _CLAUSE_BREAK.findall(sentence) + out = [] + for i, p in enumerate(parts): + p = p.strip() + if not p: + continue + out.append(p + (marks[i - 1] if 0 < i <= len(marks) else "")) + return out + + +# ---------------- 规范化 ---------------- + +def _normalize(subs: list[Subtitle], min_dur: float, max_chars: int) -> list[Subtitle]: + """合并过短条目、拆分过宽行。""" + # 1. 合并过短（< min_dur 且非末尾） + merged: list[Subtitle] = [] + for s in subs: + if merged and (s.end - s.start) < min_dur: + prev = merged[-1] + prev.text = (prev.text + " " + s.text).strip() + prev.end = s.end + else: + merged.append(Subtitle(text=s.text, start=s.start, end=s.end)) + + # 2. 拆分超过 max_chars 的行（按词折行，不改时间戳） + out: list[Subtitle] = [] + for s in merged: + if len(s.text) <= max_chars: + out.append(s) + continue + lines = _wrap_text(s.text, max_chars) + out.append(Subtitle(text=lines, start=s.start, end=s.end)) + return out + + +def _wrap_text(text: str, max_chars: int) -> str: + """按词折成 ≤2 行，每行 ≤max_chars 字符（SRT 规范）。""" + words = text.split() + lines: list[str] = [] + cur = "" + for idx, w in enumerate(words): + if cur and len(cur) + 1 + len(w) > max_chars: + lines.append(cur) + # 已有一行 + 当前行，合并剩余为一行 + cur = " ".join([w] + words[idx + 1:]) + break + else: + cur = (cur + " " + w).strip() if cur else w + if cur: + lines.append(cur) + return "\n".join(lines[:2]) diff --git a/app/services/srt_writer.py b/app/services/srt_writer.py new file mode 100644 index 0000000..930e9dd --- /dev/null +++ b/app/services/srt_writer.py @@ -0,0 +1,68 @@ +"""SRT 字幕文件写入与双语合并。""" + +from __future__ import annotations + +from pathlib import Path + +from .types import Subtitle + + +def _format_ts(seconds: float) -> str: + """秒 → SRT 时间戳 HH:MM:SS,mmm。""" + if seconds < 0: + seconds = 0.0 + ms = int(round((seconds - int(seconds)) * 1000)) + s = int(seconds) % 60 + m = (int(seconds) // 60) % 60 + h = int(seconds) // 3600 + if ms == 1000: # 四舍五入进位 + ms = 0 + s += 1 + if s == 60: + s = 0 + m += 1 + if m == 60: + m = 0 + h += 1 + return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" + + +def write_srt(subtitles: list[Subtitle], path: Path) -> Path: + """写单语 SRT。""" + path.parent.mkdir(parents=True, exist_ok=True) + lines: list[str] = [] + for i, sub in enumerate(subtitles, 1): + lines.append(str(i)) + lines.append(f"{_format_ts(sub.start)} --> {_format_ts(sub.end)}") + lines.append(sub.text) + lines.append("") + path.write_text("\n".join(lines), encoding="utf-8") + return path + + +def write_bilingual_srt( + en_subs: list[Subtitle], + zh_texts: list[str], + path: Path, +) -> Path: + """写双语合并 SRT：英文在上、中文在下，同一时间戳。 + + Args: + en_subs: 英文字幕条目 + zh_texts: 与 en_subs 等长、顺序对应的中文译文 + path: 输出路径 + """ + if len(en_subs) != len(zh_texts): + raise ValueError( + f"英文字幕数({len(en_subs)}) 与中文译文数({len(zh_texts)}) 不一致" + ) + path.parent.mkdir(parents=True, exist_ok=True) + lines: list[str] = [] + for i, (sub, zh) in enumerate(zip(en_subs, zh_texts), 1): + lines.append(str(i)) + lines.append(f"{_format_ts(sub.start)} --> {_format_ts(sub.end)}") + lines.append(sub.text) + lines.append(zh) + lines.append("") + path.write_text("\n".join(lines), encoding="utf-8") + return path diff --git a/app/services/translate_service.py b/app/services/translate_service.py new file mode 100644 index 0000000..04aad9f --- /dev/null +++ b/app/services/translate_service.py @@ -0,0 +1,60 @@ +"""翻译服务：NLLB-200，英译中。 + +通过 model_manager 加载，确保 ASR 已卸载、翻译器独占显存，从而可用大 batch_size。 +按字幕条目批量翻译，保留索引对应。 +""" + +from __future__ import annotations + +import logging + +from ..config import get_settings +from .model_manager import get_model_manager +from .types import Subtitle + +logger = logging.getLogger("audio2text.translate") + + +def translate(subtitles: list[Subtitle]) -> list[str]: + """批量翻译英文字幕为中文。 + + Args: + subtitles: 断句后的英文字幕条目 + + Returns: + list[str]，与 subtitles 等长、顺序对应的中文译文。 + 单条翻译失败时该位置回退为原英文。 + """ + if not subtitles: + return [] + + s = get_settings().translation + pipe = get_model_manager().get_translator() + batch = s.batch_size + max_len = s.max_length + + # 取纯文本（去掉折行），避免翻译把换行符当语义 + texts = [sub.text.replace("\n", " ").strip() for sub in subtitles] + logger.debug("开始翻译 %d 条字幕（batch_size=%d）...", len(texts), batch) + + results: list[str] = [] + for i in range(0, len(texts), batch): + chunk = texts[i:i + batch] + try: + out = pipe(chunk, max_length=max_len) + for item in out: + # pipeline 返回 [{"translation_text": "..."}] + results.append(item.get("translation_text", "").strip()) + except Exception as exc: # pragma: no cover + logger.warning("第 %d-%d 批翻译失败，逐条重试：%s", i, i + len(chunk), exc) + for t in chunk: + try: + out = pipe([t], max_length=max_len) + results.append(out[0].get("translation_text", "").strip()) + except Exception: + results.append(t) # 回退原文 + if (i // batch + 1) % 5 == 0: + logger.debug("已翻译 %d/%d 条。", min(i + len(chunk), len(texts)), len(texts)) + + logger.debug("翻译完成：%d 条。", len(results)) + return results diff --git a/app/services/types.py b/app/services/types.py new file mode 100644 index 0000000..220df1d --- /dev/null +++ b/app/services/types.py @@ -0,0 +1,38 @@ +"""管线各阶段共享的数据传输对象（DTO）。 + +独立于任何模型加载逻辑——segmenter（纯算法）和 srt_writer（纯 IO）可以只依赖 +本模块，不拉入 faster_whisper / torch。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass +class Word: + """ASR 识别出的单个词，带时间戳。""" + + text: str + start: float # 秒 + end: float + probability: float = 1.0 + + +@dataclass +class Segment: + """ASR 输出的一段文本，可能含词级时间戳。""" + + text: str + start: float + end: float + words: list[Word] + + +@dataclass +class Subtitle: + """断句后的一条字幕：英文文本 + 起止时间戳。""" + + text: str + start: float + end: float diff --git a/app/services/upload_service.py b/app/services/upload_service.py new file mode 100644 index 0000000..4a73b6c --- /dev/null +++ b/app/services/upload_service.py @@ -0,0 +1,249 @@ +"""分片上传服务：会话管理 + 分片落盘 + 拼接 + 创建转写任务。 + +存储布局:: + + //0.part 分片暂存 + //1.part + ... + ///. complete 后的正式视频 + +与 server 的区别：视频无需 sha256 去重（每个视频都转写），complete 直接创建 Task。 +管线触发由 controller 调用 pipeline.enqueue_task，本服务不依赖 pipeline。 +""" + +from __future__ import annotations + +import logging +import os +import shutil +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from ..config import get_settings +from ..models.task import Task +from ..models.upload_session import UploadSession +from ..schemas.task import ( + ChunkUploadResponse, + CompleteResponse, + CreateSessionRequest, + CreateSessionResponse, + SessionStatusResponse, +) + +logger = logging.getLogger("audio2text.upload") + + +class UploadService: + def __init__(self, db: Session) -> None: + self.db = db + s = get_settings() + self.upload_root = s.upload_dir() + self.work_root = s.work_dir() + self.chunk_bytes = s.storage.chunk_bytes + self.session_ttl = s.storage.chunk_session_ttl_seconds + + # ---------------- 会话生命周期 ---------------- + + def create_session(self, body: CreateSessionRequest) -> CreateSessionResponse: + upload_id = uuid.uuid4().hex + session = UploadSession( + upload_id=upload_id, + filename=body.filename, + size_bytes=body.size_bytes, + chunk_size=body.chunk_size, + total_chunks=body.total_chunks, + uploaded_chunks=[], + status="pending", + ) + self.db.add(session) + self.db.commit() + self._session_dir(upload_id).mkdir(parents=True, exist_ok=True) + return CreateSessionResponse( + upload_id=upload_id, + filename=body.filename, + size_bytes=body.size_bytes, + chunk_size=body.chunk_size, + total_chunks=body.total_chunks, + ) + + def get_status(self, upload_id: str) -> SessionStatusResponse: + session = self._require_session(upload_id) + return SessionStatusResponse( + upload_id=session.upload_id, + filename=session.filename, + size_bytes=session.size_bytes, + chunk_size=session.chunk_size, + total_chunks=session.total_chunks, + uploaded_chunks=list(session.uploaded_chunks or []), + completed=(session.status == "completed"), + task_id=session.task_id, + ) + + # ---------------- 分片写入 ---------------- + + def write_chunk(self, upload_id: str, index: int, data: bytes) -> list[int]: + session = self._require_session(upload_id) + self._validate_index(session, index) + + session_dir = self._session_dir(upload_id) + session_dir.mkdir(parents=True, exist_ok=True) + chunk_path = session_dir / f"{index}.part" + + try: + with chunk_path.open("wb") as out: + out.write(data) + out.flush() + os.fsync(out.fileno()) + except Exception: + chunk_path.unlink(missing_ok=True) + raise + + uploaded = list(session.uploaded_chunks or []) + if index not in uploaded: + uploaded.append(index) + session.uploaded_chunks = uploaded + session.updated_at = datetime.now(timezone.utc) + self.db.commit() + return sorted(uploaded) + + # ---------------- 拼接 + 创建任务 ---------------- + + def complete(self, upload_id: str) -> CompleteResponse: + session = self._require_session(upload_id) + + # 幂等：已 complete 则返回已建任务 + if session.status == "completed" and session.task_id is not None: + task = self.db.get(Task, session.task_id) + if task is not None: + return CompleteResponse( + task_id=task.id, filename=task.filename, + size_bytes=session.size_bytes, status=task.status, + ) + + uploaded = set(session.uploaded_chunks or []) + missing = [i for i in range(session.total_chunks) if i not in uploaded] + if missing: + raise HTTPException( + 409, + f"分片不齐全：缺失 {len(missing)} 个，例如 {sorted(missing)[:10]}", + ) + + final_path = self._assemble(session) + rel = str(final_path.relative_to(self.upload_root)) + + task = Task( + filename=session.filename, + source_path=rel, + status="queued", + progress=0.0, + ) + self.db.add(task) + session.status = "completed" + session.final_path = rel + session.updated_at = datetime.now(timezone.utc) + self.db.commit() + self.db.refresh(task) + # 正向关联：session → task（替代旧的 source_path 反向查找） + session.task_id = task.id + self.db.commit() + + # 清理分片暂存 + self._cleanup_session_dir(upload_id) + + logger.info("上传完成 task_id=%s file=%s size=%d", task.id, session.filename, session.size_bytes) + return CompleteResponse( + task_id=task.id, filename=task.filename, + size_bytes=session.size_bytes, status=task.status, + ) + + # ---------------- 过期会话清理 ---------------- + + def reap_stale_sessions(self) -> int: + """清理被放弃的会话：pending 且 updated_at 超过 ttl。 + + 时间比较统一用 aware UTC datetime，避免 naive datetime 的 .timestamp() + 按本地时区算导致的偏移。 + """ + cutoff = datetime.now(timezone.utc) - timedelta(seconds=self.session_ttl) + sessions = ( + self.db.query(UploadSession) + .filter(UploadSession.status == "pending") + .all() + ) + n = 0 + for session in sessions: + updated = session.updated_at + if updated is None: + continue + # SQLite 存 naive datetime，统一补 UTC 后比较 + if updated.tzinfo is None: + updated = updated.replace(tzinfo=timezone.utc) + if updated < cutoff: + self._cleanup_session_dir(session.upload_id) + self.db.delete(session) + n += 1 + logger.info("清理过期分片会话 upload_id=%s file=%s", session.upload_id, session.filename) + if n: + self.db.commit() + return n + + # ---------------- 内部 ---------------- + + def _require_session(self, upload_id: str) -> UploadSession: + if not upload_id: + raise HTTPException(400, "upload_id 不能为空") + session = ( + self.db.query(UploadSession) + .filter(UploadSession.upload_id == upload_id) + .first() + ) + if session is None: + raise HTTPException(404, f"会话不存在或已过期：{upload_id}") + return session + + @staticmethod + def _validate_index(session: UploadSession, index: int) -> None: + if index < 0 or index >= session.total_chunks: + raise HTTPException(400, f"分片下标越界：{index} 不在 [0, {session.total_chunks})") + + def _session_dir(self, upload_id: str) -> Path: + return self.work_root / upload_id + + def _assemble(self, session: UploadSession) -> Path: + """按 index 顺序拼接全部分片为正式视频文件。""" + ext = Path(session.filename).suffix or ".mp4" + now = datetime.now(timezone.utc) + sub = self.upload_root / f"{now:%Y}" / f"{now:%m}" + sub.mkdir(parents=True, exist_ok=True) + final = sub / f"{uuid.uuid4().hex}{ext}" + part_path = final.with_suffix(final.suffix + ".part") + + session_dir = self._session_dir(session.upload_id) + try: + with part_path.open("wb") as out: + for index in range(session.total_chunks): + chunk_path = session_dir / f"{index}.part" + if not chunk_path.is_file(): + raise HTTPException(409, f"拼接时发现分片缺失：{index}.part") + with chunk_path.open("rb") as src: + while buf := src.read(self.chunk_bytes): + out.write(buf) + out.flush() + os.fsync(out.fileno()) + os.replace(part_path, final) + except Exception: + part_path.unlink(missing_ok=True) + raise + return final + + def _cleanup_session_dir(self, upload_id: str) -> None: + session_dir = self._session_dir(upload_id) + try: + if session_dir.exists(): + shutil.rmtree(session_dir, ignore_errors=True) + except Exception as exc: # pragma: no cover + logger.warning("清理会话目录失败 upload_id=%s: %s", upload_id, exc) diff --git a/app/views/__init__.py b/app/views/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/views/_shared.py b/app/views/_shared.py new file mode 100644 index 0000000..73f2eb0 --- /dev/null +++ b/app/views/_shared.py @@ -0,0 +1,370 @@ +"""前端共享资产：统一样式、工具 JS、上传协议 JS、页面骨架。 + +3 个页面（home/history/logs）共用 BASE_CSS + SHARED_JS，消除 ~400 行重复 CSS +和 ~80 行重复 JS。_CSS / JS 字符串都是**普通字符串**（非 f-string），花括号用单层。 + +页面骨架 render_page() 统一 /