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