""" FreeBG.site — Complete SaaS Background Removal HuggingFace Space | Gradio 4.x Tabs: Tool | Admin (password protected) | API Docs | My Usage Admin: Add/Edit/Delete customers + Auto-update HF Secret """ import os, io, time, json, logging, random, string from datetime import datetime, timezone, timedelta from typing import Optional import numpy as np from PIL import Image logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s") logger = logging.getLogger("freebg") def utcnow(): return datetime.now(timezone.utc) # ── Config ──────────────────────────────────────────────────────────────────── HF_TOKEN = os.environ.get("HF_TOKEN", "") MASTER_KEY = os.environ.get("MASTER_API_KEY", "freebg-master-key-change-me") ADMIN_PASS = os.environ.get("ADMIN_PASSWORD", "freebg-admin-2026") HF_REPO_ID = os.environ.get("HF_REPO_ID", "freebg/background-remover") # ── HF Login ────────────────────────────────────────────────────────────────── if HF_TOKEN: try: from huggingface_hub import login login(token=HF_TOKEN, add_to_git_credential=False) logger.info("✅ HF login OK") except Exception as e: logger.warning(f"HF login: {e}") # ── API Keys ────────────────────────────────────────────────────────────────── API_KEYS: dict = {} _raw = os.environ.get("API_KEYS_JSON", "") if _raw: try: API_KEYS = json.loads(_raw) logger.info(f"Loaded {len(API_KEYS)} customer keys") except Exception as e: logger.warning(f"API_KEYS_JSON error: {e}") API_KEYS.setdefault(MASTER_KEY, { "plan": "master", "calls_today": 0, "reset_at": (utcnow() + timedelta(days=1)).timestamp(), "owner": "freebg.site", "created_at": utcnow().isoformat() }) PLAN_LIMITS = { "free": {"daily": 10, "models": ["fast"], "price": "$0"}, "starter": {"daily": 100, "models": ["fast", "quality"], "price": "$9/mo"}, "pro": {"daily": 500, "models": ["fast", "quality", "best"],"price": "$29/mo"}, "master": {"daily": 999999, "models": ["fast", "quality", "best"],"price": "Custom"}, } # ── HF Secret Auto-Update ───────────────────────────────────────────────────── def push_keys_to_hf_secret() -> tuple: if not HF_TOKEN: return False, "HF_TOKEN not set" keys_json = json.dumps(API_KEYS) try: from huggingface_hub import add_space_secret add_space_secret(HF_REPO_ID, "API_KEYS_JSON", keys_json, token=HF_TOKEN) logger.info("✅ HF Secret updated") return True, "✅ **Saved!** Space restarts in ~30s. New key will be active." except Exception as e1: logger.warning(f"add_space_secret failed: {e1}") try: import requests as req headers = {"Authorization": f"Bearer {HF_TOKEN}"} url = f"https://huggingface.co/api/spaces/{HF_REPO_ID}/secrets" for method in [req.put, req.post]: r = method(url, headers=headers, json={"key": "API_KEYS_JSON", "value": keys_json}, timeout=15) if r.status_code in (200, 201, 204): return True, "✅ **Saved!** Space restarts in ~30s." except Exception as e2: logger.error(f"API failed: {e2}") manual = ( "⚠️ **Auto-save failed** — HF_TOKEN needs **write** permission.\n\n" "**Fix:** Go to https://huggingface.co/settings/tokens\n" "→ Create new token → Type: **Write** → Replace HF_TOKEN secret with it.\n\n" "**Paste this JSON manually in HF Secret `API_KEYS_JSON`:**\n\n" + keys_json ) return False, manual # ── Key Generator ───────────────────────────────────────────────────────────── def gen_key(plan: str) -> str: chars = string.ascii_lowercase + string.digits r = lambda n: ''.join(random.choices(chars, k=n)) return f"freebg-{plan[:2]}-{r(8)}-{r(8)}" # ── Model Cache ─────────────────────────────────────────────────────────────── _cache: dict = {} def load_u2net(): if "u2net" in _cache: return _cache["u2net"] try: from rembg import new_session _cache["u2net"] = new_session("u2net") logger.info("✅ U2-Net ready"); return _cache["u2net"] except Exception as e: logger.error(f"U2-Net: {e}"); return None def load_birefnet(): if "birefnet" in _cache: return _cache["birefnet"] try: from transformers import AutoModelForImageSegmentation from torchvision import transforms m = AutoModelForImageSegmentation.from_pretrained("ZhengPeng7/BiRefNet", trust_remote_code=True) m = m.float(); m.eval() _cache["birefnet"] = m _cache["birefnet_tf"] = transforms.Compose([ transforms.Resize((1024,1024)), transforms.ToTensor(), transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225])]) logger.info("✅ BiRefNet ready"); return m except Exception as e: logger.error(f"BiRefNet: {e}"); return None def load_rmbg(): if "rmbg" in _cache: return _cache["rmbg"] try: from transformers import AutoModelForImageSegmentation from torchvision import transforms m = AutoModelForImageSegmentation.from_pretrained( "briaai/RMBG-2.0", trust_remote_code=True, token=HF_TOKEN or None) m = m.float(); m.eval() _cache["rmbg"] = m _cache["rmbg_tf"] = transforms.Compose([ transforms.Resize((1024,1024)), transforms.ToTensor(), transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225])]) logger.info("✅ RMBG-2.0 ready"); return m except Exception as e: logger.error(f"RMBG-2.0: {e}"); return None def _seg(key, img): import torch m = _cache[key]; tf = _cache[key+"_tf"]; orig = img.size inp = tf(img.convert("RGB")).unsqueeze(0).float() with torch.no_grad(): out = m(inp) pred = (out[-1] if isinstance(out,(list,tuple)) else out).sigmoid() mask = Image.fromarray((pred[0].squeeze().cpu().numpy()*255).astype(np.uint8)).resize(orig, Image.LANCZOS) r = img.convert("RGBA"); r.putalpha(mask); return r def infer_fast(img): from rembg import remove if not load_u2net(): raise RuntimeError("U2-Net not available") buf = io.BytesIO(); img.save(buf,"PNG"); buf.seek(0) return Image.open(io.BytesIO(remove(buf.read(), session=_cache["u2net"]))).convert("RGBA") def infer_quality(img): if not load_birefnet(): raise RuntimeError("BiRefNet not available") return _seg("birefnet", img) def infer_best(img): if not load_rmbg(): raise RuntimeError("BRIA RMBG-2.0 not available — check HF_TOKEN") return _seg("rmbg", img) MODEL_FN = {"fast": infer_fast, "quality": infer_quality, "best": infer_best} MODEL_INFO = { "fast": {"name":"U2-Net", "desc":"Fastest (~1-5s)"}, "quality": {"name":"BiRefNet", "desc":"High quality (~2-4min CPU)"}, "best": {"name":"BRIA RMBG-2.0", "desc":"Best quality (~2-4min CPU)"}, } # ── Auth ────────────────────────────────────────────────────────────────────── def validate_key(api_key, model): if not api_key: return False, "API key required" if api_key not in API_KEYS: return False, "Invalid API key" kd = API_KEYS[api_key]; plan = kd.get("plan","free") cfg = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"]) if utcnow().timestamp() > kd.get("reset_at",0): kd["calls_today"] = 0 kd["reset_at"] = (utcnow() + timedelta(days=1)).timestamp() if model not in cfg["models"]: return False, f"Model '{model}' not in {plan} plan. Upgrade at freebg.site" if kd.get("calls_today",0) >= cfg["daily"]: return False, f"Daily limit {cfg['daily']} reached." return True, "" def inc_usage(k): if k in API_KEYS: API_KEYS[k]["calls_today"] = API_KEYS[k].get("calls_today",0)+1 def get_usage(k): if k not in API_KEYS: return {"error":"Invalid key"} kd = API_KEYS[k]; plan = kd.get("plan","free") cfg = PLAN_LIMITS.get(plan, PLAN_LIMITS["free"]) return {"plan":plan,"calls_today":kd.get("calls_today",0), "daily_limit":cfg["daily"],"available_models":cfg["models"],"owner":kd.get("owner","")} def run_removal(api_key, img_data, model="fast", max_size=0): t0 = time.time() ok, err = validate_key(api_key, model) if not ok: return None, {}, err try: img = (Image.fromarray(img_data) if isinstance(img_data, np.ndarray) else img_data if isinstance(img_data, Image.Image) else Image.open(img_data)).convert("RGB") ow, oh = img.size if max_size>0 and max(ow,oh)>max_size: r=max_size/max(ow,oh); img=img.resize((int(ow*r),int(oh*r)),Image.LANCZOS) result = MODEL_FN[model](img) inc_usage(api_key) elapsed = round(time.time()-t0, 2) meta = {"model":MODEL_INFO[model]["name"],"model_key":model, "original_size":f"{ow}x{oh}","output_size":f"{result.size[0]}x{result.size[1]}", "processing_time_sec":elapsed,"calls_today":API_KEYS[api_key]["calls_today"]} logger.info(f"✅ {model} {ow}x{oh} {elapsed}s key=...{api_key[-6:]}") return result, meta, "" except Exception as e: logger.error(f"run_removal: {e}", exc_info=True) return None, {}, str(e) # ── Gradio Callbacks ────────────────────────────────────────────────────────── import gradio as gr _CHOICES = ["🚀 Fast (U2-Net)","⚡ High Quality (BiRefNet)","🏆 Best (BRIA RMBG-2.0)"] _CMAP = {"🚀 Fast (U2-Net)":"fast","⚡ High Quality (BiRefNet)":"quality","🏆 Best (BRIA RMBG-2.0)":"best"} def ui_remove(api_key, image, choice): m = _CMAP.get(choice,"fast") result, meta, err = run_removal(api_key, image, m) if err: return None, f"❌ **Error:** {err}" return result, (f"✅ **Done** | `{meta['model']}` | `{meta['processing_time_sec']}s` | " f"`{meta['original_size']}` → `{meta['output_size']}` | calls: `{meta['calls_today']}`") def ui_usage(api_key): info = get_usage(api_key) if "error" in info: return f"❌ {info['error']}" return (f"**Plan:** {info['plan'].upper()} \n" f"**Calls today:** {info['calls_today']} / {info['daily_limit']} \n" f"**Models:** {', '.join(info['available_models'])} \n" f"**Owner:** {info.get('owner','—')}") # ── Admin Functions ─────────────────────────────────────────────────────────── def admin_login(password): if password == ADMIN_PASS: return (gr.update(visible=False), gr.update(visible=True), "", admin_list_customers(), admin_stats()) return gr.update(visible=True), gr.update(visible=False), "❌ Wrong password", [], "" def admin_list_customers(): rows = [] for key, d in API_KEYS.items(): if key == MASTER_KEY: continue rows.append([ key, d.get("owner",""), d.get("plan","free"), d.get("calls_today",0), d.get("limit", PLAN_LIMITS.get(d.get("plan","free"),{}).get("daily",10)), d.get("created_at","")[:10] if d.get("created_at") else "" ]) return rows def admin_add_customer(email, plan, custom_limit): if not email or "@" not in email: return "❌ Valid email required", admin_list_customers(), "" if plan not in PLAN_LIMITS: return "❌ Invalid plan", admin_list_customers(), "" key = gen_key(plan) limit = int(custom_limit) if str(custom_limit).strip().isdigit() else PLAN_LIMITS[plan]["daily"] API_KEYS[key] = { "plan": plan, "owner": email, "calls_today": 0, "reset_at": 0, "limit": limit, "models": PLAN_LIMITS[plan]["models"], "created_at": utcnow().isoformat() } ok, msg = push_keys_to_hf_secret() status = f"✅ Key created for **{email}** ({plan} plan)\n\n🔑 `{key}`\n\n{msg}" return status, admin_list_customers(), key def admin_delete_customer(key): if not key or key.strip() not in API_KEYS: return "❌ Key not found", admin_list_customers() key = key.strip() if key == MASTER_KEY: return "❌ Cannot delete master key", admin_list_customers() owner = API_KEYS[key].get("owner","") del API_KEYS[key] ok, msg = push_keys_to_hf_secret() return f"✅ Deleted key for **{owner}**\n\n{msg}", admin_list_customers() def admin_upgrade_plan(key, new_plan): if not key or key.strip() not in API_KEYS: return "❌ Key not found", admin_list_customers() if new_plan not in PLAN_LIMITS: return "❌ Invalid plan", admin_list_customers() key = key.strip() API_KEYS[key]["plan"] = new_plan API_KEYS[key]["limit"] = PLAN_LIMITS[new_plan]["daily"] API_KEYS[key]["models"] = PLAN_LIMITS[new_plan]["models"] owner = API_KEYS[key].get("owner","") ok, msg = push_keys_to_hf_secret() return f"✅ **{owner}** upgraded to **{new_plan}**\n\n{msg}", admin_list_customers() def admin_refresh(): return admin_list_customers() def admin_export_json(): return json.dumps(API_KEYS, indent=2) def admin_stats(): vals = list(API_KEYS.values()) total = len([v for v in vals if v.get("owner") != "freebg.site"]) free_c = len([v for v in vals if v.get("plan")=="free"]) start = len([v for v in vals if v.get("plan")=="starter"]) pro_c = len([v for v in vals if v.get("plan") in ("pro","master") and v.get("owner")!="freebg.site"]) calls = sum(v.get("calls_today",0) for v in vals) rev = start*9 + pro_c*29 return (f"👥 **Total Customers:** {total} \n" f"🆓 **Free:** {free_c} \n" f"⭐ **Starter ($9):** {start} \n" f"🏆 **Pro ($29):** {pro_c} \n" f"📊 **Total Calls Today:** {calls} \n" f"💰 **Est. Monthly Revenue:** ${rev}") # ═══════════════════════════════════════════════════════════════════════════════ # ── NEW FEATURE FUNCTIONS (existing code untouched above) ───────────────────── # ═══════════════════════════════════════════════════════════════════════════════ import zipfile import tempfile # ── Session History Store ───────────────────────────────────────────────────── _session_history: list = [] # list of dicts: {thumb, result, name, time, model, size} def _pil_to_thumb(img: Image.Image, size=(120, 120)) -> Image.Image: """Create a small thumbnail for history display.""" thumb = img.copy() thumb.thumbnail(size, Image.LANCZOS) return thumb def _apply_background(result_rgba: Image.Image, bg_choice: str, bg_color: str, bg_image=None) -> Image.Image: """ Apply a background to a transparent RGBA image. bg_choice: 'transparent' | 'color' | 'custom_image' bg_color: hex string like '#ffffff' bg_image: PIL Image or None """ if bg_choice == "transparent": return result_rgba w, h = result_rgba.size if bg_choice == "color": try: from PIL import ImageColor rgb = ImageColor.getrgb(bg_color) except Exception: rgb = (255, 255, 255) canvas = Image.new("RGBA", (w, h), rgb + (255,)) canvas.paste(result_rgba, mask=result_rgba.split()[3]) return canvas.convert("RGB") if bg_choice == "custom_image" and bg_image is not None: bg = (Image.fromarray(bg_image) if isinstance(bg_image, np.ndarray) else bg_image).convert("RGBA") bg = bg.resize((w, h), Image.LANCZOS) bg.paste(result_rgba, mask=result_rgba.split()[3]) return bg.convert("RGB") return result_rgba # ── Batch Processing ────────────────────────────────────────────────────────── def ui_batch_remove(api_key, images, choice, bg_choice, bg_color, bg_image, progress=gr.Progress()): """ Process multiple images in a queue. Returns: gallery of results, status markdown, zip_file_path """ if not images: return [], "⚠️ No images uploaded.", None model = _CMAP.get(choice, "fast") results_gallery = [] status_lines = [] processed_pils = [] progress(0, desc="Starting batch…") for i, img_input in enumerate(images): frac = i / len(images) fname = f"image_{i+1}" # img_input from gr.File is a filepath string when type="filepath" try: if isinstance(img_input, str): pil_in = Image.open(img_input).convert("RGB") fname = os.path.basename(img_input) elif isinstance(img_input, np.ndarray): pil_in = Image.fromarray(img_input).convert("RGB") elif isinstance(img_input, Image.Image): pil_in = img_input.convert("RGB") else: status_lines.append(f"❌ `{fname}` — unsupported format") continue except Exception as e: status_lines.append(f"❌ `{fname}` — read error: {e}") progress((i + 1) / len(images), desc=f"Error on {fname}") continue progress(frac, desc=f"Processing {i+1}/{len(images)}: {fname}") result, meta, err = run_removal(api_key, pil_in, model) if err: status_lines.append(f"❌ `{fname}` — {err}") progress((i + 1) / len(images), desc=f"Failed: {fname}") continue # Apply background final = _apply_background(result, bg_choice, bg_color, bg_image) # Save to history _session_history.append({ "name": fname, "time": utcnow().strftime("%H:%M:%S"), "model": meta["model"], "size": meta["output_size"], "result": final, }) results_gallery.append(final) processed_pils.append((fname, final)) status_lines.append(f"✅ `{fname}` — {meta['processing_time_sec']}s | {meta['output_size']}") progress((i + 1) / len(images), desc=f"Done: {fname}") # Build ZIP zip_path = None if processed_pils: tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".zip") with zipfile.ZipFile(tmp.name, "w", zipfile.ZIP_DEFLATED) as zf: for fname, pil in processed_pils: buf = io.BytesIO() ext = "png" if bg_choice == "transparent" else "jpg" fmt = "PNG" if bg_choice == "transparent" else "JPEG" pil.save(buf, fmt, optimize=True) buf.seek(0) out_name = os.path.splitext(fname)[0] + f"_nobg.{ext}" zf.writestr(out_name, buf.read()) zip_path = tmp.name total = len(processed_pils) failed = len(images) - total summary = (f"### Batch Complete\n" f"✅ **{total}** processed | ❌ **{failed}** failed\n\n" + "\n".join(status_lines)) return results_gallery, summary, zip_path # ── History ─────────────────────────────────────────────────────────────────── def ui_get_history(): """Return history gallery and summary.""" if not _session_history: return [], "No images processed yet this session." gallery = [item["result"] for item in _session_history] lines = [f"| `{it['name']}` | {it['model']} | {it['size']} | {it['time']} |" for it in _session_history] table = ("| File | Model | Size | Time |\n" "|------|-------|------|------|\n" + "\n".join(lines)) return gallery, f"### Session History ({len(_session_history)} images)\n\n" + table def ui_clear_history(): _session_history.clear() return [], "🗑️ History cleared." # ── Single image with background replacement ────────────────────────────────── def ui_remove_with_bg(api_key, image, choice, bg_choice, bg_color, bg_image): """Enhanced single removal with background replacement support.""" m = _CMAP.get(choice, "fast") result, meta, err = run_removal(api_key, image, m) if err: return None, f"❌ **Error:** {err}" final = _apply_background(result, bg_choice, bg_color, bg_image) # Save to history _session_history.append({ "name": "single_image", "time": utcnow().strftime("%H:%M:%S"), "model": meta["model"], "size": meta["output_size"], "result": final, }) return final, (f"✅ **Done** | `{meta['model']}` | `{meta['processing_time_sec']}s` | " f"`{meta['original_size']}` → `{meta['output_size']}` | calls: `{meta['calls_today']}`") # ═══════════════════════════════════════════════════════════════════════════════ # ── Gradio UI ───────────────────────────────────────────────────────────────── # ═══════════════════════════════════════════════════════════════════════════════ with gr.Blocks(title="FreeBG — Background Removal API") as demo: gr.HTML("""
3 AI Models · API Key Auth · freebg.site