"""
Client-side PII redaction for Cekura observability payloads.
Detects and removes PII from the transcript *and* the audio recording before
either leaves your infrastructure, using your own LLM and speech-to-text
providers and your own API keys. Nothing is sent to Cekura until it is scrubbed.
from cekura_redact import openai_compatible_detector, openai_compatible_transcriber
from cekura_redact import redact_payload, redact_audio
detect = openai_compatible_detector(
base_url="https://api.openai.com/v1", api_key=OPENAI_KEY, model="gpt-4.1-mini")
transcribe = openai_compatible_transcriber(
base_url="https://api.openai.com/v1", api_key=OPENAI_KEY, model="whisper-1")
FIELDS = ["person_name", "phone_number", "email_address", "location", "ssn"]
payload = redact_payload(payload, FIELDS, detector=detect)
clean_audio = redact_audio(raw_audio_bytes, FIELDS, detector=detect, transcriber=transcribe)
Dependencies: none for the transcript path (standard library only).
The audio path needs `pydub` and `ffmpeg` for any format other than WAV.
"""
from __future__ import annotations
import io
import json
import re
import urllib.request
import uuid
from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
# Placeholder tokens match the ones Cekura's server-side redaction emits, so a
# call scrubbed here reads the same as one scrubbed with `redact_fields`.
ENTITY_TO_TOKEN: Dict[str, str] = {
"person_name": "<PERSON>",
"age": "<AGE>",
"phone_number": "<PHONE>",
"email_address": "<EMAIL>",
"date": "<DATE>",
"time": "<TIME>",
"location": "<LOCATION>",
"origin": "<ORIGIN>",
"gender_sexuality": "<GENDER>",
"physical_attribute": "<ATTRIBUTE>",
"occupation": "<OCCUPATION>",
"username": "<USERNAME>",
"password": "<PASSWORD>",
"ip_address": "<IP_ADDRESS>",
"url": "<URL>",
"filename": "<FILENAME>",
"event": "<EVENT>",
"vehicle_id": "<VEHICLE_ID>",
"dob": "<DOB>",
"healthcare_number": "<HEALTHCARE_NUMBER>",
"medical_professional": "<MEDICAL_PROFESSIONAL>",
"credit_card": "<CREDIT_CARD>",
"account_number": "<ACCOUNT_NUMBER>",
"bank_account": "<BANK_ACCOUNT>",
"money": "<MONEY>",
"ssn": "<SSN>",
"passport_number": "<PASSPORT>",
"driver_license": "<DRIVER_LICENSE>",
"numerical_pii": "<NUMERICAL_PII>",
}
# What each token covers. Spoken transcripts render PII in ways a regex never
# catches — dictated digits, spelled-out names, "at gmail dot com" — so the
# descriptions call those forms out explicitly.
ENTITY_DESCRIPTIONS: Dict[str, str] = {
"<PERSON>": "Names: first, last, full names, agent names in the intro (\"I'm Kate\"), spelled formats (\"b e n j a m i n\")",
"<AGE>": "Age: numeric ages (\"65\", \"25 years old\"), age thresholds in questions (\"65 or older\")",
"<PHONE>": "Phone numbers in any format, including dictated (\"4 1 6 9 1 2 0 8 5 9\") and partial (\"ending in 5 1 4 2\")",
"<EMAIL>": "Email addresses: standard, dictated (\"a at gmail dot com\"), spelled (\"g m a i l\"), partial domains",
"<DATE>": "Dates: full (\"September eighteenth 2025\"), partial (\"September eighteenth\"), standalone months, years. A date is PII even when it describes a policy or account rather than the person",
"<TIME>": "Clock times",
"<LOCATION>": "Addresses, cities, states, counties, ZIP codes (including dictated \"2 5 8 3 6\"), locations inside titles (\"Guernsey County Sheriff\")",
"<ORIGIN>": "Nationality, ethnicity",
"<GENDER>": "Gender, sexuality",
"<ATTRIBUTE>": "Physical attributes",
"<OCCUPATION>": "Job titles",
"<USERNAME>": "Usernames",
"<PASSWORD>": "Passwords",
"<IP_ADDRESS>": "IP addresses",
"<URL>": "URLs",
"<FILENAME>": "File names",
"<EVENT>": "Events",
"<VEHICLE_ID>": "Vehicle IDs",
"<DOB>": "Date of birth",
"<HEALTHCARE_NUMBER>": "Healthcare and insurance member numbers",
"<MEDICAL_PROFESSIONAL>": "Medical professional names",
"<CREDIT_CARD>": "Credit card numbers, expiry dates, CVV",
"<ACCOUNT_NUMBER>": "Account numbers",
"<BANK_ACCOUNT>": "Bank account and routing numbers",
"<MONEY>": "Monetary amounts with or without a symbol, numeric or spoken (\"$50\", \"1,200 dollars\", \"five thousand\"), in any context",
"<SSN>": "SSN, ITIN, and numerical PII such as appointment or reference IDs (\"4 8 2 1 9\")",
"<PASSPORT>": "Passport numbers",
"<DRIVER_LICENSE>": "Driver's license numbers",
"<NUMERICAL_PII>": "Any other numerical PII",
}
class RedactionError(Exception):
"""Redaction could not be completed. Callers MUST fail closed on this — never
send the payload or the recording to Cekura when redaction did not run."""
# ---------------------------------------------------------------------------
# Detection
# ---------------------------------------------------------------------------
Span = Dict[str, str] # {"original_text": ..., "replacement_token": ...}
Detector = Callable[[str], Optional[List[Span]]]
SUBMIT_TOOL = {
"type": "function",
"function": {
"name": "submit_redactions",
"description": "Submit every PII span found in the transcript.",
"parameters": {
"type": "object",
"properties": {
"redactions": {
"type": "array",
"items": {
"type": "object",
"properties": {
"original_text": {
"type": "string",
"description": "Exact PII text found in the transcript, copied verbatim.",
},
"replacement_token": {
"type": "string",
"description": "The token to replace it with (e.g. <PERSON>, <EMAIL>).",
},
},
"required": ["original_text", "replacement_token"],
},
}
},
"required": ["redactions"],
},
},
}
def build_detection_prompt(fields: Iterable[str], text: str) -> str:
"""Build the PII-detection prompt for `text`, given the entities to redact.
Entities you did not enable are listed explicitly as must-not-redact, which
keeps the model from over-redacting and destroying conversational context.
"""
enabled_tokens = {ENTITY_TO_TOKEN[f] for f in fields if f in ENTITY_TO_TOKEN}
enabled, disabled = [], []
for token in sorted(ENTITY_DESCRIPTIONS):
line = f"{token}: {ENTITY_DESCRIPTIONS[token]}"
(enabled if token in enabled_tokens else disabled).append(line)
return f"""You are identifying personally identifiable information (PII) in a conversation transcript so it can be redacted.
<transcript>
{text}
</transcript>
PII entity types you MUST redact:
<entities_to_redact>
{chr(10).join(enabled)}
</entities_to_redact>
PII entity types you MUST NOT redact:
<entities_to_ignore>
{chr(10).join(disabled) if disabled else "None"}
</entities_to_ignore>
## Rules
1. **Use context, not format.** Spoken transcripts render PII in many shapes. When
someone spells something out letter by letter or digit by digit — often after
"is that spelled..." — the spelled text is PII. "B-E-N", "B. E. N" and "b e n"
are all the same name; "2 5 8 3 6" may be a ZIP code; "a at g m a i l dot com"
is an email address.
2. **Redact every occurrence.** If a value is PII, catch every instance of it —
confirmations, repetitions and echoes included.
3. **Do not redact generic references.** "What date would you prefer?" and "Can I
get your email address?" are generic. "My email is john@example.com" is not.
4. **`original_text` must be verbatim.** Each span must be an exact, contiguous,
character-for-character copy of text in the transcript. Never paraphrase,
normalize, shorten, or abbreviate with an ellipsis — a span that does not
appear verbatim cannot be replaced, and the PII leaks.
5. **Stay inside one utterance.** The transcript may be JSON. Each span must lie
entirely within a single quoted message: never include quotes, braces or field
names, and never span two messages. Emit one entry per message instead.
6. **Emit standalone fragments too.** If part of a redacted value also appears on
its own elsewhere — the month of a redacted date, the last digits of a redacted
number, a single letter confirming a spelled name — give that occurrence its
own entry.
## Examples
- "I'm Kate, an AI assistant" -> {{"original_text": "Kate", "replacement_token": "<PERSON>"}}
- "Is that spelled b e n j a m i n?" -> {{"original_text": "b e n j a m i n", "replacement_token": "<PERSON>"}}
- "ZIP code is 2 5 8 3 6" -> {{"original_text": "2 5 8 3 6", "replacement_token": "<LOCATION>"}}
- "a at g m a i l dot com" -> {{"original_text": "a at g m a i l dot com", "replacement_token": "<EMAIL>"}}
- "the number ending in 5 1 4 2" -> {{"original_text": "5 1 4 2", "replacement_token": "<PHONE>"}}
## Before you answer
Sweep the transcript once more for (a) any remaining digit or currency symbol and
(b) any remaining full or partial occurrence of a value you already mapped. Add an
entry for each hit, or confirm it falls outside the enabled entity types. Do not
finish while an enabled entity remains unmapped.
Call `submit_redactions` with every span. An empty list is the correct answer when
the transcript contains no PII of the enabled types.
"""
def openai_compatible_detector(
base_url: str,
api_key: str,
model: str,
timeout: int = 180,
extra_headers: Optional[Mapping[str, str]] = None,
) -> Detector:
"""Detector backed by any OpenAI-compatible `/chat/completions` endpoint.
Works with OpenAI, Azure OpenAI, and self-hosted or gateway deployments
(vLLM, Ollama, Bedrock/Vertex via a compatible proxy) — anything that accepts
tool calls. Returns None on failure so the caller fails closed.
"""
def detect(prompt: str) -> Optional[List[Span]]:
body = json.dumps({
"model": model,
"messages": [{"role": "user", "content": prompt}],
"tools": [SUBMIT_TOOL],
"tool_choice": {"type": "function", "function": {"name": "submit_redactions"}},
"temperature": 0,
}).encode()
headers = {"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}", **(extra_headers or {})}
request = urllib.request.Request(
base_url.rstrip("/") + "/chat/completions", data=body, headers=headers)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
payload = json.loads(response.read())
calls = payload["choices"][0]["message"].get("tool_calls") or []
if not calls:
return None
spans = json.loads(calls[0]["function"]["arguments"] or "{}").get("redactions")
# An empty list is a valid "no PII" verdict; anything else is a failure.
return spans if isinstance(spans, list) else None
except Exception: # noqa: BLE001 — None means "failed", caller fails closed
return None
return detect
# A number attached to a currency symbol is a monetary amount by definition, so a
# deterministic sweep can back the model up: an amount it misread as a code is
# still caught. Same rationale for a month followed by a day or a year.
_CURRENCY_AMOUNT = re.compile(
r"[$€£₹]\s?\d[\d,]*(?:\.\d+)?"
r"|\b\d[\d,]*(?:\.\d+)?\s?(?:dollars?|euros?|pounds?|rupees?)\b", re.I)
_MONTH = (r"January|February|March|April|May|June|July|August|September|October"
r"|November|December|Jan|Feb|Mar|Apr|Jun|Jul|Aug|Sept?|Oct|Nov|Dec")
_MONTH_DATE = re.compile(
rf"\b(?:{_MONTH})\.?\s+\d{{1,2}}(?:st|nd|rd|th)?(?:,?\s+\d{{4}})?\b"
rf"|\b(?:{_MONTH})\.?,?\s+\d{{4}}\b")
_DETERMINISTIC = [
("money", _CURRENCY_AMOUNT),
("date", _MONTH_DATE),
("email_address", re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]{2,}\b", re.I)),
("ip_address", re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")),
("ssn", re.compile(r"(?<!\d)\d{3}-\d{2}-\d{4}(?!\d)")),
("phone_number", re.compile(
r"(?<![\d-])(?:\+\d{1,3}[\s.-]?)?(?:\(\d{3}\)|\d{3})[\s.-]?\d{3}[\s.-]?\d{4}(?![\d-])")),
]
def _deterministic_spans(text: str, fields: Sequence[str]) -> List[Span]:
"""Regex-derived spans that supplement the model. Every pattern here is
unambiguous by construction, so no model judgment is involved."""
spans: List[Span] = []
for entity, pattern in _DETERMINISTIC:
if entity not in fields:
continue
for match in set(pattern.findall(text)):
spans.append({"original_text": match,
"replacement_token": ENTITY_TO_TOKEN[entity]})
return spans
def _known_value_spans(
text: str, known_values: Optional[Mapping[str, Sequence[str]]]
) -> List[Span]:
"""Optional literal values you already hold (CRM name, address, account
number). Detection finds these on its own; passing them in is a belt-and-braces
guarantee for the values you cannot afford to miss."""
spans: List[Span] = []
for entity, values in (known_values or {}).items():
token = ENTITY_TO_TOKEN.get(entity)
if not token:
continue
for value in values:
value = (value or "").strip()
if len(value) < 3: # too short to match safely
continue
for match in set(re.findall(
rf"(?<!\w){re.escape(value)}(?!\w)", text, re.I)):
spans.append({"original_text": match, "replacement_token": token})
return spans
def detect_spans(
text: str,
fields: Iterable[str],
detector: Detector,
known_values: Optional[Mapping[str, Sequence[str]]] = None,
) -> List[Span]:
"""Detect every PII span in `text`, raising RedactionError if detection failed.
Never treat a failure as "no PII found" — that is the one bug in this whole
file that silently ships raw PII.
"""
fields = [f for f in fields if f in ENTITY_TO_TOKEN]
if not fields or not text:
return []
spans = detector(build_detection_prompt(fields, text))
if spans is None:
raise RedactionError(
"PII detection failed; refusing to return unredacted content")
spans = [s for s in spans if s.get("original_text") and s.get("replacement_token")]
return spans + _deterministic_spans(text, fields) + _known_value_spans(text, known_values)
# ---------------------------------------------------------------------------
# Transcript / payload redaction
# ---------------------------------------------------------------------------
# Top-level payload keys scrubbed key-and-value recursively. Cekura's server-side
# `redact_fields` covers the transcript and the audio only, so these are the
# fields client-side scrubbing exists for.
_FREEFORM_PAYLOAD_KEYS = ("metadata", "dynamic_variables")
def apply_spans(node: Any, spans: Sequence[Span]) -> Any:
"""Apply spans to every string leaf of a JSON-ish structure.
Replacing on the string *values* rather than on `json.dumps(node)` keeps the
result valid by construction: a span that happens to contain a quote or brace
can never corrupt the surrounding document.
"""
if isinstance(node, str):
out = node
# Longest span first: replacing "Jane" before "Jane Doe" would leave the
# surname behind as "<PERSON> Doe".
for span in sorted(spans, key=lambda s: len(s["original_text"]), reverse=True):
out = out.replace(span["original_text"], span["replacement_token"])
return out
if isinstance(node, list):
return [apply_spans(v, spans) for v in node]
if isinstance(node, dict):
return {k: apply_spans(v, spans) for k, v in node.items()}
return node
def redact_transcript(
transcript_json: Any,
fields: Iterable[str],
detector: Detector,
known_values: Optional[Mapping[str, Sequence[str]]] = None,
return_spans: bool = False,
):
"""Scrub a `transcript_json` value, leaving timings, roles and IDs intact.
Detection runs once over the whole serialized transcript — one call, full
conversational context — and the spans are applied per string leaf.
"""
if not transcript_json:
return (transcript_json, []) if return_spans else transcript_json
spans = detect_spans(json.dumps(transcript_json), fields, detector, known_values)
redacted = apply_spans(transcript_json, spans)
return (redacted, spans) if return_spans else redacted
def redact_payload(
payload: Mapping[str, Any],
fields: Iterable[str],
detector: Detector,
known_values: Optional[Mapping[str, Sequence[str]]] = None,
return_spans: bool = False,
):
"""Return a copy of an observe payload with PII replaced by placeholder tokens.
Covers `transcript_json`, `transcript`, `metadata`, `dynamic_variables` and
`customer_number`. The recording is handled separately by `redact_audio`.
"""
fields = list(fields)
out = dict(payload)
all_spans: List[Span] = []
if out.get("transcript_json") is not None:
out["transcript_json"], spans = redact_transcript(
out["transcript_json"], fields, detector, known_values, return_spans=True)
all_spans += spans
# metadata / dynamic_variables / transcript / customer_number are usually
# short: detect over them together in one more call, then apply everywhere.
side_channel = {k: out[k] for k in _FREEFORM_PAYLOAD_KEYS + ("transcript", "customer_number")
if out.get(k) is not None}
if side_channel:
side_fields = list(fields)
if isinstance(out.get("customer_number"), str) and "phone_number" not in side_fields:
side_fields.append("phone_number")
spans = detect_spans(json.dumps(side_channel), side_fields, detector, known_values)
all_spans += spans
for key in side_channel:
out[key] = apply_spans(out[key], all_spans)
return (out, all_spans) if return_spans else out
# ---------------------------------------------------------------------------
# Audio redaction
# ---------------------------------------------------------------------------
Word = Dict[str, Any] # {"text": ..., "start": <sec>, "end": <sec>}
Transcriber = Callable[[bytes], Tuple[str, List[Word]]]
def _multipart(fields: Mapping[str, str], filename: str, content: bytes) -> Tuple[bytes, str]:
boundary = uuid.uuid4().hex
buffer = io.BytesIO()
for name, value in fields.items():
buffer.write(f"--{boundary}\r\nContent-Disposition: form-data; "
f'name="{name}"\r\n\r\n{value}\r\n'.encode())
buffer.write(f"--{boundary}\r\nContent-Disposition: form-data; "
f'name="file"; filename="{filename}"\r\n'
f"Content-Type: application/octet-stream\r\n\r\n".encode())
buffer.write(content)
buffer.write(f"\r\n--{boundary}--\r\n".encode())
return buffer.getvalue(), f"multipart/form-data; boundary={boundary}"
def openai_compatible_transcriber(
base_url: str,
api_key: str,
model: str = "whisper-1",
filename: str = "call.mp3",
timeout: int = 300,
extra_fields: Optional[Mapping[str, str]] = None,
) -> Transcriber:
"""Transcriber backed by an OpenAI-compatible `/audio/transcriptions` endpoint
that returns word-level timestamps (OpenAI, Azure OpenAI, Groq, self-hosted
Whisper). Word timings are what make audio redaction possible — a transcript
without them cannot be aligned back to the waveform.
"""
def transcribe(audio_bytes: bytes) -> Tuple[str, List[Word]]:
fields = {"model": model, "response_format": "verbose_json",
"timestamp_granularities[]": "word", **(extra_fields or {})}
body, content_type = _multipart(fields, filename, audio_bytes)
request = urllib.request.Request(
base_url.rstrip("/") + "/audio/transcriptions", data=body,
headers={"Content-Type": content_type, "Authorization": f"Bearer {api_key}"})
with urllib.request.urlopen(request, timeout=timeout) as response:
payload = json.loads(response.read())
words = [{"text": w["word"], "start": float(w["start"]), "end": float(w["end"])}
for w in payload.get("words") or []]
if not words:
raise RedactionError(
"Transcription returned no word timestamps; audio cannot be aligned")
return payload.get("text", ""), words
return transcribe
def _build_word_index(words: Sequence[Word]) -> Tuple[str, List[Tuple[int, Word]]]:
"""Lay the words out as one string and record where each one starts, so a
detected span can be mapped back to the words it covers."""
parts, index, position = [], [], 0
for word in words:
text = (word.get("text") or "").strip()
if not text:
continue
parts.append(text)
index.append((position, {"text": text, "start": word["start"], "end": word["end"]}))
position += len(text) + 1
return " ".join(parts), index
def find_phrase_timestamps(
phrase: str, text: str, index: Sequence[Tuple[int, Word]]
) -> List[Tuple[float, float]]:
"""Locate every occurrence of `phrase` in `text`, ignoring case, punctuation
and spacing, and return the (start, end) audio range of each."""
normalized, positions = [], []
for position, character in enumerate(text):
if character.isalnum():
normalized.append(character.casefold())
positions.append(position)
needle = "".join(c.casefold() for c in phrase if c.isalnum())
if not needle:
return []
haystack = "".join(normalized)
timestamps, found = [], haystack.find(needle)
while found != -1:
start_pos = positions[found]
end_pos = positions[found + len(needle) - 1] + 1
covered = [word for word_pos, word in index
if word_pos < end_pos and word_pos + len(word["text"]) > start_pos]
if covered:
timestamps.append((covered[0]["start"], covered[-1]["end"]))
found = haystack.find(needle, found + len(needle))
return timestamps
def _fuzzy_phrase_timestamps(
phrase: str, index: Sequence[Tuple[int, Word]], max_ratio: float = 0.25
) -> List[Tuple[float, float]]:
"""Word-level fuzzy alignment for spans detected on a *different* transcription
of the same audio. Two transcribers spell the same speech differently ("Betty"
vs "Bette"), so exact matching misses; a bounded edit-distance match on word
runs bridges the gap. Short tokens are skipped — at that length everything is
within edit distance of everything."""
needle = "".join(c.casefold() for c in phrase if c.isalnum())
if len(needle) < 4:
return []
def distance(a: str, b: str) -> int:
if abs(len(a) - len(b)) > int(len(a) * max_ratio) + 1:
return len(a) + len(b)
previous = list(range(len(b) + 1))
for i, ca in enumerate(a, 1):
current = [i]
for j, cb in enumerate(b, 1):
current.append(min(previous[j] + 1, current[j - 1] + 1,
previous[j - 1] + (ca != cb)))
previous = current
return previous[-1]
words = [word for _, word in index]
budget = max(1, int(len(needle) * max_ratio))
timestamps = []
for start_idx in range(len(words)):
run = ""
for end_idx in range(start_idx, min(start_idx + 12, len(words))):
run += "".join(c.casefold() for c in words[end_idx]["text"] if c.isalnum())
if len(run) > len(needle) + budget:
break
if abs(len(run) - len(needle)) <= budget and distance(needle, run) <= budget:
timestamps.append((words[start_idx]["start"], words[end_idx]["end"]))
break
return timestamps
def _merge(ranges: Sequence[Tuple[float, float]]) -> List[Tuple[float, float]]:
if not ranges:
return []
merged = []
start, end = sorted(ranges)[0]
for next_start, next_end in sorted(ranges)[1:]:
if next_start <= end:
end = max(end, next_end)
else:
merged.append((start, end))
start, end = next_start, next_end
merged.append((start, end))
return merged
def apply_beeps(
audio_bytes: bytes,
ranges: Sequence[Tuple[float, float]],
audio_format: Optional[str] = None,
export_format: str = "mp3",
buffer_ms: int = 100,
beep_freq: int = 1000,
) -> bytes:
"""Replace each (start, end) range with a beep tone, padded by `buffer_ms` on
both sides so a clipped word edge cannot leak a syllable. Needs `pydub`
(and ffmpeg for any format other than WAV)."""
from pydub import AudioSegment
from pydub.generators import Sine
audio = AudioSegment.from_file(io.BytesIO(audio_bytes), format=audio_format)
if not ranges:
return audio_bytes
segments, last_end = [], 0
for start, end in ranges:
# Clamp to `last_end`: two ranges less than 2 * buffer_ms apart have
# overlapping padding, and beeping the overlap twice would lengthen the
# recording and desynchronize every timestamp after it.
start_ms = max(0, int(start * 1000) - buffer_ms, last_end)
end_ms = min(len(audio), int(end * 1000) + buffer_ms)
if end_ms <= start_ms:
continue
if start_ms > last_end:
segments.append(audio[last_end:start_ms])
beep = Sine(beep_freq).to_audio_segment(duration=end_ms - start_ms) - 10
segments.append(beep.set_frame_rate(audio.frame_rate).set_channels(audio.channels))
last_end = end_ms
if last_end < len(audio):
segments.append(audio[last_end:])
buffer = io.BytesIO()
sum(segments[1:], segments[0]).export(buffer, format=export_format)
return buffer.getvalue()
def redact_audio(
audio_bytes: bytes,
fields: Iterable[str],
detector: Detector,
transcriber: Transcriber,
known_values: Optional[Mapping[str, Sequence[str]]] = None,
supplemental_spans: Optional[Sequence[Span]] = None,
audio_format: Optional[str] = None,
export_format: str = "mp3",
) -> bytes:
"""Beep every spoken PII value out of a call recording.
Transcribes with word timestamps using your STT provider, detects PII on that
transcription with your LLM, maps each detected span back to its audio range,
and replaces those ranges with a tone.
`supplemental_spans` are spans detected on a *different* transcription of the
same call — pass the ones `redact_payload` returned. Each transcription catches
entities the other missed, so the union beeps more than either alone.
Fails closed: if detection fails, or a detected span cannot be aligned to the
waveform, RedactionError is raised rather than returning partially redacted
audio. Never fall back to uploading the original on that error.
"""
fields = [f for f in fields if f in ENTITY_TO_TOKEN]
if not fields:
return audio_bytes
_, words = transcriber(audio_bytes)
text, index = _build_word_index(words)
ranges: List[Tuple[float, float]] = []
unaligned: List[str] = []
for span in detect_spans(text, fields, detector, known_values):
found = find_phrase_timestamps(span["original_text"], text, index)
if found:
ranges.extend(found)
else:
unaligned.append(span["original_text"])
if unaligned:
raise RedactionError(
f"{len(unaligned)} detected PII span(s) could not be aligned to audio "
f"timestamps; refusing to return partially redacted audio")
# Supplemental spans come from another transcription, so exact matching often
# misses. Align fuzzily; a span that still will not align is a hard failure
# when it carries digits (phone, account, card material must never ship
# unbeeped) and is skipped otherwise — usually a name spelled differently by
# the two transcribers, where failing every time would make redaction unusable.
unaligned_numeric = []
for span in supplemental_spans or []:
original = span.get("original_text")
if not original:
continue
found = (find_phrase_timestamps(original, text, index)
or _fuzzy_phrase_timestamps(original, index))
if found:
ranges.extend(found)
elif sum(c.isdigit() for c in original) >= 4:
unaligned_numeric.append(original)
if unaligned_numeric:
raise RedactionError(
"Numeric PII found in the transcript could not be aligned to audio "
"timestamps; refusing to return partially redacted audio")
return apply_beeps(audio_bytes, _merge(ranges),
audio_format=audio_format, export_format=export_format)