smshine/mediaplayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

repaire sentences

Browse Source
This commit is contained in:
Shuming Liu
2026-10-21 19:39:04 +08:00
parent 176d722c77
commit 532d9900ba
13 changed files with 1233 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
sentence_api/tests/test_api.py
View File

@@ -1029,3 +1029,139 @@ def test_admin_can_update_course_title(tmp_path):
json={"title": "Missing course"},
)
assert missing_response.status_code == 404
def test_admin_can_repair_sentence_gaps(tmp_path, monkeypatch):
settings = replace(Settings.from_env(), data_dir=tmp_path / "data")
settings.ensure_directories()
repository = VideoRepository(settings.database_path)
repository.upsert_upload(
video_hash=VIDEO_HASH,
title="Repair gaps lesson",
filename="lesson.mp4",
stored_filename=f"{VIDEO_HASH}.mp4",
content_type="video/mp4",
size_bytes=5,
language="en",
)
(settings.videos_dir / f"{VIDEO_HASH}.mp4").write_bytes(b"fake-video")
repository.save_processing_result(
SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=18000,
algorithm_version="test",
sentences=[
SentenceBoundary(
index=0,
start_ms=0,
end_ms=2000,
text="Hello.",
language="en",
reference_speech_duration_ms=1800,
),
SentenceBoundary(
index=1,
start_ms=15000,
end_ms=18000,
text="Goodbye.",
language="en",
reference_speech_duration_ms=2500,
),
],
),
"Hello. Goodbye.",
)
repository.save_sentence_translations(VIDEO_HASH, {0: "你好。", 1: "再见。"})
from sentence_api.transcription import Transcript, TranscriptionSegment, _wav_bytes
import numpy as np
class GapTranscriber:
available = True
def transcribe(self, audio_path, language=None):
return Transcript(
text="The missing sentence.",
segments=[
TranscriptionSegment(
start_seconds=1.0,
end_seconds=5.0,
text="The missing sentence.",
)
],
)
def fake_extract(media_path, output_path, *, start_ms, duration_ms):
sample_rate = 16000
t = np.arange(int(max(1.0, duration_ms / 1000) * sample_rate)) / sample_rate
samples = (0.25 * np.sin(2 * np.pi * 220 * t)).astype(np.float32)
output_path.write_bytes(_wav_bytes(samples, sample_rate))
monkeypatch.setattr("sentence_api.processing.extract_audio_segment", fake_extract)
client = TestClient(
create_app(
BoundaryStore(tmp_path / "boundaries.json"),
settings=settings,
repository=repository,
transcriber=GapTranscriber(),
)
)
response = client.post(f"/api/v1/admin/videos/{VIDEO_HASH}/repair-gaps")
assert response.status_code == 200, response.text
payload = response.json()
assert payload["gap_count"] == 1
assert payload["inserted_count"] == 1
assert payload["sentence_count"] == 3
document = repository.get_document(VIDEO_HASH)
assert [item.text for item in document.sentences] == [
"Hello.",
"The missing sentence.",
"Goodbye.",
]
assert [item.zh_text for item in document.sentences] == ["你好。", None, "再见。"]
def test_admin_gap_repair_requires_transcriber(tmp_path):
settings = replace(Settings.from_env(), data_dir=tmp_path / "data")
settings.ensure_directories()
repository = VideoRepository(settings.database_path)
repository.upsert_upload(
video_hash=VIDEO_HASH,
title="Repair gaps lesson",
filename="lesson.mp4",
stored_filename=f"{VIDEO_HASH}.mp4",
content_type="video/mp4",
size_bytes=5,
language="en",
)
(settings.videos_dir / f"{VIDEO_HASH}.mp4").write_bytes(b"fake-video")
repository.save_processing_result(
SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=18000,
algorithm_version="test",
sentences=[
SentenceBoundary(index=0, start_ms=0, end_ms=2000, text="Hello."),
SentenceBoundary(index=1, start_ms=15000, end_ms=18000, text="Goodbye."),
],
),
None,
)
class UnavailableTranscriber:
available = False
client = TestClient(
create_app(
BoundaryStore(tmp_path / "boundaries.json"),
settings=settings,
repository=repository,
transcriber=UnavailableTranscriber(),
)
)
response = client.post(f"/api/v1/admin/videos/{VIDEO_HASH}/repair-gaps")
assert response.status_code == 503

176
sentence_api/tests/test_processing.py
View File

