votes.py

"""Vote payload validation and commit logic for the climbing-holds dataset."""

from __future__ import annotations

import hashlib
import os
import re
from collections import Counter
from typing import Any

from huggingface_hub import HfApi

import config
import hf_repo

RATING_MIN = 1
RATING_MAX = 5
ISO8601_PATTERN = re.compile(
    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:?\d{2})?$"
)


class DuplicateVoteError(Exception):
    """Raised when the same voter has already voted for a given hold."""


def infer_dominant_values(votes: list[dict[str, Any]]) -> tuple[str | None, str | None]:
    """Return the most-voted manufacturer and model from a list of vote entries.

    Only non-empty, non-whitespace values are counted. Ties are broken by
    picking the lexicographically smallest value so the result is deterministic.
    Returns (None, None) when no valid votes exist.
    """
    manufacturers: list[str] = []
    models: list[str] = []
    for vote in votes:
        if not isinstance(vote, dict):
            continue
        m = vote.get("hold_manufacturer")
        if isinstance(m, str) and m.strip():
            manufacturers.append(m.strip())
        mo = vote.get("hold_model")
        if isinstance(mo, str) and mo.strip():
            models.append(mo.strip())

    def _dominant(values: list[str]) -> str | None:
        if not values:
            return None
        counts = Counter(values)
        max_count = max(counts.values())
        candidates = sorted(k for k, v in counts.items() if v == max_count)
        return candidates[0]

    return _dominant(manufacturers), _dominant(models)


def _validate_rating(value: Any) -> int:
    if not isinstance(value, (int, float)):
        raise ValueError("hold_3d_file_rating must be a number")
    r = int(value) if isinstance(value, float) else value
    if r != value or r < RATING_MIN or r > RATING_MAX:
        raise ValueError(f"hold_3d_file_rating must be an integer between {RATING_MIN} and {RATING_MAX}")
    return r


