328 lines
13 KiB
Python
328 lines
13 KiB
Python
"""Synthetic recording recovery: durable text, exact call binding, no duplicate upload."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import wave
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from doctor_workstation.core.errors import ApiTransportError
|
|
from doctor_workstation.services import local_audio_queue as queue_module
|
|
from doctor_workstation.services.api_client import ApiClient
|
|
from doctor_workstation.services.local_audio_queue import (
|
|
LocalAudioQueueStore,
|
|
LocalAudioUploadManager,
|
|
)
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.video import recording_transcription as recording
|
|
|
|
|
|
class Repository:
|
|
def __init__(self):
|
|
self.call = {
|
|
"id": 71, "caller_id": 5, "caller_type": "doctor", "status": 2,
|
|
"local_audio_status": 2, "transcription_status": "partial",
|
|
"local_audio_urls_list": ["https://audio.example.test/recording.webm"],
|
|
"transcription_session_id": "live-session", "transcript_text": "旧的片段",
|
|
}
|
|
self.uploads = 0
|
|
self.saves = []
|
|
self.fail_save = False
|
|
self.ack_id = 71
|
|
|
|
def get_current_user(self):
|
|
return SimpleNamespace(id=5)
|
|
|
|
def list_call_records(self, _diagnosis_id):
|
|
return [self.call.copy()]
|
|
|
|
def upload_call_recording(self, **kwargs):
|
|
self.uploads += 1
|
|
return {"file_url": "https://audio.example.test/recording.webm"}
|
|
|
|
def save_local_audio_transcript(self, **payload):
|
|
self.saves.append(payload)
|
|
if self.fail_save:
|
|
raise ApiTransportError("synthetic network failure")
|
|
return {"call_record_id": self.ack_id, "transcription_status": "completed", "saved": True}
|
|
|
|
|
|
def ready(store, *, uploaded=True):
|
|
record = store.begin_recording(
|
|
session_id="synthetic-local", diagnosis_id=9, call_record_id=71, mime_type="audio/webm"
|
|
)
|
|
record.file_path.write_bytes(b"synthetic audio")
|
|
store.finalize_recording(record.id, size_bytes=15)
|
|
if uploaded:
|
|
store.mark_uploaded(record.id, "https://audio.example.test/recording.webm")
|
|
return record.id
|
|
|
|
|
|
def test_upload_automatically_recovers_missing_text_without_reupload(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store, uploaded=False)
|
|
repository = Repository()
|
|
saved = threading.Event()
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: "合成测试文字")
|
|
manager.add_upload_listener(
|
|
lambda row: saved.set() if row.transcription_status == "completed" else None
|
|
)
|
|
assert manager.submit(record_id).result(3)
|
|
assert saved.wait(3)
|
|
assert repository.uploads == 1
|
|
assert repository.saves[0]["call_record_id"] == 71
|
|
assert repository.saves[0]["expected_session_id"] == "live-session"
|
|
assert repository.saves[0]["expected_status"] == "partial"
|
|
assert store.require(record_id).transcription_status == "completed"
|
|
assert store.require(record_id).exists
|
|
|
|
|
|
def test_failed_save_keeps_draft_and_id_across_restart(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
repository.fail_save = True
|
|
recognized = []
|
|
|
|
def transcribe(path):
|
|
recognized.append(path)
|
|
return "保留原始文字，0.5 毫克。"
|
|
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=transcribe)
|
|
assert not manager.submit_transcription(record_id).result(3)
|
|
first = store.require(record_id)
|
|
assert first.status == "uploaded"
|
|
assert first.transcription_status == "failed"
|
|
assert store.transcription_draft(record_id) == "保留原始文字，0.5 毫克。"
|
|
repository.fail_save = False
|
|
restarted = LocalAudioQueueStore(tmp_path)
|
|
retry = LocalAudioUploadManager(repository, restarted, transcriber=transcribe)
|
|
assert retry.submit_transcription(record_id).result(3)
|
|
assert len(recognized) == 1
|
|
assert repository.uploads == 0
|
|
assert repository.saves[0]["recovery_id"] == repository.saves[1]["recovery_id"]
|
|
assert not restarted.transcription_draft(record_id)
|
|
|
|
|
|
def test_existing_complete_text_skips_recognition_and_save(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
repository.call["transcription_status"] = "completed"
|
|
|
|
def forbidden(_path):
|
|
pytest.fail("must not overwrite an existing completed transcript")
|
|
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=forbidden)
|
|
assert manager.submit_transcription(record_id).result(3)
|
|
assert not repository.saves
|
|
|
|
|
|
@pytest.mark.parametrize("changes", [
|
|
{"caller_id": 6}, {"status": 0}, {"local_audio_status": 0}, {"id": 72},
|
|
{"local_audio_urls_list": ["https://audio.example.test/other-call.webm"]},
|
|
])
|
|
def test_wrong_owner_active_or_missing_call_does_not_transcribe(tmp_path, changes):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
repository.call.update(changes)
|
|
recognized = []
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda path: recognized.append(path))
|
|
assert not manager.submit_transcription(record_id).result(3)
|
|
assert not recognized
|
|
assert not repository.saves
|
|
assert store.require(record_id).status == "uploaded"
|
|
|
|
|
|
@pytest.mark.parametrize("text", ["", " ", "x" * 100_001], ids=["empty", "blank", "oversized"])
|
|
def test_empty_or_oversized_asr_is_never_marked_saved(tmp_path, text):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: text)
|
|
assert not manager.submit_transcription(record_id).result(3)
|
|
assert store.require(record_id).transcription_status == "failed"
|
|
assert not repository.saves
|
|
|
|
|
|
def test_wrong_call_ack_preserves_draft_as_failed(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
repository.ack_id = 72
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: "合成文字")
|
|
assert not manager.submit_transcription(record_id).result(3)
|
|
assert store.transcription_draft(record_id) == "合成文字"
|
|
assert store.require(record_id).transcription_status == "failed"
|
|
|
|
|
|
def test_duplicate_submit_recognizes_once(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
entered, release = threading.Event(), threading.Event()
|
|
|
|
def transcribe(_path):
|
|
entered.set()
|
|
assert release.wait(3)
|
|
return "合成文字"
|
|
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=transcribe)
|
|
first = manager.submit_transcription(record_id)
|
|
assert entered.wait(3)
|
|
assert manager.submit_transcription(record_id) is first
|
|
release.set()
|
|
assert first.result(3)
|
|
assert len(repository.saves) == 1
|
|
|
|
|
|
def test_interrupted_work_resumes_with_saved_draft(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
claimed = store.claim_transcription(record_id)
|
|
store.save_transcription_draft(record_id, "识别已完成但进程意外退出")
|
|
queue_module._RECOVERED_DATABASES.discard(store.database_path)
|
|
restarted = LocalAudioQueueStore(tmp_path)
|
|
assert restarted.require(record_id).transcription_status == "pending"
|
|
assert restarted.require(record_id).transcription_recovery_id == claimed.transcription_recovery_id
|
|
manager = LocalAudioUploadManager(Repository(), restarted)
|
|
assert manager.submit_pending_transcriptions(diagnosis_id=9)[0].result(3)
|
|
assert restarted.require(record_id).transcription_status == "completed"
|
|
|
|
|
|
def test_upload_finishes_before_call_end_then_text_recovers_automatically(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(queue_module, "_TRANSCRIPTION_END_RETRY_DELAYS", (0.01, 0.01))
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store, uploaded=False)
|
|
repository = Repository()
|
|
calls = []
|
|
saved = threading.Event()
|
|
|
|
def call_state(_diagnosis_id):
|
|
calls.append(True)
|
|
return [{**repository.call, "status": 1 if len(calls) == 1 else 2}]
|
|
|
|
repository.list_call_records = call_state
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: "合成文字")
|
|
manager.add_upload_listener(
|
|
lambda row: saved.set() if row.transcription_status == "completed" else None
|
|
)
|
|
assert manager.submit(record_id).result(3)
|
|
assert saved.wait(3)
|
|
# One poll while the call is still ending, one after it ended, one refresh right before saving.
|
|
assert len(calls) == 3
|
|
assert repository.uploads == 1
|
|
assert len(repository.saves) == 1
|
|
|
|
|
|
def test_save_compares_against_server_state_refreshed_after_recognition(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
states = [
|
|
{**repository.call, "transcription_status": "running", "transcript_text": ""},
|
|
{**repository.call, "transcription_status": "partial", "transcript_text": "旧的片段"},
|
|
]
|
|
repository.list_call_records = lambda _diagnosis_id: [states.pop(0)]
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: "合成文字")
|
|
assert manager.submit_transcription(record_id).result(3)
|
|
# The realtime finish landed while local recognition ran; the CAS uses the new status.
|
|
assert repository.saves[0]["expected_status"] == "partial"
|
|
assert repository.saves[0]["expected_session_id"] == "live-session"
|
|
assert store.require(record_id).transcription_status == "completed"
|
|
|
|
|
|
def test_text_completed_during_recognition_skips_save_without_failing(tmp_path):
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
states = [
|
|
{**repository.call, "transcription_status": "partial"},
|
|
{**repository.call, "transcription_status": "completed", "transcript_text": "医生：已经保存"},
|
|
]
|
|
repository.list_call_records = lambda _diagnosis_id: [states.pop(0)]
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda _path: "合成文字")
|
|
assert manager.submit_transcription(record_id).result(3)
|
|
assert repository.saves == []
|
|
assert store.require(record_id).transcription_status == "completed"
|
|
|
|
|
|
def test_waiting_for_call_end_is_bounded_and_retry_remains_available(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(queue_module, "_TRANSCRIPTION_END_RETRY_DELAYS", (0.01, 0.01))
|
|
store = LocalAudioQueueStore(tmp_path)
|
|
record_id = ready(store)
|
|
repository = Repository()
|
|
repository.call["status"] = 1
|
|
recognized = []
|
|
manager = LocalAudioUploadManager(repository, store, transcriber=lambda path: recognized.append(path))
|
|
assert not manager.submit_transcription(record_id).result(3)
|
|
assert not recognized
|
|
assert store.require(record_id).transcription_status == "failed"
|
|
repository.call["status"] = 2
|
|
manager._transcriber = lambda _path: "合成文字"
|
|
assert manager.submit_transcription(record_id).result(3)
|
|
|
|
|
|
def wav(path: Path, samples: int, *, rate=16_000, channels=1):
|
|
with wave.open(str(path), "wb") as output:
|
|
output.setnchannels(channels)
|
|
output.setsampwidth(2)
|
|
output.setframerate(rate)
|
|
output.writeframes(b"\x10\x00" * samples * channels)
|
|
|
|
|
|
def test_decoder_keeps_tail_and_resamples_stereo(tmp_path):
|
|
pytest.importorskip("av")
|
|
source = tmp_path / "synthetic.wav"
|
|
wav(source, 48_000 * 25, rate=48_000, channels=2)
|
|
chunks = list(recording.recording_pcm(source))
|
|
assert [len(chunk) for chunk in chunks] == [384_000, 384_000, 32_000]
|
|
|
|
|
|
def test_decoder_rejects_corrupt_or_overlong_audio(tmp_path, monkeypatch):
|
|
pytest.importorskip("av")
|
|
source = tmp_path / "synthetic.wav"
|
|
source.write_bytes(b"not audio")
|
|
with pytest.raises(recording.RecordingTranscriptionError, match="解码失败"):
|
|
list(recording.recording_pcm(source))
|
|
wav(source, 32_000)
|
|
monkeypatch.setattr(recording, "MAX_RECORDING_SECONDS", 1)
|
|
with pytest.raises(recording.RecordingTranscriptionError, match="超过"):
|
|
list(recording.recording_pcm(source))
|
|
|
|
|
|
def test_recording_transcriber_preserves_text_and_never_labels_speakers(tmp_path):
|
|
pytest.importorskip("av")
|
|
source = tmp_path / "synthetic.wav"
|
|
wav(source, 16_000 * 13)
|
|
responses = iter(["沒有過敏，0.5 mg。", "不是 5 mg。"])
|
|
engine = SimpleNamespace(transcribe=lambda _pcm: next(responses))
|
|
assert recording.RecordingTranscriber(engine)(source) == "没有过敏，0.5 mg。\n不是 5 mg。"
|
|
|
|
|
|
def test_repository_sends_exact_binding_and_compare_state():
|
|
calls = []
|
|
|
|
def handle(request):
|
|
assert request.url.path == "/adminapi/tcm.diagnosis/saveLocalAudioTranscript"
|
|
calls.append(json.loads(request.content))
|
|
return httpx.Response(200, json={"code": 1, "data": {
|
|
"call_record_id": 71, "transcription_status": "completed", "saved": True,
|
|
}})
|
|
|
|
client = ApiClient("https://example.test/adminapi/", transport=httpx.MockTransport(handle))
|
|
repository = RemoteDoctorRepository(client)
|
|
payload = dict(
|
|
diagnosis_id=9, call_record_id=71, recovery_id="audio-" + "a" * 32,
|
|
expected_session_id="live-session", expected_status="partial",
|
|
audio_url="https://audio.example.test/recording.webm", transcript_text="合成测试文字",
|
|
)
|
|
assert repository.save_local_audio_transcript(**payload)["saved"]
|
|
assert calls == [payload]
|