"""Offline adapters for the selected native records.

Media bindings are prepared separately with real codecs and audited rights.
An adapter never downloads an image URL or assumes a dataset card licenses it.
All reversals of a pair keep the same source group and media assets.
"""
from __future__ import annotations

import json
import ast
from .curated import digest, validate_calls


def _json(x):
    return json.loads(x) if isinstance(x, str) else x


def normalize(dataset: str, row: dict, provenance: dict, binding: dict | None = None) -> list[dict]:
    """`provenance` supplies id/revision/license/attribution/source_url/split.

    binding: {packet: encoded modality packet, assets: [...rights/checksums...],
              group: source object/video/speaker/uploader ID, caption: aligned caption}
    For media, `group` is mandatory: no clip-level or view-level random split.
    """
    def record(task, ins, outs, group):
        return {**provenance, "dataset": dataset, "task": task, "group": group, "inputs": ins, "outputs": outs,
                **({"assets": binding["assets"], "codec": binding.get("codec", {})} if binding else {})}
    text = lambda s: {"type": "text", "text": str(s).strip()}
    if dataset == "mbpp":
        source_split = provenance.get("source_split")
        expected = {"train": "train", "validation": "val", "test": "test"}
        if source_split not in expected or provenance.get("split") != expected[source_split]:
            raise ValueError("MBPP official train/validation/test provenance must be preserved")
        code = row["code"].strip()
        ast.parse(code)  # syntax validation only; dataset code is never executed by this adapter
        examples = "\n".join(row.get("test_list", []))
        prompt = row["text"].strip() + ("\n\nExamples:\n" + examples if examples else "")
        return [record("coding", [text(prompt)], [text("```python\n" + code + "\n```")], str(row["task_id"]))]
    if dataset in ("builtin", "user"):
        if not row.get("group"): raise ValueError("local records require a stable source group")
        return [record(row["task"], row["inputs"], row["outputs"], row["group"])]
    if dataset == "fineweb_edu":
        return [record("text", [], [text(row["text"])], row.get("url") or digest(row["text"]))]
    if dataset == "smoltalk":
        messages = row["messages"]
        if not messages or messages[-1]["role"] != "assistant":
            raise ValueError("SmolTalk row must end with a complete assistant response")
        context = "\n".join(f"{m['role']}: {m['content']}" for m in messages[:-1])
        if not context.strip(): raise ValueError("conversation has no conditioning")
        return [record("chat_text", [text(context)], [text(messages[-1]["content"])], digest(messages))]
    if dataset == "xlam":
        tools = _json(row["tools"])
        calls = _json(row["answers"])
        validate_calls(tools, calls)
        # Whole tool schemas (including function descriptions/parameters) define a heldout family.
        group = digest(sorted(tools, key=lambda x: x.get("function", x)["name"]))
        return [record("tools", [{"type": "tools", "value": tools}, text(row["query"])], [{"type": "calls", "value": calls}], group)]
    if binding is None or not binding.get("group") or not binding.get("assets"):
        raise ValueError("media adapter requires an encoded packet, group ID and audited asset binding")
    packet, group = binding["packet"], binding["group"]
    if dataset == "openimages_narratives":
        if not str(row["dataset_id"]).startswith("open_images"):
            raise ValueError("only the Open Images subset is selected; COCO/Flickr/ADE rights differ")
        cap, a, b, kind = row["caption"], "t2i", "i2t", "image"
    elif dataset == "finevideo":
        # Required exact-window caption is selected from scene/activity metadata, not a whole-video title.
        if not binding.get("caption_aligned") or not binding.get("window_seconds"):
            raise ValueError("FineVideo needs scene-aligned window caption and timestamps")
        cap, a, b, kind = binding["caption"], "t2v", "v2t", "video"
    elif dataset == "libritts_r":
        if not binding.get("complete_utterance"):
            raise ValueError("speech pair must retain the complete aligned utterance")
        cap, a, b, kind = row["text_normalized"], "t2a", "a2t", "audio"
    elif dataset == "fsd50k":
        # Weak class-label conditioning; never report this as free-form human captioning.
        cap, a, b, kind = "sound of " + ", ".join(row["labels"]), "t2a", "a2t", "audio"
    elif dataset == "objaverse_cap3d":
        cap, a, b, kind = row["caption"], "t2m", "m2t", "mesh"
    else:
        raise ValueError(f"no enabled adapter for {dataset}")
    if packet["type"] != kind or not str(cap).strip(): raise ValueError("packet kind or caption mismatch")
    out = [record(a, [text(cap)], [packet], group), record(b, [packet], [text(cap)], group)]
    if dataset == "finevideo" and len(packet["codes"]) >= 4:
        mid = len(packet["codes"]) // 2
        out.append(record("vcont", [{"type":"video", "codes":packet["codes"][:mid]}],
                          [{"type":"video", "codes":packet["codes"][mid:]}], group))
    return out