""" Logo Extractor — Full Stack SaaS v2.0 ======================================== Fixes in this version: 1. MASTER_API_KEY secret support (admin can use it as X-API-Key directly) 2. _save_users() runs in background thread — never blocks requests 3. Usage counter update is fire-and-forget (no HF push on every extract) 4. Model cached to /tmp/logo_detector.onnx — survives across soft restarts 5. HF Hub model pull uses correct repo path 6. Admin panel fully wired — approve/reject/revoke/restore/plan-change 7. Startup model export wrapped with timeout guard 8. All auth paths unified (X-API-Key accepts both user keys AND master key) HF Space Secrets to set (Settings → Variables and secrets): ADMIN_PASSWORD — Admin panel login password MASTER_API_KEY — Super API key that always works (for your own testing) APP_SECRET_KEY — Backend-server → HF secret (Phase 2) HF_TOKEN — Your HF token (model download + users.json push) HF_REPO_ID — e.g. freebg/logo-extractor """ from __future__ import annotations import os, gc, io, time, base64, json, logging, secrets, threading, uuid from contextlib import asynccontextmanager from typing import Optional from datetime import datetime, timezone import cv2 import numpy as np from fastapi import (FastAPI, File, Header, HTTPException, UploadFile, Request) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse import onnxruntime as ort from PIL import Image # ─── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") log = logging.getLogger("logo-ex") # ─── Config ───────────────────────────────────────────────────────────────── MODEL_PATH = os.getenv("MODEL_PATH", "/tmp/logo_detector.onnx") MODEL_URL = os.getenv("MODEL_URL", "") APP_SECRET_KEY = os.getenv("APP_SECRET_KEY", "") # backend→HF MASTER_API_KEY = os.getenv("MASTER_API_KEY", "") # always-valid super key ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "changeme") HF_TOKEN = os.getenv("HF_TOKEN", "") HF_REPO_ID = os.getenv("HF_REPO_ID", "") # e.g. freebg/logo-extractor CONF_THRESHOLD = float(os.getenv("CONF_THRESHOLD", "0.35")) IOU_THRESHOLD = float(os.getenv("IOU_THRESHOLD", "0.45")) CROP_PADDING = int(os.getenv("CROP_PADDING", "10")) MAX_IMAGE_SIZE = int(os.getenv("MAX_IMAGE_SIZE", "1920")) INPUT_SIZE = 640 USERS_FILE = "/tmp/users.json" # /tmp survives soft restarts, no perm issues # ─── Global state ──────────────────────────────────────────────────────────── ort_session: Optional[ort.InferenceSession] = None _admin_tokens: set[str] = set() _stats: dict = { "total_requests": 0, "total_logos": 0, "errors": 0, "start_time": datetime.now(timezone.utc).isoformat() } # ════════════════════════════════════════════════════════════════════════════ # USER DATABASE (/tmp/users.json ←→ HF repo) # ════════════════════════════════════════════════════════════════════════════ def _empty_db() -> dict: return {"pending": [], "active": [], "revoked": []} def _load_users() -> dict: if os.path.exists(USERS_FILE): try: with open(USERS_FILE) as f: return json.load(f) except Exception: pass return _empty_db() def _push_to_repo(path: str) -> None: """Push users.json to HF repo — called in background thread.""" if not (HF_TOKEN and HF_REPO_ID): return try: from huggingface_hub import HfApi HfApi(token=HF_TOKEN).upload_file( path_or_fileobj=path, path_in_repo="users.json", repo_id=HF_REPO_ID, repo_type="space", commit_message="auto: update users.json", ) log.info("users.json pushed to repo") except Exception as e: log.warning("Repo push failed (local OK): %s", e) def _save_users(db: dict) -> None: """Save locally then push to HF repo in background (non-blocking).""" with open(USERS_FILE, "w") as f: json.dump(db, f, indent=2) t = threading.Thread(target=_push_to_repo, args=(USERS_FILE,), daemon=True) t.start() def _pull_from_repo() -> None: """Pull users.json from HF repo at startup.""" if not (HF_TOKEN and HF_REPO_ID): log.warning("HF_TOKEN/HF_REPO_ID not set — using local users.json only") return try: from huggingface_hub import hf_hub_download path = hf_hub_download( repo_id=HF_REPO_ID, filename="users.json", repo_type="space", token=HF_TOKEN, local_dir="/tmp", local_dir_use_symlinks=False, ) # hf_hub_download may save as /tmp/users.json already if path != USERS_FILE and os.path.exists(path): import shutil; shutil.copy(path, USERS_FILE) log.info("users.json pulled from repo (%d bytes)", os.path.getsize(USERS_FILE)) except Exception as e: log.warning("Pull failed (first run?): %s", e) def _gen_key() -> str: return f"logo-{secrets.token_hex(4)}-{secrets.token_hex(4)}-{secrets.token_hex(3)}" # ════════════════════════════════════════════════════════════════════════════ # AUTH # ════════════════════════════════════════════════════════════════════════════ def verify_api_key(key: str) -> Optional[dict]: """ Returns user record if valid active key. Also accepts MASTER_API_KEY (returns None as user = master access). Raises 401 otherwise. """ if not key: raise HTTPException(401, "X-API-Key header required") # Master key — always valid if MASTER_API_KEY and key == MASTER_API_KEY: return None # None = master/admin caller # User keys db = _load_users() for u in db.get("active", []): if u.get("api_key") == key: return u raise HTTPException(401, "Invalid or inactive API key") def verify_backend_secret(secret: str) -> None: if not APP_SECRET_KEY: raise HTTPException(503, "APP_SECRET_KEY not configured") if secret != APP_SECRET_KEY: raise HTTPException(401, "Invalid backend secret") def _check_admin(token: str) -> None: if not token or token not in _admin_tokens: raise HTTPException(401, "Invalid admin session — please login again") # ════════════════════════════════════════════════════════════════════════════ # MODEL # ════════════════════════════════════════════════════════════════════════════ def ensure_model() -> None: if os.path.exists(MODEL_PATH): log.info("Model found: %s (%.1f MB)", MODEL_PATH, os.path.getsize(MODEL_PATH) / 1e6) return # Option 1: direct URL if MODEL_URL: import urllib.request log.info("Downloading model from MODEL_URL …") urllib.request.urlretrieve(MODEL_URL, MODEL_PATH) log.info("Downloaded → %s", MODEL_PATH) return # Option 2: openfoodfacts HF model (logo-specific, best quality) log.info("Trying openfoodfacts/universal-logo-detector …") try: import shutil import onnx # noqa — must be importable before ultralytics export from ultralytics import YOLO m = YOLO("hf_hub:openfoodfacts/universal-logo-detector") exported = m.export(format="onnx", imgsz=INPUT_SIZE, opset=17, simplify=True, dynamic=False) found = None for c in [str(exported) if exported else "", "universal-logo-detector.onnx", "best.onnx"]: if c and os.path.exists(c): found = c; break if not found: for root, _, files in os.walk("."): for fn in files: if fn.endswith(".onnx"): found = os.path.join(root, fn); break if found: shutil.move(found, MODEL_PATH) log.info("Logo model ready → %s (%.1f MB)", MODEL_PATH, os.path.getsize(MODEL_PATH) / 1e6) return except Exception as e: log.warning("HF logo model failed: %s", e) # Option 3: YOLOv8n fallback (general detector — logo class not present) log.warning("Falling back to YOLOv8n (general detector) …") import shutil import onnx # noqa from ultralytics import YOLO m = YOLO("yolov8n.pt") m.export(format="onnx", imgsz=INPUT_SIZE, opset=17, simplify=True, dynamic=False) for c in ["yolov8n.onnx", "yolov8n/yolov8n.onnx"]: if os.path.exists(c): shutil.move(c, MODEL_PATH) log.info("Fallback model saved → %s", MODEL_PATH) return def load_ort() -> ort.InferenceSession: opts = ort.SessionOptions() opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL opts.intra_op_num_threads = 2 opts.inter_op_num_threads = 1 sess = ort.InferenceSession(MODEL_PATH, sess_options=opts, providers=["CPUExecutionProvider"]) log.info("ONNX session ready | in=%s out=%s", sess.get_inputs()[0].shape, sess.get_outputs()[0].shape) return sess # ════════════════════════════════════════════════════════════════════════════ # IMAGE PROCESSING # ════════════════════════════════════════════════════════════════════════════ def preprocess(img: np.ndarray): h, w = img.shape[:2] scale = min(INPUT_SIZE / w, INPUT_SIZE / h) nw, nh = int(w * scale), int(h * scale) resized = cv2.resize(img, (nw, nh)) px = (INPUT_SIZE - nw) // 2 py = (INPUT_SIZE - nh) // 2 padded = cv2.copyMakeBorder( resized, py, INPUT_SIZE - nh - py, px, INPUT_SIZE - nw - px, cv2.BORDER_CONSTANT, value=(114, 114, 114)) rgb = cv2.cvtColor(padded, cv2.COLOR_BGR2RGB) t = rgb.astype(np.float32) / 255.0 return np.expand_dims(np.transpose(t, (2, 0, 1)), 0), scale, px, py def _nms(boxes: np.ndarray, scores: np.ndarray, thresh: float) -> list[int]: if not len(boxes): return [] x1, y1, x2, y2 = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3] areas = (x2 - x1) * (y2 - y1) order = scores.argsort()[::-1] keep: list[int] = [] while order.size: i = order[0]; keep.append(int(i)) xx1 = np.maximum(x1[i], x1[order[1:]]) yy1 = np.maximum(y1[i], y1[order[1:]]) xx2 = np.minimum(x2[i], x2[order[1:]]) yy2 = np.minimum(y2[i], y2[order[1:]]) inter = np.maximum(0, xx2 - xx1) * np.maximum(0, yy2 - yy1) iou = inter / (areas[i] + areas[order[1:]] - inter + 1e-6) order = order[1:][iou <= thresh] return keep def postprocess(raw: np.ndarray, ow: int, oh: int, scale: float, px: int, py: int) -> list[dict]: pred = raw[0].T cx, cy, bw, bh = pred[:, 0], pred[:, 1], pred[:, 2], pred[:, 3] conf = pred[:, 4:].max(1) if pred.shape[1] > 5 else pred[:, 4] mask = conf >= CONF_THRESHOLD if not mask.any(): return [] cx, cy, bw, bh, conf = (cx[mask], cy[mask], bw[mask], bh[mask], conf[mask]) x1, y1 = cx - bw / 2, cy - bh / 2 x2, y2 = cx + bw / 2, cy + bh / 2 boxes = np.stack([x1, y1, x2, y2], 1) keep = _nms(boxes, conf, IOU_THRESHOLD) out = [] for box, score in zip(boxes[keep], conf[keep]): ox1 = max(0, int((box[0] - px) / scale)) oy1 = max(0, int((box[1] - py) / scale)) ox2 = min(ow, int((box[2] - px) / scale)) oy2 = min(oh, int((box[3] - py) / scale)) if ox2 > ox1 and oy2 > oy1: out.append({"x1": ox1, "y1": oy1, "x2": ox2, "y2": oy2, "confidence": round(float(score), 4)}) return out def crop_b64(img: np.ndarray, box: dict, fmt: str = "PNG") -> tuple: h, w = img.shape[:2] x1 = max(0, box["x1"] - CROP_PADDING) y1 = max(0, box["y1"] - CROP_PADDING) x2 = min(w, box["x2"] + CROP_PADDING) y2 = min(h, box["y2"] + CROP_PADDING) crop = img[y1:y2, x1:x2] if fmt == "TRANSPARENT": rgb = cv2.cvtColor(crop, cv2.COLOR_BGR2RGB) mask = np.all(rgb > 240, 2).astype(np.uint8) * 255 rgba = np.dstack([rgb, 255 - mask]) pil = Image.fromarray(rgba, "RGBA") else: pil = Image.fromarray(cv2.cvtColor(crop, cv2.COLOR_BGR2RGB)) buf = io.BytesIO() pil.save(buf, format="PNG" if fmt != "JPEG" else "JPEG") return base64.b64encode(buf.getvalue()).decode(), x2 - x1, y2 - y1 # ════════════════════════════════════════════════════════════════════════════ # HTML UI # ════════════════════════════════════════════════════════════════════════════ HTML = r""" Logo Extractor API
checking…
🖼
Drop or click to upload
JPEG · PNG · WEBP · max 15 MB
Result
✂
Upload image & click
Extract Logos
Request an API Key
Submit your details. Admin approves and your key is auto-generated — usually within 24 h.
Check Request Status
Plans
PlanLimitPrice
Free50/day$0
Starter500/day$9/mo
ProUnlimited$29/mo
Logo Extractor API
Extract logos from any image. Returns cropped logo images as base64 JSON. Supports PNG, JPEG, and transparent background.
Endpoint
POST /extract-logo
X-API-Key: logo-xxxx-xxxx-xxxx
Content-Type: multipart/form-data

