| """ |
| 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) |
