diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,63 +1,89 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -ZERO COST PROJECT v3.0.0 — RELAY SERVER (CPU) -Arsitektur baru: CPU relay + ZeroGPU workers - -Server ini handle SEMUA management: -- Token pool (85 akun HF untuk auth ke worker) -- User token economy (quota harian, ban, admin) -- Safety filter (hard-block + soft-score) -- Spy log dengan TTL -- Broadcast system -- Admin panel endpoints -- Routing ke Backend 1 (normal) atau Backend 2 (uncensored) -- Blur processing untuk uncensored -- Passkey unlock -- Legacy frontend guard - -Backend ZeroGPU 1 & 2 sekarang PURE PEKERJA — hanya expose -endpoint worker_generate(secret, prompt, neg, w, h, seed, batch). +ZERO COST PROJECT v3.0.0 — RELAY SERVER (Docker + FastAPI) +============================================================ +CPU relay yang handle SEMUA management: + - Token pool (85 akun HF untuk ZeroGPU auth) + - User token economy (quota harian, ban, admin) + - Safety filter (hard-block + soft-score) + - Spy log dengan TTL + - Broadcast system + - Admin panel endpoints + - Routing ke Backend 1 (normal) / Backend 2 (uncensored) + - Blur processing + passkey unlock + - Legacy frontend guard middleware + +Backend ZeroGPU 1 & 2 = PURE PEKERJA (image generation only). """ -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [0] EARLY SETUP ║ -# ╚══════════════════════════════════════════════════════════════╝ -import os, sys, threading, re, time, json, io, base64, gc, random -import traceback, math +import os +import io +import re +import json +import base64 +import time +import random +import threading from datetime import datetime, timedelta, timezone +from typing import Optional, Dict, List, Any +from fastapi import FastAPI, Request, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +import httpx from PIL import Image, ImageFilter -import requests as http_requests +from pydantic import BaseModel -# ── [0a] UTC HELPER (no deprecation warning) ── -def _utcnow(): - return datetime.now(timezone.utc).replace(tzinfo=None) +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [0] CONFIG — HARDCODED ║ +# ╚══════════════════════════════════════════════════════════════╝ +VERSION = "3.0.0-relay-docker" +# Backend ZeroGPU workers (pure image generators) +BACKEND1_URL = "https://bl4ckspaces-zimageturbo-backend.hf.space" # normal +BACKEND2_URL = "https://bl4ckspaces-mps.hf.space" # uncensored MPS -def _utc_str(): - return _utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") +# Shared secret antara relay ↔ worker (untuk verify request internal) +WORKER_SECRET = "zc_worker_k7x2m9q4_bl4ck" +# Passkeys +UNCENSORED_PASSKEY = "Rusdi6967" +ADMIN_PASSKEY = "Ambalabu69" -# ── [0b] LOG THROTTLE ── -_log_seen = {} -_log_throttle_lock = threading.Lock() +# Generation params (dikirim ke worker) +LOCKED_STEPS = 8 +LOCKED_CFG_SCALE = 0.0 +LOCKED_MAX_SEQUENCE_LENGTH = 256 +MIN_BATCH = 1 +MAX_BATCH = 2 +MAX_PIXELS = 1024 * 1024 +MIN_PIXELS = 512 * 512 +MIN_SIDE = 512 +MAX_SIDE = 2048 +RESOLUTION_STEP = 8 +MAX_ASPECT = 3.0 +# Token economy +TOKEN_DAILY_QUOTA = 15 +COST_NORMAL = 1 +COST_UNCENSORED = 3 +BLUR_RADIUS = 30 -def _log_throttled(key, msg, interval=300): - now = time.time() - with _log_throttle_lock: - last = _log_seen.get(key, 0) - if now - last < interval: - return False - _log_seen[key] = now - print(msg) - return True +# Spy log +SPY_MAX_ENTRIES = 40 +SPY_FULL_TTL_SECONDS = 900 # 15 menit +# Worker timeouts (second) +WORKER_CONNECT_TIMEOUT = 20 +WORKER_READ_TIMEOUT = 600 # 10 menit (generate butuh waktu) +WORKER_POOL_LIMIT = 50 # max koneksi ke worker paralel -# ── [0c] HF TOKEN POOL (85 akun) ── -# Digunakan untuk Authorization header saat relay ke backend ZeroGPU + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [0a] HF TOKEN POOL (85 akun) ║ +# ╚══════════════════════════════════════════════════════════════╝ _RAW_PAYLOAD = ( "PiRCDDtPcPFMLWkTkVaZmzoleHOunXnLIA" "-BHvZXGICstaktSwycmwNmzHGrTNmKxnlRZ" @@ -146,21 +172,56 @@ _RAW_PAYLOAD = ( "-JybhNvxSGmhEMJTxnLktzLdempTyAEXYuu" ) POOL_85 = ["hf_" + seg.strip() for seg in _RAW_PAYLOAD.split("-")] -assert len(POOL_85) == 85, f"Expected 85 tokens, got {len(POOL_85)}" +assert len(POOL_85) == 85, f"Token pool should have 85, got {len(POOL_85)}" + + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [1] UTILITY FUNCTIONS ║ +# ╚══════════════════════════════════════════════════════════════╝ +def _utcnow(): + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _utc_str(): + return _utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") + + +_log_seen = {} +_log_throttle_lock = threading.Lock() + + +def _log_throttled(key: str, msg: str, interval: int = 300) -> bool: + now = time.time() + with _log_throttle_lock: + last = _log_seen.get(key, 0) + if now - last < interval: + return False + _log_seen[key] = now + print(msg, flush=True) + return True + + +def _check_quota_error(error_msg: str) -> bool: + msg = str(error_msg).lower() + return any(kw in msg for kw in [ + "exceeded your zerogpu quota", "quota", + "0s left", "authenticate with a hugging face token" + ]) # ╔══════════════════════════════════════════════════════════════╗ -# ║ HF TOKEN POOL MANAGER ║ +# ║ [2] TOKEN POOL MANAGER ║ # ╚══════════════════════════════════════════════════════════════╝ class TokenPoolManager: - def __init__(self, tokens): + def __init__(self, tokens: List[str]): self.tokens = tokens self._lock = threading.Lock() self._current_index = 0 - self._status = {} - for i in range(len(tokens)): - self._status[i] = {"state": "active", "rest_until": None, - "exhausted_at": None, "use_count": 0, "error_count": 0} + self._status = { + i: {"state": "active", "rest_until": None, + "exhausted_at": None, "use_count": 0, "error_count": 0} + for i in range(len(tokens)) + } def _reactivate_rested(self): now = time.time() @@ -169,7 +230,7 @@ class TokenPoolManager: s["state"] = "active" s["rest_until"] = None s["exhausted_at"] = None - print(f"[TokenPool] Token #{i} reactivated") + print(f"[TokenPool] Token #{i} reactivated", flush=True) def get_next_token(self): with self._lock: @@ -182,28 +243,29 @@ class TokenPoolManager: return idx, self.tokens[idx] return None, None - def mark_exhausted(self, index, error_msg=""): + def mark_exhausted(self, index: int, error_msg: str = ""): with self._lock: rest = self._parse_rest_time(error_msg) self._status[index].update({ - "state": "resting", "exhausted_at": time.time(), + "state": "resting", + "exhausted_at": time.time(), "rest_until": time.time() + rest, "error_count": self._status[index]["error_count"] + 1, }) - print(f"[TokenPool] Token #{index} exhausted → resting {self._format_duration(rest)}") + print(f"[TokenPool] Token #{index} exhausted → resting {self._format_duration(rest)}", flush=True) - def is_token_active(self, index): + def is_token_active(self, index: int) -> bool: with self._lock: self._reactivate_rested() return self._status[index]["state"] == "active" - def get_token_index(self, token): + def get_token_index(self, token: str): try: return self.tokens.index(token) except ValueError: return None - def _parse_rest_time(self, error_msg): + def _parse_rest_time(self, error_msg: str) -> int: msg = str(error_msg).lower() mult = {"minute": 60, "hour": 3600, "day": 86400} m = re.search(r"in\s+(\d+)\s+(minute|hour|day)s?", msg) @@ -214,23 +276,21 @@ class TokenPoolManager: return int(m.group(1)) * mult.get(m.group(2), 3600) now = _utcnow() tomorrow = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) - return (tomorrow - now).total_seconds() + return int((tomorrow - now).total_seconds()) @staticmethod - def _format_duration(seconds): + def _format_duration(seconds: float) -> str: if seconds <= 0: return "0s" seconds = int(seconds) - h = seconds // 3600 - m = (seconds % 3600) // 60 - s = seconds % 60 + h, m, s = seconds // 3600, (seconds % 3600) // 60, seconds % 60 if h > 0: return f"{h}h {m}m" if m > 0: return f"{m}m {s}s" return f"{s}s" - def get_pool_status(self): + def get_pool_status(self) -> Dict: with self._lock: self._reactivate_rested() now = time.time() @@ -256,96 +316,16 @@ pool_manager = TokenPoolManager(POOL_85) # ╔══════════════════════════════════════════════════════════════╗ -# ║ STANDARD IMPORTS ║ -# ╚══════════════════════════════════════════════════════════════╝ -import gradio as gr -print(f"[Relay] Gradio {gr.__version__}") -from fastapi.middleware.cors import CORSMiddleware -from fastapi import Request -from fastapi.responses import JSONResponse - - -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [1] PATHS ║ +# ║ [3] USER TOKEN MANAGER ║ # ╚══════════════════════════════════════════════════════════════╝ -def _find_storage_dir(): - candidates = ["/data", "/home/user/data", os.path.join(os.getcwd(), "data"), - os.path.join(os.path.expanduser("~"), "relay_data")] - for path in candidates: - try: - os.makedirs(path, exist_ok=True) - tf = os.path.join(path, ".write_test") - with open(tf, "w") as f: - f.write("test") - os.remove(tf) - print(f"[Relay] Storage: {path}") - return path - except (PermissionError, OSError): - continue - import tempfile - path = os.path.join(tempfile.gettempdir(), "relay_data") - os.makedirs(path, exist_ok=True) - return path - - -STORAGE_DIR = _find_storage_dir() +STORAGE_DIR = os.environ.get("RELAY_STORAGE", "/data") USER_TOKENS_FILE = os.path.join(STORAGE_DIR, "user_tokens.json") BROADCAST_FILE = os.path.join(STORAGE_DIR, "broadcast.json") +os.makedirs(STORAGE_DIR, exist_ok=True) -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [2] CONSTANTS ║ -# ╚══════════════════════════════════════════════════════════════╝ -# Backend ZeroGPU workers (pure image generators) -BACKEND1_URL = os.environ.get("ZC_BACKEND1_URL", - "https://bl4ckspaces-zimageturbo-backend.hf.space").rstrip("/") -BACKEND2_URL = os.environ.get("ZC_BACKEND2_URL", - "https://bl4ckspaces-mps.hf.space").rstrip("/") - -# Worker secret - harus sama dengan yang di-set di kedua backend -WORKER_SECRET = os.environ.get("ZC_WORKER_SECRET", "zc_worker_k7x2m9q4_bl4ck") - -# Passkeys -UNCENSORED_PASSKEY = os.environ.get("ZC_UNCENSORED_PASSKEY", "Rusdi6967") -ADMIN_PASSKEY = os.environ.get("ZC_ADMIN_PASSKEY", "Ambalabu69") - -# Generation params (dikirim ke worker) -LOCKED_STEPS = 8 -LOCKED_CFG_SCALE = 0.0 -LOCKED_MAX_SEQUENCE_LENGTH = 256 -MIN_BATCH = 1 -MAX_BATCH = 2 -MAX_PIXELS = 1024 * 1024 -MIN_PIXELS = 512 * 512 -MIN_SIDE = 512 -MAX_SIDE = 2048 -RESOLUTION_STEP = 8 -MAX_ASPECT = 3.0 - -# Token economy -TOKEN_DAILY_QUOTA = 15 -COST_NORMAL = 1 -COST_UNCENSORED = 3 -BLUR_RADIUS = 30 - -# Spy log -SPY_MAX_ENTRIES = 40 -SPY_FULL_TTL_SECONDS = 900 -_spy_lock = threading.Lock() -_spy_log = [] -_spy_counter = 0 - -# Worker timeouts -WORKER_TIMEOUT_CONNECT = 15 -WORKER_TIMEOUT_READ = 600 # 10 menit untuk generate -WORKER_TOTAL_DEADLINE = 660 - - -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [3] USER TOKEN MANAGER ║ -# ╚══════════════════════════════════════════════════════════════╝ class UserTokenManager: - def __init__(self, filepath): + def __init__(self, filepath: str): self.filepath = filepath self._lock = threading.Lock() self._data = {} @@ -356,14 +336,13 @@ class UserTokenManager: if os.path.exists(self.filepath): with open(self.filepath, "r") as f: self._data = json.load(f) - print(f"[UserTokens] Loaded {len(self._data)} users") + print(f"[UserTokens] Loaded {len(self._data)} users", flush=True) except Exception as e: - print(f"[UserTokens] Load failed: {e}") + print(f"[UserTokens] Load failed: {e}", flush=True) self._data = {} def _save(self): try: - os.makedirs(os.path.dirname(self.filepath), exist_ok=True) tmp = self.filepath + ".tmp" with open(tmp, "w") as f: json.dump(self._data, f, indent=2) @@ -371,7 +350,7 @@ class UserTokenManager: except Exception as e: _log_throttled("ut_save_fail", f"[UserTokens] Save failed: {e}", 300) - def _should_reset(self, user): + def _should_reset(self, user: Dict) -> bool: lr = user.get("last_reset", "") if not lr: return True @@ -381,7 +360,7 @@ class UserTokenManager: except Exception: return True - def get_user(self, ip): + def get_user(self, ip: str) -> Dict: with self._lock: if ip not in self._data: self._data[ip] = { @@ -398,7 +377,7 @@ class UserTokenManager: self._save() return dict(user) - def is_banned(self, ip): + def is_banned(self, ip: str): user = self.get_user(ip) if not user.get("banned_until"): return False, None @@ -416,7 +395,7 @@ class UserTokenManager: except Exception: return False, None - def consume(self, ip, cost): + def consume(self, ip: str, cost: int): with self._lock: user = self._data.get(ip) if not user: @@ -431,13 +410,13 @@ class UserTokenManager: self._save() return True, user["tokens"] - def mark_uncensored(self, ip): + def mark_uncensored(self, ip: str): with self._lock: if ip in self._data: self._data[ip]["uncensored_used"] = self._data[ip].get("uncensored_used", 0) + 1 self._save() - def grant_tokens(self, ip, amount, by="system"): + def grant_tokens(self, ip: str, amount: int, by: str = "system"): with self._lock: self.get_user(ip) self._data[ip]["tokens"] = self._data[ip].get("tokens", 0) + int(amount) @@ -445,7 +424,7 @@ class UserTokenManager: self._save() return self._data[ip]["tokens"] - def ban_ip(self, ip, hours, reason="", by="admin"): + def ban_ip(self, ip: str, hours: float, reason: str = "", by: str = "admin"): with self._lock: self.get_user(ip) until = _utcnow() + timedelta(hours=float(hours)) @@ -455,7 +434,7 @@ class UserTokenManager: self._save() return self._data[ip]["banned_until"] - def unban_ip(self, ip): + def unban_ip(self, ip: str): with self._lock: if ip in self._data: self._data[ip]["banned_until"] = None @@ -464,7 +443,7 @@ class UserTokenManager: return True return False - def promote_admin(self, ip, by="system"): + def promote_admin(self, ip: str, by: str = "system"): with self._lock: self.get_user(ip) self._data[ip]["is_admin"] = True @@ -472,7 +451,7 @@ class UserTokenManager: self._data[ip]["promoted_at"] = _utc_str() self._save() - def demote_admin(self, ip): + def demote_admin(self, ip: str): with self._lock: if ip in self._data: self._data[ip]["is_admin"] = False @@ -482,7 +461,7 @@ class UserTokenManager: return True return False - def list_users(self, limit=300): + def list_users(self, limit: int = 300): with self._lock: items = [{"ip": ip, **data} for ip, data in self._data.items()] items.sort(key=lambda x: x.get("gens", 0), reverse=True) @@ -522,7 +501,6 @@ def _load_broadcast(): def _save_broadcast(): try: - os.makedirs(os.path.dirname(BROADCAST_FILE), exist_ok=True) with open(BROADCAST_FILE, "w") as f: json.dump(_broadcast, f, indent=2) except Exception: @@ -532,7 +510,7 @@ def _save_broadcast(): _load_broadcast() -def broadcast_set(message, btype="info", by="admin"): +def broadcast_set(message, btype: str = "info", by: str = "admin"): global _broadcast with _broadcast_lock: _broadcast = { @@ -586,8 +564,7 @@ class SmartSafetyFilter: "see-through": 5, "wet t-shirt": 6, "upskirt": 9, "downblouse": 7, "spread legs": 6, "ahegao": 7, "tentacle": 5, "striptease": 6, "stripper": 5, "camgirl": 5, "onlyfans": 3, - "bondage": 6, "bdsm": 5, "fetish": 4, - "gore": 8, "murder": 5, + "bondage": 6, "bdsm": 5, "fetish": 4, "gore": 8, "murder": 5, "r18": 6, "r-18": 6, "dildo": 7, "sex toy": 7, } SAFE_CONTEXT = { @@ -605,17 +582,15 @@ class SmartSafetyFilter: THRESHOLD = 15 @classmethod - def check(cls, prompt, negative_prompt="", hard_only=False): + def check(cls, prompt: str, negative_prompt: str = "", hard_only: bool = False): combined = ((prompt or "") + " " + (negative_prompt or "")).lower() for phrase in cls.HARD_BLOCK_PHRASES: if phrase in combined: - return (False, - f"Prohibited content detected: '{phrase.strip()}'. Not allowed in any mode.", + return (False, f"Prohibited content detected: '{phrase.strip()}'. Not allowed in any mode.", 100, [phrase.strip()]) for word in cls.HARD_BLOCK_WORDS: if re.search(r"\b" + re.escape(word) + r"\b", combined): - return (False, - f"Prohibited content detected: '{word}'. Not allowed in any mode.", + return (False, f"Prohibited content detected: '{word}'. Not allowed in any mode.", 100, [word]) if hard_only: return True, "OK (hard-blocks only)", 0, [] @@ -640,8 +615,7 @@ class SmartSafetyFilter: top = sorted(hits, key=lambda x: -x[1])[:5] top_str = ", ".join(f"'{t}'(+{w})" for t, w in top) return (False, - f"Content scored {score}/{cls.THRESHOLD}. Flagged: {top_str}. " - f"Try a milder prompt or use Uncensored mode.", + f"Content scored {score}/{cls.THRESHOLD}. Flagged: {top_str}. Try milder prompt or Uncensored mode.", score, [t for t, _ in top]) return True, "OK", score, [] @@ -650,15 +624,14 @@ safety_filter = SmartSafetyFilter() # ╔══════════════════════════════════════════════════════════════╗ -# ║ [6] STATS ║ +# ║ [6] STATS + SPY LOG ║ # ╚══════════════════════════════════════════════════════════════╝ _generation_stats = { "total_generations": 0, "total_time_seconds": 0.0, "last_generation_time": None, "last_resolution": None, "last_seed": None, "started_at": time.time(), "errors": 0, "safety_blocks": 0, - "normal_gens": 0, "uncensored_gens": 0, - "tokens_consumed": 0, "relay_failures": 0, "relay_busy": 0, - "legacy_blocked": 0, + "normal_gens": 0, "uncensored_gens": 0, "tokens_consumed": 0, + "relay_failures": 0, "legacy_blocked": 0, } @@ -675,145 +648,12 @@ def _update_stats(width, height, elapsed, seed, mode="normal", tokens=1): _generation_stats["normal_gens"] += 1 -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [7] WORKER RELAY (pure HTTP ke backend ZeroGPU) ║ -# ╚══════════════════════════════════════════════════════════════╝ -_worker_status = { - "backend1": {"reachable": None, "last_check": 0, "last_error": None, - "ok": 0, "fail": 0}, - "backend2": {"reachable": None, "last_check": 0, "last_error": None, - "ok": 0, "fail": 0}, -} -_ws_lock = threading.Lock() - - -def _check_worker(backend_key, force=False): - url = BACKEND1_URL if backend_key == "backend1" else BACKEND2_URL - now = time.time() - with _ws_lock: - if not force and now - _worker_status[backend_key]["last_check"] < 120: - return dict(_worker_status[backend_key]) - ok, err = False, None - try: - r = http_requests.get(url + "/", timeout=6) - ok = (r.status_code == 200) - if not ok: - err = f"HTTP {r.status_code}" - except Exception as e: - err = str(e)[:150] - with _ws_lock: - _worker_status[backend_key].update({ - "reachable": ok, "last_check": time.time(), "last_error": err - }) - return dict(_worker_status[backend_key]) - - -def _relay_to_worker(backend_key, prompt, neg, width, height, seed, batch, hf_token): - """ - Relay request ke backend ZeroGPU. - Backend expose endpoint: /gradio_api/call/worker_generate - Args: [WORKER_SECRET, prompt, neg, width, height, seed, batch] - Auth: Bearer (untuk ZeroGPU quota) - """ - url = BACKEND1_URL if backend_key == "backend1" else BACKEND2_URL - data = [WORKER_SECRET, str(prompt), str(neg or ""), - int(width), int(height), int(seed), int(batch)] - headers = { - "Content-Type": "application/json", - "Authorization": "Bearer " + hf_token, - } - post_url = url + "/gradio_api/call/worker_generate" - - deadline = time.monotonic() + WORKER_TOTAL_DEADLINE - - # POST - event_id = None - last_err = None - for attempt in range(3): - if time.monotonic() > deadline: - raise RuntimeError(f"{backend_key} deadline exceeded (POST phase)") - try: - r = http_requests.post(post_url, json={"data": data}, - headers=headers, timeout=WORKER_TIMEOUT_CONNECT) - if r.status_code in (502, 503): - last_err = f"{backend_key} cold-starting (HTTP {r.status_code})" - print(f"[Worker] {last_err} — retry {attempt+1}/3 in 10s") - time.sleep(10) - continue - r.raise_for_status() - event_id = r.json().get("event_id") - if not event_id: - raise RuntimeError(f"No event_id from {backend_key}: " + r.text[:200]) - break - except http_requests.exceptions.Timeout: - last_err = f"{backend_key} POST timeout" - print(f"[Worker] {last_err} — retry {attempt+1}/3") - time.sleep(3) - continue - except http_requests.exceptions.HTTPError as e: - raise RuntimeError(f"{backend_key} POST failed: {e}") - if not event_id: - raise RuntimeError(last_err or f"{backend_key} unreachable") - - # SSE stream - get_url = post_url + "/" + event_id - current_event = None - with http_requests.get(get_url, headers=headers, stream=True, - timeout=(15, WORKER_TIMEOUT_READ)) as resp: - resp.raise_for_status() - resp.encoding = "utf-8" - for raw in resp.iter_lines(decode_unicode=True): - if time.monotonic() > deadline: - raise RuntimeError(f"{backend_key} deadline exceeded (SSE phase)") - if raw is None: - continue - line = raw.strip() - if not line: - current_event = None - continue - if line.startswith("event:"): - current_event = line[6:].strip() - continue - if not line.startswith("data:"): - continue - payload = line[5:].strip() - if not payload: - continue - try: - float(payload) - continue # heartbeat - except ValueError: - pass - try: - parsed = json.loads(payload) - except json.JSONDecodeError: - continue - if current_event == "complete" or isinstance(parsed, list): - if parsed and isinstance(parsed[0], str): - return json.loads(parsed[0]) - raise RuntimeError(f"{backend_key} unexpected output format") - if isinstance(parsed, dict) and parsed.get("msg"): - msg = parsed["msg"] - if msg == "process_completed": - if parsed.get("success") is False: - raise RuntimeError(f"{backend_key}: {parsed.get('error') or 'process failed'}") - out = parsed.get("output") or {} - arr = out.get("data") or parsed.get("data") or [] - if arr and isinstance(arr[0], str): - return json.loads(arr[0]) - raise RuntimeError(f"{backend_key} empty output") - if msg == "process_failed": - raise RuntimeError(f"{backend_key}: {parsed.get('error') or 'process_failed'}") - if msg == "queue_full": - raise RuntimeError(f"{backend_key} queue full") - continue - raise RuntimeError(f"{backend_key} stream ended without output") +_spy_lock = threading.Lock() +_spy_log = [] +_spy_counter = 0 -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [8] BLUR + SPY ║ -# ╚══════════════════════════════════════════════════════════════╝ -def _blur_data_url(data_url, radius=BLUR_RADIUS): +def _blur_data_url(data_url: str, radius: int = BLUR_RADIUS): try: _, b64 = data_url.split(",", 1) raw = base64.b64decode(b64) @@ -825,11 +665,11 @@ def _blur_data_url(data_url, radius=BLUR_RADIUS): img.save(buf, format="JPEG", quality=90) return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode("utf-8") except Exception as e: - print(f"[Relay] blur failed: {e}") + print(f"[Relay] blur failed: {e}", flush=True) return None -def _make_thumbnail(data_url, max_size=224): +def _make_thumbnail(data_url: str, max_size: int = 224): try: _, b64 = data_url.split(",", 1) raw = base64.b64decode(b64) @@ -844,7 +684,7 @@ def _make_thumbnail(data_url, max_size=224): return None -def _add_spy_entry(entry): +def _add_spy_entry(entry: Dict) -> int: global _spy_counter entry["time_ts"] = time.time() with _spy_lock: @@ -863,32 +703,8 @@ def _add_spy_entry(entry): return entry["id"] -def _extract_client_info(request): - ip, ua = "unknown", "unknown" - try: - if request and hasattr(request, "headers"): - hdrs = dict(request.headers) - xff = hdrs.get("x-forwarded-for", "") - if xff: - ip = xff.split(",")[0].strip() - else: - ip = hdrs.get("x-real-ip", "").strip() - if not ip: - try: - if request.client and request.client.host: - ip = request.client.host - except Exception: - pass - ua = hdrs.get("user-agent", "unknown") - if len(ua) > 300: - ua = ua[:300] + "..." - except Exception: - pass - return ip or "unknown", ua or "unknown" - - -def _log_to_spy(client_ip, client_ua, token_idx, prompt, neg, width, height, - batch, result_dict, stage="generate", mode="normal", worker_key="backend1"): +def _log_to_spy(client_ip, client_ua, token_idx, prompt, neg, width, height, batch, + result_dict, stage="generate", mode="normal", worker_key="backend1"): try: images_spy = [] if result_dict.get("success"): @@ -926,86 +742,569 @@ def _log_to_spy(client_ip, client_ua, token_idx, prompt, neg, width, height, return None -def _check_quota_error(error_msg): - msg = str(error_msg).lower() - return any(kw in msg for kw in ["exceeded your zerogpu quota", "quota", - "0s left", "authenticate with a hugging face token"]) +def _build_feed_payload(since: int = 0): + with _spy_lock: + snapshot = list(_spy_log) + if since > 0: + picked = [e for e in snapshot if e.get("id", 0) > since] + picked.reverse() + else: + picked = list(reversed(snapshot)) + feed = [] + for e in picked: + ec = dict(e) + ec.pop("_full_expires", None) + ec["images"] = [{"seed": im.get("seed"), "thumb": im.get("thumb"), + "is_blurred": im.get("is_blurred", False)} + for im in e.get("images", [])] + feed.append(ec) + return feed # ╔══════════════════════════════════════════════════════════════╗ -# ║ [9] VALIDATION ║ +# ║ [7] HTTPX CLIENT (ASYNC, CONNECTION POOLING) ║ # ╚══════════════════════════════════════════════════════════════╝ -def _validate_batch(batch): - try: - batch = int(batch) - except (ValueError, TypeError): - return None, "Batch must be an integer (1 or 2)" - if batch < MIN_BATCH or batch > MAX_BATCH: - return None, f"Batch size {batch} REJECTED. Allowed: {MIN_BATCH}-{MAX_BATCH} only." - return batch, None +_http_client: Optional[httpx.AsyncClient] = None + + +async def get_http_client() -> httpx.AsyncClient: + global _http_client + if _http_client is None or _http_client.is_closed: + _http_client = httpx.AsyncClient( + timeout=httpx.Timeout( + connect=WORKER_CONNECT_TIMEOUT, + read=WORKER_READ_TIMEOUT, + write=30.0, + pool=10.0, + ), + limits=httpx.Limits( + max_connections=WORKER_POOL_LIMIT, + max_keepalive_connections=20, + keepalive_expiry=30, + ), + follow_redirects=True, + ) + return _http_client -def _validate_resolution(width, height): - if width % RESOLUTION_STEP != 0 or height % RESOLUTION_STEP != 0: - return f"Dimensions must be multiples of {RESOLUTION_STEP}" - if not (MIN_SIDE <= width <= MAX_SIDE) or not (MIN_SIDE <= height <= MAX_SIDE): - return f"Each side must be between {MIN_SIDE} and {MAX_SIDE}" - pixels = width * height - if pixels > MAX_PIXELS: - return f"REJECTED: {pixels:,} px exceeds 1024×1024 spec (max {MAX_PIXELS:,})" - if pixels < MIN_PIXELS: - return f"REJECTED: {pixels:,} px below minimum ({MIN_PIXELS:,})" - ratio = max(width, height) / max(1, min(width, height)) - if ratio > MAX_ASPECT: - return f"REJECTED: aspect ratio {ratio:.2f}:1 too extreme (max {MAX_ASPECT}:1)" - return None +_worker_status = { + "backend1": {"reachable": None, "last_check": 0, "last_error": None, "ok": 0, "fail": 0}, + "backend2": {"reachable": None, "last_check": 0, "last_error": None, "ok": 0, "fail": 0}, +} +_ws_lock = asyncio.Lock() if False else threading.Lock() # placeholder, pakai sync lock -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [10] SMART GENERATE (ROUTER) ║ -# ╚══════════════════════════════════════════════════════════════╝ -def smart_generate(prompt, negative_prompt, width, height, seed, - batch_or_mode=None, mode=None, request: gr.Request = None): +async def _check_worker(backend_key: str, force: bool = False) -> Dict: + url = BACKEND1_URL if backend_key == "backend1" else BACKEND2_URL + now = time.time() + with _ws_lock: + if not force and now - _worker_status[backend_key]["last_check"] < 120: + return dict(_worker_status[backend_key]) + ok, err = False, None + try: + client = await get_http_client() + r = await client.get(url + "/", timeout=8.0) + ok = (r.status_code == 200) + if not ok: + err = f"HTTP {r.status_code}" + except Exception as e: + err = str(e)[:150] + with _ws_lock: + _worker_status[backend_key].update({ + "reachable": ok, "last_check": time.time(), "last_error": err + }) + return dict(_worker_status[backend_key]) + + +async def _relay_to_worker(backend_key: str, prompt: str, neg: str, + width: int, height: int, seed: int, batch: int, + hf_token: str) -> Dict: """ - Router utama: validasi → safety → route ke worker → blur jika uncensored. - Backward compatible dengan frontend v1.5.1 (6 args dengan hf_token di arg ke-6). + Relay request ke backend ZeroGPU. + Backend expose: POST /gradio_api/call/worker_generate + Args: [WORKER_SECRET, prompt, neg, width, height, seed, batch] """ - # ── Detect frontend version ── - legacy_mode = False - if batch_or_mode is not None and isinstance(batch_or_mode, str): - s = str(batch_or_mode).strip() - if s.startswith("hf_") or len(s) > 20: - legacy_mode = True - batch = 1 - mode = "normal" - elif s in ("normal", "uncensored"): - batch = 1 - mode = s + url = BACKEND1_URL if backend_key == "backend1" else BACKEND2_URL + data = [WORKER_SECRET, str(prompt), str(neg or ""), + int(width), int(height), int(seed), int(batch)] + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {hf_token}", + } + post_url = f"{url}/gradio_api/call/worker_generate" + + client = await get_http_client() + + # POST — get event_id + last_err = None + event_id = None + for attempt in range(3): + try: + r = await client.post(post_url, json={"data": data}, headers=headers) + if r.status_code in (502, 503): + last_err = f"{backend_key} cold-starting (HTTP {r.status_code})" + print(f"[Worker] {last_err} — retry {attempt+1}/3 in 10s", flush=True) + await asyncio.sleep(10) + continue + r.raise_for_status() + body = r.json() + event_id = body.get("event_id") + if not event_id: + raise RuntimeError(f"No event_id from {backend_key}: {r.text[:200]}") + break + except httpx.TimeoutException: + last_err = f"{backend_key} POST timeout" + print(f"[Worker] {last_err} — retry {attempt+1}/3", flush=True) + await asyncio.sleep(3) + continue + except httpx.HTTPStatusError as e: + raise RuntimeError(f"{backend_key} POST failed: HTTP {e.response.status_code}") + except Exception as e: + raise RuntimeError(f"{backend_key} POST error: {type(e).__name__}: {e}") + + if not event_id: + raise RuntimeError(last_err or f"{backend_key} unreachable after 3 retries") + + # SSE stream + get_url = f"{post_url}/{event_id}" + current_event = None + try: + async with client.stream("GET", get_url, headers=headers) as resp: + resp.raise_for_status() + async for line in resp.aiter_lines(): + if not line: + current_event = None + continue + if line.startswith("event:"): + current_event = line[6:].strip() + continue + if not line.startswith("data:"): + continue + payload = line[5:].strip() + if not payload: + continue + try: + float(payload) + continue # heartbeat + except ValueError: + pass + try: + parsed = json.loads(payload) + except json.JSONDecodeError: + continue + if current_event == "complete" or isinstance(parsed, list): + if parsed and isinstance(parsed[0], str): + return json.loads(parsed[0]) + raise RuntimeError(f"{backend_key} unexpected output format") + if isinstance(parsed, dict) and parsed.get("msg"): + msg = parsed["msg"] + if msg == "process_completed": + if parsed.get("success") is False: + raise RuntimeError(f"{backend_key}: {parsed.get('error') or 'process failed'}") + out = parsed.get("output") or {} + arr = out.get("data") or parsed.get("data") or [] + if arr and isinstance(arr[0], str): + return json.loads(arr[0]) + raise RuntimeError(f"{backend_key} empty output") + if msg == "process_failed": + raise RuntimeError(f"{backend_key}: {parsed.get('error') or 'process_failed'}") + if msg == "queue_full": + raise RuntimeError(f"{backend_key} queue full") + continue + except httpx.TimeoutException: + raise RuntimeError(f"{backend_key} SSE timeout") + except httpx.HTTPStatusError as e: + raise RuntimeError(f"{backend_key} SSE HTTP {e.response.status_code}") + + raise RuntimeError(f"{backend_key} stream ended without output") + + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [8] FASTAPI APP ║ +# ╚══════════════════════════════════════════════════════════════╝ +import asyncio + +app = FastAPI( + title="ZeroCost Relay", + version=VERSION, + docs_url=None, # disable Swagger UI + redoc_url=None, # disable ReDoc +) + +# CORS +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + expose_headers=["*"], +) + + +# ── Helper: extract client info ── +def _extract_client_info(request: Request): + ip = "unknown" + ua = "unknown" + try: + xff = request.headers.get("x-forwarded-for", "") + if xff: + ip = xff.split(",")[0].strip() else: - try: - batch = int(batch_or_mode) - mode = mode or "normal" - except (ValueError, TypeError): - batch = 1 - mode = "normal" - elif batch_or_mode is not None: + ip = request.headers.get("x-real-ip", "") or (request.client.host if request.client else "unknown") + ua = request.headers.get("user-agent", "unknown") + if len(ua) > 300: + ua = ua[:300] + "..." + except Exception: + pass + return ip or "unknown", ua or "unknown" + + +# ── Helper: extract token from Authorization header ── +def _extract_token(request: Request): + auth = request.headers.get("authorization", "") + if auth.lower().startswith("bearer "): + return auth[7:].strip() + return auth.strip() if auth else None + + +# ── Helper: admin auth ── +def _admin_auth(key: str) -> bool: + return str(key).strip() == ADMIN_PASSKEY + + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [9] MIDDLEWARE — LEGACY FRONTEND GUARD ║ +# ╚══════════════════════════════════════════════════════════════╝ +@app.middleware("http") +async def legacy_frontend_guard(request: Request, call_next): + """Block request dari frontend v1.5.1 di pintu masuk.""" + path = request.url.path + method = request.method + + # Block /get_tokens (plural) — endpoint v1.5.1 + if method == "POST" and path == "/get_tokens": + _generation_stats["legacy_blocked"] += 1 + _log_throttled( + "guard_legacy_get_tokens", + f"[Guard] 🚫 Blocked legacy /get_tokens from {request.client.host if request.client else '?'}", + 60, + ) + return JSONResponse( + status_code=410, + content={ + "error": "OUTDATED_FRONTEND", + "message": "Your frontend is outdated (v1.5.1). Hard-refresh: Ctrl+Shift+R", + "required_version": "3.0.0", + "blocked_by": "legacy_guard", + } + ) + + # Block /generate dengan 6 args (hf_token di arg ke-6) + if method == "POST" and path == "/generate": try: - batch = int(batch_or_mode) - except (ValueError, TypeError): - batch = 1 - mode = mode or "normal" + body_bytes = await request.body() + if body_bytes: + body = json.loads(body_bytes) + # Gradio-style: { "data": [...] } + if isinstance(body, dict) and "data" in body: + payload = body["data"] + if isinstance(payload, list) and len(payload) == 6: + sixth = payload[5] + if isinstance(sixth, str) and sixth.startswith("hf_"): + _generation_stats["legacy_blocked"] += 1 + _log_throttled( + "guard_legacy_generate", + f"[Guard] 🚫 Blocked legacy generate (6 args) from {request.client.host if request.client else '?'}", + 30, + ) + return JSONResponse( + status_code=410, + content={ + "error": "OUTDATED_FRONTEND", + "message": "Frontend v1.5.1 detected. Hard-refresh to load v3.0.0", + "required_version": "3.0.0", + "blocked_by": "legacy_guard", + } + ) + except (json.JSONDecodeError, Exception): + pass + # Restore body untuk downstream handler + async def receive_wrapper(): + return {"type": "http.request", "body": body_bytes, "more_body": False} + request._receive = receive_wrapper + + return await call_next(request) + + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [10] PUBLIC ENDPOINTS ║ +# ╚══════════════════════════════════════════════════════════════╝ +@app.get("/") +async def root(): + return {"service": "ZeroCost Relay", "version": VERSION, "status": "online"} + + +@app.get("/info") +async def info(): + """Compatibility endpoint (frontend pakai ini untuk detect API).""" + return { + "version": VERSION, + "named_endpoints": { + "/generate": {}, "/get_token": {}, "/get_tokens": {}, + "/pool_status": {}, "/user_status": {}, "/get_config": {}, + "/health": {}, "/get_broadcast": {}, "/unlock_image": {}, + "/diagnose": {}, "/get_stats": {}, "/reset_stats": {}, + "/admin_feed": {}, "/admin_feed_since": {}, "/admin_users": {}, + "/admin_ban": {}, "/admin_unban": {}, "/admin_grant": {}, + "/admin_promote": {}, "/admin_demote": {}, "/admin_broadcast": {}, + "/admin_image": {}, "/admin_log": {}, + } + } + + +@app.post("/get_token") +async def get_token(): + idx, token = pool_manager.get_next_token() + if idx is None: + status = pool_manager.get_pool_status() + soonest = None + for d in status.get("resting_details", []): + if soonest is None or d["remaining_seconds"] < soonest: + soonest = d["remaining_seconds"] + return { + "success": False, + "error": f"All HF tokens resting. Next in ~{TokenPoolManager._format_duration(soonest or 0)}.", + "pool_status": status, + } + return { + "success": True, "token": token, "token_index": idx, + "pool_active": pool_manager.get_pool_status()["active"], + "pool_total": len(POOL_85), + } + + +@app.post("/get_tokens") +async def get_tokens_legacy(): + """Stub untuk frontend v1.5.1 — return POOL_85 array.""" + _log_throttled("legacy_get_tokens", + "[Relay] Legacy 'get_tokens' called (served pool)", 600) + return POOL_85 + + +@app.post("/pool_status") +async def pool_status(): + return pool_manager.get_pool_status() + + +@app.post("/user_status") +async def user_status(request: Request): + client_ip, _ = _extract_client_info(request) + user = user_tokens.get_user(client_ip) + banned, ban_info = user_tokens.is_banned(client_ip) + return { + "ip": client_ip, "tokens": user.get("tokens", 0), + "is_admin": user.get("is_admin", False), + "total_used": user.get("total_used", 0), + "uncensored_used": user.get("uncensored_used", 0), + "gens": user.get("gens", 0), "last_reset": user.get("last_reset"), + "banned": banned, "ban_info": ban_info, + "costs": {"normal": COST_NORMAL, "uncensored": COST_UNCENSORED}, + "daily_quota": TOKEN_DAILY_QUOTA, + } + + +@app.post("/get_config") +async def get_config(): + b1 = await _check_worker("backend1") + b2 = await _check_worker("backend2") + return { + "version": VERSION, + "architecture": "Docker relay + ZeroGPU workers", + "modes": { + "normal": {"label": "Normal (HF Official)", "cost": COST_NORMAL, + "nsfw_filter": "full", "blur": False, "worker": "backend1"}, + "uncensored": {"label": "Uncensored (CivitAI)", "cost": COST_UNCENSORED, + "nsfw_filter": "hard-blocks only", "blur": True, + "passkey_required": True, "worker": "backend2"}, + }, + "token_economy": {"daily_quota": TOKEN_DAILY_QUOTA, "cost_normal": COST_NORMAL, + "cost_uncensored": COST_UNCENSORED, "reset": "daily (24h)"}, + "resolution_policy": {"max_pixels": MAX_PIXELS, "min_pixels": MIN_PIXELS, + "min_side": MIN_SIDE, "max_side": MAX_SIDE, + "max_aspect": MAX_ASPECT, "step": RESOLUTION_STEP}, + "batch_policy": {"min": MIN_BATCH, "max": MAX_BATCH}, + "workers": {"backend1": b1, "backend2": b2}, + } + + +@app.post("/health") +async def health(): + b1 = await _check_worker("backend1") + b2 = await _check_worker("backend2") + with _spy_lock: + spy_count = len(_spy_log) + return { + "status": "healthy", + "version": VERSION, + "architecture": "Docker relay + ZeroGPU workers", + "backend1": b1, "backend2": b2, + "hf_pool": pool_manager.get_pool_status(), + "token_economy": user_tokens.stats(), + "generation_stats": { + "total": _generation_stats["total_generations"], + "normal": _generation_stats["normal_gens"], + "uncensored": _generation_stats["uncensored_gens"], + "safety_blocks": _generation_stats["safety_blocks"], + "tokens_consumed": _generation_stats["tokens_consumed"], + "relay_failures": _generation_stats["relay_failures"], + "legacy_blocked": _generation_stats["legacy_blocked"], + }, + "spy_entries": spy_count, + } + + +@app.post("/get_broadcast") +async def get_broadcast(): + b = broadcast_get() + if not b.get("active"): + return {"active": False} + return b + + +@app.post("/diagnose") +async def diagnose(request: Request): + headers_dict = {} + try: + for key, value in request.headers.items(): + k = key.lower() + if k in ("authorization", "cookie") and len(value) > 20: + headers_dict[key] = value[:20] + "..." + else: + headers_dict[key] = value + except Exception as e: + headers_dict["_error"] = str(e) + client_ip, _ = _extract_client_info(request) + b1 = await _check_worker("backend1") + b2 = await _check_worker("backend2") + return { + "all_headers": headers_dict, "client_ip": client_ip, + "user": user_tokens.get_user(client_ip), + "pool_status": pool_manager.get_pool_status(), + "broadcast": broadcast_get(), + "workers": {"backend1": b1, "backend2": b2}, + } + + +@app.post("/get_stats") +async def get_stats(): + stats = dict(_generation_stats) + stats["uptime_seconds"] = round(time.time() - _generation_stats["started_at"], 1) + stats["avg_time_seconds"] = (round(stats["total_time_seconds"] / stats["total_generations"], 2) + if stats["total_generations"] > 0 else 0) + return stats + + +@app.post("/reset_stats") +async def reset_stats(): + global _generation_stats + _generation_stats = { + "total_generations": 0, "total_time_seconds": 0.0, + "last_generation_time": None, "last_resolution": None, "last_seed": None, + "started_at": time.time(), "errors": 0, "safety_blocks": 0, + "normal_gens": 0, "uncensored_gens": 0, "tokens_consumed": 0, + "relay_failures": 0, "legacy_blocked": 0, + } + return {"success": True, "message": "Stats reset"} + + +@app.post("/unlock_image") +async def unlock_image(request: Request): + body = await request.json() + passkey = body.get("passkey", "") + entry_id = body.get("entry_id") + if str(passkey).strip() != UNCENSORED_PASSKEY: + return {"success": False, "error": "Invalid passkey"} + try: + entry_id = int(entry_id) + except (ValueError, TypeError): + return {"success": False, "error": "Bad entry id"} + with _spy_lock: + for e in _spy_log: + if e["id"] == entry_id: + imgs = [{"seed": im.get("seed"), "full": im.get("full")} + for im in e.get("images", []) if im.get("full")] + if not imgs: + return {"success": False, + "error": "Image expired (15-min TTL). Please regenerate."} + return {"success": True, "entry_id": entry_id, "images": imgs} + return {"success": False, "error": "Image expired from buffer. Please regenerate."} + + +# ╔════════════════════���═════════════════════════════════════════╗ +# ║ [11] VALIDATION HELPERS ║ +# ╚══════════════════════════════════════════════════════════════╝ +def _validate_resolution(width: int, height: int): + if width % RESOLUTION_STEP != 0 or height % RESOLUTION_STEP != 0: + return f"Dimensions must be multiples of {RESOLUTION_STEP}" + if not (MIN_SIDE <= width <= MAX_SIDE) or not (MIN_SIDE <= height <= MAX_SIDE): + return f"Each side must be between {MIN_SIDE} and {MAX_SIDE}" + pixels = width * height + if pixels > MAX_PIXELS: + return f"REJECTED: {pixels:,} px exceeds 1024×1024 spec (max {MAX_PIXELS:,})" + if pixels < MIN_PIXELS: + return f"REJECTED: {pixels:,} px below minimum ({MIN_PIXELS:,})" + ratio = max(width, height) / max(1, min(width, height)) + if ratio > MAX_ASPECT: + return f"REJECTED: aspect ratio {ratio:.2f}:1 too extreme (max {MAX_ASPECT}:1)" + return None + + +# ╔══════════════════════════════════════════════════════════════╗ +# ║ [12] /generate — MAIN ROUTER ║ +# ╚══════════════════════════════════════════════════════════════╝ +@app.post("/generate") +async def generate(request: Request): + body = await request.json() + # Accept both array and object format + if isinstance(body, dict) and "data" in body: + payload = body["data"] + elif isinstance(body, list): + payload = body else: - batch = 1 - mode = "normal" + payload = [ + body.get("prompt", ""), + body.get("negative_prompt", ""), + body.get("width", 1024), + body.get("height", 1024), + body.get("seed", -1), + body.get("batch", 1), + body.get("mode", "normal"), + ] + + # Backward compat: detect legacy v1.5.1 (6 args dengan hf_token di posisi ke-6) + legacy_mode = False + if len(payload) == 6: + sixth = payload[5] + if isinstance(sixth, str) and sixth.startswith("hf_"): + legacy_mode = True if legacy_mode: + prompt, neg, width, height, seed, _ = payload + batch, mode = 1, "normal" _log_throttled("legacy_generate", "[Relay] Legacy frontend (v1.5.1) — auto-adapting to normal mode", 60) _generation_stats["legacy_blocked"] += 1 + else: + prompt = payload[0] if len(payload) > 0 else "" + neg = payload[1] if len(payload) > 1 else "" + width = payload[2] if len(payload) > 2 else 1024 + height = payload[3] if len(payload) > 3 else 1024 + seed = payload[4] if len(payload) > 4 else -1 + batch = payload[5] if len(payload) > 5 else 1 + mode = payload[6] if len(payload) > 6 else "normal" - # ── Client info ── client_ip, client_ua = _extract_client_info(request) - mode = str(mode or "normal").lower().strip() if mode not in ("normal", "uncensored"): mode = "normal" @@ -1016,9 +1315,8 @@ def smart_generate(prompt, negative_prompt, width, height, seed, rej = {"success": False, "error": f"Banned until {ban_info['until']}. Reason: {ban_info['reason']}", "retryable": False, "banned": True, "ban_info": ban_info} - _log_to_spy(client_ip, client_ua, -1, prompt, negative_prompt, - 0, 0, 0, rej, stage="banned", mode=mode) - return json.dumps(rej) + _log_to_spy(client_ip, client_ua, -1, prompt, neg, 0, 0, 0, rej, stage="banned", mode=mode) + return rej # ── Batch validation ── try: @@ -1034,14 +1332,14 @@ def smart_generate(prompt, negative_prompt, width, height, seed, height = int(height) seed = int(seed) if seed is not None else -1 except (ValueError, TypeError) as e: - return json.dumps({"success": False, "error": f"Bad param: {e}", "retryable": False}) + return {"success": False, "error": f"Bad param: {e}", "retryable": False} res_err = _validate_resolution(width, height) if res_err: rej = {"success": False, "error": res_err, "retryable": False, "resolution_rejected": True} - _log_to_spy(client_ip, client_ua, -1, prompt, negative_prompt, - width, height, batch_val, rej, stage="resolution_rejected", mode=mode) - return json.dumps(rej) + _log_to_spy(client_ip, client_ua, -1, prompt, neg, width, height, batch_val, rej, + stage="resolution_rejected", mode=mode) + return rej # ── User tokens ── user = user_tokens.get_user(client_ip) @@ -1054,31 +1352,31 @@ def smart_generate(prompt, negative_prompt, width, height, seed, "error": f"Insufficient tokens. Need {total_cost}, have {user.get('tokens', 0)}. Resets daily.", "retryable": False, "insufficient_tokens": True, "tokens_have": user.get("tokens", 0), "tokens_needed": total_cost} - _log_to_spy(client_ip, client_ua, -1, prompt, negative_prompt, - width, height, batch_val, rej, stage="insufficient_tokens", mode=mode) - return json.dumps(rej) + _log_to_spy(client_ip, client_ua, -1, prompt, neg, width, height, batch_val, rej, + stage="insufficient_tokens", mode=mode) + return rej # ── Safety filter ── is_safe, safety_reason, safety_score, safety_hits = safety_filter.check( - prompt, negative_prompt, hard_only=(mode == "uncensored")) + prompt, neg, hard_only=(mode == "uncensored")) if not is_safe: _generation_stats["safety_blocks"] += 1 - print(f"[Relay] 🛡️ SAFETY BLOCKED ({mode}) from {client_ip}") + print(f"[Relay] 🛡️ SAFETY BLOCKED ({mode}) from {client_ip}", flush=True) rej = {"success": False, "error": safety_reason, "retryable": False, "safety_blocked": True, "safety_score": safety_score, "safety_hits": safety_hits, "_safety_score": safety_score} - _log_to_spy(client_ip, client_ua, -1, prompt, negative_prompt, - width, height, batch_val, rej, stage="safety_filtered", mode=mode) - return json.dumps(rej) + _log_to_spy(client_ip, client_ua, -1, prompt, neg, width, height, batch_val, rej, + stage="safety_filtered", mode=mode) + return rej # ── Consume user tokens ── remaining = 999999 if not is_admin: ok, remaining = user_tokens.consume(client_ip, total_cost) if not ok: - return json.dumps({"success": False, - "error": f"Insufficient tokens (race). Need {total_cost}.", - "retryable": True, "insufficient_tokens": True}) + return {"success": False, + "error": f"Insufficient tokens (race). Need {total_cost}.", + "retryable": True, "insufficient_tokens": True} if mode == "uncensored": user_tokens.mark_uncensored(client_ip) @@ -1088,21 +1386,16 @@ def smart_generate(prompt, negative_prompt, width, height, seed, user_tokens.grant_tokens(client_ip, total_cost, by="refund") except Exception: pass - fail = {"success": False, "error": err_msg, "retryable": True, - "tokens_refunded": total_cost} + fail = {"success": False, "error": err_msg, "retryable": True, "tokens_refunded": total_cost} if extra: fail.update(extra) - _log_to_spy(client_ip, client_ua, -1, prompt, negative_prompt, - width, height, batch_val, fail, stage=stage, mode=mode) - return json.dumps(fail) + _log_to_spy(client_ip, client_ua, -1, prompt, neg, width, height, batch_val, fail, + stage=stage, mode=mode) + return fail # ── Pick worker ── - if mode == "uncensored": - worker_key = "backend2" - worker_label = "MPS (uncensored)" - else: - worker_key = "backend1" - worker_label = "ZeroGPU (normal)" + worker_key = "backend2" if mode == "uncensored" else "backend1" + worker_label = "MPS (uncensored)" if mode == "uncensored" else "ZeroGPU (normal)" # ── Get HF token for ZeroGPU auth ── hf_idx, hf_token = pool_manager.get_next_token() @@ -1116,11 +1409,10 @@ def smart_generate(prompt, negative_prompt, width, height, seed, seed = random.randint(0, 2**32 - 1) # ── Relay to worker ── - print(f"[Relay] 🚀 {worker_label} | {width}x{height} b={batch_val} | {client_ip} | token #{hf_idx}") + print(f"[Relay] 🚀 {worker_label} | {width}x{height} b={batch_val} | {client_ip} | token #{hf_idx}", flush=True) t0 = time.time() try: - result = _relay_to_worker(worker_key, prompt, negative_prompt, - width, height, seed, batch_val, hf_token) + result = await _relay_to_worker(worker_key, prompt, neg, width, height, seed, batch_val, hf_token) except Exception as e: emsg = str(e) _generation_stats["relay_failures"] += 1 @@ -1128,7 +1420,7 @@ def smart_generate(prompt, negative_prompt, width, height, seed, _worker_status[worker_key]["fail"] += 1 if _check_quota_error(emsg): - print(f"[Relay] ⚠️ ZeroGPU quota exhausted for HF token #{hf_idx}") + print(f"[Relay] ⚠️ ZeroGPU quota exhausted for HF token #{hf_idx}", flush=True) pool_manager.mark_exhausted(hf_idx, emsg) if not is_admin: try: @@ -1141,33 +1433,25 @@ def smart_generate(prompt, negative_prompt, width, height, seed, "try_different_token": True, "pool_status": pool_manager.get_pool_status(), "tokens_refunded": total_cost} - _log_to_spy(client_ip, client_ua, hf_idx, prompt, negative_prompt, - width, height, batch_val, fail, + _log_to_spy(client_ip, client_ua, hf_idx, prompt, neg, width, height, batch_val, fail, stage="quota_exhausted", mode=mode, worker_key=worker_key) - return json.dumps(fail) + return fail - print(f"[Relay] ❌ Worker failed: {emsg[:180]}") - return _refund_and_respond( - f"{worker_label} worker error: {emsg[:300]}", "error") + print(f"[Relay] ❌ Worker failed: {emsg[:180]}", flush=True) + return _refund_and_respond(f"{worker_label} worker error: {emsg[:300]}", "error") if not isinstance(result, dict) or not result.get("success"): err = (result or {}).get("error", f"{worker_key} returned failure") if _check_quota_error(err): pool_manager.mark_exhausted(hf_idx, err) - if not is_admin: - try: - user_tokens.grant_tokens(client_ip, total_cost, by="refund") - except Exception: - pass fail = {"success": False, "error": f"GPU quota exhausted: {err}", "retryable": True, "quota_exhausted": True, "try_different_token": True, "pool_status": pool_manager.get_pool_status(), "tokens_refunded": total_cost} - _log_to_spy(client_ip, client_ua, hf_idx, prompt, negative_prompt, - width, height, batch_val, fail, + _log_to_spy(client_ip, client_ua, hf_idx, prompt, neg, width, height, batch_val, fail, stage="quota_exhausted", mode=mode, worker_key=worker_key) - return json.dumps(fail) + return fail return _refund_and_respond(str(err), "error") with _ws_lock: @@ -1189,18 +1473,17 @@ def smart_generate(prompt, negative_prompt, width, height, seed, _update_stats(width, height, elapsed, md.get("seed", seed), mode=mode, tokens=0 if is_admin else total_cost) - # ── Spy log (gambar tajam disimpan server-side untuk unlock) ── - entry_id = _log_to_spy(client_ip, client_ua, hf_idx, prompt, negative_prompt, + # ── Spy log ── + entry_id = _log_to_spy(client_ip, client_ua, hf_idx, prompt, neg, width, height, batch_val, result, stage="generate", mode=mode, worker_key=worker_key) - # ── Blur jika uncensored & bukan admin ── + # ── Blur if uncensored & not admin ── if mode == "uncensored" and not is_admin: for img in result.get("images", []): blurred = _blur_data_url(img.get("data", "")) if blurred is None: - return _refund_and_respond( - "Blur failed. Tokens refunded.", "error") + return _refund_and_respond("Blur failed. Tokens refunded.", "error") img["data"] = blurred img["is_blurred"] = True md["is_blurred"] = True @@ -1211,576 +1494,247 @@ def smart_generate(prompt, negative_prompt, width, height, seed, md["unlock_id"] = None result["tokens_remaining"] = remaining - print(f"[Relay] ✅ OK | {elapsed:.2f}s | entry_id={entry_id} | blurred={mode=='uncensored' and not is_admin}") - return json.dumps(result) - - -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [11] API FUNCTIONS ║ -# ╚══════════════════════════════════════════════════════════════╝ -print("\n[Relay] Building Gradio Blocks...") - - -def _api_get_token(): - idx, token = pool_manager.get_next_token() - if idx is None: - status = pool_manager.get_pool_status() - soonest = None - for d in status.get("resting_details", []): - if soonest is None or d["remaining_seconds"] < soonest: - soonest = d["remaining_seconds"] - return json.dumps({ - "success": False, - "error": f"All HF tokens resting. Next in ~{TokenPoolManager._format_duration(soonest or 0)}.", - "pool_status": status}) - return json.dumps({"success": True, "token": token, "token_index": idx, - "pool_active": pool_manager.get_pool_status()["active"], - "pool_total": len(POOL_85)}) - - -def _api_get_tokens_legacy(): - """Stub untuk frontend v1.5.1 - return POOL_85 array.""" - _log_throttled("legacy_get_tokens", - "[Relay] Legacy 'get_tokens' called (served pool)", 600) - return json.dumps(POOL_85) - - -def _api_pool_status(): - return json.dumps(pool_manager.get_pool_status()) - - -def _api_user_status(request: gr.Request): - client_ip, _ = _extract_client_info(request) - user = user_tokens.get_user(client_ip) - banned, ban_info = user_tokens.is_banned(client_ip) - return json.dumps({ - "ip": client_ip, "tokens": user.get("tokens", 0), - "is_admin": user.get("is_admin", False), - "total_used": user.get("total_used", 0), - "uncensored_used": user.get("uncensored_used", 0), - "gens": user.get("gens", 0), "last_reset": user.get("last_reset"), - "banned": banned, "ban_info": ban_info, - "costs": {"normal": COST_NORMAL, "uncensored": COST_UNCENSORED}, - "daily_quota": TOKEN_DAILY_QUOTA, - }) - - -def _api_get_config(): - return json.dumps({ - "version": "3.0.0-relay", - "architecture": "CPU relay + ZeroGPU workers", - "modes": { - "normal": {"label": "Normal (HF Official)", "cost": COST_NORMAL, - "nsfw_filter": "full", "blur": False, "worker": "backend1"}, - "uncensored": {"label": "Uncensored (CivitAI)", "cost": COST_UNCENSORED, - "nsfw_filter": "hard-blocks only", "blur": True, - "passkey_required": True, "worker": "backend2"}, - }, - "token_economy": {"daily_quota": TOKEN_DAILY_QUOTA, "cost_normal": COST_NORMAL, - "cost_uncensored": COST_UNCENSORED, "reset": "daily (24h)"}, - "resolution_policy": {"max_pixels": MAX_PIXELS, "min_pixels": MIN_PIXELS, - "min_side": MIN_SIDE, "max_side": MAX_SIDE, - "max_aspect": MAX_ASPECT, "step": RESOLUTION_STEP}, - "batch_policy": {"min": MIN_BATCH, "max": MAX_BATCH}, - "workers": { - "backend1": _check_worker("backend1"), - "backend2": _check_worker("backend2"), - }, - }) - - -def _api_health(): - b1 = _check_worker("backend1") - b2 = _check_worker("backend2") - with _spy_lock: - spy_count = len(_spy_log) - return json.dumps({ - "status": "healthy", - "version": "3.0.0-relay", - "architecture": "CPU relay + ZeroGPU workers", - "backend1": b1, - "backend2": b2, - "hf_pool": pool_manager.get_pool_status(), - "token_economy": user_tokens.stats(), - "generation_stats": { - "total": _generation_stats["total_generations"], - "normal": _generation_stats["normal_gens"], - "uncensored": _generation_stats["uncensored_gens"], - "safety_blocks": _generation_stats["safety_blocks"], - "tokens_consumed": _generation_stats["tokens_consumed"], - "relay_failures": _generation_stats["relay_failures"], - "legacy_blocked": _generation_stats["legacy_blocked"], - }, - "spy_entries": spy_count, - }) - - -def _api_get_broadcast(): - b = broadcast_get() - if not b.get("active"): - return json.dumps({"active": False}) - return json.dumps(b) - - -def _api_unlock_image(passkey, entry_id): - if str(passkey).strip() != UNCENSORED_PASSKEY: - return json.dumps({"success": False, "error": "Invalid passkey"}) - try: - entry_id = int(entry_id) - except (ValueError, TypeError): - return json.dumps({"success": False, "error": "Bad entry id"}) - with _spy_lock: - for e in _spy_log: - if e["id"] == entry_id: - imgs = [{"seed": im.get("seed"), "full": im.get("full")} - for im in e.get("images", []) if im.get("full")] - if not imgs: - return json.dumps({"success": False, - "error": "Image expired (15-min TTL). Please regenerate."}) - return json.dumps({"success": True, "entry_id": entry_id, "images": imgs}) - return json.dumps({"success": False, - "error": "Image expired from buffer. Please regenerate."}) - - -def _api_diagnose(request: gr.Request): - headers_dict = {} - try: - if request and hasattr(request, "headers"): - for key, value in request.headers.items(): - k = key.lower() - headers_dict[key] = (value[:20] + "...") if k in ("authorization", "cookie") and len(value) > 20 else value - except Exception as e: - headers_dict["_error"] = str(e) - client_ip, _ = _extract_client_info(request) - return json.dumps({ - "all_headers": headers_dict, "client_ip": client_ip, - "user": user_tokens.get_user(client_ip), - "pool_status": pool_manager.get_pool_status(), - "broadcast": broadcast_get(), - "workers": { - "backend1": _check_worker("backend1"), - "backend2": _check_worker("backend2"), - }, - }) - - -def _api_get_stats(): - stats = dict(_generation_stats) - stats["uptime_seconds"] = round(time.time() - _generation_stats["started_at"], 1) - stats["avg_time_seconds"] = (round(stats["total_time_seconds"] / stats["total_generations"], 2) - if stats["total_generations"] > 0 else 0) - return json.dumps(stats) - - -def _api_reset_stats(): - global _generation_stats - _generation_stats = { - "total_generations": 0, "total_time_seconds": 0.0, - "last_generation_time": None, "last_resolution": None, "last_seed": None, - "started_at": time.time(), "errors": 0, "safety_blocks": 0, - "normal_gens": 0, "uncensored_gens": 0, - "tokens_consumed": 0, "relay_failures": 0, "relay_busy": 0, - "legacy_blocked": 0, - } - return json.dumps({"success": True, "message": "Stats reset"}) + print(f"[Relay] ✅ OK | {elapsed:.2f}s | entry_id={entry_id} | blurred={mode=='uncensored' and not is_admin}", flush=True) + return result # ╔══════════════════════════════════════════════════════════════╗ -# ║ [12] ADMIN ENDPOINTS ║ +# ║ [13] ADMIN ENDPOINTS ║ # ╚══════════════════════════════════════════════════════════════╝ -def _admin_auth(key): - return str(key).strip() == ADMIN_PASSKEY - - _admin_action_log = [] _admin_log_lock = threading.Lock() -def _log_admin_action(action, details): +def _log_admin_action(action: str, details: Dict): with _admin_log_lock: _admin_action_log.append({"time": _utc_str(), "action": action, "details": details}) if len(_admin_action_log) > 200: del _admin_action_log[:len(_admin_action_log) - 200] - print(f"[Admin] {action}: {details}") + print(f"[Admin] {action}: {details}", flush=True) -def _build_feed_payload(since=0): - with _spy_lock: - snapshot = list(_spy_log) - if since > 0: - picked = [e for e in snapshot if e.get("id", 0) > since] - picked.reverse() - else: - picked = list(reversed(snapshot)) - feed = [] - for e in picked: - ec = dict(e) - ec.pop("_full_expires", None) - ec["images"] = [{"seed": im.get("seed"), "thumb": im.get("thumb"), - "is_blurred": im.get("is_blurred", False)} - for im in e.get("images", [])] - feed.append(ec) - return feed - - -def _api_admin_feed(admin_key): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) - return json.dumps({ +@app.post("/admin_feed") +async def admin_feed(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + return { "success": True, "entries": _build_feed_payload(0), "incremental": False, "latest_id": (_spy_counter if _spy_log else 0), "stats": _generation_stats, "user_stats": user_tokens.stats(), "pool": pool_manager.get_pool_status(), - "workers": { - "backend1": _check_worker("backend1"), - "backend2": _check_worker("backend2"), - }, + "workers": {"backend1": await _check_worker("backend1"), + "backend2": await _check_worker("backend2")}, "broadcast": broadcast_get(), "server_time": _utc_str(), - }) + } -def _api_admin_feed_since(admin_key, since_id): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_feed_since") +async def admin_feed_since(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} try: - since = int(since_id) if since_id not in (None, "", 0) else 0 + since = int(body.get("since_id", 0)) except (ValueError, TypeError): since = 0 - return json.dumps({ + return { "success": True, "entries": _build_feed_payload(since), "incremental": since > 0, "latest_id": (_spy_counter if _spy_log else 0), "stats": _generation_stats, "user_stats": user_tokens.stats(), "pool": pool_manager.get_pool_status(), - "workers": { - "backend1": _check_worker("backend1"), - "backend2": _check_worker("backend2"), - }, + "workers": {"backend1": await _check_worker("backend1"), + "backend2": await _check_worker("backend2")}, "broadcast": broadcast_get(), "server_time": _utc_str(), - }) + } -def _api_admin_users(admin_key): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) - return json.dumps({"success": True, "users": user_tokens.list_users(limit=300)}) +@app.post("/admin_users") +async def admin_users(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + return {"success": True, "users": user_tokens.list_users(limit=300)} -def _api_admin_ban(admin_key, ip, hours, reason): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_ban") +async def admin_ban(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + ip = body.get("ip", "") try: - hours = float(hours) + hours = float(body.get("hours", 0)) except (ValueError, TypeError): - return json.dumps({"success": False, "error": "hours must be a number"}) + return {"success": False, "error": "hours must be a number"} if hours <= 0: - return json.dumps({"success": False, "error": "hours must be > 0"}) + return {"success": False, "error": "hours must be > 0"} + reason = body.get("reason", "") until = user_tokens.ban_ip(str(ip), hours, str(reason or "No reason"), by="admin_panel") _log_admin_action("ban", {"ip": ip, "hours": hours, "reason": reason, "until": until}) - return json.dumps({"success": True, "ip": ip, "banned_until": until, "reason": reason}) + return {"success": True, "ip": ip, "banned_until": until, "reason": reason} -def _api_admin_unban(admin_key, ip): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_unban") +async def admin_unban(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + ip = body.get("ip", "") ok = user_tokens.unban_ip(str(ip)) _log_admin_action("unban", {"ip": ip}) - return json.dumps({"success": ok, "ip": ip}) + return {"success": ok, "ip": ip} -def _api_admin_grant(admin_key, ip, amount): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_grant") +async def admin_grant(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + ip = body.get("ip", "") try: - amount = int(amount) + amount = int(body.get("amount", 0)) except (ValueError, TypeError): - return json.dumps({"success": False, "error": "amount must be integer"}) + return {"success": False, "error": "amount must be integer"} bal = user_tokens.grant_tokens(str(ip), amount, by="admin_panel") _log_admin_action("grant", {"ip": ip, "amount": amount, "new_balance": bal}) - return json.dumps({"success": True, "ip": ip, "amount": amount, "new_balance": bal}) + return {"success": True, "ip": ip, "amount": amount, "new_balance": bal} -def _api_admin_promote(admin_key, ip): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_promote") +async def admin_promote(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + ip = body.get("ip", "") user_tokens.promote_admin(str(ip), by="admin_panel") _log_admin_action("promote", {"ip": ip}) - return json.dumps({"success": True, "ip": ip, "is_admin": True}) + return {"success": True, "ip": ip, "is_admin": True} -def _api_admin_demote(admin_key, ip): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_demote") +async def admin_demote(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + ip = body.get("ip", "") user_tokens.demote_admin(str(ip)) _log_admin_action("demote", {"ip": ip}) - return json.dumps({"success": True, "ip": ip, "is_admin": False}) + return {"success": True, "ip": ip, "is_admin": False} -def _api_admin_broadcast(admin_key, message, btype): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_broadcast") +async def admin_broadcast(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} + message = body.get("message", "") + btype = body.get("type", "info") if not message or not str(message).strip(): result = broadcast_clear() _log_admin_action("broadcast_clear", {}) else: result = broadcast_set(str(message), str(btype or "info"), by="admin_panel") _log_admin_action("broadcast_set", {"message": str(message)[:200], "type": btype}) - return json.dumps({"success": True, "broadcast": result}) + return {"success": True, "broadcast": result} -def _api_admin_image(admin_key, entry_id): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) +@app.post("/admin_image") +async def admin_image(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} try: - entry_id = int(entry_id) + entry_id = int(body.get("entry_id", 0)) except (ValueError, TypeError): - return json.dumps({"success": False, "error": "Bad entry id"}) + return {"success": False, "error": "Bad entry id"} with _spy_lock: for e in _spy_log: if e["id"] == entry_id: imgs = [{"seed": im.get("seed"), "full": im.get("full")} for im in e.get("images", [])] if all(not im.get("full") for im in imgs): - return json.dumps({"success": False, - "error": f"Full image expired ({SPY_FULL_TTL_SECONDS//60}-min TTL)."}) - return json.dumps({ - "success": True, "entry_id": entry_id, "images": imgs, - "entry": {k: v for k, v in e.items() - if k not in ("images", "_full_expires")}, - }) - return json.dumps({"success": False, "error": "Entry not found (rotated out)"}) - - -def _api_admin_log(admin_key): - if not _admin_auth(admin_key): - return json.dumps({"success": False, "error": "Invalid admin key"}) + return {"success": False, + "error": f"Full image expired ({SPY_FULL_TTL_SECONDS//60}-min TTL)."} + return {"success": True, "entry_id": entry_id, "images": imgs, + "entry": {k: v for k, v in e.items() + if k not in ("images", "_full_expires")}} + return {"success": False, "error": "Entry not found (rotated out)"} + + +@app.post("/admin_log") +async def admin_log(request: Request): + body = await request.json() + key = body.get("admin_key", "") if isinstance(body, dict) else "" + if not _admin_auth(key): + return {"success": False, "error": "Invalid admin key"} with _admin_log_lock: - return json.dumps({"success": True, "log": list(reversed(_admin_action_log))}) - - -# ╔══════════════════════════════════════════════════════════════╗ -# ║ [13] GRADIO BLOCKS ║ -# ╚══════════════════════════════════════════════════════════════╝ -def _click(btn, fn, inputs, outputs, api_name, concurrency_limit=None): - kwargs = dict(fn=fn, inputs=inputs, outputs=outputs, api_name=api_name) - if concurrency_limit is not None: - kwargs["concurrency_limit"] = concurrency_limit - try: - return btn.click(**kwargs) - except TypeError: - kwargs.pop("concurrency_limit", None) - try: - return btn.click(**kwargs) - except Exception: - return None - - -GENERATE_CONCURRENCY = 50 # CPU bisa handle banyak paralel -API_CONCURRENCY = 100 - -with gr.Blocks(title="ZeroCost Relay v3.0.0") as demo: - gr.Markdown( - "# ⚡ ZeroCost Relay v3.0.0 (CPU Server)\n" - f"Management + routing → {BACKEND1_URL} (normal) / {BACKEND2_URL} (uncensored)" - ) - with gr.Tab("🎨 Generate"): - with gr.Row(): - with gr.Column(scale=1): - ui_prompt = gr.Textbox(label="Prompt", lines=3) - ui_neg = gr.Textbox(label="Negative Prompt", lines=2) - with gr.Row(): - ui_width = gr.Number(label="Width", value=1024) - ui_height = gr.Number(label="Height", value=1024) - ui_seed = gr.Number(label="Seed (-1=random)", value=-1) - ui_batch = gr.Number(label="Batch (1-2)", value=1) - ui_mode = gr.Dropdown(label="Mode", choices=["normal", "uncensored"], value="normal") - with gr.Column(scale=1): - ui_output = gr.Textbox(label="Result (JSON)", lines=10) - ui_btn = gr.Button("⚡ Generate", variant="primary") - _click(ui_btn, smart_generate, - [ui_prompt, ui_neg, ui_width, ui_height, ui_seed, ui_batch, ui_mode], - ui_output, "generate", GENERATE_CONCURRENCY) - - with gr.Tab("API", visible=False): - _gt_out = gr.Textbox(visible=False); _gt_btn = gr.Button(visible=False) - _click(_gt_btn, _api_get_token, None, _gt_out, "get_token", API_CONCURRENCY) - - _ps_out = gr.Textbox(visible=False); _ps_btn = gr.Button(visible=False) - _click(_ps_btn, _api_pool_status, None, _ps_out, "pool_status", API_CONCURRENCY) - - _us_out = gr.Textbox(visible=False); _us_btn = gr.Button(visible=False) - _click(_us_btn, _api_user_status, None, _us_out, "user_status", API_CONCURRENCY) - - _c_out = gr.Textbox(visible=False); _c_btn = gr.Button(visible=False) - _click(_c_btn, _api_get_config, None, _c_out, "get_config", API_CONCURRENCY) - - _h_out = gr.Textbox(visible=False); _h_btn = gr.Button(visible=False) - _click(_h_btn, _api_health, None, _h_out, "health", API_CONCURRENCY) - - _bc_out = gr.Textbox(visible=False); _bc_btn = gr.Button(visible=False) - _click(_bc_btn, _api_get_broadcast, None, _bc_out, "get_broadcast", API_CONCURRENCY) - - _d_out = gr.Textbox(visible=False); _d_btn = gr.Button(visible=False) - _click(_d_btn, _api_diagnose, None, _d_out, "diagnose", API_CONCURRENCY) - - _st_out = gr.Textbox(visible=False); _st_btn = gr.Button(visible=False) - _click(_st_btn, _api_get_stats, None, _st_out, "get_stats", API_CONCURRENCY) - - _rs_out = gr.Textbox(visible=False); _rs_btn = gr.Button(visible=False) - _click(_rs_btn, _api_reset_stats, None, _rs_out, "reset_stats", API_CONCURRENCY) - - # Legacy stub - _gtl_out = gr.Textbox(visible=False); _gtl_btn = gr.Button(visible=False) - _click(_gtl_btn, _api_get_tokens_legacy, None, _gtl_out, "get_tokens", API_CONCURRENCY) - - _pk_in = gr.Textbox(visible=False); _eid_in = gr.Number(visible=False) - _ul_out = gr.Textbox(visible=False); _ul_btn = gr.Button(visible=False) - _click(_ul_btn, _api_unlock_image, [_pk_in, _eid_in], _ul_out, "unlock_image", API_CONCURRENCY) - - _ak = gr.Textbox(visible=False) - _af_out = gr.Textbox(visible=False); _af_btn = gr.Button(visible=False) - _click(_af_btn, _api_admin_feed, [_ak], _af_out, "admin_feed", API_CONCURRENCY) - - _since = gr.Number(visible=False) - _afs_out = gr.Textbox(visible=False); _afs_btn = gr.Button(visible=False) - _click(_afs_btn, _api_admin_feed_since, [_ak, _since], _afs_out, "admin_feed_since", API_CONCURRENCY) - - _au_out = gr.Textbox(visible=False); _au_btn = gr.Button(visible=False) - _click(_au_btn, _api_admin_users, [_ak], _au_out, "admin_users", API_CONCURRENCY) - - _ip1 = gr.Textbox(visible=False); _hrs = gr.Textbox(visible=False); _rsn = gr.Textbox(visible=False) - _ban_out = gr.Textbox(visible=False); _ban_btn = gr.Button(visible=False) - _click(_ban_btn, _api_admin_ban, [_ak, _ip1, _hrs, _rsn], _ban_out, "admin_ban", API_CONCURRENCY) - - _ip2 = gr.Textbox(visible=False) - _ub_out = gr.Textbox(visible=False); _ub_btn = gr.Button(visible=False) - _click(_ub_btn, _api_admin_unban, [_ak, _ip2], _ub_out, "admin_unban", API_CONCURRENCY) - - _ip3 = gr.Textbox(visible=False); _amt = gr.Textbox(visible=False) - _gr_out = gr.Textbox(visible=False); _gr_btn = gr.Button(visible=False) - _click(_gr_btn, _api_admin_grant, [_ak, _ip3, _amt], _gr_out, "admin_grant", API_CONCURRENCY) - - _ip4 = gr.Textbox(visible=False) - _pr_out = gr.Textbox(visible=False); _pr_btn = gr.Button(visible=False) - _click(_pr_btn, _api_admin_promote, [_ak, _ip4], _pr_out, "admin_promote", API_CONCURRENCY) - - _ip5 = gr.Textbox(visible=False) - _dm_out = gr.Textbox(visible=False); _dm_btn = gr.Button(visible=False) - _click(_dm_btn, _api_admin_demote, [_ak, _ip5], _dm_out, "admin_demote", API_CONCURRENCY) - - _msg = gr.Textbox(visible=False); _btype = gr.Textbox(visible=False) - _bcs_out = gr.Textbox(visible=False); _bcs_btn = gr.Button(visible=False) - _click(_bcs_btn, _api_admin_broadcast, [_ak, _msg, _btype], _bcs_out, "admin_broadcast", API_CONCURRENCY) - - _eid2 = gr.Number(visible=False) - _im_out = gr.Textbox(visible=False); _im_btn = gr.Button(visible=False) - _click(_im_btn, _api_admin_image, [_ak, _eid2], _im_out, "admin_image", API_CONCURRENCY) - - _al_out = gr.Textbox(visible=False); _al_btn = gr.Button(visible=False) - _click(_al_btn, _api_admin_log, [_ak], _al_out, "admin_log", API_CONCURRENCY) - -print("[Relay] Blocks created: 14 public + 11 admin endpoints") + return {"success": True, "log": list(reversed(_admin_action_log))} # ╔══════════════════════════════════════════════════════════════╗ -# ║ [14] CORS + LEGACY GUARD MIDDLEWARE ║ +# ║ [14] STARTUP ║ # ╚══════════════════════════════════════════════════════════════╝ -app = demo.app -try: - app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, - allow_methods=["*"], allow_headers=["*"], expose_headers=["*"]) - print("[Relay] CORS added") -except Exception as e: - print(f"[Relay] CORS warning: {e}") - - -@app.middleware("http") -async def legacy_frontend_guard(request: Request, call_next): - """Satpam pintu masuk — block frontend v1.5.1 di level middleware.""" - path = request.url.path - method = request.method - - # /get_tokens (plural) - endpoint lama - if method == "POST" and "/call/get_tokens" in path and "get_token" not in path: - _generation_stats["legacy_blocked"] += 1 - _log_throttled("guard_legacy_get_tokens", - f"[Guard] 🚫 Blocked legacy /get_tokens from {request.client.host if request.client else '?'}", 60) - return JSONResponse(status_code=410, content={ - "event_id": None, "error": "OUTDATED_FRONTEND", - "message": "Your frontend is outdated. Hard-refresh: Ctrl+Shift+R", - "required_version": "3.0.0", "blocked_by": "legacy_guard", - }) - - if method == "GET" and "/call/get_tokens/" in path and "get_token" not in path: - return JSONResponse(status_code=410, content={ - "error": "OUTDATED_FRONTEND", "blocked_by": "legacy_guard" - }) - - # /generate dengan 6 args (hf_token di posisi ke-6) - if method == "POST" and path.endswith("/call/generate"): - try: - body = await request.body() - if body: - data = json.loads(body) - payload = data.get("data", []) - if isinstance(payload, list) and len(payload) == 6: - sixth = payload[5] - if isinstance(sixth, str) and sixth.startswith("hf_"): - _generation_stats["legacy_blocked"] += 1 - _log_throttled("guard_legacy_generate", - f"[Guard] 🚫 Blocked legacy generate (6 args) from {request.client.host if request.client else '?'}", 30) - return JSONResponse(status_code=410, content={ - "event_id": None, "error": "OUTDATED_FRONTEND", - "message": "Frontend v1.5.1 detected. Hard-refresh to load v3.0.0", - "required_version": "3.0.0", "blocked_by": "legacy_guard", - }) - except (json.JSONDecodeError, Exception): - pass - - return await call_next(request) - - -print("[Relay] 🛡️ Legacy frontend guard installed") +@app.on_event("startup") +async def startup(): + print("\n" + "=" * 60, flush=True) + print(f" ZERO COST v3.0.0 RELAY (Docker + FastAPI) - STARTING", flush=True) + print("=" * 60, flush=True) + print(f" Storage: {STORAGE_DIR}", flush=True) + print(f" Backend 1: {BACKEND1_URL}", flush=True) + print(f" Backend 2: {BACKEND2_URL}", flush=True) + print(f" Worker Secret: set ({len(WORKER_SECRET)} chars)", flush=True) + print(f" HF Pool: {len(POOL_85)} tokens", flush=True) + print(f" Token economy: {TOKEN_DAILY_QUOTA}/user/day | normal={COST_NORMAL} | uncensored={COST_UNCENSORED}", flush=True) + print(f" Safety: hard-block ALL modes; soft-score normal only", flush=True) + print(f" Spy log: {SPY_MAX_ENTRIES} entries, TTL {SPY_FULL_TTL_SECONDS//60} min", flush=True) + print(f" Worker timeout: connect={WORKER_CONNECT_TIMEOUT}s, read={WORKER_READ_TIMEOUT}s", flush=True) + print("-" * 60, flush=True) + print(" Endpoints:", flush=True) + print(" POST /generate, /get_token, /pool_status, /user_status", flush=True) + print(" POST /get_config, /health, /get_broadcast, /unlock_image", flush=True) + print(" POST /diagnose, /get_stats, /reset_stats", flush=True) + print(" POST /admin_feed, /admin_users, /admin_ban, etc.", flush=True) + print("-" * 60, flush=True) + print(" Initial worker health check:", flush=True) + b1 = await _check_worker("backend1", force=True) + b2 = await _check_worker("backend2", force=True) + print(f" Backend 1: {'✅ reachable' if b1['reachable'] else '❌ ' + str(b1['last_error'])}", flush=True) + print(f" Backend 2: {'✅ reachable' if b2['reachable'] else '❌ ' + str(b2['last_error'])}", flush=True) + print("=" * 60 + "\n", flush=True) + + +@app.on_event("shutdown") +async def shutdown(): + global _http_client + if _http_client and not _http_client.is_closed: + await _http_client.aclose() # ╔══════════════════════════════════════════════════════════════╗ -# ║ [15] STARTUP ║ +# ║ [15] LAUNCH (for local testing) ║ # ╚══════════════════════════════════════════════════════════════╝ -def startup(): - print("\n" + "=" * 60) - print(" ZERO COST v3.0.0 RELAY (CPU) - STARTING") - print("=" * 60) - print(f" Storage: {STORAGE_DIR}") - print(f" Backend 1: {BACKEND1_URL}") - print(f" Backend 2: {BACKEND2_URL}") - print(f" Worker Secret: {'set' if WORKER_SECRET else 'NOT SET'}") - print(f" HF Pool: {len(POOL_85)} tokens") - print(f" Token economy: {TOKEN_DAILY_QUOTA}/user/day | normal={COST_NORMAL} | uncensored={COST_UNCENSORED}") - print(f" Safety: hard-block ALL modes; soft-score normal only") - print(f" Spy log: {SPY_MAX_ENTRIES} entries, TTL {SPY_FULL_TTL_SECONDS//60} min") - print(f" Generate conc: {GENERATE_CONCURRENCY} slots") - print(f" API conc: {API_CONCURRENCY} slots") - print(" " + "-" * 56) - print(" Initial worker health check:") - b1 = _check_worker("backend1", force=True) - b2 = _check_worker("backend2", force=True) - print(f" Backend 1: {'✅ reachable' if b1['reachable'] else '❌ ' + str(b1['last_error'])}") - print(f" Backend 2: {'✅ reachable' if b2['reachable'] else '❌ ' + str(b2['last_error'])}") - print("=" * 60 + "\n") - - -startup() - - if __name__ == "__main__": - try: - demo.queue(default_concurrency_limit=1, max_threads=500) - except TypeError: - try: - demo.queue(default_concurrency_limit=1) - except TypeError: - demo.queue() - demo.launch(server_name="0.0.0.0", server_port=7860, share=False) \ No newline at end of file + import uvicorn + uvicorn.run( + "app:app", + host="0.0.0.0", + port=7860, + workers=4, + limit_concurrency=500, + timeout_keep_alive=65, + ) \ No newline at end of file