File size: 3,391 Bytes
40bd7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """
VTX License Server — FastAPI application for Hugging Face Spaces.
POST /api/validate
GET /api/health
Keys are loaded from the private ManChildTechnologies/VTX-BetaKeys dataset.
"""
from __future__ import annotations
import os
import threading
from typing import Any
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
import uvicorn
from beta_keys_schema import today_iso, validation_result
from beta_keys_store import keys_status, load_keys, update_key_record
app = FastAPI(
title="VTX License Server",
description="Beta key validation API for VTX Studio",
version="1.1.0",
)
_refresh_timer: threading.Timer | None = None
class ValidateRequest(BaseModel):
beta_key: str = Field(
...,
description="VTX Studio beta license key",
examples=["VTX-4D8A9B", "VTX-BETA-XXXX-XXXX"],
)
def validate_beta_key(beta_key: str, *, track_activation: bool = True) -> dict[str, Any]:
key = str(beta_key or "").strip()
if not key:
return {"valid": False, "active": False, "status": "INVALID"}
record = load_keys().get(key)
result = validation_result(record)
if not result.get("valid") or not result.get("active"):
return result
if track_activation:
_schedule_activation_update(key)
return result
def _schedule_activation_update(beta_key: str) -> None:
def _worker() -> None:
try:
update_key_record(
beta_key,
lambda record: {**record, "last_activation": today_iso()},
commit_message=f"Track activation for {beta_key}",
)
except Exception:
pass
thread = threading.Thread(target=_worker, name=f"vtx-activation-{beta_key}", daemon=True)
thread.start()
def _background_refresh() -> None:
global _refresh_timer
try:
load_keys(force=True)
except Exception:
pass
_refresh_timer = threading.Timer(float(os.environ.get("VTX_KEYS_REFRESH_SECONDS", "30") or 30), _background_refresh)
_refresh_timer.daemon = True
_refresh_timer.start()
@app.on_event("startup")
async def startup_refresh() -> None:
try:
load_keys(force=True)
except Exception:
pass
_background_refresh()
@app.on_event("shutdown")
async def shutdown_refresh() -> None:
global _refresh_timer
if _refresh_timer is not None:
_refresh_timer.cancel()
_refresh_timer = None
@app.post("/api/validate")
async def api_validate(payload: ValidateRequest) -> dict[str, Any]:
return validate_beta_key(payload.beta_key)
@app.get("/api/health")
async def api_health() -> JSONResponse:
try:
keys = load_keys()
status = keys_status()
status["ok"] = True
status["service"] = "VTX-License-Server"
status["keys_loaded"] = len(keys)
return JSONResponse(content=status)
except Exception as exc:
status = keys_status()
status["ok"] = bool(status.get("keys_loaded"))
status["service"] = "VTX-License-Server"
status["error"] = str(exc)
code = 200 if status["ok"] else 500
return JSONResponse(content=status, status_code=code)
if __name__ == "__main__":
port = int(os.environ.get("PORT", os.environ.get("HF_SPACE_PORT", "7860")))
uvicorn.run(app, host="0.0.0.0", port=port) |