""" FastAPI Service for SanitizeAI - AI Watermark Remover & Metadata Sanitizer. Provides secured, high-performance REST API endpoints for external automations (n8n, Make, custom scripts), the SanitizeAI Web Frontend, and the dedicated Admin Control Panel. """ from __future__ import annotations import io import os import sys import json import time import smtplib import secrets import tempfile from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Literal, Optional, Dict, Any, List from fastapi import ( FastAPI, File, Form, Header, HTTPException, Security, UploadFile, status, Body, ) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response, FileResponse from fastapi.security import APIKeyHeader from pydantic import BaseModel # Ensure local directory is on path BASE_DIR = Path(__file__).parent.resolve() if str(BASE_DIR) not in sys.path: sys.path.insert(0, str(BASE_DIR)) from engine import ( CleanMode, clean_image, clean_text_content, clean_video, sanitize_video_mp4, get_engine_status, normalize_clean_mode, add_sanitizeai_watermark, FFMPEG_PATH, ) # Persistent configuration file path CONFIG_FILE = BASE_DIR / "app_config.json" DEFAULT_DEV_API_KEY = "dev-secret-key-12345" # Global runtime state class AppState: def __init__(self): self.master_api_key: str = os.getenv("SANITIZER_API_KEY", DEFAULT_DEV_API_KEY) self.agency_api_keys: Dict[str, Dict[str, Any]] = {} # {api_key: {user_email, expires_at, created_at}} self.smtp_config: Dict[str, Any] = { "smtp_host": "smtp.gmail.com", "smtp_port": 465, "sender_email": "vibeautomationn@gmail.com", "app_password": "", "sender_name": "SanitizeAI Security", "is_configured": False, } self.stats = { "total_clean_requests": 0, "images_cleaned": 0, "videos_cleaned": 0, "texts_cleaned": 0, "bytes_processed": 0, "active_users_count": 0, "recent_traffic_log": [], } self.users_db: Dict[str, Dict[str, Any]] = {} self.load_config() def load_config(self): if CONFIG_FILE.exists(): try: data = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) self.master_api_key = data.get("master_api_key", self.master_api_key) self.smtp_config.update(data.get("smtp_config", {})) self.agency_api_keys = data.get("agency_api_keys", {}) self.users_db = data.get("users_db", {}) loaded_stats = data.get("stats", {}) self.stats.update(loaded_stats) except Exception as e: print(f"[WARN] Failed to load config: {e}") def save_config(self): try: data = { "master_api_key": self.master_api_key, "smtp_config": self.smtp_config, "agency_api_keys": self.agency_api_keys, "users_db": self.users_db, "stats": self.stats, } CONFIG_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8") except Exception as e: print(f"[WARN] Failed to save config: {e}") def record_traffic(self, action: str, format_type: str, file_size_bytes: int, user_email: str = "guest", plan: str = "free"): self.stats["total_clean_requests"] += 1 self.stats["bytes_processed"] += file_size_bytes if format_type == "image": self.stats["images_cleaned"] += 1 elif format_type == "video": self.stats["videos_cleaned"] += 1 elif format_type == "text": self.stats["texts_cleaned"] += 1 log_entry = { "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"), "action": action, "format": format_type, "size_kb": round(file_size_bytes / 1024, 1), "user": user_email, "plan": plan, } self.stats["recent_traffic_log"].insert(0, log_entry) if len(self.stats["recent_traffic_log"]) > 50: self.stats["recent_traffic_log"] = self.stats["recent_traffic_log"][:50] self.save_config() state = AppState() api_key_header_scheme = APIKeyHeader(name="X-API-Key", auto_error=False) def verify_api_key(api_key: str | None = Security(api_key_header_scheme)) -> str: """Verify master API Key or active Agency Plan API Key.""" if not api_key: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or missing API Key. Provide a valid 'X-API-Key' header.", headers={"WWW-Authenticate": "ApiKey"}, ) # Check Master Key if api_key == state.master_api_key: return api_key # Check Active Agency Plan User Keys if api_key in state.agency_api_keys: info = state.agency_api_keys[api_key] exp_str = info.get("expires_at") if exp_str: try: exp_dt = datetime.fromisoformat(exp_str) if datetime.now(timezone.utc) > exp_dt: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Agency API Key has expired. Please renew your subscription.", ) except ValueError: pass return api_key raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or revoked API Key. Provide a valid 'X-API-Key' header.", headers={"WWW-Authenticate": "ApiKey"}, ) # Email Delivery Helper via linked Gmail SMTP def send_smtp_email(to_email: str, subject: str, html_body: str) -> bool: cfg = state.smtp_config if not cfg.get("sender_email") or not cfg.get("app_password"): print(f"[SMTP] Email sending skipped (Gmail not configured in Admin Panel). To: {to_email}") return False try: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = f"{cfg.get('sender_name', 'SanitizeAI')} <{cfg['sender_email']}>" msg["To"] = to_email html_part = MIMEText(html_body, "html") msg.attach(html_part) port = int(cfg.get("smtp_port", 465)) host = cfg.get("smtp_host", "smtp.gmail.com") if port == 465: with smtplib.SMTP_SSL(host, port, timeout=10) as server: server.login(cfg["sender_email"], cfg["app_password"]) server.sendmail(cfg["sender_email"], to_email, msg.as_string()) else: with smtplib.SMTP(host, port, timeout=10) as server: server.starttls() server.login(cfg["sender_email"], cfg["app_password"]) server.sendmail(cfg["sender_email"], to_email, msg.as_string()) print(f"[SMTP] Successfully sent email to {to_email} (Subject: {subject})") return True except Exception as e: print(f"[SMTP ERROR] Failed to send email to {to_email}: {e}") return False # Initialize FastAPI app app = FastAPI( title="SanitizeAI API", description="SanitizeAI - Automated AI Watermark Remover, Metadata Stripper & Content Sanitizer API with Admin Control Suite.", version="0.1.29", docs_url="/docs", redoc_url="/redoc", ) # CORS middleware for web frontend & external workflows app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/", tags=["Health"]) @app.get("/health", tags=["Health"]) async def health_check(): """Health check endpoint to verify service status.""" return { "status": "healthy", "service": "SanitizeAI API", "version": "0.1.29", "engine": get_engine_status(), } @app.get("/v1/info", tags=["Info"]) async def get_info(api_key: str = Security(verify_api_key)): """Retrieve system status and supported features (Authenticated).""" return { "service": "SanitizeAI API", "version": "0.1.29", "supported_modes": [m.value for m in CleanMode], "default_mode": CleanMode.SYNTHID_DESTRUCTOR.value, "engine_status": get_engine_status(), } # ============================================================================= # ADMIN PANEL ENDPOINTS # ============================================================================= @app.get("/v1/admin/stats", tags=["Admin"]) async def get_admin_stats(api_key: str = Security(verify_api_key)): """Retrieve live server traffic, revenue, media breakdown, and engine telemetry for the Admin Panel.""" plan_distribution = {"free_trial": 0, "starter_5": 0, "creator_9": 0, "agency_29": 0, "admin": 1} for u in state.users_db.values(): p = u.get("subscription_status", "free_trial") plan_distribution[p] = plan_distribution.get(p, 0) + 1 return { "stats": state.stats, "engine": get_engine_status(), "smtp_configured": state.smtp_config.get("is_configured", False), "sender_email": state.smtp_config.get("sender_email", ""), "total_registered_users": len(state.users_db), "plan_distribution": plan_distribution, "master_api_key_masked": state.master_api_key[:4] + "••••••••" + state.master_api_key[-4:] if len(state.master_api_key) > 8 else "••••••••", } @app.get("/v1/admin/users", tags=["Admin"]) async def get_admin_users(api_key: str = Security(verify_api_key)): """List all registered users, subscription statuses, and active API keys.""" return {"users": list(state.users_db.values())} class UpdateUserPlanRequest(BaseModel): user_id: str email: str full_name: Optional[str] = "User" plan: Literal["free_trial", "starter_5", "creator_9", "agency_29", "admin"] duration_days: int = 30 send_email_notification: bool = True @app.post("/v1/admin/user/plan", tags=["Admin"]) async def update_user_plan(req: UpdateUserPlanRequest, api_key: str = Security(verify_api_key)): """Admin endpoint to manually grant, upgrade, or revoke user subscription plans.""" now = datetime.now(timezone.utc) exp_dt = now + timedelta(days=req.duration_days) if req.plan != "free_trial" else None exp_str = exp_dt.isoformat() if exp_dt else None # Generate or manage Agency API Key if $29 Agency Plan user_api_key = None if req.plan == "agency_29": # Check existing key or generate new one existing = [k for k, v in state.agency_api_keys.items() if v.get("user_email") == req.email] if existing: user_api_key = existing[0] state.agency_api_keys[user_api_key]["expires_at"] = exp_str else: user_api_key = f"san_agency_{secrets.token_hex(16)}" state.agency_api_keys[user_api_key] = { "user_email": req.email, "created_at": now.isoformat(), "expires_at": exp_str, } else: # If downgraded from Agency, revoke any active keys to_del = [k for k, v in state.agency_api_keys.items() if v.get("user_email") == req.email] for k in to_del: del state.agency_api_keys[k] user_record = { "id": req.user_id, "email": req.email, "full_name": req.full_name, "subscription_status": req.plan, "subscription_start": now.isoformat(), "subscription_end": exp_str, "api_key": user_api_key, "updated_at": now.isoformat(), } state.users_db[req.user_id] = user_record state.save_config() # Optional email notification if req.send_email_notification: plan_names = { "free_trial": "Free Trial Plan", "starter_5": "Starter Plan ($5/mo)", "creator_9": "Creator Pro Plan ($9/mo)", "agency_29": "Agency & Developer API Plan ($29/mo)", "admin": "Administrator Master Access", } plan_name = plan_names.get(req.plan, req.plan) email_body = f"""

