File size: 8,180 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 216 | """Dataset auditing and portable manifest construction."""
from __future__ import annotations
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Mapping
from .audio import AudioPayloadError, inspect_audio
from .grouping import GroupingConfig, attach_group_ids, derive_group_keys, group_sizes
from .manifest import MANIFEST_SCHEMA_VERSION
from .schema import ValidationIssue, normalize_record_with_issues
def _category(value: Any) -> str:
if value is None:
return "<null>"
if isinstance(value, bool):
return "true" if value else "false"
text = str(value).strip()
return text or "<empty>"
def _source_base_dir(source_file: str | None) -> Path | None:
if source_file is None or source_file.startswith("hf://"):
return None
path = Path(source_file)
return path.parent if path.suffix else path
def build_manifest_row(
raw_record: Mapping[str, Any],
*,
grouping_config: GroupingConfig | None = None,
) -> dict[str, Any]:
"""Validate, fingerprint and reduce a raw row to a lightweight manifest row."""
result = normalize_record_with_issues(raw_record)
record = result.record
issues = list(result.issues)
audio_fields: dict[str, Any] = {
"audio_sha256": None,
"audio_num_bytes": None,
"audio_format": None,
"sample_rate": None,
"num_channels": None,
"num_frames": None,
"bits_per_sample": None,
"duration_seconds": None,
"audio_path": None,
}
if record.audio is not None:
try:
audio_fields.update(
inspect_audio(record.audio, base_dir=_source_base_dir(record.source_file)).to_dict()
)
except AudioPayloadError as exc:
issues.append(ValidationIssue("error", "invalid_audio", str(exc), "audio"))
group_source = dict(raw_record)
group_source.setdefault("record_id", record.record_id)
group_source.setdefault("dataset", record.dataset)
group_source.setdefault("spoken_text", record.spoken_text)
group_source["audio_path"] = audio_fields["audio_path"]
group_keys = derive_group_keys(
group_source,
audio_sha256=audio_fields["audio_sha256"],
config=grouping_config,
)
return {
"record_id": record.record_id,
"source_file": record.source_file,
"source_row": record.source_row,
"language": record.language,
"endpoint": record.endpoint,
"midfiller": record.midfiller,
"endfiller": record.endfiller,
"synthetic": record.synthetic,
"dataset": record.dataset,
"spoken_text": record.spoken_text,
**audio_fields,
"group_keys": group_keys,
"group_id": None,
"validation_errors": [item.to_dict() for item in issues if item.severity == "error"],
"validation_warnings": [item.to_dict() for item in issues if item.severity == "warning"],
}
def _append_duplicate_conflict_warnings(rows: list[dict[str, Any]]) -> None:
labels_by_hash: dict[str, set[bool]] = defaultdict(set)
indices_by_hash: dict[str, list[int]] = defaultdict(list)
for index, row in enumerate(rows):
digest = row.get("audio_sha256")
endpoint = row.get("endpoint")
if isinstance(digest, str):
indices_by_hash[digest].append(index)
if isinstance(endpoint, bool):
labels_by_hash[digest].add(endpoint)
for digest, labels in labels_by_hash.items():
if len(labels) <= 1:
continue
issue = ValidationIssue(
"warning",
"conflicting_duplicate_label",
f"exact audio hash {digest[:12]}… has conflicting endpoint labels",
"endpoint",
).to_dict()
for index in indices_by_hash[digest]:
rows[index]["validation_warnings"].append(issue)
def build_audit_report(rows: Iterable[Mapping[str, Any]]) -> dict[str, Any]:
"""Aggregate manifest rows into a JSON-serializable data quality report."""
materialized = list(rows)
distributions: dict[str, Counter[str]] = {
field: Counter(_category(row.get(field)) for row in materialized)
for field in ("endpoint", "language", "dataset", "synthetic", "midfiller", "endfiller", "audio_format")
}
errors = Counter(
str(issue.get("code", "unknown"))
for row in materialized
for issue in row.get("validation_errors", [])
if isinstance(issue, Mapping)
)
warnings = Counter(
str(issue.get("code", "unknown"))
for row in materialized
for issue in row.get("validation_warnings", [])
if isinstance(issue, Mapping)
)
invalid_rows = sum(bool(row.get("validation_errors")) for row in materialized)
audio_hash_counts = Counter(
str(row["audio_sha256"]) for row in materialized if row.get("audio_sha256")
)
labels_by_hash: dict[str, set[Any]] = defaultdict(set)
for row in materialized:
if row.get("audio_sha256"):
labels_by_hash[str(row["audio_sha256"])].add(row.get("endpoint"))
group_counts = group_sizes(materialized)
duration_values = [
float(row["duration_seconds"])
for row in materialized
if isinstance(row.get("duration_seconds"), (int, float))
]
total_bytes = sum(
int(row["audio_num_bytes"])
for row in materialized
if isinstance(row.get("audio_num_bytes"), int)
)
duplicate_record_ids = Counter(
str(row["record_id"]) for row in materialized if row.get("record_id")
)
return {
"manifest_schema_version": MANIFEST_SCHEMA_VERSION,
"generated_at": datetime.now(timezone.utc).isoformat(),
"records": {
"total": len(materialized),
"valid": len(materialized) - invalid_rows,
"invalid": invalid_rows,
},
"distributions": {
field: dict(sorted(counts.items())) for field, counts in distributions.items()
},
"audio": {
"total_encoded_bytes": total_bytes,
"duration_observed_records": len(duration_values),
"total_duration_seconds": sum(duration_values),
"minimum_duration_seconds": min(duration_values) if duration_values else None,
"maximum_duration_seconds": max(duration_values) if duration_values else None,
},
"duplicates": {
"unique_audio_hashes": len(audio_hash_counts),
"duplicate_audio_hashes": sum(count > 1 for count in audio_hash_counts.values()),
"duplicate_audio_rows_beyond_first": sum(count - 1 for count in audio_hash_counts.values() if count > 1),
"conflicting_label_audio_hashes": sum(
len({value for value in labels if isinstance(value, bool)}) > 1
for labels in labels_by_hash.values()
),
"duplicate_record_ids": sum(count > 1 for count in duplicate_record_ids.values()),
},
"groups": {
"total": len(group_counts),
"multirow": sum(count > 1 for count in group_counts.values()),
"largest_rows": max(group_counts.values(), default=0),
},
"validation": {
"errors": dict(sorted(errors.items())),
"warnings": dict(sorted(warnings.items())),
},
}
def audit_records(
records: Iterable[Mapping[str, Any]],
*,
grouping_config: GroupingConfig | None = None,
limit: int | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
"""Build a leakage-grouped manifest and aggregate report from raw rows."""
if limit is not None and limit < 0:
raise ValueError("limit cannot be negative")
rows: list[dict[str, Any]] = []
for index, raw_record in enumerate(records):
if limit is not None and index >= limit:
break
rows.append(build_manifest_row(raw_record, grouping_config=grouping_config))
attach_group_ids(rows)
_append_duplicate_conflict_warnings(rows)
return rows, build_audit_report(rows)
|