""" 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", }