| """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) |
|
|
