80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from services.shared.core import new_id, utc_now_iso
|
|
from services.shared.sql_models import VoiceTranscriptSegmentRow
|
|
|
|
|
|
_SEQUENCE_CONFLICT_MARKERS = (
|
|
"uq_voice_transcript_segments_session_sequence",
|
|
"voice_transcript_segments.session_id, voice_transcript_segments.sequence_no",
|
|
"voice_transcript_segments.session_id, sequence_no",
|
|
)
|
|
|
|
|
|
def next_transcript_sequence(session, session_id: str) -> int:
|
|
row = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.order_by(VoiceTranscriptSegmentRow.sequence_no.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
return 1
|
|
return int(row.sequence_no or 0) + 1
|
|
|
|
|
|
def _is_sequence_conflict(exc: IntegrityError) -> bool:
|
|
message = str(exc).lower()
|
|
return any(marker in message for marker in _SEQUENCE_CONFLICT_MARKERS)
|
|
|
|
|
|
def add_transcript_segment(
|
|
session,
|
|
*,
|
|
session_id: str,
|
|
call_id: str,
|
|
interaction_id: str | None,
|
|
speaker: str,
|
|
source_type: str,
|
|
text: str,
|
|
confidence: float | None = None,
|
|
payload: dict | None = None,
|
|
is_final: bool = True,
|
|
barge_in_interrupted: bool = False,
|
|
sequence_no: int | None = None,
|
|
created_at: str | None = None,
|
|
max_attempts: int = 5,
|
|
) -> VoiceTranscriptSegmentRow:
|
|
for attempt in range(max_attempts):
|
|
candidate_sequence = sequence_no if attempt == 0 and sequence_no is not None else next_transcript_sequence(session, session_id)
|
|
row = VoiceTranscriptSegmentRow(
|
|
segment_id=new_id("vts"),
|
|
session_id=session_id,
|
|
call_id=call_id,
|
|
interaction_id=interaction_id,
|
|
speaker=speaker,
|
|
source_type=source_type,
|
|
sequence_no=candidate_sequence,
|
|
text=text,
|
|
confidence=confidence,
|
|
is_final=is_final,
|
|
barge_in_interrupted=barge_in_interrupted,
|
|
payload_json=json.dumps(payload or {}, ensure_ascii=False),
|
|
created_at=created_at or utc_now_iso(),
|
|
)
|
|
try:
|
|
with session.begin_nested():
|
|
session.add(row)
|
|
session.flush()
|
|
return row
|
|
except IntegrityError as exc:
|
|
if not _is_sequence_conflict(exc) or attempt >= max_attempts - 1:
|
|
raise
|
|
session.expire_all()
|
|
raise RuntimeError("Voice transcript segment insert retry exhausted")
|