@@ -4,6 +4,7 @@ from sentence_api.processing import (
MOSS_ALGORITHM_VERSION,
VideoProcessor,
document_from_transcript,
find_sentence_gaps,
split_sentence_text,
)
from sentence_api.models import SentenceBoundary, SentenceBoundaryDocument
@@ -381,3 +382,178 @@ def test_mark_processing_clears_previous_sentence_data(tmp_path):
assert document.sentences == []
assert repository.get_moss_segments(VIDEO_HASH) == []
def test_find_sentence_gaps_detects_leading_middle_and_trailing():
document = SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=60000,
algorithm_version="test",
sentences=[
SentenceBoundary(index=0, start_ms=10000, end_ms=12000, text="A"),
SentenceBoundary(index=1, start_ms=25000, end_ms=27000, text="B"),
],
)
assert find_sentence_gaps(document, min_gap_ms=8000) == [
(0, 10000),
(12000, 25000),
(27000, 60000),
]
def test_find_sentence_gaps_ignores_short_pauses():
document = SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=9000,
algorithm_version="test",
sentences=[
SentenceBoundary(index=0, start_ms=0, end_ms=4000, text="A"),
SentenceBoundary(index=1, start_ms=4500, end_ms=9000, text="B"),
],
)
assert find_sentence_gaps(document) == []
def test_repair_gaps_inserts_transcribed_sentences_and_keeps_existing(tmp_path, monkeypatch):
from dataclasses import replace
from sentence_api.config import Settings
settings = replace(Settings.from_env(), data_dir=tmp_path / "data")
settings.ensure_directories()
repository = VideoRepository(settings.database_path)
repository.upsert_upload(
video_hash=VIDEO_HASH,
title="Gap repair lesson",
filename="lesson.mp4",
stored_filename=f"{VIDEO_HASH}.mp4",
content_type="video/mp4",
size_bytes=5,
language="en",
)
(settings.videos_dir / f"{VIDEO_HASH}.mp4").write_bytes(b"fake-video")
repository.save_processing_result(
SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=18000,
algorithm_version="test",
sentences=[
SentenceBoundary(
index=0,
start_ms=0,
end_ms=2000,
text="Hello.",
language="en",
reference_speech_duration_ms=1800,
),
SentenceBoundary(
index=1,
start_ms=15000,
end_ms=18000,
text="Goodbye.",
language="en",
reference_speech_duration_ms=2500,
),
],
),
"Hello. Goodbye.",
)
repository.save_sentence_translations(VIDEO_HASH, {0: "你好。", 1: "再见。"})
class GapTranscriber:
available = True
calls = 0
def transcribe(self, audio_path, language=None):
self.calls += 1
return Transcript(
text="The missing sentence.",
segments=[
TranscriptionSegment(
start_seconds=1.0,
end_seconds=5.0,
text="The missing sentence.",
)
],
)
def fake_extract(media_path, output_path, *, start_ms, duration_ms):
output_path.write_bytes(_wav_bytes(_speech(max(1.0, duration_ms / 1000)), SAMPLE_RATE))
monkeypatch.setattr("sentence_api.processing.extract_audio_segment", fake_extract)
transcriber = GapTranscriber()
result = VideoProcessor(settings, repository, transcriber).repair_gaps(VIDEO_HASH)
assert result["gap_count"] == 1
assert result["transcribed_gaps"] == 1
assert result["inserted_count"] == 1
assert transcriber.calls == 1
document = repository.get_document(VIDEO_HASH)
assert [sentence.text for sentence in document.sentences] == [
"Hello.",
"The missing sentence.",
"Goodbye.",
]
assert document.sentences[0].zh_text == "你好。"
assert document.sentences[1].zh_text is None
assert document.sentences[2].zh_text == "再见。"
assert document.sentences[0].end_ms <= document.sentences[1].start_ms
assert document.sentences[1].end_ms <= document.sentences[2].start_ms
assert document.sentences[1].start_ms >= 2000
assert document.sentences[1].end_ms <= 15000
def test_repair_gaps_skips_silent_gaps(tmp_path, monkeypatch):
from dataclasses import replace
from sentence_api.config import Settings
settings = replace(Settings.from_env(), data_dir=tmp_path / "data")
settings.ensure_directories()
repository = VideoRepository(settings.database_path)
repository.upsert_upload(
video_hash=VIDEO_HASH,
title="Silent gap lesson",
filename="lesson.mp4",
stored_filename=f"{VIDEO_HASH}.mp4",
content_type="video/mp4",
size_bytes=5,
language="en",
)
(settings.videos_dir / f"{VIDEO_HASH}.mp4").write_bytes(b"fake-video")
repository.save_processing_result(
SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=18000,
algorithm_version="test",
sentences=[
SentenceBoundary(index=0, start_ms=0, end_ms=2000, text="Hello."),
SentenceBoundary(index=1, start_ms=15000, end_ms=18000, text="Goodbye."),
],
),
None,
)
class TrackingTranscriber:
available = True
calls = 0
def transcribe(self, audio_path, language=None):
self.calls += 1
return Transcript(text="hallucinated", segments=[])
def fake_extract(media_path, output_path, *, start_ms, duration_ms):
output_path.write_bytes(_wav_bytes(_silence(max(1.0, duration_ms / 1000)), SAMPLE_RATE))
monkeypatch.setattr("sentence_api.processing.extract_audio_segment", fake_extract)
transcriber = TrackingTranscriber()
result = VideoProcessor(settings, repository, transcriber).repair_gaps(VIDEO_HASH)
assert result["gap_count"] == 1
assert result["inserted_count"] == 0
assert transcriber.calls == 0
document = repository.get_document(VIDEO_HASH)
assert [sentence.text for sentence in document.sentences] == ["Hello.", "Goodbye."]