From 856aa380263f088d39742d263b54b566bb19daaf Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Sun, 30 Aug 2026 06:52:14 +0800 Subject: [PATCH] add volcengine OSS support --- sentence_api/.env.example | 10 +++ sentence_api/README.md | 9 +++ sentence_api/config.py | 49 ++++++++++++++ sentence_api/main.py | 24 +++++-- sentence_api/oss.py | 77 +++++++++++++++++++++ sentence_api/processing.py | 29 +++++++- sentence_api/repository.py | 28 +++++++- sentence_api/requirements.txt | 3 + sentence_api/tests/test_api.py | 120 +++++++++++++++++++++++++++++++++ 9 files changed, 339 insertions(+), 10 deletions(-) create mode 100644 sentence_api/oss.py diff --git a/sentence_api/.env.example b/sentence_api/.env.example index 02b8baa..b753695 100644 --- a/sentence_api/.env.example +++ b/sentence_api/.env.example @@ -16,3 +16,13 @@ MOSS_TIMEOUT_SECONDS=1800 MOSS_MAX_NEW_TOKENS=65536 # 每句结束时间向后顺延的毫秒数，避免句子末尾发音被切掉；按需在 200-500 之间调整 MOSS_END_PADDING_MS=300 + +# 火山引擎对象存储（TOS / 火山 OSS）。桶请设置为公共读，或把下方地址配置为可公开访问的 CDN 域名。 +VOLCANO_OSS_ACCESS_KEY=replace-with-volcano-access-key +VOLCANO_OSS_SECRET_KEY=replace-with-volcano-secret-key +VOLCANO_OSS_BUCKET=your-course-videos +VOLCANO_OSS_ENDPOINT=tos-cn-beijing.volces.com +VOLCANO_OSS_REGION=cn-beijing +VOLCANO_OSS_KEY_PREFIX=videos/ +# 可选：绑定自定义域名或 CDN 后填写，例如 https://media.example.com +# VOLCANO_OSS_PUBLIC_BASE_URL=https://media.example.com diff --git a/sentence_api/README.md b/sentence_api/README.md index 833ff65..424a0d7 100644 --- a/sentence_api/README.md +++ b/sentence_api/README.md @@ -4,6 +4,9 @@ FastAPI service for the Android oral-training SDK. It provides: - chunked video upload with SHA-256 calculation; - files stored under `data/v/` and metadata stored in SQLite; +- after the first processing pass, videos can be uploaded automatically to + Volcano Engine Object Storage (TOS / 火山 OSS) and the catalog/share UI will + use that public URL; - background sentence transcription through MOSS-Transcribe-Diarize; - a browser-based administration page at `/admin`; - video catalog and HTTP Range playback; @@ -33,6 +36,12 @@ http://127.0.0.1:8000/docs When `MOSS_TRANSCRIBE_URL` is empty, uploaded videos still receive silence-based boundaries, but no reference transcript is produced and assessment is disabled. +When all `VOLCANO_OSS_*` credentials in `.env.example` are configured, the first +processing pass saves sentence boundaries, uploads the source video, and only +then marks it ready. `stream_url` and the legacy `/content` endpoint then point +to the OSS/CDN URL. If the upload fails, the video is marked failed so clients +do not fall back to server-local playback. + ## Main Endpoints ```text diff --git a/sentence_api/config.py b/sentence_api/config.py index beb25f8..1694d8e 100644 --- a/sentence_api/config.py +++ b/sentence_api/config.py @@ -1,6 +1,7 @@ import os from dataclasses import dataclass from pathlib import Path +from urllib.parse import quote def _bool_env(name: str, default: bool) -> bool: @@ -26,6 +27,13 @@ class Settings: moss_max_new_tokens: int moss_end_padding_ms: int pass_score: float + volcano_oss_access_key: str + volcano_oss_secret_key: str + volcano_oss_bucket: str + volcano_oss_endpoint: str + volcano_oss_region: str + volcano_oss_key_prefix: str + volcano_oss_public_base_url: str @classmethod def from_env(cls) -> "Settings": @@ -54,6 +62,19 @@ class Settings: moss_max_new_tokens=int(os.getenv("MOSS_MAX_NEW_TOKENS", "65536")), moss_end_padding_ms=int(os.getenv("MOSS_END_PADDING_MS", "300")), pass_score=float(os.getenv("ASSESSMENT_PASS_SCORE", "70")), + volcano_oss_access_key=os.getenv("VOLCANO_OSS_ACCESS_KEY", "").strip(), + volcano_oss_secret_key=os.getenv("VOLCANO_OSS_SECRET_KEY", "").strip(), + volcano_oss_bucket=os.getenv("VOLCANO_OSS_BUCKET", "").strip(), + volcano_oss_endpoint=os.getenv( + "VOLCANO_OSS_ENDPOINT", "tos-cn-beijing.volces.com" + ).strip().strip("/"), + volcano_oss_region=os.getenv("VOLCANO_OSS_REGION", "cn-beijing").strip(), + volcano_oss_key_prefix=Settings._normalize_key_prefix( + os.getenv("VOLCANO_OSS_KEY_PREFIX", "videos/") + ), + volcano_oss_public_base_url=os.getenv( + "VOLCANO_OSS_PUBLIC_BASE_URL", "" + ).strip().rstrip("/"), ) @property @@ -76,6 +97,34 @@ class Settings: def database_path(self) -> Path: return self.data_dir / "oral_trainer.sqlite3" + @property + def volcano_oss_enabled(self) -> bool: + return all( + ( + self.volcano_oss_access_key, + self.volcano_oss_secret_key, + self.volcano_oss_bucket, + self.volcano_oss_endpoint, + self.volcano_oss_region, + ) + ) + + def volcano_oss_public_url(self, object_key: str) -> str: + quoted_key = quote(object_key) + if self.volcano_oss_public_base_url: + return f"{self.volcano_oss_public_base_url}/{quoted_key}" + endpoint = self.volcano_oss_endpoint + if "://" in endpoint: + scheme, host = endpoint.split("://", 1) + else: + scheme, host = "https", endpoint + return f"{scheme}://{self.volcano_oss_bucket}.{host}/{quoted_key}" + + @staticmethod + def _normalize_key_prefix(value: str) -> str: + normalized = value.strip().strip("/") + return f"{normalized}/" if normalized else "" + def ensure_directories(self) -> None: for path in (self.data_dir, self.videos_dir, self.work_dir, self.attempts_dir, self.dub_shares_dir): path.mkdir(parents=True, exist_ok=True) diff --git a/sentence_api/main.py b/sentence_api/main.py index c1173af..e6d78c6 100644 --- a/sentence_api/main.py +++ b/sentence_api/main.py @@ -22,7 +22,7 @@ from fastapi import ( Request, UploadFile, ) -from fastapi.responses import FileResponse, Response +from fastapi.responses import FileResponse, RedirectResponse, Response from fastapi.staticfiles import StaticFiles from starlette.concurrency import run_in_threadpool from starlette.requests import ClientDisconnect @@ -40,6 +40,7 @@ from .models import ( VideoSummary, VideoUploadResponse, ) +from .oss import VolcanoOSSUploader from .processing import VideoProcessor from .repository import VideoRepository from .store import BoundaryStore @@ -67,6 +68,7 @@ def create_app( settings: Optional[Settings] = None, repository: Optional[VideoRepository] = None, transcriber: Optional[Transcriber] = None, + oss_uploader: Optional[VolcanoOSSUploader] = None, ) -> FastAPI: service_settings = settings or Settings.from_env() service_settings.ensure_directories() @@ -78,7 +80,8 @@ def create_app( timeout_seconds=service_settings.moss_timeout_seconds, max_new_tokens=service_settings.moss_max_new_tokens, ) - processor = VideoProcessor(service_settings, video_repository, moss) + oss = oss_uploader or VolcanoOSSUploader(service_settings) + processor = VideoProcessor(service_settings, video_repository, moss, oss) assessment_service = AssessmentService( video_repository, moss, @@ -94,6 +97,7 @@ def create_app( application.state.boundary_store = legacy_store application.state.video_repository = video_repository application.state.transcriber = moss + application.state.oss_uploader = oss application.state.video_processor = processor application.state.assessment_service = assessment_service @@ -130,6 +134,7 @@ def create_app( "video_count": len(videos) + legacy_store.count(), "managed_video_count": len(videos), "moss_configured": moss.available, + "volcano_oss_configured": oss.enabled, "scoring_version": "asr-fluency-v1", } @@ -157,10 +162,12 @@ def create_app( ) @application.get("/api/v1/videos/{video_hash}/content") - def stream_video(video_hash: str = SHA256_PATH) -> FileResponse: + def stream_video(video_hash: str = SHA256_PATH): row = video_repository.get_video(video_hash) if row is None: raise HTTPException(status_code=404, detail="Video was not found.") + if row.get("remote_url"): + return RedirectResponse(row["remote_url"], status_code=307) media_path = service_settings.videos_dir / row["stored_filename"] if not media_path.is_file(): raise HTTPException(status_code=404, detail="Stored video file is missing.") @@ -702,8 +709,15 @@ def _process_safely(processor: VideoProcessor, video_hash: str) -> None: def _video_summary(row: Dict[str, Any], settings: Settings) -> VideoSummary: - relative_stream_url = f"/api/v1/videos/{row['video_hash']}/content" - stream_url = f"{settings.public_base_url}{relative_stream_url}" if settings.public_base_url else relative_stream_url + if row.get("remote_url"): + stream_url = row["remote_url"] + else: + relative_stream_url = f"/api/v1/videos/{row['video_hash']}/content" + stream_url = ( + f"{settings.public_base_url}{relative_stream_url}" + if settings.public_base_url + else relative_stream_url + ) return VideoSummary( video_hash=row["video_hash"], title=row["title"], diff --git a/sentence_api/oss.py b/sentence_api/oss.py new file mode 100644 index 0000000..f1ee652 --- /dev/null +++ b/sentence_api/oss.py @@ -0,0 +1,77 @@ +import logging +from pathlib import Path +from typing import Optional +from urllib.parse import quote + +from .config import Settings + + +logger = logging.getLogger(__name__) + + +class VolcanoOSSUploader: + """Upload course videos to Volcano Engine Object Storage (TOS).""" + + def __init__(self, settings: Settings): + self.settings = settings + + @property + def enabled(self) -> bool: + return self.settings.volcano_oss_enabled + + def object_key(self, video_hash: str, stored_filename: str) -> str: + normalized_hash = video_hash.strip().lower() + suffix = Path(stored_filename).suffix or ".mp4" + return f"{self.settings.volcano_oss_key_prefix}{normalized_hash}{suffix}" + + def public_url(self, object_key: str) -> str: + return self.settings.volcano_oss_public_url(object_key) + + def upload( + self, + *, + video_hash: str, + stored_filename: str, + local_path: Path, + content_type: Optional[str] = None, + ) -> str: + """Upload one local file and return its permanent public URL.""" + if not self.enabled: + raise RuntimeError("Volcano OSS is not configured.") + if not local_path.is_file(): + raise FileNotFoundError(f"Cannot upload a missing video: {local_path}") + + # Keep the SDK import optional for development machines that do not run + # the production service. It is declared in sentence_api/requirements.txt. + try: + import tos + except ImportError as exc: + raise RuntimeError( + "The Volcano TOS SDK is not installed. Run: pip install tos" + ) from exc + + object_key = self.object_key(video_hash, stored_filename) + endpoint = self.settings.volcano_oss_endpoint + client = tos.TosClientV2( + self.settings.volcano_oss_access_key, + self.settings.volcano_oss_secret_key, + endpoint, + self.settings.volcano_oss_region, + ) + try: + client.put_object_from_file( + self.settings.volcano_oss_bucket, + object_key, + str(local_path), + ) + except Exception: + logger.exception( + "Volcano OSS upload failed: bucket=%s key=%s endpoint=%s", + self.settings.volcano_oss_bucket, + object_key, + endpoint, + ) + raise + + logger.info("Uploaded %s to Volcano OSS object %s", local_path.name, object_key) + return self.public_url(object_key) diff --git a/sentence_api/processing.py b/sentence_api/processing.py index 6ba30c7..807ea99 100644 --- a/sentence_api/processing.py +++ b/sentence_api/processing.py @@ -14,6 +14,7 @@ from .audio_metrics import ( from .config import Settings from .generate_boundaries import ALGORITHM_VERSION, make_entry from .models import SentenceBoundary, SentenceBoundaryDocument +from .oss import VolcanoOSSUploader from .repository import VideoRepository from .transcription import Transcript, Transcriber, split_sentences_at_punctuation @@ -27,10 +28,12 @@ class VideoProcessor: settings: Settings, repository: VideoRepository, transcriber: Transcriber, + oss_uploader: Optional[VolcanoOSSUploader] = None, ): self.settings = settings self.repository = repository self.transcriber = transcriber + self.oss_uploader = oss_uploader def process(self, video_hash: str) -> None: video = self.repository.get_video(video_hash) @@ -57,7 +60,7 @@ class VideoProcessor: ) if not document.sentences: raise RuntimeError("MOSS returned no timestamped speech segments.") - self.repository.save_processing_result(document, transcript.text) + self._finish_processing(document, transcript.text) else: entry, _ = make_entry(media_path, video_hash=video_hash) document = SentenceBoundaryDocument( @@ -66,7 +69,7 @@ class VideoProcessor: algorithm_version=ALGORITHM_VERSION, sentences=entry["sentences"], ) - self.repository.save_processing_result(document, None) + self._finish_processing(document, None) except Exception as exc: self.repository.mark_failed(video_hash, str(exc)) raise @@ -74,6 +77,28 @@ class VideoProcessor: if work_path is not None: work_path.unlink(missing_ok=True) + def _finish_processing( + self, + document: SentenceBoundaryDocument, + transcription: Optional[str], + ) -> None: + if self.oss_uploader is None or not self.oss_uploader.enabled: + self.repository.save_processing_result(document, transcription, ready=True) + return + + # Keep the course invisible while the mandatory OSS upload is running. + self.repository.save_processing_result(document, transcription, ready=False) + video = self.repository.get_video(document.video_hash) + if video is None: + raise ValueError(f"Unknown video: {document.video_hash}") + remote_url = self.oss_uploader.upload( + video_hash=document.video_hash, + stored_filename=video["stored_filename"], + local_path=self.settings.videos_dir / video["stored_filename"], + content_type=video.get("content_type"), + ) + self.repository.mark_oss_uploaded(document.video_hash, remote_url) + def extract_audio(media_path: Path, output_path: Path) -> None: ffmpeg = shutil.which("ffmpeg") diff --git a/sentence_api/repository.py b/sentence_api/repository.py index 7f99fe6..85dc137 100644 --- a/sentence_api/repository.py +++ b/sentence_api/repository.py @@ -43,6 +43,7 @@ class VideoRepository: error_message TEXT, algorithm_version TEXT, transcription TEXT, + remote_url TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); @@ -113,6 +114,14 @@ class VideoRepository: "ALTER TABLE dub_share_segments ADD COLUMN score_details TEXT NOT NULL DEFAULT '{}'" ) + with self._connect() as connection: + video_columns = { + row["name"] + for row in connection.execute("PRAGMA table_info(videos)") + } + if "remote_url" not in video_columns: + connection.execute("ALTER TABLE videos ADD COLUMN remote_url TEXT") + def upsert_upload( self, *, @@ -137,8 +146,8 @@ class VideoRepository: INSERT INTO videos ( video_hash, title, filename, stored_filename, content_type, size_bytes, duration_ms, language, status, error_message, - algorithm_version, transcription, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, 'uploaded', NULL, NULL, NULL, ?, ?) + algorithm_version, transcription, remote_url, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, 'uploaded', NULL, NULL, NULL, NULL, ?, ?) ON CONFLICT(video_hash) DO UPDATE SET title = excluded.title, filename = excluded.filename, @@ -180,6 +189,7 @@ class VideoRepository: self, document: SentenceBoundaryDocument, transcription: Optional[str], + ready: bool = True, ) -> None: with self._connect() as connection: connection.execute( @@ -208,12 +218,13 @@ class VideoRepository: ) connection.execute( """ - UPDATE videos SET duration_ms = ?, status = 'ready', error_message = NULL, + UPDATE videos SET duration_ms = ?, status = ?, error_message = NULL, algorithm_version = ?, transcription = ?, updated_at = ? WHERE video_hash = ? """, ( document.duration_ms, + "ready" if ready else "processing", document.algorithm_version, transcription, utc_now(), @@ -221,6 +232,17 @@ class VideoRepository: ), ) + def mark_oss_uploaded(self, video_hash: str, remote_url: str) -> None: + with self._connect() as connection: + connection.execute( + """ + UPDATE videos + SET remote_url = ?, status = 'ready', error_message = NULL, updated_at = ? + WHERE video_hash = ? + """, + (remote_url, utc_now(), normalize_video_hash(video_hash)), + ) + def list_videos(self) -> List[Dict[str, Any]]: with self._connect() as connection: rows = connection.execute( diff --git a/sentence_api/requirements.txt b/sentence_api/requirements.txt index a310817..06d7eda 100644 --- a/sentence_api/requirements.txt +++ b/sentence_api/requirements.txt @@ -6,3 +6,6 @@ httpx>=0.27,<1 python-multipart>=0.0.9,<1 av>=12.0 numpy>=1.26 +tos>=2.8,<3 +requests>=2.31,<3 +Deprecated>=1.2,<2 diff --git a/sentence_api/tests/test_api.py b/sentence_api/tests/test_api.py index 49c6052..2b4c3a0 100644 --- a/sentence_api/tests/test_api.py +++ b/sentence_api/tests/test_api.py @@ -5,6 +5,9 @@ from fastapi.testclient import TestClient from sentence_api.main import create_app from sentence_api.config import Settings +from sentence_api.oss import VolcanoOSSUploader +from sentence_api.models import SentenceBoundary, SentenceBoundaryDocument +from sentence_api.processing import VideoProcessor from sentence_api.repository import VideoRepository from sentence_api.store import BoundaryStore @@ -68,3 +71,120 @@ def test_invalid_hash_is_rejected(tmp_path): response = client.get("/api/v1/videos/not-a-sha256/sentence-boundaries") assert response.status_code == 422 + + +def test_oss_video_url_is_used_for_catalog_and_content(tmp_path): + index_path = tmp_path / "boundaries.json" + index_path.write_text("{}", encoding="utf-8") + settings = replace( + Settings.from_env(), + data_dir=tmp_path / "data", + legacy_boundaries_file=index_path, + ) + settings.ensure_directories() + repository = VideoRepository(settings.database_path) + repository.upsert_upload( + video_hash=VIDEO_HASH, + title="OSS lesson", + filename="lesson.mp4", + stored_filename=f"{VIDEO_HASH}.mp4", + content_type="video/mp4", + size_bytes=1, + language=None, + ) + remote_url = f"https://course-media.example.com/videos/{VIDEO_HASH}.mp4" + repository.mark_oss_uploaded(VIDEO_HASH, remote_url) + client = TestClient( + create_app( + BoundaryStore(index_path), + settings=settings, + repository=repository, + ) + ) + + catalog = client.get("/api/v1/videos") + stream = client.get(f"/api/v1/videos/{VIDEO_HASH}/content", follow_redirects=False) + + assert catalog.status_code == 200 + assert catalog.json()["videos"][0]["stream_url"] == remote_url + assert stream.status_code == 307 + assert stream.headers["location"] == remote_url + + +def test_volcano_oss_object_key_and_public_url(tmp_path): + settings = replace( + Settings.from_env(), + volcano_oss_access_key="ak", + volcano_oss_secret_key="sk", + volcano_oss_bucket="course-media", + volcano_oss_endpoint="tos-cn-beijing.volces.com", + volcano_oss_region="cn-beijing", + volcano_oss_key_prefix="lessons/videos/", + ) + uploader = VolcanoOSSUploader(settings) + key = uploader.object_key(VIDEO_HASH, f"{VIDEO_HASH}.mp4") + + assert settings.volcano_oss_enabled is True + assert key == f"lessons/videos/{VIDEO_HASH}.mp4" + assert uploader.public_url(key) == ( + f"https://course-media.tos-cn-beijing.volces.com/" + f"lessons/videos/{VIDEO_HASH}.mp4" + ) + + +def test_processor_marks_ready_only_after_oss_upload(tmp_path, monkeypatch): + settings = replace(Settings.from_env(), data_dir=tmp_path / "data") + settings.ensure_directories() + media_path = settings.videos_dir / f"{VIDEO_HASH}.mp4" + media_path.write_bytes(b"video") + repository = VideoRepository(settings.database_path) + repository.upsert_upload( + video_hash=VIDEO_HASH, + title="OSS lesson", + filename="lesson.mp4", + stored_filename=f"{VIDEO_HASH}.mp4", + content_type="video/mp4", + size_bytes=5, + language=None, + ) + document = SentenceBoundaryDocument( + video_hash=VIDEO_HASH, + duration_ms=1000, + algorithm_version="test", + sentences=[SentenceBoundary(index=0, start_ms=0, end_ms=1000)], + ) + uploads = [] + + class EnabledUploader: + enabled = True + + def upload(self, **kwargs): + uploads.append(kwargs) + return "https://media.example.com/lesson.mp4" + + class UnavailableTranscriber: + available = False + + monkeypatch.setattr( + "sentence_api.processing.make_entry", + lambda *args, **kwargs: ( + { + "duration_ms": 1000, + "sentences": [{"index": 0, "start_ms": 0, "end_ms": 1000}], + }, + {}, + ), + ) + processor = VideoProcessor( + settings, + repository, + UnavailableTranscriber(), + EnabledUploader(), + ) + + processor.process(VIDEO_HASH) + + video = repository.get_video(VIDEO_HASH) + assert uploads[0]["stored_filename"] == f"{VIDEO_HASH}.mp4" + assert video["status"] == "ready" + assert video["remote_url"] == "https://media.example.com/lesson.mp4"