"""Accept named score fields only; never execute or extract uploaded files."""
import copy
import math
import re
from urllib.parse import urlsplit


class Invalid(ValueError):
    pass


def text(value, label, maximum=200, required=True):
    if not isinstance(value, str) or len(value.strip()) > maximum:
        raise Invalid(f"{label} must be text, at most {maximum} characters.")
    value = value.strip()
    if (required and not value) or any(ord(c) < 32 and c not in '\n\t' for c in value):
        raise Invalid(f"Please provide a valid {label.lower()}.")
    return value


def url(value, label):
    value = text(value, label, 1000)
    try:
        parsed = urlsplit(value)
        if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
            raise ValueError()
    except ValueError:
        raise Invalid(f"{label} must be an https:// link without credentials.")
    return value


def score(value, label):
    if value is None:
        return None
    if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 100:
        raise Invalid(f"{label} must be a number between 0 and 100, or null.")
    if abs(value * 10 - round(value * 10)) > 1e-7:
        raise Invalid(f"{label} must use at most one decimal place, as in the published tables.")
    return float(value)


def simple_scores(raw, columns, label):
    if not isinstance(raw, dict):
        raise Invalid(f"{label} must be an object with named scores.")
    lookup = {c["label"]: c["key"] for c in columns}
    lookup.update({c["key"]: c["key"] for c in columns})
    result = {c["key"]: None for c in columns}
    seen = set()
    for key, value in raw.items():
        if key not in lookup or lookup[key] in seen:
            raise Invalid(f"Unknown or duplicate {label} field: {key}")
        seen.add(lookup[key])
        result[lookup[key]] = score(value, key)
    return result


def nested_scores(raw, columns, label):
    if not isinstance(raw, dict):
        raise Invalid(f"{label} must group full score names by reasoning dimension.")
    result = {c["key"]: None for c in columns}
    seen = set()
    for dimension, values in raw.items():
        subset = [c for c in columns if c["dimension"].casefold() == dimension.casefold()]
        if not subset or dimension.casefold() in seen:
            raise Invalid(f"Unknown or duplicate {label} dimension: {dimension}")
        seen.add(dimension.casefold())
        result.update(simple_scores(values, subset, label))
    return result


def validate_submission(raw, data):
    if not isinstance(raw, dict):
        raise Invalid("The submission must be a JSON object.")
    allowed = {"model_name", "project_url", "evidence_url", "team", "email", "group", "benchmark",
               "language", "multi_image", "revision_of", "notes", "full_dataset", "results"}
    if set(raw) - allowed:
        raise Invalid("Unknown submission fields: " + ", ".join(sorted(set(raw) - allowed)))
    benchmark = next((b for b in data["benchmarks"] if b["id"] == raw.get("benchmark")), None)
    if not benchmark or raw.get("language") not in benchmark["results_by_language"]:
        raise Invalid("Select a supported benchmark and instruction language.")
    if raw.get("full_dataset") is not True:
        raise Invalid("Confirm that results use the full benchmark, not the demonstration data.")
    if type(raw.get("multi_image")) is not bool:
        raise Invalid("Select whether the model supports multiple input images.")
    if benchmark["id"] == "risebench" and raw["multi_image"]:
        raise Invalid("RISEBench evaluates single-image inputs only.")
    if raw.get("group") not in {"closed", "open", "agentic"}:
        raise Invalid("Select a model category.")
    result = {key: raw[key] for key in ("benchmark", "language", "group", "multi_image")}
    for key, label, size, required in [("model_name", "Model name", 160, True), ("team", "Team", 120, False),
                                       ("notes", "Notes", 3000, False), ("revision_of", "Existing model identifier", 160, False)]:
        result[key] = text(raw.get(key, ""), label, size, required)
    result["project_url"] = url(raw.get("project_url"), "Project link")
    result["evidence_url"] = url(raw.get("evidence_url"), "Evaluation artifacts link")
    result["email"] = text(raw.get("email"), "Contact email", 254)
    if not re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", result["email"]):
        raise Invalid("Provide a valid contact email.")
    uploaded = raw.get("results")
    if not isinstance(uploaded, dict) or set(uploaded) - {"accuracy", "metrics", "subcategories", "task_types"}:
        raise Invalid("Upload a result JSON matching the downloadable template.")
    result["accuracy"] = simple_scores(uploaded.get("accuracy"), benchmark["accuracy_columns"], "accuracy")
    required_keys = [d["key"] for d in benchmark["dimensions"]]
    if benchmark["id"] == "risebench-plusplus":
        required_keys += ["single"] + (["multi", "total"] if result["multi_image"] else [])
        if not result["multi_image"] and any(result["accuracy"][key] is not None for key in ("multi", "total")):
            raise Invalid("Single-image-only models must leave Total and Multi null.")
    else:
        required_keys += ["total"]
    if any(result["accuracy"][key] is None for key in required_keys):
        raise Invalid("Fill in every reasoning-dimension accuracy and the applicable summary scores.")
    result["metrics"] = simple_scores(uploaded.get("metrics", {}), benchmark["metrics"], "evaluation score")
    for field, columns in (("subcategories", "subcategory_columns"), ("task_types", "task_type_columns")):
        if columns not in benchmark:
            if uploaded.get(field):
                raise Invalid("Fine-grained submissions are supported only for RISEBench++.")
            continue
        result[field] = nested_scores(uploaded.get(field, {}), benchmark[columns], field)
    if benchmark["id"] == "risebench-plusplus" and not result["multi_image"]:
        for column in benchmark["task_type_columns"]:
            if column["case_count"] == column["multi_image_case_count"] and result["task_types"][column["key"]] is not None:
                raise Invalid(f"{column['label']} is a multi-image-only task; leave it null for this model.")
    result["full_dataset"] = True
    return result


def template(benchmark):
    result = {"accuracy": {c["label"]: None for c in benchmark["accuracy_columns"]},
              "metrics": {c["label"]: None for c in benchmark["metrics"]}}
    for field, key in (("subcategories", "subcategory_columns"), ("task_types", "task_type_columns")):
        if key in benchmark:
            result[field] = {}
            for c in benchmark[key]:
                result[field].setdefault(c["dimension"].title(), {})[c["label"]] = None
    return result


def public_leaderboard(base, state):
    data = copy.deepcopy(base)
    for b in data["benchmarks"]:
        for language, entry in b["results_by_language"].items():
            models = {m["id"]: m for m in entry["models"]}
            approved = sorted((r for r in state["submissions"].values() if r["status"] == "approved"
                               and r["payload"]["benchmark"] == b["id"] and r["payload"]["language"] == language),
                              key=lambda r: r["reviewed_at"])
            for record in approved:
                p = record["payload"]
                model_id = p["revision_of"] or "community-" + record["id"]
                previous = models.get(model_id)
                model = {"id": model_id, "name": p["model_name"], "group": p["group"], "multi_image": p["multi_image"],
                         "source_order": previous["source_order"] if previous else len(models) + 1}
                # An explicit public allowlist: no email, notes, receipt tokens or review history.
                for field in ("accuracy", "metrics", "subcategories", "task_types"):
                    if field in p:
                        model[field] = copy.deepcopy(p[field])
                models[model_id] = model
            entry["models"] = list(models.values())
            entry["model_count"] = len(models)
            for group in entry["groups"]:
                group["count"] = sum(m["group"] == group["id"] for m in models.values())
    return data