mkbk
65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from .models import AudioFrame
|
|
|
|
|
|
def build_fake_full_duplex_audio_fixture() -> list[AudioFrame]:
|
|
return [
|
|
AudioFrame(b"\x01\x00" * 160, 16000, 1, 0, 1, {"duration_ms": 20, "assistant_echo": True, "speech": True}),
|
|
AudioFrame(b"\x02\x00" * 800, 16000, 1, 100, 2, {"duration_ms": 100, "speech": True, "partial": "你"}),
|
|
AudioFrame(b"\x03\x00" * 800, 16000, 1, 200, 3, {"duration_ms": 100, "speech": True, "partial": "你好"}),
|
|
AudioFrame(b"\x00\x00" * 800, 16000, 1, 300, 4, {"duration_ms": 100, "speech": False}),
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PerformanceMetric:
|
|
name: str
|
|
duration_ms: int
|
|
payload: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class PerformanceMetricRecorder:
|
|
def __init__(self) -> None:
|
|
self.metrics: list[PerformanceMetric] = []
|
|
|
|
def record(self, name: str, *, started_at_ms: int, finished_at_ms: int, payload: dict[str, Any] | None = None) -> None:
|
|
self.metrics.append(
|
|
PerformanceMetric(
|
|
name=name,
|
|
duration_ms=max(0, finished_at_ms - started_at_ms),
|
|
payload=sanitize_diagnostics(payload or {}),
|
|
)
|
|
)
|
|
|
|
def summary(self) -> dict[str, int]:
|
|
return {metric.name: metric.duration_ms for metric in self.metrics}
|
|
|
|
|
|
def sanitize_diagnostics(data: dict[str, Any]) -> dict[str, Any]:
|
|
sanitized: dict[str, Any] = {}
|
|
for key, value in data.items():
|
|
normalized = key.lower()
|
|
if any(part in normalized for part in {"api_key", "authorization", "raw_audio", "pcm", "secret", "token"}):
|
|
sanitized[key] = "[redacted]"
|
|
elif isinstance(value, dict):
|
|
sanitized[key] = sanitize_diagnostics(value)
|
|
elif isinstance(value, str):
|
|
sanitized[key] = re.sub(r"(?:sk|tp)-[A-Za-z0-9_\-]{16,}", "[redacted]", value)
|
|
else:
|
|
sanitized[key] = value
|
|
return sanitized
|
|
|
|
|
|
def diagnostics_contain_sensitive_data(data: Any) -> bool:
|
|
if isinstance(data, dict):
|
|
return any(diagnostics_contain_sensitive_data(value) for value in data.values())
|
|
if isinstance(data, (list, tuple, set)):
|
|
return any(diagnostics_contain_sensitive_data(value) for value in data)
|
|
text = str(data)
|
|
return bool(re.search(r"(?:sk|tp)-[A-Za-z0-9_\-]{16,}", text))
|