beta_keys_schema.py

"""
VTX beta key schema, migration, and validation helpers.
"""
from __future__ import annotations

import secrets
import string
from datetime import date
from typing import Any

VALID_STATUSES = frozenset({"ACTIVE", "REVOKED", "EXPIRED", "BANNED", "INVALID"})
KEY_CHARS = string.ascii_uppercase + string.digits
NEW_KEY_PREFIX = "VTX-"


def today_iso() -> str:
    return date.today().isoformat()


def _random_key_segment(length: int = 6) -> str:
    return "".join(secrets.choice(KEY_CHARS) for _ in range(length))


def generate_unique_key(existing: dict[str, Any]) -> str:
    for _ in range(5000):
        candidate = f"{NEW_KEY_PREFIX}{_random_key_segment(6)}"
        if candidate not in existing:
            return candidate
    raise RuntimeError("Unable to generate a unique beta key.")


def normalize_record(key: str, record: dict[str, Any] | None) -> dict[str, Any]:
    data = dict(record or {})
    key_text = str(data.get("key") or key or "").strip()
    active = bool(data.get("active"))
    status = str(data.get("status") or "").strip().upper()
    if not status:
        status = "ACTIVE" if active else "REVOKED"
    if status not in VALID_STATUSES:
        status = "ACTIVE" if active else "REVOKED"
    if status != "ACTIVE":
        active = False
    created = str(data.get("created") or data.get("created_date") or "").strip()
    last_activation = data.get("last_activation")
    if last_activation is not None and last_activation != "":
        last_activation = str(last_activation).strip()
    else:
        last_activation = None
    return {
        "key": key_text,
        "name": str(data.get("name") or data.get("owner") or "").strip(),
        "notes": str(data.get("notes") or "").strip(),
        "created": created,
        "last_activation": last_activation,
        "active": active,
        "status": status,
    }


def migrate_keys_payload(payload: Any) -> dict[str, dict[str, Any]]:
    if payload is None:
        return {}
    if isinstance(payload, list):
        migrated: dict[str, dict[str, Any]] = {}
        for item in payload:
            if not isinstance(item, dict):
                continue
            key = str(item.get("key") or "").strip()
            if not key:
                continue
            migrated[key] = normalize_record(key, item)
        return migrated
    if not isinstance(payload, dict):
        return {}

    if "keys" in payload and isinstance(payload.get("keys"), list):
        return migrate_keys_payload(payload["keys"])

    migrated: dict[str, dict[str, Any]] = {}
    for raw_key, raw_record in payload.items():
        key = str(raw_key or "").strip()
        if not key:
            continue
        if isinstance(raw_record, str):
            migrated[key] = normalize_record(
                key,
                {"key": key, "active": True, "status": "ACTIVE", "notes": raw_record},
            )
            continue
        if not isinstance(raw_record, dict):
            continue
        migrated[key] = normalize_record(key, raw_record)
    return migrated


def validation_result(record: dict[str, Any] | None) -> dict[str, Any]:
    if not record:
        return {"valid": False, "active": False, "status": "INVALID"}
    normalized = normalize_record(str(record.get("key") or ""), record)
    active = bool(normalized.get("active"))
    status = str(normalized.get("status") or "").strip().upper()
    if status == "ACTIVE" and active:
        return {"valid": True, "active": True, "status": "ACTIVE"}
    if status == "REVOKED" or (not active and status == "ACTIVE"):
        return {"valid": True, "active": False, "status": "REVOKED"}
    if status in {"EXPIRED", "BANNED"}:
        return {"valid": True, "active": False, "status": status}
    return {"valid": True, "active": active, "status": status}


def new_record(name: str, notes: str = "") -> dict[str, Any]:
    return {
        "name": str(name or "").strip(),
        "notes": str(notes or "").strip(),
        "created": today_iso(),
        "last_activation": None,
        "active": True,
        "status": "ACTIVE",
    }