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

fixed bugs

Browse Source
This commit is contained in:
Shuming Liu
2026-10-29 11:32:33 +08:00
parent 031f3c6f21
commit a87228cdc9
7 changed files with 229 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
sentence_api/tests/test_api.py
View File

@@ -824,7 +824,7 @@ def test_admin_can_merge_adjacent_sentences_and_reindexes_following_sentences(tm
assert [item.index for item in document.sentences] == [0, 1, 2]
assert [item.start_ms for item in document.sentences] == [0, 2000, 9000]
assert [item.end_ms for item in document.sentences] == [2000, 9000, 12000]
assert all(not (item.zh_text or "").strip() for item in document.sentences)
assert [item.zh_text for item in document.sentences] == ["之前。", None, "下一句。"]
last_response = client.put(
f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/2/merge",
@@ -936,7 +936,73 @@ def test_admin_batch_applies_merge_and_split_once(tmp_path, monkeypatch):
assert all(not (item.zh_text or "").strip() for item in document.sentences)
def test_admin_sentence_merge_restarts_subtitle_translation(tmp_path, monkeypatch):
def test_admin_batch_can_persist_provisional_split_without_processing(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="Provisional split lesson",
filename="lesson.mp4",
stored_filename=f"{VIDEO_HASH}.mp4",
content_type="video/mp4",
size_bytes=5,
language="en",
)
repository.save_processing_result(
SentenceBoundaryDocument(
video_hash=VIDEO_HASH,
duration_ms=12000,
algorithm_version="test",
sentences=[
SentenceBoundary(index=0, start_ms=0, end_ms=5000, text="First part second part", language="en", reference_speech_duration_ms=4600),
SentenceBoundary(index=1, start_ms=5000, end_ms=12000, text="Unchanged.", language="en", reference_speech_duration_ms=6400),
],
),
None,
)
repository.save_sentence_translations(VIDEO_HASH, {1: "未变化。"})
calls = []
class MustNotTranscribe:
available = True
def transcribe(self, audio_path, language=None):
calls.append(audio_path)
raise AssertionError("provisional save must not transcribe")
client = TestClient(
create_app(
BoundaryStore(tmp_path / "boundaries.json"),
settings=settings,
repository=repository,
transcriber=MustNotTranscribe(),
)
)
response = client.put(
f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/batch",
json={
"process_timings": False,
"sentences": [
{"start_ms": 0, "end_ms": 2500, "text": "First part", "language": "en", "zh_text": None, "reference_speech_duration_ms": 2300, "source_start_ms": 0, "source_end_ms": 5000},
{"start_ms": 2500, "end_ms": 5000, "text": "second part", "language": "en", "zh_text": None, "reference_speech_duration_ms": 2300, "source_start_ms": 0, "source_end_ms": 5000},
{"start_ms": 5000, "end_ms": 12000, "text": "Unchanged.", "language": "en", "zh_text": "未变化。", "reference_speech_duration_ms": 6400},
],
},
)
assert response.status_code == 200, response.text
payload = response.json()
assert [item["text"] for item in payload] == ["First part", "second part", "Unchanged."]
assert [item["start_ms"] for item in payload] == [0, 2500, 5000]
assert [item["zh_text"] for item in payload] == [None, None, "未变化。"]
assert calls == []
document = repository.get_document(VIDEO_HASH)
assert [item.text for item in document.sentences] == ["First part", "second part", "Unchanged."]
assert [item.zh_text for item in document.sentences] == [None, None, "未变化。"]
def test_admin_sentence_merge_does_not_start_subtitle_translation(tmp_path, monkeypatch):
client = make_client(tmp_path)
_ready_video(client.app.state.video_repository)
requested = []
@@ -966,66 +1032,15 @@ def test_admin_sentence_merge_restarts_subtitle_translation(tmp_path, monkeypatc
assert merge.status_code == 200, merge.text
assert merge.json()["text"] == "Hello. Nice to meet you."
finished = wait_for_subtitle_translation(client, VIDEO_HASH)
assert finished.json()["complete"] is True
assert finished.json()["translations"] == [
{"index": 0, "zh_text": "中文:Hello. Nice to meet you."},
]
snapshot = client.get(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation")
assert snapshot.status_code == 200
assert snapshot.json()["status"] == "idle"
assert snapshot.json()["complete"] is False
assert snapshot.json()["translations"] == []
assert len(requested) == 1
document = client.app.state.video_repository.get_document(VIDEO_HASH)
assert [sentence.text for sentence in document.sentences] == ["Hello. Nice to meet you."]
assert [sentence.zh_text for sentence in document.sentences] == ["中文:Hello. Nice to meet you."]
assert requested[-1] == [(0, "Hello. Nice to meet you.")]
def test_admin_sentence_merge_drops_stale_in_flight_translation(tmp_path, monkeypatch):
client = make_client(tmp_path)
_ready_video(client.app.state.video_repository)
started = threading.Event()
release = threading.Event()
requested = []
def translate_sentences(sentences, **kwargs):
call_index = len(requested)
requested.append([(index, text) for index, text in sentences])
persist = kwargs.get("persist")
if call_index == 0:
if persist is not None:
persist({0: "旧0", 1: "旧1"})
started.set()
assert release.wait(timeout=5)
if persist is not None:
persist({0: "晚0", 1: "晚1"})
return {0: "晚0", 1: "晚1"}
return {index: f"新:{text}" for index, text in sentences}
class StubAITutor:
available = True
stub = StubAITutor()
stub.translate_sentences = translate_sentences
monkeypatch.setattr(client.app.state, "ai_tutor", stub)
client.app.state.subtitle_translations.ai_tutor = stub
client.app.state.subtitle_translations.post_wait_seconds = 0.2
response = client.post(f"/api/v1/videos/{VIDEO_HASH}/subtitle-translation")
assert started.wait(timeout=2)
assert response.status_code == 200, response.text
merge = client.put(f"/api/v1/admin/videos/{VIDEO_HASH}/sentences/0/merge")
assert merge.status_code == 200, merge.text
release.set()
finished = wait_for_subtitle_translation(client, VIDEO_HASH)
assert finished.json()["complete"] is True
assert finished.json()["translations"] == [
{"index": 0, "zh_text": "新:Hello. Nice to meet you."},
]
document = client.app.state.video_repository.get_document(VIDEO_HASH)
assert [sentence.zh_text for sentence in document.sentences] == [
"新:Hello. Nice to meet you."
]
assert requested[0] == [(0, "Hello."), (1, "Nice to meet you.")]
assert requested[-1] == [(0, "Hello. Nice to meet you.")]
assert [sentence.zh_text for sentence in document.sentences] == [None]
def test_admin_boundary_adjustment_updates_linked_next_start(tmp_path):