SanitizeAI - Subscription Updated

Hello {req.full_name or req.email},

Your account access has been upgraded to {plan_name}.

Active Tier: {plan_name}

Valid Until: {exp_dt.strftime('%B %d, %Y') if exp_dt else 'Unlimited / Free'}

{f'

Your Developer API Key: {user_api_key}

' if user_api_key else ''}

You can start using your enhanced features immediately at SanitizeAI Web App.

Thank you for using SanitizeAI.

""" send_smtp_email(req.email, f"Your SanitizeAI Plan: {plan_name}", email_body) return {"status": "success", "user": user_record} # ============================================================================= # PADDLE BILLING INTEGRATION # ============================================================================= PADDLE_CLIENT_TOKEN = "live_0418ead7737d36cfc416800f9e0" PADDLE_API_KEY = "pdl_live_apikey_01m4bykvywb8d523g52qk1gcsp_7r68JMnXVVEnz0Wy20f8ZX_A1e" PADDLE_PRICE_MAP = { "pri_01m4by5njgky7p81zeghnmp73x": "starter_5", "pri_01m4by961q9tjgm21wn9jrynfz": "creator_9", "pri_01m4bybbzsj9s13fapzkm01zwb": "agency_29", } class PaddleActivateRequest(BaseModel): user_email: str price_id: str plan_tier: Optional[str] = None transaction_id: Optional[str] = None customer_id: Optional[str] = None full_name: Optional[str] = None @app.get("/v1/billing/paddle/config", tags=["Billing"]) async def get_paddle_config(): """Retrieve public Paddle Billing configuration for frontend initialization.""" return { "client_token": PADDLE_CLIENT_TOKEN, "prices": { "starter_5": "pri_01m4by5njgky7p81zeghnmp73x", "creator_9": "pri_01m4by961q9tjgm21wn9jrynfz", "agency_29": "pri_01m4bybbzsj9s13fapzkm01zwb", } } @app.post("/v1/billing/paddle/activate", tags=["Billing"]) async def activate_paddle_payment(req: PaddleActivateRequest): """Activate subscription and provision user privileges upon Paddle checkout completion.""" tier = PADDLE_PRICE_MAP.get(req.price_id, req.plan_tier or "creator_9") user_id = "user_" + secrets.token_hex(6) # Check if user already exists for uid, u in state.users_db.items(): if u.get("email") == req.user_email: user_id = uid break update_req = UpdateUserPlanRequest( user_id=user_id, email=req.user_email, full_name=req.full_name or req.user_email.split("@")[0], plan=tier, duration_days=30, send_email_notification=True, ) result = await update_user_plan(update_req, api_key=state.master_api_key) return { "status": "success", "plan": tier, "transaction_id": req.transaction_id, "user": result.get("user", {}), } class SmtpConfigRequest(BaseModel): smtp_host: str = "smtp.gmail.com" smtp_port: int = 465 sender_email: str app_password: str sender_name: str = "SanitizeAI Security" @app.post("/v1/admin/settings/smtp", tags=["Admin"]) async def save_smtp_settings(req: SmtpConfigRequest, api_key: str = Security(verify_api_key)): """Save Gmail SMTP settings for automated transactional notifications.""" state.smtp_config = { "smtp_host": req.smtp_host, "smtp_port": req.smtp_port, "sender_email": req.sender_email, "app_password": req.app_password, "sender_name": req.sender_name, "is_configured": True, } state.save_config() return {"status": "success", "message": "Gmail SMTP configuration saved."} @app.post("/v1/admin/settings/api-key", tags=["Admin"]) async def update_master_api_key(new_key: str = Form(...), api_key: str = Security(verify_api_key)): """Update the master API key dynamically.""" if len(new_key.strip()) < 8: raise HTTPException(status_code=400, detail="API Key must be at least 8 characters long.") state.master_api_key = new_key.strip() state.save_config() return {"status": "success", "message": "Master API Key updated successfully."} class EmailTriggerRequest(BaseModel): to_email: str email_type: Literal["welcome", "upgrade", "reminder", "custom"] full_name: Optional[str] = "User" plan_name: Optional[str] = "Creator Pro ($9/mo)" api_key: Optional[str] = None custom_subject: Optional[str] = None custom_body: Optional[str] = None @app.post("/v1/admin/email/send", tags=["Admin"]) async def send_system_email(req: EmailTriggerRequest, api_key: str = Security(verify_api_key)): """Trigger system email (Welcome, Upgrade, Expiry reminder, or Custom).""" if req.email_type == "welcome": subject = "Welcome to SanitizeAI - Strip AI Footprints in Seconds" body = f"""

Welcome to SanitizeAI! 🛡️

Hi {req.full_name or 'there'},

Thank you for signing up for SanitizeAI. You can now purge invisible SynthID watermarks, strip C2PA digital signatures, inpaint on-screen logos, and sanitize AI text artifacts in seconds.

Your Free Trial Quota:

Launch SanitizeAI

Need unlimited cleans or API access? Check out our Pro and Agency tiers anytime.

""" elif req.email_type == "reminder": subject = "Friendly Reminder: Your SanitizeAI Subscription is Expiring Soon" body = f"""

Subscription Renewal Notice ⚡

Hello {req.full_name or req.to_email},

This is a friendly reminder that your {req.plan_name} subscription is nearing its expiration date.

To avoid any disruption to your unlimited sanitization and API workflow, please renew your subscription.

Grace Period Policy: If unpaid, your account receives a 1-day grace period before automatic downgrade to the Free Trial quota.

Renew Subscription

""" else: subject = req.custom_subject or "SanitizeAI Notification" body = req.custom_body or "

Message from SanitizeAI Administrator.

" success = send_smtp_email(req.to_email, subject, body) if not success: return {"status": "failed", "message": "Email could not be delivered. Please check Gmail SMTP credentials in Admin Panel."} return {"status": "success", "message": f"Email delivered to {req.to_email}."} # ============================================================================= # SANITIZATION ENDPOINTS # ============================================================================= @app.post( "/v1/clean", tags=["Image Sanitizer"], summary="Clean and Sanitize Image", ) @app.post( "/v1/sanitize/image", tags=["Image Sanitizer"], summary="Sanitize Image (Frontend & Workflow Endpoint)", ) async def clean_image_endpoint( file: UploadFile = File(..., description="Image file to clean (PNG, JPG, JPEG, WEBP)"), mode: str = Form( default=CleanMode.SYNTHID_DESTRUCTOR.value, description="Cleaning mode: 'synthid_destructor', 'visual_purge', 'metadata_only', 'deep_diffusion'", ), intensity: float = Form( default=0.5, ge=0.01, le=1.0, description="Denoising / watermark destruction strength (0.01 to 1.0)", ), logo_location: str = Form( default="bottom_right", description="Visual watermark logo location ('bottom_right', 'bottom_left', 'top_right', 'top_left', 'all_corners')", ), keep_standard_metadata: bool = Form( default=True, description="Preserve non-AI standard metadata like camera/DPI/author", ), plan: str = Form( default="free_trial", description="User plan tier: 'starter_5' applies SanitizeAI watermark and preserves AI logo", ), user_email: str = Form( default="guest", description="User email for traffic accounting", ), api_key: str = Security(verify_api_key), ): """ Sanitize an uploaded image by stripping AI metadata (SD, ComfyUI, Midjourney, C2PA) and destroying invisible or visual watermarks (SynthID, StableSignature, corner logos). """ filename = file.filename or "image.png" ext = Path(filename).suffix.lower() valid_exts = {".png", ".jpg", ".jpeg", ".webp"} if ext not in valid_exts and (not file.content_type or not file.content_type.startswith("image/")): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported file type: '{ext}'. Supported formats: PNG, JPG, JPEG, WEBP.", ) try: raw_bytes = await file.read() if not raw_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Empty file uploaded.", ) start_time = time.time() # Check if $5 plan watermark policy applies apply_wm = (plan in ("starter_5", "basic_5", "plan_5")) # Process image through modular engine cleaned_bytes, mime_type = clean_image( image_input=raw_bytes, mode=mode, intensity=intensity, logo_location=logo_location, keep_standard_metadata=keep_standard_metadata, apply_watermark=apply_wm, ) duration_ms = int((time.time() - start_time) * 1000) state.record_traffic("clean_image", "image", len(raw_bytes), user_email=user_email, plan=plan) return Response( content=cleaned_bytes, media_type=mime_type, headers={ "X-Clean-Mode": str(normalize_clean_mode(mode).value), "X-Processing-Time-Ms": str(duration_ms), "X-Watermarked": "true" if apply_wm else "false", "Content-Disposition": f'attachment; filename="sanitized_{Path(filename).name}"', }, ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Image processing failed: {str(e)}", ) @app.post( "/v1/sanitize/video", tags=["Video Sanitizer"], summary="Sanitize Video File", ) async def clean_video_endpoint( file: UploadFile = File(..., description="Video file to clean (MP4, MOV, AVI, MKV, WEBM)"), mode: str = Form( default="synthid_destructor", description="Cleaning mode: 'synthid_destructor', 'visual_purge', 'metadata_only'", ), logo_location: str = Form( default="bottom_right", description="Location of visible watermark logo", ), crf: int = Form( default=18, description="Output video quality CRF (default 18 for visually lossless)", ), plan: str = Form( default="free_trial", description="User plan tier", ), user_email: str = Form( default="guest", description="User email for accounting", ), api_key: str = Security(verify_api_key), ): """ Sanitize an uploaded video by stripping C2PA metadata manifests, rebuilding a clean ISO BMFF MP4 container, and optionally purging visible logos and SynthID watermarks. """ # $5 Plan restriction: Video cleaning not available if plan in ("starter_5", "basic_5", "plan_5"): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Video cleaning is not available on the $5 Starter Plan. Please upgrade to Creator Pro ($9/mo) or Agency ($29/mo).", ) if not FFMPEG_PATH: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="FFmpeg binary not available on server.", ) filename = file.filename or "video.mp4" norm_mode = normalize_clean_mode(mode) do_logo = (norm_mode == CleanMode.VISUAL_PURGE) do_synth = (norm_mode in (CleanMode.SYNTHID_DESTRUCTOR, CleanMode.VISUAL_PURGE)) do_meta = True try: raw_bytes = await file.read() if not raw_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Empty video file uploaded.", ) with tempfile.TemporaryDirectory() as tmp_dir: tmp_in = Path(tmp_dir) / filename tmp_out = Path(tmp_dir) / f"sanitized_{Path(filename).stem}.mp4" tmp_in.write_bytes(raw_bytes) clean_video( input_video=tmp_in, output_video=tmp_out, remove_visible_logo=do_logo, remove_synthid=do_synth, remove_metadata=do_meta, logo_location=logo_location, crf_quality=crf, ) if not tmp_out.exists() or tmp_out.stat().st_size == 0: raise RuntimeError("Video sanitization did not generate an output file.") cleaned_bytes = tmp_out.read_bytes() state.record_traffic("clean_video", "video", len(raw_bytes), user_email=user_email, plan=plan) return Response( content=cleaned_bytes, media_type="video/mp4", headers={ "X-Clean-Mode": str(norm_mode.value), "Content-Disposition": f'attachment; filename="sanitized_{Path(filename).stem}.mp4"', }, ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Video sanitization failed: {str(e)}", ) @app.post("/v1/clean-text", tags=["Text Sanitizer"]) @app.post("/v1/sanitize/text", tags=["Text Sanitizer"]) async def clean_text_endpoint( text: str = Form(..., description="AI-generated text to sanitize"), remove_ai_phrases: bool = Form( default=True, description="Remove conversational intro/outro markers", ), plan: str = Form( default="free_trial", description="User plan tier", ), user_email: str = Form( default="guest", description="User email for accounting", ), api_key: str = Security(verify_api_key), ): """ Sanitize text content by removing zero-width spaces, invisible Unicode tracking markers, and homoglyphs. """ cleaned_text, stats = clean_text_content(text=text, remove_ai_phrases=remove_ai_phrases) state.record_traffic("clean_text", "text", len(text.encode("utf-8")), user_email=user_email, plan=plan) return { "cleaned_text": cleaned_text, "stats": stats, } if __name__ == "__main__": import uvicorn uvicorn.run("api:app", host="0.0.0.0", port=8000, reload=True)