def _validate_vote_datetime(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("vote_datetime must be a non-empty string")
    s = value.strip()
    if not ISO8601_PATTERN.match(s):
        raise ValueError("vote_datetime must be ISO 8601 format (e.g. 2025-03-12T14:30:00.000Z)")
    return s


def validate_vote_payload(
    body: dict[str, Any],
    hf_token: str | None = None,
) -> tuple[dict[str, Any], str | None]:
    """
    Validate vote payload and build the stored vote entry (no token).
    Returns (vote_entry, hf_token_or_none). Raises ValueError on validation error.
    The user's hf_token is passed via the Authorization header and must never be persisted.
    """
    required = ("hold_id", "hold_manufacturer", "hold_model", "hold_3d_file_rating", "vote_datetime", "anonymous")
    for key in required:
        if key not in body:
            raise ValueError(f"Missing required field: {key}")

    hold_id = body["hold_id"]
    if not isinstance(hold_id, str) or not hold_id.strip():
        raise ValueError("hold_id must be a non-empty string")
    hold_id = hold_id.strip()

    hold_manufacturer = body["hold_manufacturer"]
    if not isinstance(hold_manufacturer, str):
        raise ValueError("hold_manufacturer must be a string")
    hold_manufacturer = hold_manufacturer.strip()

    hold_model = body["hold_model"]
    if not isinstance(hold_model, str):
        raise ValueError("hold_model must be a string")
    hold_model = hold_model.strip()

    rating = _validate_rating(body["hold_3d_file_rating"])
    vote_datetime = _validate_vote_datetime(body["vote_datetime"])
    anonymous = body["anonymous"]
    if not isinstance(anonymous, bool):
        raise ValueError("anonymous must be a boolean")

    resolved_token: str | None = None
    if not anonymous:
        if not hf_token or not hf_token.strip():
            raise ValueError("Authorization header with Bearer token is required when anonymous is false")
        resolved_token = hf_token.strip()

    entry = {
        "hold_id": hold_id,
        "hold_manufacturer": hold_manufacturer,
        "hold_model": hold_model,
        "hold_3d_file_rating": rating,
        "vote_datetime": vote_datetime,
        "anonymous": anonymous,
    }
    return entry, resolved_token


def compute_voter_fingerprint(
    client_ip: str,
    anonymous: bool,
    user_token: str | None,
) -> str:
    """Build a unique, stable fingerprint for a voter.

    - Non-anonymous with a valid HF token: resolve the HF username via whoami.
    - Anonymous or token resolution fails: SHA-256 of the client IP.
    """
    if not anonymous and user_token:
        try:
            info = HfApi().whoami(token=user_token)
            username = info.get("name") or info.get("user")
            if username:
                return f"hf:{username}"
        except Exception:
            pass
    return f"ip:{hashlib.sha256(client_ip.encode()).hexdigest()}"


def _resolve_commit_token(anonymous: bool, user_token: str | None) -> str:
    hf_token = os.environ.get("HF_TOKEN")
    if anonymous or not user_token:
        if not hf_token:
            raise RuntimeError("HF_TOKEN is not set (required for anonymous votes or when user token is missing)")
        return hf_token
    return user_token


def _has_existing_vote(hold_votes: list[dict[str, Any]], fingerprint: str) -> bool:
    return any(v.get("voter_fingerprint") == fingerprint for v in hold_votes)


def _build_metadata_update(
    repo_id: str,
    token: str,
    revision: str | None,
    hold_id: str,
    hold_votes: list[dict[str, Any]],
) -> tuple[str, dict[str, Any]] | None:
    """Load metadata.json for the hold and apply dominant manufacturer/model from votes.

    Returns (metadata_path, updated_metadata) if any field changed, else None.
    """
    metadata_path = f"{hold_id}/{config.METADATA_FILENAME}"
    metadata = hf_repo.load_json_file_optional(
        repo_id, metadata_path, token, revision, default=None
    )
    if not isinstance(metadata, dict):
        config.logger.warning(
            "Could not load metadata for hold '%s'; skipping metadata update from votes.", hold_id
        )
        return None

    dominant_manufacturer, dominant_model = infer_dominant_values(hold_votes)
    updated = dict(metadata)
    changed = False

    if dominant_manufacturer and updated.get("manufacturer") != dominant_manufacturer:
        updated["manufacturer"] = dominant_manufacturer
        changed = True

    if dominant_model and updated.get("model") != dominant_model:
        updated["model"] = dominant_model
        changed = True

    if not changed:
        return None

    config.logger.info(
        "Updating hold '%s' metadata from votes: manufacturer=%r, model=%r",
        hold_id,
        dominant_manufacturer,
        dominant_model,
    )
    return metadata_path, updated


def process_vote(
    api: HfApi,
    repo_id: str,
    revision: str | None,
    vote_entry: dict[str, Any],
    user_token: str | None,
    client_ip: str,
) -> dict[str, Any]:
    """
    Load hold votes, check for duplicates, append entry, commit votes.json and
    metadata.json (with dominant manufacturer/model from all votes) together.
    On commit failure with user token, retries with HF_TOKEN.
    Raises DuplicateVoteError if the same voter already voted for this hold.
    """
    anonymous = vote_entry.get("anonymous", True)
    token = _resolve_commit_token(anonymous, user_token)
    hold_id = vote_entry["hold_id"]
    hold_votes_path = f"{hold_id}/{config.VOTES_FILENAME}"

    fingerprint = compute_voter_fingerprint(client_ip, anonymous, user_token)

    hold_votes: list[Any] = hf_repo.load_json_file_optional(
        repo_id, hold_votes_path, token, revision, default=[]
    )
    if not isinstance(hold_votes, list):
        hold_votes = []

    if _has_existing_vote(hold_votes, fingerprint):
        raise DuplicateVoteError("You have already voted for this hold")

    vote_entry["voter_fingerprint"] = fingerprint
    hold_votes.append(vote_entry)
    hold_votes_map = {hold_votes_path: hold_votes}

    metadata_update = _build_metadata_update(repo_id, token, revision, hold_id, hold_votes)

    def _do_commit(commit_token: str) -> None:
        hf_repo.commit_vote_updates(
            api,
            repo_id=repo_id,
            token=commit_token,
            revision=revision,
            hold_votes=hold_votes_map,
            metadata_update=metadata_update,
        )

    try:
        _do_commit(token)
        return {"status": "success", "message": "Vote recorded"}
    except Exception as exc:
        if not anonymous and user_token and token == user_token:
            hf_token = os.environ.get("HF_TOKEN")
            if hf_token and hf_token != user_token:
                config.logger.warning("Commit with user token failed, retrying with HF_TOKEN: %s", exc)
                _do_commit(hf_token)
                return {"status": "success", "message": "Vote recorded"}
        raise