File size: 7,745 Bytes
35d483e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 | """Leakage-aware grouping using exact audio and metadata linkages."""
from __future__ import annotations
from dataclasses import dataclass
import hashlib
import json
import re
import unicodedata
from typing import Any, Iterable, Mapping, MutableMapping, Sequence
@dataclass(frozen=True)
class GroupingConfig:
"""Controls which stable identifiers link examples into components."""
include_record_id: bool = True
include_audio_path: bool = True
include_text: bool = True
minimum_text_characters: int = 12
minimum_text_tokens: int = 3
_FIELD_FAMILIES: dict[str, tuple[str, ...]] = {
"conversation": ("conversation_id", "conversation", "dialogue_id", "dialog_id", "session_id", "call_id"),
"speaker": ("speaker_id", "speaker", "user_id", "actor_id"),
"voice": ("voice_id", "voice", "tts_voice", "speaker_voice"),
"recording": ("recording_id", "source_recording_id", "audio_id", "clip_id"),
"prompt": ("prompt_id", "source_prompt_id", "template_id", "script_id", "parent_id"),
}
def _canonical_value(value: Any) -> str | None:
if value is None:
return None
if isinstance(value, str):
normalized = unicodedata.normalize("NFKC", value).strip().casefold()
return normalized or None
if isinstance(value, (bool, int, float)):
return json.dumps(value, sort_keys=True, allow_nan=False)
try:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
except (TypeError, ValueError):
normalized = str(value).strip().casefold()
return normalized or None
def _private_key(family: str, namespace: str, value: Any) -> str | None:
canonical = _canonical_value(value)
if canonical is None:
return None
digest = hashlib.sha256(f"{namespace}\0{canonical}".encode("utf-8")).hexdigest()[:32]
return f"{family}:{digest}"
def _lookup(record: Mapping[str, Any], field: str) -> Any:
if field in record:
return record[field]
metadata = record.get("metadata")
if isinstance(metadata, Mapping) and field in metadata:
return metadata[field]
return None
def _normalize_text(value: str) -> str:
value = unicodedata.normalize("NFKC", value).casefold()
return " ".join(re.findall(r"[\w']+", value, flags=re.UNICODE))
def derive_group_keys(
record: Mapping[str, Any],
*,
audio_sha256: str | None = None,
config: GroupingConfig | None = None,
) -> list[str]:
"""Derive privacy-preserving exact-audio and metadata linkage keys.
Every shared key induces an edge. :func:`attach_group_ids` computes the
transitive connected components, so an audio duplicate linked to a shared
speaker cannot be split indirectly.
"""
config = config or GroupingConfig()
keys: set[str] = set()
audio_hash = audio_sha256 or _lookup(record, "audio_sha256")
if isinstance(audio_hash, str) and audio_hash:
keys.add(f"audio:{audio_hash.lower()}")
dataset = _canonical_value(
_lookup(record, "dataset") or _lookup(record, "source_dataset") or _lookup(record, "data_source")
)
namespace = dataset or "unknown-dataset"
for family, fields in _FIELD_FAMILIES.items():
for field in fields:
value = _lookup(record, field)
key = _private_key(family, namespace, value)
if key:
keys.add(key)
if config.include_record_id:
explicit_id = _lookup(record, "id")
if explicit_id is None:
# Audit manifests canonicalize the upstream id as record_id.
explicit_id = _lookup(record, "record_id")
key = _private_key("record", namespace, explicit_id)
if key:
keys.add(key)
if config.include_audio_path and not audio_hash:
audio_path = _lookup(record, "audio_path")
if audio_path is None:
audio = _lookup(record, "audio")
if isinstance(audio, Mapping):
audio_path = audio.get("path")
if audio_path:
# Basename is portable across machines and catches copied manifests.
basename = str(audio_path).replace("\\", "/").rsplit("/", 1)[-1]
key = _private_key("audio_path", namespace, basename)
if key:
keys.add(key)
if config.include_text:
text = _lookup(record, "spoken_text") or _lookup(record, "transcript") or _lookup(record, "text")
if isinstance(text, str):
normalized_text = _normalize_text(text)
if (
len(normalized_text) >= config.minimum_text_characters
and len(normalized_text.split()) >= config.minimum_text_tokens
):
key = _private_key("text", "global", normalized_text)
if key:
keys.add(key)
return sorted(keys)
class _UnionFind:
def __init__(self, size: int) -> None:
self.parent = list(range(size))
self.rank = [0] * size
def find(self, item: int) -> int:
while self.parent[item] != item:
self.parent[item] = self.parent[self.parent[item]]
item = self.parent[item]
return item
def union(self, left: int, right: int) -> None:
left_root = self.find(left)
right_root = self.find(right)
if left_root == right_root:
return
if self.rank[left_root] < self.rank[right_root]:
left_root, right_root = right_root, left_root
self.parent[right_root] = left_root
if self.rank[left_root] == self.rank[right_root]:
self.rank[left_root] += 1
def attach_group_ids(
rows: Sequence[MutableMapping[str, Any]],
*,
group_keys_field: str = "group_keys",
) -> Sequence[MutableMapping[str, Any]]:
"""Attach deterministic component ``group_id`` values to manifest rows.
Rows are modified in place to avoid duplicating a full 270k-row manifest.
The resulting IDs do not depend on row order.
"""
union_find = _UnionFind(len(rows))
first_row_for_key: dict[str, int] = {}
normalized_keys: list[list[str]] = []
for index, row in enumerate(rows):
raw_keys = row.get(group_keys_field) or []
if isinstance(raw_keys, str):
raw_keys = [raw_keys]
keys = sorted({str(key) for key in raw_keys if key})
if not keys:
fallback = hashlib.sha256(
f"{row.get('record_id', '')}\0{row.get('source_file', '')}\0{row.get('source_row', index)}".encode(
"utf-8"
)
).hexdigest()
keys = [f"row:{fallback}"]
row[group_keys_field] = keys
normalized_keys.append(keys)
for key in keys:
previous = first_row_for_key.setdefault(key, index)
union_find.union(index, previous)
component_keys: dict[int, set[str]] = {}
for index, keys in enumerate(normalized_keys):
root = union_find.find(index)
component_keys.setdefault(root, set()).update(keys)
component_ids = {
root: "grp_" + hashlib.sha256("\n".join(sorted(keys)).encode("utf-8")).hexdigest()[:24]
for root, keys in component_keys.items()
}
for index, row in enumerate(rows):
row[group_keys_field] = normalized_keys[index]
row["group_id"] = component_ids[union_find.find(index)]
return rows
def group_sizes(rows: Iterable[Mapping[str, Any]]) -> dict[str, int]:
"""Return the number of rows in every leakage component."""
counts: dict[str, int] = {}
for row in rows:
group_id = str(row.get("group_id") or "")
if group_id:
counts[group_id] = counts.get(group_id, 0) + 1
return counts
|