Body:
  image          file    JPEG/PNG/WEBP, max 15 MB
  output_format  string  PNG | JPEG | TRANSPARENT
  max_logos      int     default 20
Response
{
  "success": true,
  "count": 2,
  "elapsed_ms": 280.4,
  "logos": [{
    "base64": "iVBORw0KGgo...",
    "mime_type": "image/png",
    "width": 180, "height": 90,
    "confidence": 0.912,
    "bbox": {"x1":45,"y1":12,"x2":225,"y2":102}
  }]
}
Python
import requests, base64
r = requests.post(
    "https://YOUR-SPACE.hf.space/extract-logo",
    headers={"X-API-Key": "logo-xxxx-xxxx-xxxx"},
    files={"image": open("img.jpg","rb")},
    params={"output_format":"PNG"}
)
for i,l in enumerate(r.json()["logos"]):
    open(f"logo_{i}.png","wb").write(base64.b64decode(l["base64"]))
cURL
curl -X POST https://YOUR-SPACE.hf.space/extract-logo \
  -H "X-API-Key: logo-xxxx-xxxx-xxxx" \
  -F "image=@logo.jpg"
🔐 Admin Login

✅ Approve Request

✏ Change Plan

""" # ════════════════════════════════════════════════════════════════════════════ # FASTAPI APP # ════════════════════════════════════════════════════════════════════════════ @asynccontextmanager async def lifespan(app: FastAPI): global ort_session log.info("=== Startup ===") _pull_from_repo() ensure_model() try: ort_session = load_ort() except Exception as e: log.error("ORT load failed: %s", e) yield if ort_session: del ort_session gc.collect() app = FastAPI(title="Logo Extractor", version="2.0.0", lifespan=lifespan, docs_url="/api/docs", redoc_url=None) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # ─── UI & Health ───────────────────────────────────────────────────────────── @app.get("/", response_class=HTMLResponse) async def root(): return HTML @app.get("/health") async def health(): return {"status": "ok" if ort_session else "degraded", "model_loaded": ort_session is not None, "version": "2.0.0"} # ─── Public: register & status ─────────────────────────────────────────────── @app.post("/api/register") async def register(request: Request): b = await request.json() name = b.get("name", "").strip() email = b.get("email", "").strip().lower() plan = b.get("plan", "free") msg = b.get("message", "").strip() if not name or not email: raise HTTPException(400, "Name and email required") if "@" not in email: raise HTTPException(400, "Valid email required") db = _load_users() all_emails = ([u["email"] for u in db["pending"]] + [u["email"] for u in db["active"]] + [u["email"] for u in db["revoked"]]) if email in all_emails: raise HTTPException(409, "A request with this email already exists") db["pending"].append({ "id": str(uuid.uuid4()), "email": email, "name": name, "plan": plan, "message": msg, "requested_at": datetime.now(timezone.utc).isoformat(), }) _save_users(db) log.info("New request: %s (%s)", email, plan) return {"message": f"Request received for {email}. You'll be notified when approved."} @app.get("/api/status") async def status_check(email: str): email = email.strip().lower() db = _load_users() for u in db.get("active", []): if u["email"] == email: return {"status": "active", "plan": u["plan"], "api_key": u["api_key"]} for u in db.get("pending", []): if u["email"] == email: return {"status": "pending"} for u in db.get("revoked", []): if u["email"] == email: return {"status": "revoked"} return {"status": "not_found"} @app.post("/api/verify-key") async def verify_key_route(request: Request): b = await request.json() key = b.get("key", "") db = _load_users() is_master = bool(MASTER_API_KEY and key == MASTER_API_KEY) is_user = any(u.get("api_key") == key for u in db.get("active", [])) return {"valid": is_master or is_user, "type": "master" if is_master else ("user" if is_user else "invalid")} # ─── Admin ──────────────────────────────────────────────────────────────────── @app.post("/admin/login") async def admin_login(request: Request): b = await request.json() if b.get("password") != ADMIN_PASSWORD: raise HTTPException(401, "Wrong password") token = secrets.token_hex(24) _admin_tokens.add(token) return {"token": token} @app.get("/admin/data") async def admin_data(x_admin_token: str = Header(...)): _check_admin(x_admin_token) db = _load_users() inp = out = "N/A" if ort_session: inp = str(ort_session.get_inputs()[0].shape) out = str(ort_session.get_outputs()[0].shape) return { "stats": _stats, "pending": db.get("pending", []), "active": db.get("active", []), "revoked": db.get("revoked", []), "model": {"loaded": ort_session is not None, "input_shape": inp, "output_shape": out}, } @app.post("/admin/approve") async def admin_approve(request: Request, x_admin_token: str = Header(...)): _check_admin(x_admin_token) b = await request.json() uid = b.get("user_id", "") plan = b.get("plan", "free") db = _load_users() user = next((u for u in db["pending"] if u["id"] == uid), None) if not user: raise HTTPException(404, "Pending user not found") api_key = _gen_key() db["pending"] = [u for u in db["pending"] if u["id"] != uid] user.update({ "api_key": api_key, "plan": plan, "approved_at": datetime.now(timezone.utc).isoformat(), "usage": 0, "last_used": None, }) db["active"].append(user) _save_users(db) log.info("Approved %s → %s…", user["email"], api_key[:16]) return {"message": f"Approved {user['email']}", "api_key": api_key, "email": user["email"]} @app.post("/admin/reject") async def admin_reject(request: Request, x_admin_token: str = Header(...)): _check_admin(x_admin_token) uid = (await request.json()).get("user_id", "") db = _load_users() db["pending"] = [u for u in db["pending"] if u["id"] != uid] _save_users(db) return {"message": "Rejected"} @app.post("/admin/revoke") async def admin_revoke(request: Request, x_admin_token: str = Header(...)): _check_admin(x_admin_token) uid = (await request.json()).get("user_id", "") db = _load_users() user = next((u for u in db["active"] if u["id"] == uid), None) if not user: raise HTTPException(404, "User not found") db["active"] = [u for u in db["active"] if u["id"] != uid] db["revoked"].append(user) _save_users(db) log.info("Revoked key for %s", user["email"]) return {"message": f"Revoked {user['email']}"} @app.post("/admin/reactivate") async def admin_reactivate(request: Request, x_admin_token: str = Header(...)): _check_admin(x_admin_token) uid = (await request.json()).get("user_id", "") db = _load_users() user = next((u for u in db["revoked"] if u["id"] == uid), None) if not user: raise HTTPException(404, "Revoked user not found") db["revoked"] = [u for u in db["revoked"] if u["id"] != uid] db["active"].append(user) _save_users(db) return {"message": f"Restored {user['email']}"} @app.post("/admin/change-plan") async def admin_change_plan(request: Request, x_admin_token: str = Header(...)): _check_admin(x_admin_token) b = await request.json() uid = b.get("user_id", "") plan = b.get("plan", "free") if plan not in ("free", "starter", "pro"): raise HTTPException(400, "Plan must be free/starter/pro") db = _load_users() for u in db["active"]: if u["id"] == uid: u["plan"] = plan _save_users(db) return {"message": f"Plan updated to {plan}"} raise HTTPException(404, "User not found") # ─── Main extract endpoint ──────────────────────────────────────────────────── @app.post("/extract-logo") async def extract_logo( image: UploadFile = File(...), output_format: str = "PNG", max_logos: int = 20, x_api_key: Optional[str] = Header(None, alias="X-API-Key"), x_custom_secret: Optional[str] = Header(None, alias="X-Custom-Secret"), ): t0 = time.perf_counter() # ── Auth ────────────────────────────────────────────────────────────── user = None if x_api_key: user = verify_api_key(x_api_key) # raises 401 if invalid elif x_custom_secret: verify_backend_secret(x_custom_secret) else: raise HTTPException(401, "Provide X-API-Key (user key or master key) or X-Custom-Secret") if ort_session is None: _stats["errors"] += 1 raise HTTPException(503, "Model not loaded — check Space logs") # ── Read image ──────────────────────────────────────────────────────── raw = await image.read() if len(raw) > 15 * 1024 * 1024: raise HTTPException(413, "Image too large — max 15 MB") nparr = np.frombuffer(raw, np.uint8) img = cv2.imdecode(nparr, cv2.IMREAD_COLOR) if img is None: raise HTTPException(400, "Cannot decode image") oh, ow = img.shape[:2] if max(ow, oh) > MAX_IMAGE_SIZE: r = MAX_IMAGE_SIZE / max(ow, oh) img = cv2.resize(img, None, fx=r, fy=r) oh, ow = img.shape[:2] # ── Inference ───────────────────────────────────────────────────────── tensor, scale, px, py = preprocess(img) inp_name = ort_session.get_inputs()[0].name raw_out = ort_session.run(None, {inp_name: tensor}) dets = postprocess(raw_out[0], ow, oh, scale, px, py)[:max_logos] # ── Crop & encode ───────────────────────────────────────────────────── fmt = output_format.upper() result_logos = [] for det in dets: b64, cw, ch = crop_b64(img, det, fmt) mime = "image/jpeg" if fmt == "JPEG" else "image/png" result_logos.append({ "base64": b64, "mime_type": mime, "width": cw, "height": ch, "confidence": det["confidence"], "bbox": {k: det[k] for k in ("x1", "y1", "x2", "y2")}, }) # ── Update usage counter (background, non-blocking) ─────────────────── if x_api_key and user is not None: # user=None means master key api_key_snap = x_api_key def _update_usage(): try: db = _load_users() for u in db["active"]: if u.get("api_key") == api_key_snap: u["usage"] = u.get("usage", 0) + 1 u["last_used"] = datetime.now(timezone.utc).isoformat() break # Save locally only — don't push to HF on every request with open(USERS_FILE, "w") as f: json.dump(db, f, indent=2) except Exception: pass threading.Thread(target=_update_usage, daemon=True).start() del img, tensor, raw_out, nparr, raw gc.collect() elapsed = round((time.perf_counter() - t0) * 1000, 1) _stats["total_requests"] += 1 _stats["total_logos"] += len(result_logos) return JSONResponse({ "success": True, "count": len(result_logos), "elapsed_ms": elapsed, "logos": result_logos, }) if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)