import os import sys import json import base64 import tempfile import shutil import time import socket import sqlite3 import hashlib import hmac import queue import threading import uuid # Automatically include local venv site-packages if present venv_site = os.path.join(os.path.dirname(__file__), '..', 'venv', 'lib', f'python{sys.version_info.major}.{sys.version_info.minor}', 'site-packages') if os.path.exists(venv_site) and venv_site not in sys.path: sys.path.insert(0, os.path.abspath(venv_site)) def sanitize_and_ensure_transparent_subject(img_path, client=None): """ Verifies if an image has a clean transparent background for 3D generation. If the image lacks transparency (alpha < 10%), performs an automatic center-weighted crop fallback and re-preprocesses it to isolate the central subject. """ try: from PIL import Image import numpy as np except ImportError as ie: print(f"[Backend Preprocessing] Pillow or numpy not installed: {ie}. Skipping advanced transparency sanitation.") return img_path try: if not os.path.exists(img_path): return img_path img = Image.open(img_path).convert('RGBA') width, height = img.size # Calculate alpha coverage alpha_channel = np.array(img.split()[3]) transparent_ratio = np.mean(alpha_channel < 30) print(f"[Backend Preprocessing] Alpha transparency ratio: {transparent_ratio * 100:.2f}%") # If image is > 90% solid (less than 10% transparency), remote RMBG failed if transparent_ratio < 0.10: print("[Backend Preprocessing] Solid image detected (RMBG failed or no transparency). Applying smart center-crop fallback...") # Crop central 80% to eliminate edge distractions (pillows, beds, frames) crop_margin_w = int(width * 0.10) crop_margin_h = int(height * 0.10) cropped_img = img.crop((crop_margin_w, crop_margin_h, width - crop_margin_w, height - crop_margin_h)) # Save cropped temporary file cropped_temp_path = img_path.replace(".png", "_cropped_fallback.png").replace(".jpg", "_cropped_fallback.png") cropped_img.save(cropped_temp_path, "PNG") # Try re-running remote preprocess_image on the cropped subject if client: try: from gradio_client import handle_file res = client.predict(handle_file(cropped_temp_path), True, api_name="/preprocess_image") path_val = res.get('path') if isinstance(res, dict) else res if path_val and os.path.exists(path_val): img = Image.open(path_val).convert('RGBA') print("[Backend Preprocessing] Re-preprocessing with central focus succeeded!") else: img = cropped_img except Exception as e: print(f"[Backend Preprocessing] Re-preprocessing fallback warning: {e}") img = cropped_img else: img = cropped_img # Scale subject down slightly so it occupies ~80% of the canvas with generous margins (prevents border distortions) max_dim = max(img.size[0], img.size[1]) target_size = int(max_dim * 0.82) ratio = min(target_size / img.size[0], target_size / img.size[1]) new_w = max(1, int(img.size[0] * ratio)) new_h = max(1, int(img.size[1] * ratio)) img_resized = img.resize((new_w, new_h), Image.Resampling.LANCZOS) # Pad and center on a square transparent canvas with margin square_canvas = Image.new('RGBA', (max_dim, max_dim), (0, 0, 0, 0)) offset_x = (max_dim - new_w) // 2 offset_y = (max_dim - new_h) // 2 square_canvas.paste(img_resized, (offset_x, offset_y), img_resized) final_1024 = square_canvas.resize((1024, 1024), Image.Resampling.LANCZOS) out_path = img_path.replace(".png", "_preprocessed_clean.png").replace(".jpg", "_preprocessed_clean.png") if out_path == img_path: out_path = img_path + "_clean.png" final_1024.save(out_path, "PNG") print(f"[Backend Preprocessing] Clean 1024x1024 padded transparent image prepared: {out_path}") return out_path except Exception as err: print(f"[Backend Preprocessing] Exception in sanitize_and_ensure_transparent_subject: {err}") return img_path # Set a generous timeout (5 minutes) to allow sleeping Hugging Face Spaces to wake up socket.setdefaulttimeout(300) # Check if persistent volume is mounted on Hugging Face (/data) PERSISTENT_DIR = '/data' if (os.path.exists('/data') and os.path.isdir('/data')) else None def get_db_path(): if PERSISTENT_DIR: return os.path.join(PERSISTENT_DIR, 'users.db') return os.path.join(os.path.dirname(__file__), 'data', 'users.db') def get_generated_dir(subfolder, username=None): if PERSISTENT_DIR: base = os.path.join(PERSISTENT_DIR, 'generated', subfolder) else: base = os.path.join(os.path.dirname(__file__), 'generated', subfolder) if username: return os.path.join(base, username) return base def calculate_3d_cost(resolution, texture_size): cost = 5 try: res_val = int(resolution) if res_val >= 1536: cost += 2 except ValueError: if str(resolution) == '1536': cost += 2 try: tex_val = int(texture_size) if tex_val >= 4096: cost += 3 except ValueError: pass return cost def get_db_connection(): """ Returns a thread-safe SQLite connection configured with WAL (Write-Ahead Logging) and a generous timeout to support simultaneous concurrent database access across multiple worker threads. """ db_path = get_db_path() conn = sqlite3.connect(db_path, timeout=30.0) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA synchronous=NORMAL;") return conn def init_db(): db_path = get_db_path() os.makedirs(os.path.dirname(db_path), exist_ok=True) conn = get_db_connection() cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, salt TEXT NOT NULL, created_at REAL NOT NULL ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, username TEXT NOT NULL, type TEXT NOT NULL, status TEXT NOT NULL, progress INTEGER DEFAULT 0, message TEXT, result TEXT, created_at REAL NOT NULL, updated_at REAL NOT NULL ) ''') # Check and add 'credits' column if not exists (database migration) cursor.execute("PRAGMA table_info(users)") columns = [row[1] for row in cursor.fetchall()] if 'credits' not in columns: print("[Database] Migrating: Adding 'credits' column to users table.") cursor.execute("ALTER TABLE users ADD COLUMN credits INTEGER DEFAULT 20") conn.commit() conn.close() job_queue = queue.Queue() def update_job_status(job_id, status, progress=None, message=None, result=None): try: conn = get_db_connection() cursor = conn.cursor() now = time.time() updates = [("status", status), ("updated_at", now)] if progress is not None: updates.append(("progress", progress)) if message is not None: updates.append(("message", message)) if result is not None: if isinstance(result, (dict, list)): result_str = json.dumps(result) else: result_str = str(result) updates.append(("result", result_str)) set_clause = ", ".join([f"{col} = ?" for col, _ in updates]) values = [val for _, val in updates] values.append(job_id) cursor.execute(f"UPDATE jobs SET {set_clause} WHERE id = ?", values) conn.commit() conn.close() except Exception as e: print(f"[Backend Error updating job status] job={job_id} error={e}") def refund_credits(username, amount): try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("UPDATE users SET credits = credits + ? WHERE username = ?", (amount, username)) conn.commit() conn.close() print(f"[Backend] Successfully refunded {amount} credits to user {username}") except Exception as e: print(f"[Backend Error refunding credits] user={username} error={e}") def execute_job_3d(job_id, username, params): temp_img_path = None client = None try: update_job_status(job_id, 'processing', progress=10, message="Iniciando generación de malla 3D...") version = params.get('version', 'v2') image_data_b64 = params.get('image') # Base64 encoded image hf_token = params.get('token', '') seed = float(params.get('seed', 0)) resolution = params.get('resolution', '1024') decimation_target = int(params.get('decimation_target', 300000)) texture_size = int(params.get('texture_size', 2048)) ss_guidance = float(params.get('ss_guidance', 7.5)) ss_steps = int(params.get('ss_steps', 12)) slat_guidance = float(params.get('slat_guidance', 3.0)) slat_steps = int(params.get('slat_steps', 12)) auto_optimize = params.get('auto_optimize', False) quad_target_faces = int(params.get('quad_target_faces', 60000)) prompt = params.get('prompt', '') remesh_method = params.get('remeshMethod', 'cleanup') if not image_data_b64: raise Exception("No image data provided") if isinstance(image_data_b64, str) and ('generated_images' in image_data_b64 or 'generated_models' in image_data_b64): clean_url = image_data_b64.split('?')[0] parts = [p for p in clean_url.split('/') if p] disk_path = None if len(parts) >= 3 and parts[-3] == 'generated_images': disk_path = os.path.join(get_generated_dir('images', parts[-2]), parts[-1]) elif len(parts) >= 3 and parts[-3] == 'generated_models': disk_path = os.path.join(get_generated_dir('models', parts[-2]), parts[-1]) else: rel_path = clean_url.lstrip('/') if os.path.exists(rel_path): disk_path = rel_path if disk_path and os.path.exists(disk_path): with open(disk_path, 'rb') as f: image_bytes = f.read() else: raise Exception(f"No se encontró la imagen en el servidor: {clean_url}") elif isinstance(image_data_b64, str) and os.path.exists(image_data_b64): with open(image_data_b64, 'rb') as f: image_bytes = f.read() else: if ',' in image_data_b64: image_data_b64 = image_data_b64.split(',')[1] image_bytes = base64.b64decode(image_data_b64) # Save to temporary file with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as temp_img: temp_img.write(image_bytes) temp_img_path = temp_img.name # Setup Connection options connect_options = {} current_token = os.environ.get('HF_TOKEN', '').strip() hf_token_clean = str(hf_token).strip() if hf_token else '' if hf_token_clean in ('null', 'undefined'): hf_token_clean = '' is_hf_space = 'SPACE_ID' in os.environ if is_hf_space: token_to_use = hf_token_clean if hf_token_clean else current_token else: token_to_use = current_token if token_to_use == 'PON_TU_TOKEN_AQUI': token_to_use = '' token_to_use = token_to_use.strip() if token_to_use: connect_options['token'] = token_to_use else: raise Exception("Falta el Token de Hugging Face. Por favor, asegúrate de que esté configurado.") target_space = os.environ.get('HF_SPACE', 'LogicalTrue/TRELLIS.2') update_job_status(job_id, 'processing', progress=20, message="Verificando estado del servidor de IA...") stage = get_space_status(target_space, token_to_use) if stage == "PAUSED": raise Exception(f"El Space '{target_space}' está PAUSADO.") elif stage in ("STOPPED", "ERROR"): raise Exception(f"El Space '{target_space}' está APAGADO o tiene un ERROR (Estado: {stage}).") elif stage == "SLEEPING": update_job_status(job_id, 'processing', progress=25, message="Despertando servidor de IA (esto demora 2-3 minutos)...") update_job_status(job_id, 'processing', progress=30, message="Conectando al servidor de IA...") client = Client(target_space, **connect_options) try: client.predict(api_name="/start_session") except Exception as se: print(f"[Backend] Remote session initialization warning: {se}") # Preprocessing & Background Removal Pipeline preprocessed_img_path = temp_img_path try: update_job_status(job_id, 'processing', progress=40, message="Removiendo fondo de imagen (Pre-procesamiento)...") preprocess_result = client.predict(handle_file(temp_img_path), True, api_name="/preprocess_image") path_val = preprocess_result.get('path') if isinstance(preprocess_result, dict) else preprocess_result if path_val and os.path.exists(str(path_val)): preprocessed_img_path = str(path_val) except Exception as pe: print(f"[Backend] Primary Trellis /preprocess_image failed or missing argument: {pe}. Trying RMBG-1.4 fallback...") try: rmbg_client = Client("briaai/BRIA-RMBG-1.4", **connect_options) rmbg_res = rmbg_client.predict(handle_file(temp_img_path), api_name="/rmbg") path_val = rmbg_res.get('path') if isinstance(rmbg_res, dict) else rmbg_res if path_val and os.path.exists(str(path_val)): preprocessed_img_path = str(path_val) print("[Backend] RMBG-1.4 dedicated background removal succeeded!") except Exception as rmbg_err: print(f"[Backend] Dedicated RMBG-1.4 fallback failed: {rmbg_err}") # Sanitize and ensure transparent subject padding for 3D reconstruction preprocessed_img_path = sanitize_and_ensure_transparent_subject(preprocessed_img_path, client) update_job_status(job_id, 'processing', progress=50, message="Construyendo representación 3D (Inferencia de IA)...") # Ensure steps and resolutions are balanced for ZeroGPU stability safe_ss_steps = min(int(ss_steps), 12) safe_slat_steps = min(int(slat_steps), 12) job = client.submit( handle_file(preprocessed_img_path), seed, resolution, ss_guidance, 0.7, safe_ss_steps, 5.0, slat_guidance, 0.5, safe_slat_steps, 3.0, 1.0, 0.0, 12, 3.0, api_name="/image_to_3d" ) job.result() update_job_status(job_id, 'processing', progress=75, message="Extrayendo texturas PBR y generando archivo GLB...") extract_job = client.submit( decimation_target, texture_size, api_name="/extract_glb" ) extract_result = extract_job.result() print(f"[Backend Forensics] extract_result type={type(extract_result)} content={extract_result}") if hasattr(extract_result, 'data'): print(f"[Backend Forensics] extract_result.data={extract_result.data}") if hasattr(extract_result, 'data') and extract_result.data and len(extract_result.data) >= 2: gltf_file = extract_result.data[0] glb_file = extract_result.data[1] elif isinstance(extract_result, (list, tuple)) and len(extract_result) >= 2: gltf_file = extract_result[0] glb_file = extract_result[1] elif isinstance(extract_result, str): gltf_file = extract_result glb_file = extract_result elif isinstance(extract_result, dict): gltf_file = extract_result.get('value', extract_result.get('path', extract_result)) glb_file = gltf_file else: raise Exception(f"extract_glb retorno un formato inesperado: {type(extract_result)} -> {extract_result}") print(f"[Backend Forensics] gltf_file={gltf_file} glb_file={glb_file}") output_dir = get_generated_dir("models", username) os.makedirs(output_dir, exist_ok=True) gltf_local_path = gltf_file.get('path') if isinstance(gltf_file, dict) else (gltf_file if isinstance(gltf_file, str) else None) glb_local_path = glb_file.get('path') if isinstance(glb_file, dict) else (glb_file if isinstance(glb_file, str) else None) print(f"[Backend Forensics] raw paths: gltf_local_path={gltf_local_path} glb_local_path={glb_local_path}") # Distinguish real GLB (glTF binary) vs FBX (Kaydara FBX) real_glb_path = None real_fbx_path = None for candidate in [gltf_local_path, glb_local_path]: if candidate and os.path.exists(candidate): try: with open(candidate, 'rb') as f_cand: magic = f_cand.read(16) print(f"[Backend Forensics] File {candidate} size={os.path.getsize(candidate)} magic={magic[:16]}") if len(magic) >= 4 and magic[:4] == b'glTF': real_glb_path = candidate elif magic.startswith(b'Kaydara FB'): real_fbx_path = candidate except Exception as ex: print(f"[Backend Forensics] Error inspecting {candidate}: {ex}") # If real_glb_path wasn't found by magic bytes, fallback to gltf_local_path if not real_glb_path: real_glb_path = gltf_local_path if (gltf_local_path and os.path.exists(gltf_local_path)) else glb_local_path filename = f"model_{int(time.time())}.glb" dest_path = os.path.join(output_dir, filename) fbx_filename = filename.replace(".glb", ".fbx") dest_fbx_path = os.path.join(output_dir, fbx_filename) if real_glb_path and os.path.exists(real_glb_path): shutil.copy(real_glb_path, dest_path) with open(dest_path, 'rb') as f_check: copied_header = f_check.read(16) print(f"[Backend Forensics] Dest file {dest_path} saved! Header: {copied_header}") # Save direct FBX file if returned by IA if real_fbx_path and os.path.exists(real_fbx_path): shutil.copy(real_fbx_path, dest_fbx_path) print(f"[Backend Forensics] Dest FBX file {dest_fbx_path} saved from AI!") update_job_status(job_id, 'processing', progress=85, message="Generando versión FBX lista para descargas...") blender_path = os.environ.get('BLENDER_PATH', '') if not blender_path or not os.path.exists(blender_path): blender_path = shutil.which("blender") or "" if blender_path and os.path.exists(blender_path): import subprocess script_path = os.path.join(os.path.dirname(__file__), "scripts", "blender", "clean_mesh_blender.py") clean_dest_path = dest_path.replace(".glb", "_clean.glb") cmd = [blender_path, "--background", "--python", script_path, "--", dest_path, clean_dest_path, str(quad_target_faces), remesh_method] print(f"[Backend Background Worker] Generating FBX immediately: {' '.join(cmd)}") try: subprocess.run(cmd, capture_output=True, text=True, timeout=90) except Exception as be: print(f"[Backend Background Worker] Immediate FBX export note: {be}") gltf_url = f"/generated_models/{username}/{filename}" glb_url = f"/generated_models/{username}/{filename}" fbx_filename = filename.replace(".glb", ".fbx") fbx_url = f"/generated_models/{username}/{fbx_filename}" if os.path.exists(os.path.join(output_dir, fbx_filename)) else None else: gltf_url = gltf_file.get('url') if isinstance(gltf_file, dict) else gltf_local_path glb_url = glb_file.get('url') if isinstance(glb_file, dict) else glb_local_path fbx_url = None update_job_status(job_id, 'processing', progress=95, message="Clasificando especie del modelo...") detected_category = classify_species(image_bytes, prompt, token_to_use) if glb_local_path and os.path.exists(glb_local_path): metadata_path = dest_path.replace(".glb", ".json") try: with open(metadata_path, "w", encoding="utf-8") as meta_f: json.dump({ "detectedCategory": detected_category, "prompt": prompt, "timestamp": time.time() }, meta_f, indent=2) except Exception as me: print(f"[Backend] Metadata warning: {me}") result_payload = { "gltfUrl": gltf_url, "glbUrl": glb_url, "fbxUrl": fbx_url, "detectedCategory": detected_category } update_job_status(job_id, 'completed', progress=100, message="Generación 3D completada con éxito.", result=result_payload) except Exception as ex: print(f"[Backend Job Error] job={job_id} error={ex}") update_job_status(job_id, 'failed', message=f"Fallo en la generación: {str(ex)}") # Refund credits refund_credits(username, params.get('cost', 5)) finally: if temp_img_path: try: os.unlink(temp_img_path) except: pass if client: try: client.close() except: pass def execute_job_2d(job_id, username, params): from gradio_client import Client as GradioClient try: update_job_status(job_id, 'processing', progress=20, message="Conectando al servidor FLUX de imágenes...") prompt = params.get('prompt', '') hf_token = params.get('token', '') current_token = os.environ.get('HF_TOKEN', '').strip() hf_token_clean = str(hf_token).strip() if hf_token else '' if hf_token_clean in ('null', 'undefined'): hf_token_clean = '' is_hf_space = 'SPACE_ID' in os.environ if is_hf_space: token_to_use = hf_token_clean if hf_token_clean else current_token else: token_to_use = current_token if token_to_use == 'PON_TU_TOKEN_AQUI': token_to_use = '' token_to_use = token_to_use.strip() connect_options = {} if token_to_use: connect_options['token'] = token_to_use else: raise Exception("Falta el Token de Hugging Face.") spaces_to_try = [ "black-forest-labs/FLUX.1-schnell", "multimodalart/FLUX.1-schnell" ] success = False response_data = None last_error = None for space_name in spaces_to_try: client = None try: update_job_status(job_id, 'processing', progress=40, message=f"Generando imagen vía {space_name}...") client = GradioClient(space_name, **connect_options) result = client.predict( prompt=prompt, seed=0, randomize_seed=True, width=1024, height=1024, num_inference_steps=4, api_name="/infer" ) if isinstance(result, (list, tuple)) and len(result) > 0: img_local_path = result[0] elif isinstance(result, dict) and 'path' in result: img_local_path = result['path'] else: img_local_path = result if img_local_path and os.path.exists(img_local_path): with open(img_local_path, "rb") as img_file: img_bytes = img_file.read() images_dir = get_generated_dir("images", username) os.makedirs(images_dir, exist_ok=True) img_filename = f"image_{int(time.time())}.png" img_dest_path = os.path.join(images_dir, img_filename) with open(img_dest_path, "wb") as f: f.write(img_bytes) img_b64 = base64.b64encode(img_bytes).decode('utf-8') response_data = { "image": f"data:image/png;base64,{img_b64}", "imageUrl": f"/generated_images/{username}/{img_filename}", "model_used": space_name } success = True break else: raise Exception(f"La ruta devuelta no existe: {img_local_path}") except Exception as ex: last_error = str(ex) finally: if client: try: client.close() except: pass if success and response_data: update_job_status(job_id, 'completed', progress=100, message="Generación de imagen completada.", result=response_data) else: raise Exception(f"Fallaron todos los Spaces de FLUX. Último error: {last_error}") except Exception as ex: print(f"[Backend Job Error] job={job_id} error={ex}") update_job_status(job_id, 'failed', message=f"Error al generar imagen 2D: {str(ex)}") refund_credits(username, params.get('cost', 1)) def background_worker(worker_id): print(f"[Backend Background Worker #{worker_id}] Starting worker thread...") while True: try: job = job_queue.get() if job is None: break job_id = job["id"] username = job["username"] job_type = job["type"] params = job["params"] print(f"[Backend Background Worker #{worker_id}] Processing job={job_id} user={username} type={job_type}") if job_type == '3d': execute_job_3d(job_id, username, params) elif job_type == '2d': execute_job_2d(job_id, username, params) job_queue.task_done() except Exception as we: print(f"[Backend Background Worker #{worker_id} Exception] {we}") time.sleep(1) # Spawn a pool of worker threads for parallel job processing NUM_WORKER_THREADS = 4 worker_threads = [] for i in range(NUM_WORKER_THREADS): t = threading.Thread(target=background_worker, args=(i + 1,), daemon=True) t.start() worker_threads.append(t) from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer # pyrefly: ignore [missing-import] from gradio_client import Client, handle_file if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") # Simple environment loader to avoid external dependencies def load_dotenv(): env_path = os.path.join(os.path.dirname(__file__), '.env') if os.path.exists(env_path): with open(env_path, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, val = line.split('=', 1) key_str = key.strip() val_str = val.strip() is_hf_space = 'SPACE_ID' in os.environ current_val = os.environ.get(key_str, '').strip() # Update/overwrite if: # 1. Variable not already set in environment # 2. Or current value is empty/placeholder # 3. Or we are running locally (not HF Spaces) if (key_str not in os.environ or current_val in ('', 'PON_TU_TOKEN_AQUI', 'null', 'undefined') or not is_hf_space): # Avoid overwriting a valid token in the environment with a placeholder from .env if not (val_str == 'PON_TU_TOKEN_AQUI' and current_val.startswith('hf_')): os.environ[key_str] = val_str # Initialize configuration load_dotenv() HF_TOKEN = os.environ.get('HF_TOKEN', '') HF_SPACE = os.environ.get('HF_SPACE', 'LogicalTrue/TRELLIS.2') # Hugging Face Spaces always runs on port 7860 if 'SPACE_ID' in os.environ: PORT = 7860 print(f"[Backend] Running inside Hugging Face Space. Forcing PORT to {PORT}") else: PORT = int(os.environ.get('PORT', '8000')) if not HF_TOKEN or HF_TOKEN == 'PON_TU_TOKEN_AQUI': print("\n[⚠️ WARNING] HF_TOKEN is not configured or has default placeholder value in .env file.") print("Please open the '.env' file and insert your Hugging Face Token (hf_...) to access your private Space.\n") def get_space_status(space_id, token=None): """ Checks the current status of a Hugging Face Space. Returns the stage string, e.g. 'RUNNING', 'SLEEPING', 'PAUSED', 'STOPPED', 'ERROR', or 'UNKNOWN'. """ import requests url = f"https://huggingface.co/api/spaces/{space_id}" headers = {} if token: headers["Authorization"] = f"Bearer {token}" try: r = requests.get(url, headers=headers, timeout=5) if r.status_code == 200: data = r.json() runtime = data.get("runtime", {}) stage = runtime.get("stage", "UNKNOWN").upper() return stage else: print(f"[Space Status] Failed to fetch status for {space_id}: HTTP {r.status_code}") return "UNKNOWN" except Exception as e: print(f"[Space Status] Error checking status for {space_id}: {e}") return "UNKNOWN" def classify_species(image_bytes, prompt_text, hf_token): p = prompt_text.lower() if prompt_text else "" # Spider / Insect keywords spider_words = ["spider", "araña", "aracnido", "arachnid", "tarantula", "insect", "insecto", "crab", "cangrejo", "scorpion", "escorpion", "bug"] if any(w in p for w in spider_words): print(f"[Classifier] Detected 'unsupported' category from prompt: '{prompt_text}'") return "unsupported" # Quadruped keywords quad_words = ["horse", "caballo", "dog", "perro", "cat", "gato", "wolf", "lobo", "lion", "leon", "tiger", "tigre", "cow", "vaca", "sheep", "oveja", "pig", "cerdo", "fox", "zorro", "deer", "ciervo", "bear", "oso", "quadruped", "cuadrupedo", "animal", "camel", "camello", "elephant", "elefante", "giraffe", "jirafa"] if any(w in p for w in quad_words): print(f"[Classifier] Detected 'local_quadruped' category from prompt: '{prompt_text}'") return "local_quadruped" # Humanoid keywords humanoid_words = ["human", "humano", "man", "hombre", "woman", "mujer", "boy", "chico", "girl", "chica", "character", "personaje", "soldier", "soldado", "warrior", "guerrero", "wizard", "mago", "hero", "heroe", "knight", "caballero", "robot", "biped", "bipedo", "alien", "cyborg", "golem"] if any(w in p for w in humanoid_words): print(f"[Classifier] Detected 'ai' category from prompt: '{prompt_text}'") return "ai" # 2. Image classification fallback via CLIP on HF if not hf_token or hf_token in ('null', 'undefined'): print("[Classifier] No HF Token for image classification. Defaulting to 'ai'.") return "ai" try: import requests headers = {"Authorization": f"Bearer {hf_token}"} urls_to_try = [ "https://api-inference.huggingface.co/models/openai/clip-vit-large-patch14", "https://router.huggingface.co/hf-inference/models/openai/clip-vit-large-patch14", "https://api-inference.hf.co/models/openai/clip-vit-large-patch14" ] img_b64 = base64.b64encode(image_bytes).decode('utf-8') payload = { "image": img_b64, "parameters": { "candidate_labels": [ "a bipedal humanoid character or person", "a four-legged animal or quadruped", "a spider or multi-legged insect", "an object, prop or static furniture" ] } } print("[Classifier] Querying CLIP zero-shot classification on Hugging Face...") for api_url in urls_to_try: try: response = requests.post(api_url, headers=headers, json=payload, timeout=8) if response.status_code == 200: res_data = response.json() if isinstance(res_data, list) and len(res_data) > 0: best_label = res_data[0].get("label", "") score = res_data[0].get("score", 0.0) print(f"[Classifier] CLIP result: {best_label} (score: {score:.3f})") if "bipedal" in best_label: return "ai" elif "four-legged" in best_label: return "local_quadruped" elif "spider" in best_label: return "unsupported" else: return "unsupported" else: err_preview = response.text[:200] if response.text else "" if " 1 else '' return os.path.join(get_generated_dir('images'), subpath) elif parts[0] == 'generated_models': subpath = os.path.join(*parts[1:]) if len(parts) > 1 else '' resolved_file = os.path.join(get_generated_dir('models'), subpath) return resolved_file elif parts[0] in ('app.js', 'styles.css', 'index.html', 'viewer.html'): return os.path.join(os.path.dirname(__file__), '..', 'frontend', parts[0]) return os.path.join(os.path.dirname(__file__), '..', 'frontend', *parts) def end_headers(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization') self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate, max-age=0') self.send_header('Pragma', 'no-cache') self.send_header('Expires', '0') super().end_headers() def do_OPTIONS(self): self.send_response(200, "OK") self.end_headers() def get_logged_in_user(self): cookie_header = self.headers.get('Cookie', '') if cookie_header: cookies = {} for item in cookie_header.split(';'): item = item.strip() if '=' in item: k, v = item.split('=', 1) cookies[k.strip()] = v.strip() return cookies.get('session_user') return None def do_GET(self): if self.path == '/api/gallery': self.handle_get_gallery() elif self.path.startswith('/api/space-status'): self.handle_space_status() elif self.path.startswith('/api/job-status'): self.handle_job_status() elif self.path == '/api/user-active-job': self.handle_user_active_job() elif self.path == '/api/me': username = self.get_logged_in_user() credits = 0 nick = None full_name = None avatar = None email = None if username: try: db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT credits, nick, full_name, avatar, email FROM users WHERE username = ?", (username,)) row = cursor.fetchone() conn.close() if row: credits = row[0] if row[0] is not None else 0 nick = row[1] full_name = row[2] avatar = row[3] email = row[4] except Exception as e: print(f"[Backend] Error checking user profile: {e}") self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({ "username": username, "credits": credits, "nick": nick or username, "full_name": full_name or "", "avatar": avatar or "", "email": email or "" }).encode('utf-8')) else: super().do_GET() def do_POST(self): if self.path == '/api/register': self.handle_register() elif self.path == '/api/login': self.handle_login() elif self.path == '/api/logout': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Set-Cookie', 'session_user=; Path=/; Expires=Thu, 01 Jan 1970 00:00:00 GMT; Max-Age=0; SameSite=Lax') self.send_header('Set-Cookie', 'session_user=; Path=/; Expires=Thu, 01 Jan 1970 00:00:00 GMT; Max-Age=0; SameSite=None; Secure') self.end_headers() self.wfile.write(json.dumps({"success": True}).encode('utf-8')) elif self.path == '/api/update-profile': self.handle_update_profile() elif self.path == '/api/update-privacy': self.handle_update_privacy() elif self.path == '/api/generate-3d': self.handle_generate_3d() elif self.path == '/api/optimize-3d': self.handle_optimize_3d() elif self.path == '/api/rig-3d': self.handle_rig_3d() elif self.path == '/api/generate-2d': self.handle_generate_2d() elif self.path == '/api/delete-gallery': self.handle_delete_gallery() elif self.path == '/api/save-weights': self.handle_save_weights() elif self.path == '/api/topup': self.handle_topup() elif self.path == '/api/create-checkout-session': self.handle_create_checkout_session() elif self.path == '/api/lemonsqueezy-webhook': self.handle_lemonsqueezy_webhook() else: self.send_error(404, "Endpoint not found") def handle_register(self): try: content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) username = params.get('username', '').strip().lower() password = params.get('password', '') if not username or not password: self.send_error_response(400, "Nombre de usuario y contraseña son obligatorios.") return if not username.isalnum() or len(username) < 3: self.send_error_response(400, "El nombre de usuario debe ser alfanumérico y de al menos 3 caracteres.") return if len(password) < 4: self.send_error_response(400, "La contraseña debe tener al menos 4 caracteres.") return db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT id FROM users WHERE username = ?", (username,)) if cursor.fetchone(): conn.close() self.send_error_response(400, "El nombre de usuario ya está registrado.") return salt = base64.b64encode(os.urandom(16)).decode('utf-8') hasher = hashlib.sha256() hasher.update((password + salt).encode('utf-8')) password_hash = hasher.hexdigest() cursor.execute( "INSERT INTO users (username, password_hash, salt, created_at, credits) VALUES (?, ?, ?, ?, 20)", (username, password_hash, salt, time.time()) ) conn.commit() conn.close() os.makedirs(get_generated_dir("images", username), exist_ok=True) os.makedirs(get_generated_dir("models", username), exist_ok=True) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True}).encode('utf-8')) except Exception as e: self.send_error_response(500, str(e)) def handle_update_profile(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) nick = params.get('nick', '').strip() full_name = params.get('full_name', '').strip() avatar = params.get('avatar', '').strip() if not nick: self.send_error_response(400, "El apodo / nick no puede estar vacío.") return db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute( "UPDATE users SET nick = ?, full_name = ?, avatar = ? WHERE username = ?", (nick, full_name, avatar, username) ) conn.commit() conn.close() self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True, "nick": nick, "full_name": full_name, "avatar": avatar}).encode('utf-8')) except Exception as e: print(f"[Backend] Error updating profile: {e}") self.send_error_response(500, str(e)) def handle_update_privacy(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) email = params.get('email', '').strip() current_password = params.get('current_password', '') new_password = params.get('new_password', '') db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() if new_password: cursor.execute("SELECT password_hash, salt FROM users WHERE username = ?", (username,)) user_row = cursor.fetchone() if not user_row: conn.close() self.send_error_response(404, "Usuario no encontrado.") return stored_hash, salt = user_row[0], user_row[1] hasher = hashlib.sha256() hasher.update((current_password + salt).encode('utf-8')) if hasher.hexdigest() != stored_hash: conn.close() self.send_error_response(400, "La contraseña actual es incorrecta.") return new_salt = base64.b64encode(os.urandom(16)).decode('utf-8') new_hasher = hashlib.sha256() new_hasher.update((new_password + new_salt).encode('utf-8')) new_hash = new_hasher.hexdigest() cursor.execute( "UPDATE users SET email = ?, password_hash = ?, salt = ? WHERE username = ?", (email, new_hash, new_salt, username) ) else: cursor.execute( "UPDATE users SET email = ? WHERE username = ?", (email, username) ) conn.commit() conn.close() self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True}).encode('utf-8')) except Exception as e: print(f"[Backend] Error updating privacy: {e}") self.send_error_response(500, str(e)) def handle_login(self): try: content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) username = params.get('username', '').strip().lower() password = params.get('password', '') if not username or not password: self.send_error_response(400, "Nombre de usuario y contraseña son obligatorios.") return db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT password_hash, salt FROM users WHERE username = ?", (username,)) row = cursor.fetchone() conn.close() if not row: self.send_error_response(400, "Usuario o contraseña incorrectos.") return db_hash, salt = row hasher = hashlib.sha256() hasher.update((password + salt).encode('utf-8')) login_hash = hasher.hexdigest() if login_hash != db_hash: self.send_error_response(400, "Usuario o contraseña incorrectos.") return os.makedirs(get_generated_dir("images", username), exist_ok=True) os.makedirs(get_generated_dir("models", username), exist_ok=True) # Query credits credits = 0 try: db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) row = cursor.fetchone() conn.close() if row: credits = row[0] except Exception as e: print(f"[Backend] Error checking login credits: {e}") self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Set-Cookie', f'session_user={username}; Path=/; Max-Age=2592000; SameSite=Lax') self.end_headers() self.wfile.write(json.dumps({"success": True, "username": username, "credits": credits}).encode('utf-8')) except Exception as e: self.send_error_response(500, str(e)) def handle_topup(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) amount = int(params.get('amount', 50)) db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("UPDATE users SET credits = credits + ? WHERE username = ?", (amount, username)) conn.commit() cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) credits_row = cursor.fetchone() conn.close() new_credits = credits_row[0] if credits_row else 0 self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True, "credits": new_credits}).encode('utf-8')) except Exception as e: self.send_error_response(500, str(e)) def handle_create_checkout_session(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) pack_type = str(params.get('pack_type', '25')) api_key = os.environ.get('LEMON_SQUEEZY_API_KEY', '').strip() store_id = os.environ.get('LEMON_SQUEEZY_STORE_ID', '').strip() variant_25 = os.environ.get('LEMON_SQUEEZY_VARIANT_25', '').strip() variant_100 = os.environ.get('LEMON_SQUEEZY_VARIANT_100', '').strip() host = self.headers.get('Host', 'localhost:8000') protocol = 'https' if 'hf.space' in host or 'huggingface.co' in host else 'http' base_url = f"{protocol}://{host}" amount_credits = 100 if pack_type == '100' else 25 variant_id = variant_100 if pack_type == '100' else variant_25 if not api_key or not store_id or not variant_id: print("[⚠️ Lemon Squeezy] API keys/Variant IDs missing. Simulating checkout url.") mock_url = f"{base_url}/?payment=success" db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("UPDATE users SET credits = credits + ? WHERE username = ?", (amount_credits, username)) conn.commit() conn.close() self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"url": mock_url}).encode('utf-8')) return import urllib.request import urllib.error url = "https://api.lemonsqueezy.com/v1/checkouts" req_data = { "data": { "type": "checkouts", "attributes": { "product_options": { "redirect_url": f"{base_url}/?payment=success" }, "checkout_data": { "custom": { "username": username, "amount": str(amount_credits) } } }, "relationships": { "store": { "data": { "type": "stores", "id": str(store_id) } }, "variant": { "data": { "type": "variants", "id": str(variant_id) } } } } } req = urllib.request.Request( url, data=json.dumps(req_data).encode('utf-8'), headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/vnd.api+json", "Accept": "application/vnd.api+json" }, method="POST" ) try: with urllib.request.urlopen(req) as response: res_body = response.read().decode('utf-8') res_json = json.loads(res_body) checkout_url = res_json["data"]["attributes"]["url"] self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"url": checkout_url}).encode('utf-8')) except urllib.error.HTTPError as http_err: err_content = http_err.read().decode('utf-8') print(f"[Lemon Squeezy API Error] {http_err.code}: {err_content}") self.send_error_response(http_err.code, f"Error de Lemon Squeezy: {err_content}") except Exception as e: print(f"[Lemon Squeezy Checkout Error] {e}") self.send_error_response(500, str(e)) def handle_lemonsqueezy_webhook(self): try: content_length = int(self.headers.get('Content-Length', 0)) payload = self.rfile.read(content_length) sig_header = self.headers.get('X-Signature', '') webhook_secret = os.environ.get('LEMON_SQUEEZY_WEBHOOK_SECRET', '').strip() if webhook_secret and webhook_secret != 'PON_TU_WEBHOOK_SECRET_AQUI': digest = hmac.new( webhook_secret.encode('utf-8'), payload, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(digest, sig_header): print("[⚠️ Lemon Squeezy Webhook] Invalid signature verification.") self.send_response(400) self.end_headers() return else: print("[⚠️ Lemon Squeezy Webhook] Webhook secret not configured. Bypassing signature check (Developer Mode).") event = json.loads(payload.decode('utf-8')) event_name = event.get('meta', {}).get('event_name') if event_name == 'order_created': custom_data = event.get('meta', {}).get('custom_data', {}) username = custom_data.get('username') amount = custom_data.get('amount') if username and amount: try: amount = int(amount) db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("UPDATE users SET credits = credits + ? WHERE username = ?", (amount, username)) conn.commit() conn.close() print(f"[Lemon Squeezy Webhook] Successfully credited {amount} credits to user: {username}") except Exception as db_err: print(f"[Lemon Squeezy Webhook Database Error] {db_err}") self.send_response(500) self.end_headers() return else: print(f"[Lemon Squeezy Webhook Warning] Webhook custom_data missing username/amount: {custom_data}") self.send_response(200) self.end_headers() except Exception as e: print(f"[Lemon Squeezy Webhook Exception] {e}") self.send_response(500) self.end_headers() def handle_generate_3d(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return load_dotenv() # Reload env dynamically content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) resolution = params.get('resolution', '1024') texture_size = int(params.get('texture_size', 2048)) required_credits = calculate_3d_cost(resolution, texture_size) params['cost'] = required_credits # store cost in params for potential refund # Check credits dynamically db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) row = cursor.fetchone() if not row or row[0] < required_credits: conn.close() self.send_response(402) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"error": f"Créditos insuficientes. Esta generación 3D con ajustes seleccionados cuesta {required_credits} créditos."}).encode('utf-8')) return # Deduct credits immediately cursor.execute("UPDATE users SET credits = MAX(0, credits - ?) WHERE username = ?", (required_credits, username)) # Query the updated credits cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) credits_row = cursor.fetchone() new_credits = credits_row[0] if credits_row else 0 # Create asynchronous job job_id = str(uuid.uuid4()) now = time.time() cursor.execute( "INSERT INTO jobs (id, username, type, status, progress, message, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (job_id, username, '3d', 'pending', 0, 'En cola de espera...', now, now) ) conn.commit() conn.close() # Push to background worker queue job_queue.put({ "id": job_id, "username": username, "type": "3d", "params": params }) response_data = { "success": True, "job_id": job_id, "status": "pending", "credits": new_credits } self.send_response(202) # 202 Accepted self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response_data).encode('utf-8')) except Exception as e: print(f"[Backend] Error initiating 3D generation job: {e}") self.send_error_response(500, str(e)) def handle_optimize_3d(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return load_dotenv() content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) model_url = params.get('modelUrl', '') # e.g. "/generated_models/model_1782268665.glb" quad_target_faces = int(params.get('quad_target_faces', 60000)) remesh_method = params.get('remeshMethod', 'cleanup') if not model_url: self.send_error_response(400, "No modelUrl provided") return filename = os.path.basename(model_url) output_dir = get_generated_dir("models", username) dest_path = os.path.join(output_dir, filename) if not os.path.exists(dest_path): self.send_error_response(404, f"Model file {filename} not found") return # Output to a new file (_quad.glb) to keep the original source model intact in gallery if "_quad" in filename: clean_filename = filename else: clean_filename = filename.replace(".glb", "_quad.glb") clean_dest_path = os.path.join(output_dir, clean_filename) # Copy metadata json if exists so the remeshed model retains species category meta_src = dest_path.replace(".glb", ".json") meta_dest = clean_dest_path.replace(".glb", ".json") if os.path.exists(meta_src) and not os.path.exists(meta_dest): try: shutil.copy(meta_src, meta_dest) except Exception as me: print(f"[Backend] QuadriFlow metadata copy note: {me}") blender_path = os.environ.get('BLENDER_PATH', '') if not blender_path or not os.path.exists(blender_path): blender_path = shutil.which("blender") or "" if blender_path and os.path.exists(blender_path): print(f"[Backend] Local optimization requested. Running Blender...") import subprocess script_path = os.path.join(os.path.dirname(__file__), "scripts", "blender", "clean_mesh_blender.py") cmd = [blender_path, "--background", "--python", script_path, "--", dest_path, clean_dest_path, str(quad_target_faces), remesh_method] print(f"[Backend] Executing: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) print(f"[Backend] Blender Output:\n{result.stdout}") if result.stderr: print(f"[Backend] Blender Errors:\n{result.stderr}") fbx_filename = clean_filename.replace(".glb", ".fbx") fbx_url = f"/generated_models/{username}/{fbx_filename}" if os.path.exists(os.path.join(output_dir, fbx_filename)) else None if result.returncode == 0 and os.path.exists(clean_dest_path): response_data = { "success": True, "glbUrl": f"/generated_models/{username}/{clean_filename}", "gltfUrl": f"/generated_models/{username}/{clean_filename}", "fbxUrl": fbx_url } else: raise Exception(f"Blender failed with exit status {result.returncode}") else: raise Exception("BLENDER_PATH is not configured or executable not found locally.") self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response_data).encode('utf-8')) except Exception as e: print(f"[Backend] Error during 3D optimization: {e}") self.send_error_response(500, str(e)) def handle_rig_3d(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return load_dotenv() content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) model_url = params.get('modelUrl', '') # e.g. "/generated_models/model_1782268665.glb" rig_method = params.get('rigMethod', 'ai') hf_token = params.get('token', '') if not model_url: self.send_error_response(400, "No modelUrl provided") return filename = os.path.basename(model_url) output_dir = get_generated_dir("models", username) dest_path = os.path.join(output_dir, filename) if not os.path.exists(dest_path): self.send_error_response(404, f"Model file {filename} not found") return # Rigging target paths base_name, _ = os.path.splitext(filename) rigged_fbx_filename = f"{base_name}_rigged.fbx" rigged_fbx_path = os.path.join(output_dir, rigged_fbx_filename) if rig_method == 'ai': rigged_glb_filename = f"{base_name}_rigged.glb" rigged_glb_path = os.path.join(output_dir, rigged_glb_filename) print(f"[Backend] AI Rigging requested via Hugging Face...") # Get the correct token current_token = os.environ.get('HF_TOKEN', '') # Clean token from spaces/quotes hf_token_clean = str(hf_token).strip() if hf_token else '' if hf_token_clean in ('null', 'undefined'): hf_token_clean = '' # If running on HF Spaces, prioritize token sent by the client. If running locally, only use the .env token. is_hf_space = 'SPACE_ID' in os.environ if is_hf_space: token_to_use = hf_token_clean if hf_token_clean else current_token else: token_to_use = current_token if token_to_use == 'PON_TU_TOKEN_AQUI': token_to_use = '' token_to_use = token_to_use.strip() print(f"[Backend] Client token length: {len(hf_token_clean)}, Env token length: {len(current_token)}, Token to use length: {len(token_to_use)}") connect_options = {} if token_to_use: connect_options['token'] = token_to_use connect_options['headers'] = {"Authorization": f"Bearer {token_to_use}"} # Check Space status before calling Gradio unirig_space = "LogicalTrue/Unirig" print(f"[Backend] Checking status of Hugging Face Space: '{unirig_space}'...") stage = get_space_status(unirig_space, token_to_use) print(f"[Backend] Checked Space stage: '{stage}'") if stage == "PAUSED": self.send_error_response(503, f"El Space de Rigging '{unirig_space}' está PAUSADO. Por favor, reanúdalo en la consola de Hugging Face.") return elif stage in ("STOPPED", "ERROR"): self.send_error_response(503, f"El Space de Rigging '{unirig_space}' está APAGADO o tiene un ERROR (Estado actual: {stage}).") return elif stage == "SLEEPING": print(f"[Backend] ¡Atención! El Space de Rigging '{unirig_space}' está DORMIDO (SLEEPING). Gradio intentará despertarlo (esto puede demorar de 2 a 3 minutos)...") # UniRig API call with retry on 429 from gradio_client import Client, handle_file print(f"[Backend] Connecting to '{unirig_space}'...") client = Client(unirig_space, **connect_options) print(f"[Backend] Submitting {filename} to UniRig...") res_path = None max_retries = 3 for attempt in range(max_retries): try: res_path = client.predict( handle_file(dest_path), # archivo_3d 12345, # seed api_name="/rig_mesh" ) if res_path and os.path.exists(res_path): break except Exception as pe: err_str = str(pe) if "429" in err_str or "Too Many Requests" in err_str: if attempt < max_retries - 1: wait_sec = (attempt + 1) * 3 print(f"[Backend] Rate limit 429 detected during rigging. Retrying in {wait_sec}s (Attempt {attempt+1}/{max_retries})...") time.sleep(wait_sec) continue raise pe if res_path and os.path.exists(res_path): shutil.copy(res_path, rigged_glb_path) print(f"[Backend] ✓ AI Rigging completed successfully. Saved to: {rigged_glb_path}") response_data = { "success": True, "riggedFbxUrl": f"/generated_models/{username}/{rigged_glb_filename}" } else: raise Exception("AI Rigging failed: could not retrieve the generated rigged GLB model from Hugging Face Space.") else: # Local procedural rigging using Blender blender_path = os.environ.get('BLENDER_PATH', '') if not blender_path or not os.path.exists(blender_path): blender_path = shutil.which("blender") or "" if blender_path and os.path.exists(blender_path): script_name = "rig_quadruped_blender.py" if rig_method == "local_quadruped" else "rig_mesh_blender.py" print(f"[Backend] Local procedural rigging ({rig_method}) requested. Running Blender with {script_name}...") import subprocess script_path = os.path.join(os.path.dirname(__file__), "scripts", "blender", script_name) cmd = [blender_path, "--background", "--python", script_path, "--", dest_path, rigged_fbx_path] print(f"[Backend] Executing: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) print(f"[Backend] Blender Output:\n{result.stdout}") if result.stderr: print(f"[Backend] Blender Errors:\n{result.stderr}") rigged_glb_filename = f"{base_name}_rigged.glb" rigged_glb_path = os.path.join(output_dir, rigged_glb_filename) if result.returncode == 0 and os.path.exists(rigged_fbx_path): has_glb = os.path.exists(rigged_glb_path) response_data = { "success": True, "gltfUrl": f"/generated_models/{username}/{rigged_glb_filename}" if has_glb else None, "glbUrl": f"/generated_models/{username}/{rigged_glb_filename}" if has_glb else None, "fbxUrl": f"/generated_models/{username}/{rigged_fbx_filename}", "riggedFbxUrl": f"/generated_models/{username}/{rigged_glb_filename}" if has_glb else f"/generated_models/{username}/{rigged_fbx_filename}" } else: raise Exception(f"Blender rigging failed with exit status {result.returncode}") else: raise Exception("BLENDER_PATH is not configured or executable not found locally.") self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response_data).encode('utf-8')) except Exception as e: print(f"[Backend] Error during rigging: {e}") self.send_error_response(500, str(e)) def handle_generate_2d(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) prompt = params.get('prompt', '') params['cost'] = 1 # 2D image cost is 1 credit if not prompt: self.send_error_response(400, "No prompt provided") return # Check credits (needs 1) db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) row = cursor.fetchone() if not row or row[0] < 1: conn.close() self.send_response(402) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"error": "Créditos insuficientes. Generar una imagen cuesta 1 crédito."}).encode('utf-8')) return # Deduct credits immediately cursor.execute("UPDATE users SET credits = MAX(0, credits - 1) WHERE username = ?", (username,)) # Query updated credits cursor.execute("SELECT credits FROM users WHERE username = ?", (username,)) credits_row = cursor.fetchone() new_credits = credits_row[0] if credits_row else 0 # Create async job job_id = str(uuid.uuid4()) now = time.time() cursor.execute( "INSERT INTO jobs (id, username, type, status, progress, message, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (job_id, username, '2d', 'pending', 0, 'En cola de espera...', now, now) ) conn.commit() conn.close() # Push to background worker queue job_queue.put({ "id": job_id, "username": username, "type": "2d", "params": params }) response_data = { "success": True, "job_id": job_id, "status": "pending", "credits": new_credits } self.send_response(202) # 202 Accepted self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response_data).encode('utf-8')) except Exception as e: print(f"[Backend] Error during 2D generation initialization: {e}") self.send_error_response(500, str(e)) def handle_space_status(self): try: from urllib.parse import urlparse, parse_qs parsed_path = urlparse(self.path) query_params = parse_qs(parsed_path.query) space_type = query_params.get('type', ['3d'])[0] token = query_params.get('token', [''])[0] load_dotenv() current_token = os.environ.get('HF_TOKEN', '').strip() hf_token_clean = token.strip() if token else '' if hf_token_clean in ('null', 'undefined'): hf_token_clean = '' # If running on HF Spaces, prioritize token sent by the client. If running locally, only use the .env token. is_hf_space = 'SPACE_ID' in os.environ if is_hf_space: token_to_use = hf_token_clean if hf_token_clean else current_token else: token_to_use = current_token if token_to_use == 'PON_TU_TOKEN_AQUI': token_to_use = '' token_to_use = token_to_use.strip() if space_type == 'rig': target_space = "LogicalTrue/Unirig" else: target_space = os.environ.get('HF_SPACE', 'LogicalTrue/TRELLIS.2') stage = get_space_status(target_space, token_to_use) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"stage": stage, "space": target_space}).encode('utf-8')) except Exception as e: self.send_error_response(500, str(e)) def handle_job_status(self): try: from urllib.parse import urlparse, parse_qs parsed_path = urlparse(self.path) query_params = parse_qs(parsed_path.query) job_id_list = query_params.get('job_id') if not job_id_list: self.send_error_response(400, "Missing job_id parameter") return job_id = job_id_list[0] db_path = get_db_path() conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT id, username, type, status, progress, message, result FROM jobs WHERE id = ?", (job_id,)) row = cursor.fetchone() conn.close() if not row: self.send_error_response(404, f"Job {job_id} not found") return job_data = { "job_id": row[0], "username": row[1], "type": row[2], "status": row[3], "progress": row[4], "message": row[5], "result": json.loads(row[6]) if row[6] and (row[6].startswith('{') or row[6].startswith('[')) else row[6] } self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(job_data).encode('utf-8')) except Exception as e: print(f"[Backend Error in handle_job_status] {e}") self.send_error_response(500, str(e)) def handle_user_active_job(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return # Clean up stale / orphan jobs older than 3 minutes (180 seconds) now_ts = time.time() cursor.execute( "SELECT id, created_at, status FROM jobs WHERE username = ? AND status IN ('pending', 'processing')", (username,) ) all_running = cursor.fetchall() for r_job in all_running: job_id_val, c_time, j_status = r_job # If job was created > 180 seconds ago and not completed, mark as failed/timeout if (now_ts - float(c_time or 0)) > 180: print(f"[Backend Cleanup] Terminating orphan stale job {job_id_val} (age: {int(now_ts - float(c_time or 0))}s)") cursor.execute( "UPDATE jobs SET status = 'failed', message = 'Tiempo de espera agotado (Proceso finalizado)', progress = 0 WHERE id = ?", (job_id_val,) ) conn.commit() cursor.execute( "SELECT id, username, type, status, progress, message, result FROM jobs WHERE username = ? AND status IN ('pending', 'processing') ORDER BY created_at DESC LIMIT 1", (username,) ) row = cursor.fetchone() conn.close() if not row: self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"has_active": False}).encode('utf-8')) return job_data = { "has_active": True, "job_id": row[0], "username": row[1], "type": row[2], "status": row[3], "progress": row[4], "message": row[5], "result": json.loads(row[6]) if row[6] and (row[6].startswith('{') or row[6].startswith('[')) else row[6] } self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps(job_data).encode('utf-8')) except Exception as e: self.send_error_response(500, str(e)) def handle_get_gallery(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return images_dir = get_generated_dir("images", username) models_dir = get_generated_dir("models", username) os.makedirs(images_dir, exist_ok=True) os.makedirs(models_dir, exist_ok=True) items = [] # Read 2D images for f in os.listdir(images_dir): if f.endswith(('.png', '.jpg', '.jpeg', '.webp')): path = os.path.join(images_dir, f) mtime = os.path.getmtime(path) items.append({ "name": f, "type": "image", "url": f"/generated_images/{username}/{f}", "mtime": mtime }) # Read 3D models for f in os.listdir(models_dir): if f.endswith('.glb') and not f.endswith('_dirty.glb') and not f.endswith('_temp.glb'): path = os.path.join(models_dir, f) # Sanitize: verify magic header of GLB file try: with open(path, 'rb') as f_magic: magic_bytes = f_magic.read(16) # glTF binary magic is 0x46546C67 ('glTF') if not (len(magic_bytes) >= 4 and magic_bytes[:4] == b'glTF'): print(f"[Gallery] Skipping non-GLB or corrupt file: {f} (magic={magic_bytes[:8]})") continue except Exception: continue mtime = os.path.getmtime(path) fbx_filename = f.replace('.glb', '.fbx') has_fbx = os.path.exists(os.path.join(models_dir, fbx_filename)) # Read metadata if exists meta_path = path.replace(".glb", ".json") detected_category = "unknown" if os.path.exists(meta_path): try: with open(meta_path, "r", encoding="utf-8") as meta_f: meta_data = json.load(meta_f) detected_category = meta_data.get("detectedCategory", "unknown") except Exception as me: print(f"[Gallery] Error reading metadata for {f}: {me}") else: # Extract timestamp/ID from model name (e.g. model_1784353123.glb -> 1784353123) base_clean = f.replace("_quad.glb", "").replace("_clean.glb", "").replace(".glb", "") parts = base_clean.split("_") timestamp = "" for part in parts: if part.isdigit() and len(part) >= 9: timestamp = part break if timestamp: image_name = f"image_{timestamp}.png" image_path = os.path.join(images_dir, image_name) if os.path.exists(image_path): print(f"[Gallery] Backfilling missing metadata for {f} using {image_name}...") try: with open(image_path, "rb") as img_f: image_bytes = img_f.read() # Load token load_dotenv() current_token = os.environ.get('HF_TOKEN', '').strip() detected_category = classify_species(image_bytes, "", current_token) # Save metadata JSON file with open(meta_path, "w", encoding="utf-8") as meta_f: json.dump({ "detectedCategory": detected_category, "prompt": "", "timestamp": mtime }, meta_f, indent=2) except Exception as c_err: print(f"[Gallery] Failed backfilling metadata: {c_err}") items.append({ "name": f, "type": "model", "url": f"/generated_models/{username}/{f}", "fbxUrl": f"/generated_models/{username}/{fbx_filename}" if has_fbx else None, "detectedCategory": detected_category, "mtime": mtime }) # Sort items by creation time (newest first) items.sort(key=lambda x: x["mtime"], reverse=True) self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() try: self.wfile.write(json.dumps({"items": items}).encode('utf-8')) except (BrokenPipeError, ConnectionResetError): pass except Exception as e: print(f"[Backend] Error getting gallery: {e}") try: self.send_error_response(500, str(e)) except (BrokenPipeError, ConnectionResetError): pass def handle_delete_gallery(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) filename = params.get('name', '') item_type = params.get('type', '') if not filename or not item_type: self.send_error_response(400, "Missing name or type") return if item_type == "image": target_dir = get_generated_dir("images", username) elif item_type == "model": target_dir = get_generated_dir("models", username) else: self.send_error_response(400, "Invalid type") return # Security check: avoid directory traversal clean_name = os.path.basename(filename) file_path = os.path.join(target_dir, clean_name) if os.path.exists(file_path): try: os.remove(file_path) print(f"[Backend] Deleted file: {file_path}") except Exception as file_err: raise Exception(f"El archivo está siendo usado por otro programa (ej: Blender). Detalles: {file_err}") # If it was a 3D model, clean up associated files (.obj, .mtl, .fbx, _dirty.glb, _texture.png) if item_type == "model" and clean_name.endswith(".glb"): prefix = clean_name.replace(".glb", "") for ext in [".obj", ".mtl", ".fbx", "_dirty.glb", "_clean.glb", "_clean.fbx", "_texture.png", "_rigged.fbx", "_rigged.glb", ".json"]: assoc_file = os.path.join(target_dir, prefix + ext) if os.path.exists(assoc_file): try: os.remove(assoc_file) print(f"[Backend] Deleted associated file: {assoc_file}") except Exception as assoc_err: print(f"[Backend] Warning: could not delete associated file {assoc_file}: {assoc_err}") self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True}).encode('utf-8')) else: self.send_error_response(404, "File not found") except Exception as e: print(f"[Backend] Error deleting gallery item: {e}") self.send_error_response(500, str(e)) def handle_save_weights(self): try: username = self.get_logged_in_user() if not username: self.send_error_response(401, "No has iniciado sesión.") return content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) params = json.loads(post_data.decode('utf-8')) model_url = params.get('modelUrl', '') glb_base64 = params.get('glbBase64', '') if not model_url or not glb_base64: self.send_error_response(400, "Missing modelUrl or glbBase64 data") return filename = os.path.basename(model_url) output_dir = get_generated_dir("models", username) dest_path = os.path.join(output_dir, filename) if not os.path.exists(dest_path): self.send_error_response(404, f"Model file {filename} not found") return # Extract base64 binary if ',' in glb_base64: glb_base64 = glb_base64.split(',')[1] glb_bytes = base64.b64decode(glb_base64) # Write updated GLB with open(dest_path, "wb") as f: f.write(glb_bytes) print(f"[Backend] Saved updated GLB weights for: {dest_path}") # Check if there is an associated FBX (regenerate it) fbx_filename = filename.replace(".glb", ".fbx") fbx_dest_path = os.path.join(output_dir, fbx_filename) blender_path = os.environ.get('BLENDER_PATH', '') if not blender_path or not os.path.exists(blender_path): blender_path = shutil.which("blender") or "" if blender_path and os.path.exists(blender_path): print(f"[Backend] Regenerating FBX from updated GLB weights...") import subprocess script_path = os.path.join(os.path.dirname(__file__), "scripts", "blender", "glb_to_fbx_weights.py") with open(script_path, "w", encoding="utf-8") as f_script: f_script.write('''import bpy import sys import json def strip_gltf_extensions(glb_path): try: with open(glb_path, "rb") as f: data = f.read() if len(data) < 20 or data[:4] != b'glTF': return json_len = int.from_bytes(data[12:16], byteorder='little') json_bytes = data[20:20+json_len] gltf_json = json.loads(json_bytes.decode('utf-8', errors='ignore')) modified = False for key in ['extensionsRequired', 'extensionsUsed']: if key in gltf_json and 'EXT_texture_webp' in gltf_json[key]: gltf_json[key].remove('EXT_texture_webp') modified = True if modified: new_bytes = json.dumps(gltf_json).encode('utf-8') if len(new_bytes) <= len(json_bytes): new_bytes = new_bytes.ljust(len(json_bytes), b' ') new_data = data[:20] + new_bytes + data[20+len(json_bytes):] with open(glb_path, "wb") as f: f.write(new_data) print(f"[Blender] Stripped EXT_texture_webp extension requirement from GLB.") except Exception as e: print(f"[Blender] Extension strip note: {e}") args = sys.argv[sys.argv.index("--") + 1:] glb_in = args[0] fbx_out = args[1] strip_gltf_extensions(glb_in) bpy.ops.wm.read_factory_settings(use_empty=True) print(f"Importing GLB: {glb_in}") bpy.ops.import_scene.gltf(filepath=glb_in) print(f"Exporting FBX: {fbx_out}") bpy.ops.export_scene.fbx( filepath=fbx_out, use_selection=False, object_types={'ARMATURE', 'MESH'}, use_mesh_modifiers=True, add_leaf_bones=False, bake_anim=False ) print("FBX conversion completed successfully.") ''') cmd = [blender_path, "--background", "--python", script_path, "--", dest_path, fbx_dest_path] print(f"[Backend] Executing: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) print(f"[Backend] Blender Output:\n{result.stdout}") if result.stderr: print(f"[Backend] Blender Errors:\n{result.stderr}") try: os.remove(script_path) except: pass self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"success": True, "fbxUrl": f"/generated_models/{username}/{fbx_filename}" if os.path.exists(fbx_dest_path) else None}).encode('utf-8')) except Exception as e: print(f"[Backend] Error saving weights: {e}") self.send_error_response(500, str(e)) def send_error_response(self, code, message): self.send_response(code) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({"error": message}).encode('utf-8')) def run_server(): init_db() server_address = ('', PORT) httpd = ThreadingHTTPServer(server_address, F23DHTTPRequestHandler) print(f"[Backend] 23DFactory server running at http://localhost:{PORT}") try: httpd.serve_forever() except KeyboardInterrupt: print("\n[Backend] Server shutting down.") httpd.server_close() if __name__ == '__main__': run_server()