""" edit_api.py — headless throughput API for Qwen-Image-Edit Rapid-AIO (v23 Q8 GGUF) running on top of a local ComfyUI server. Flow per request: image + prompt -> upload to ComfyUI -> inject into the workflow graph -> queue -> poll until done -> return the edited PNG. Run ComfyUI first (run_comfyui.sh), then this service (start_api.sh). """ import io import os os.environ["HF_HUB_OFFLINE"] = "1" os.environ["TRANSFORMERS_OFFLINE"] = "1" import json import time import uuid import random import copy import threading import csv import subprocess try: from . import database from . import embeddings from . import naming except ImportError: import database import embeddings import naming import requests from PIL import Image from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel import shutil import re from typing import Union, Any # --- config ----------------------------------------------------------------- _HERE = os.path.dirname(os.path.abspath(__file__)) CONFIG_PATH = os.path.join(_HERE, "config.json") WD_MODEL = os.environ.get("WD_MODEL", "SmilingWolf/wd-vit-tagger-v3") COMFY = os.environ.get("COMFY_URL", "http://127.0.0.1:8188").rstrip("/") WORKFLOW_PATH = os.environ.get( "WORKFLOW_PATH", os.path.join(os.path.dirname(os.path.abspath(__file__)), "workflow_qwen_edit.json"), ) # Default target pixel area for the output latent. # We currently cap at ~1MP by default; raise via MAX_AREA env if you want bigger output. # A6000 48GB is not VRAM-bound here, so default to a ~2MP output budget. # This comfortably allows full-HD-ish outputs like 1920x1080. # Override via MAX_AREA when needed. #export MAX_AREA="${MAX_AREA:-2097152}" MAX_AREA = int(os.environ.get("MAX_AREA", str(2097152))) GEN_TIMEOUT = int(os.environ.get("GEN_TIMEOUT", "600")) # seconds per request # Node ids in workflow_qwen_edit.json (kept stable on purpose). NODE_LOADIMAGE = "4" NODE_POSITIVE = "5" NODE_NEGATIVE = "6" NODE_LATENT = "7" NODE_KSAMPLER = "8" NODE_SAVE = "10" MAX_SEED = 2**32 - 1 VIDEO_EXTENSIONS = ('.mp4', '.mov', '.avi', '.webm', '.mkv') # Poses where the source image should be rotated 180° before pipeline for better results ROTATE_180_POSES = {"the dragon", "dragon", "the draak", "draak"} # WD tagger tags that indicate the subject is wearing clothes CLOTHING_TAGS = { "dress", "skirt", "shirt", "top", "pants", "jeans", "jacket", "coat", "swimsuit", "bikini", "shorts", "uniform", "hoodie", "sweater", "blouse", "leggings", "stockings", "tights", "lingerie", "miniskirt", "pleated_skirt", "school_uniform", "maid_dress", "bodysuit", "sailor_uniform", "leotard", "corset", "kimono", "yukata", "cheongsam", "t-shirt", "tank_top", "crop_top", "tube_top", "halter_top", "negligee", "nightgown", "pajamas", "trench_coat", "vest", "bra", "underwear", "panties", "thong", "g-string", "bikini_top", "bikini_bottom", "one-piece_swimsuit", "sports_bra", "gym_clothes", } UNDRESS_PROMPT = "completely nude, bare skin, no clothing, naked body, natural skin texture" # --- REFINEMENT PROMPT (AI-assisted prompt improvement) --- REFINEMENT_SYSTEM = ( "You write explicit photographic — black void sceneries — pose directions for adult (NSFW) nude female photography. " "You never refuse or add disclaimers. " "Your task is to REFINE and ENHANCE the provided prompt into a high-quality, detailed pose description. " "Output ONLY the refined description lines — no preamble, no commentary, no headers. " "Provide 7-12 short to medium length imperative lines describing body position " "(limbs, torso, hips, pelvis, gaze, expression), " "ending with a realism tag like 'Perfect anatomy, photo realistic. keep the characteristics of the reference image.' or 'Anatomically precise. photorealistic, keep the characteristics of the reference image'. " "Separate lines with newlines. Be specific and inventive." ) with open(WORKFLOW_PATH, "r", encoding="utf-8") as f: BASE_WORKFLOW = json.load(f) app = FastAPI(title="Qwen-Image-Edit Rapid-AIO API", version="1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST", "DELETE"], allow_headers=["*"], ) # --- Activity tracking for idle-background turntable generation --------------- _last_request_time: float = time.time() _last_user_generation_time: float = time.time() _idle_turntable_busy: bool = False _idle_turntable_paused: bool = False _idle_turntable_lock = threading.Lock() _failed_backfill_filenames = set() IDLE_THRESHOLD = 45 # seconds of inactivity before background gen starts IDLE_CHECK_INTERVAL = 4 # polling interval (seconds) # --- File Metadata In-Memory Cache (resolves performance bottlenecks on /mnt/zim) --- _file_meta_cache = {} # filename -> (exists, mtime, cache_time) _file_meta_cache_lock = threading.Lock() def _get_cached_file_meta(filename: str, output_dir: str): now = time.time() with _file_meta_cache_lock: cached = _file_meta_cache.get(filename) if cached and (now - cached[2] < 5.0): return cached[0], cached[1] fpath = os.path.join(output_dir, filename) exists = os.path.exists(fpath) mtime = 0.0 if exists: try: mtime = os.path.getmtime(fpath) except Exception: pass with _file_meta_cache_lock: _file_meta_cache[filename] = (exists, mtime, now) return exists, mtime def _update_cached_file_meta(filename: str, exists: bool = True, mtime: float = None): if mtime is None: mtime = time.time() with _file_meta_cache_lock: _file_meta_cache[filename] = (exists, mtime, time.time()) def _clear_cached_file_meta(filename: str): with _file_meta_cache_lock: if filename in _file_meta_cache: del _file_meta_cache[filename] _last_preloaded_images_set = None @app.middleware("http") async def _track_activity(request, call_next): global _last_request_time _last_request_time = time.time() return await call_next(request) def _sync_preloaded_images(): """Update PRELOADED_IMAGES in car.html (both source and output) to match current DB state.""" global _last_preloaded_images_set try: output_dir = _load_output_dir() paths = [ os.path.join(_HERE, "car.html"), os.path.join(output_dir, "car.html") ] persons = database.list_persons(include_archived=False) # Only include if file actually exists on disk, using the fast cache db_images = [] for p in persons: exists, mtime = _get_cached_file_meta(p[0], output_dir) if exists: db_images.append(p[0]) # Sort by mtime, newest first db_images.sort(key=lambda x: _get_cached_file_meta(x, output_dir)[1], reverse=True) # Avoid redundant HTML rewrites if the preloaded set is unchanged current_set = set(db_images) if _last_preloaded_images_set is not None and _last_preloaded_images_set == current_set: return _last_preloaded_images_set = current_set images_json = json.dumps(db_images, indent=12).strip() # Format for JS insertion images_json = images_json.replace('\n', '\n ') pattern = r'// --- HYDRATION_START ---.*?// --- HYDRATION_END ---' replacement = f'// --- HYDRATION_START ---\n const PRELOADED_IMAGES = {images_json};\n // --- HYDRATION_END ---' for p in paths: if not os.path.exists(p): continue with open(p, 'r') as f: content = f.read() new_content = re.sub(pattern, replacement, content, flags=re.DOTALL) with open(p, 'w') as f: f.write(new_content) print(f"[static] Updated {p} with {len(db_images)} preloaded images") except Exception as e: print(f"[static] Failed to update car.html hydration: {e}") def _sync_frontend(): for name in ["car.html", "trash.html"]: src = os.path.join(_HERE, name) if not os.path.exists(src): continue try: dest = os.path.join(_load_output_dir(), name) shutil.copy2(src, dest) print(f"[{name}] synced → {dest}") except Exception as e: print(f"[{name}] sync warning: {e}") def _watch_frontend(): files = ["car.html", "trash.html"] last_mtimes = {} for name in files: src = os.path.join(_HERE, name) if os.path.exists(src): last_mtimes[name] = os.path.getmtime(src) while True: time.sleep(1) for name in files: src = os.path.join(_HERE, name) if not os.path.exists(src): continue try: mtime = os.path.getmtime(src) if mtime != last_mtimes.get(name): last_mtimes[name] = mtime dest = os.path.join(_load_output_dir(), name) shutil.copy2(src, dest) print(f"[{name}] change detected → synced to {dest}") except Exception: pass def _load_wireframe_dir() -> str: with open(CONFIG_PATH, "r") as f: conf = json.load(f) d = conf.get("wireframe_dir", "/mnt/zim/tour-comfy/wireframe") return os.path.expanduser(d) def _load_faceswap_model_path() -> str: with open(CONFIG_PATH, "r") as f: conf = json.load(f) return os.path.expanduser(conf.get("faceswap_model", "~/.insightface/models/inswapper_128.onnx")) def _run_consistency_check(): """Identifies DB records with missing files, files on disk with no DB record, and archived items.""" try: output_dir = _load_output_dir() data_dir = os.path.join(output_dir, "_data") os.makedirs(data_dir, exist_ok=True) persons = database.list_persons(include_archived=True) missing_files = [] archived_items = [] missing_group = [] for p in persons: filename = p[0] group_id = p[2] if not group_id: missing_group.append({ "filename": filename, "name": p[1] }) is_archived = bool(p[14]) if p[14] else False if is_archived: archived_items.append({ "filename": filename, "group_id": group_id, "name": p[1] }) fpath = os.path.join(output_dir, filename) if not os.path.exists(fpath): missing_files.append({ "filename": filename, "group_id": group_id, "name": p[1] }) # Untracked files db_filenames = set(p[0] for p in persons) untracked_files = [] if os.path.isdir(output_dir): for f in os.listdir(output_dir): if f.lower().endswith(('.png', '.jpg', '.jpeg', '.webp', '.mp4')): # Skip internal folders and data files if f.startswith('_') or f == 'car.html' or f == 'trash.html': continue if f not in db_filenames: untracked_files.append(f) report = { "timestamp": time.time(), "missing_files": missing_files, "untracked_files": untracked_files, "archived_items": archived_items, "missing_group": missing_group } _write_json(os.path.join(data_dir, "inconsistencies.json"), report) print(f"[consistency] check complete: {len(missing_files)} missing, {len(untracked_files)} untracked, {len(archived_items)} archived, {len(missing_group)} missing group") return report except Exception as e: print(f"[consistency] check failed: {e}") return None def _consistency_check_daemon(): """Runs daily consistency check.""" # Wait for startup time.sleep(30) while True: _run_consistency_check() # Sleep for 24 hours time.sleep(86400) def _idle_turntable_daemon(): """ Background daemon: when the API has been idle > IDLE_THRESHOLD seconds, generate the next missing turntable view for the next group. Yields after each view so activity check can stop it promptly. """ global _idle_turntable_busy import sys as _sys if _HERE not in _sys.path: _sys.path.insert(0, _HERE) time.sleep(60) # wait for full startup before touching ComfyUI while True: time.sleep(IDLE_CHECK_INTERVAL) if _idle_turntable_paused: continue if time.time() - _last_user_generation_time < IDLE_THRESHOLD: continue try: output_dir = _load_output_dir() persons = database.list_persons() except Exception as e: print(f"[turntable-bg] db/config error: {e}") continue # Build {group_id: (preferred_filename, best_sort_order)} groups: dict = {} for row in persons: fname, group_id, sort_order = row[0], row[2], row[6] if not group_id: continue if fname.startswith("_turntable/"): continue if group_id not in groups: groups[group_id] = (fname, sort_order) else: cur_sort = groups[group_id][1] if sort_order is not None and (cur_sort is None or sort_order < cur_sort): groups[group_id] = (fname, sort_order) import turntable_cache as tc generated_one = False for group_id, (preferred_fname, _) in groups.items(): src_path = os.path.join(output_dir, preferred_fname) if not os.path.exists(src_path): continue state = tc.load_state(output_dir, group_id) if state is None: state = tc.init_state(output_dir, group_id, src_path, preferred_fname) elif state.get("completed"): continue elif state.get("preferred_filename") != preferred_fname: # Preferred image changed — restart turntable for this group print(f"[turntable-bg] {group_id}: preferred changed, resetting") state = tc.init_state(output_dir, group_id, src_path, preferred_fname) deg = tc.next_missing_angle(state) if deg is None: continue # all angles done but no video yet — build it below # Re-check idle right before the expensive generation if time.time() - _last_user_generation_time < IDLE_THRESHOLD or _idle_turntable_paused: break print(f"[turntable-bg] {group_id}: rendering {deg:.0f}° " f"({len(state['views'])}/{state['n_views']})…") with _idle_turntable_lock: _idle_turntable_busy = True _write_turntable_static() try: from orbit_qwen import yaw_prompt, _autocrop_alpha import io as _io from PIL import Image as _Image base_pil = _Image.open(src_path).convert("RGB") prompt = yaw_prompt(deg) png = _run_pipeline( base_pil, prompt, state["seed"], MAX_AREA, steps=state["steps"] ) view_pil = _Image.open(_io.BytesIO(png)).convert("RGBA") view_pil = _autocrop_alpha(view_pil) views_dir = os.path.join(tc.cache_dir(output_dir, group_id), "views") os.makedirs(views_dir, exist_ok=True) angle_idx = state["angles"].index(deg) vpath = os.path.join(views_dir, f"view_{angle_idx:03d}_{int(deg):03d}deg.png") view_pil.save(vpath) tc.mark_view_done(output_dir, group_id, state, deg, vpath) n_done = len(state["views"]) print(f"[turntable-bg] {group_id}: {deg:.0f}° saved " f"({n_done}/{state['n_views']})") # Register frame in database for filmstrip visibility + Info tab links try: vname = os.path.relpath(vpath, output_dir).replace("\\", "/") database.upsert_person( vname, filepath=vpath, group_id=group_id, prompt=prompt, source_refs=json.dumps([preferred_fname]), sort_order=200 + angle_idx, pose=f"Orbit {int(deg)}°", tags=["ORBIT"] ) _update_cached_file_meta(vname, exists=True) except Exception as db_err: print(f"[turntable-bg] DB registration error: {db_err}") generated_one = True _write_turntable_static() if n_done >= state["n_views"]: _finalize_turntable(output_dir, group_id, state) except Exception as e: import traceback print(f"[turntable-bg] {group_id}: error at {deg:.0f}°: {e}") print(traceback.format_exc()) finally: with _idle_turntable_lock: _idle_turntable_busy = False _write_turntable_static() break # one view per cycle; re-check idle on next loop # If no turntable views were generated, check if any legacy image needs metadata backfill if not generated_one: legacy_candidate = None for row in persons: fname = row[0] content_type = row[12] if len(row) > 12 else None people_count = row[19] if len(row) > 19 else None if fname.startswith("_turntable/"): continue if content_type == 'video': continue if not fname.lower().endswith(('.png', '.jpg', '.jpeg', '.webp')): continue fpath = os.path.join(output_dir, fname) if not os.path.exists(fpath): continue if people_count is None and fname not in _failed_backfill_filenames: legacy_candidate = fname break if legacy_candidate: print(f"[metadata-bg] Idle backfill: processing {legacy_candidate}…") try: res = _process_image_for_metadata(legacy_candidate) if res is None: _failed_backfill_filenames.add(legacy_candidate) except Exception as ex: print(f"[metadata-bg] Error processing legacy backfill for {legacy_candidate}: {ex}") _failed_backfill_filenames.add(legacy_candidate) def _finalize_turntable(output_dir: str, group_id: str, state: dict): """Mark state completed (without building the MP4).""" import turntable_cache as tc try: tc.mark_completed(output_dir, group_id, state, "") print(f"[turntable-bg] {group_id}: complete (custom frame-loop only)") _write_turntable_static() except Exception as e: import traceback print(f"[turntable-bg] {group_id}: finalize error: {e}\n{traceback.format_exc()}") @app.on_event("startup") def on_startup(): try: database.migrate_schema() except Exception as e: print(f"DB migration warning: {e}") _sync_frontend() threading.Thread(target=_watch_frontend, daemon=True).start() threading.Thread(target=_idle_turntable_daemon, daemon=True).start() threading.Thread(target=_consistency_check_daemon, daemon=True).start() threading.Thread(target=_privacy_monitor_daemon, daemon=True).start() # Mount wireframe static dir for browser video preview try: wf_dir = _load_wireframe_dir() if os.path.isdir(wf_dir): app.mount("/wireframe", StaticFiles(directory=wf_dir), name="wireframe") print(f"[wireframe] mounted {wf_dir} → /wireframe") except Exception as e: print(f"[wireframe] mount warning: {e}") # Mount output dir so images can be served via HTTP (/output/filename.png) try: out_dir = _load_output_dir() if os.path.isdir(out_dir): app.mount("/output", StaticFiles(directory=out_dir), name="output") print(f"[output] mounted {out_dir} → /output") except Exception as e: print(f"[output] mount warning: {e}") # Write initial static data files (synchronous — ensures files exist before first request) _write_all_static() # Trigger pose index backfill try: if _load_pose_estimator(): build_pose_index() except Exception: pass # --- helpers ---------------------------------------------------------------- def _round16(x: int) -> int: return max(16, int(round(x / 16.0)) * 16) def _target_size(w: int, h: int, max_area: int) -> tuple[int, int]: """Scale (w, h) to ~max_area preserving aspect, rounded to /16.""" scale = (max_area / float(w * h)) ** 0.5 return _round16(w * scale), _round16(h * scale) def _prep_image(pil: Image.Image, max_area: int) -> tuple[Image.Image, int, int]: """ Prepare image for ComfyUI: 1. Scale (up or down) to fit area while preserving aspect. 2. Ensure dimensions are rounded to 16. """ w, h = pil.width, pil.height rw, rh = _target_size(w, h, max_area) if rw != w or rh != h: pil = pil.resize((rw, rh), resample=Image.LANCZOS) return pil, rw, rh def _comfy_upload(img_bytes: bytes, filename: str) -> str: """Upload an image to ComfyUI's input dir; return the stored name.""" r = requests.post( f"{COMFY}/upload/image", files={"image": (filename, img_bytes, "image/png")}, data={"overwrite": "true", "type": "input"}, timeout=60, ) r.raise_for_status() j = r.json() name = j["name"] sub = j.get("subfolder", "") return f"{sub}/{name}" if sub else name def _comfy_queue(graph: dict, client_id: str) -> str: r = requests.post( f"{COMFY}/prompt", json={"prompt": graph, "client_id": client_id}, timeout=60, ) if r.status_code != 200: raise HTTPException(502, f"ComfyUI rejected workflow: {r.text}") return r.json()["prompt_id"] def _comfy_wait(prompt_id: str, deadline: float) -> dict: """Poll /history until the prompt finishes; return its outputs dict.""" global _last_user_generation_time while time.time() < deadline: if not _idle_turntable_busy: _last_user_generation_time = time.time() r = requests.get(f"{COMFY}/history/{prompt_id}", timeout=30) if r.status_code == 200: hist = r.json() if prompt_id in hist: entry = hist[prompt_id] status = entry.get("status", {}) if status.get("status_str") == "error": raise HTTPException(500, f"ComfyUI execution error: {json.dumps(status)}") outputs = entry.get("outputs", {}) if outputs: return outputs time.sleep(0.5) raise HTTPException(504, f"Generation timed out after {GEN_TIMEOUT}s") def _comfy_fetch_image(outputs: dict) -> bytes: node_out = outputs.get(NODE_SAVE) or next( (v for v in outputs.values() if "images" in v), None ) if not node_out or not node_out.get("images"): raise HTTPException(500, "No output image produced") img = node_out["images"][0] r = requests.get( f"{COMFY}/view", params={ "filename": img["filename"], "subfolder": img.get("subfolder", ""), "type": img.get("type", "output"), }, timeout=60, ) r.raise_for_status() return r.content # --- WD tagger (lazy) ------------------------------------------------------- _tagger = None # (model, transform, labels) once loaded _tagger_lock = threading.Lock() def _load_tagger(): global _tagger if _tagger is not None: return _tagger with _tagger_lock: if _tagger is not None: return _tagger import torch import timm from timm.data import create_transform, resolve_data_config import huggingface_hub model = timm.create_model(f"hf_hub:{WD_MODEL}", pretrained=True).eval() if torch.cuda.is_available(): model = model.cuda() cfg = resolve_data_config(model.pretrained_cfg, model=model) transform = create_transform(**cfg) lpath = huggingface_hub.hf_hub_download(WD_MODEL, "selected_tags.csv", local_files_only=True) with open(lpath, newline="") as f: rows = list(csv.DictReader(f)) # category 0=general 4=character 9=rating labels = [(r["name"], int(r.get("category", 9))) for r in rows] _tagger = (model, transform, labels) return _tagger def _run_tagger(pil_img: Image.Image, threshold: float = 0.35): import torch model, transform, labels = _load_tagger() with embeddings._gpu_lock: tensor = transform(pil_img.convert("RGB")).unsqueeze(0) if torch.cuda.is_available(): tensor = tensor.cuda() with torch.no_grad(): scores = torch.sigmoid(model(tensor))[0].cpu().tolist() tags = [ {"tag": name, "score": round(score, 3), "cat": cat} for (name, cat), score in zip(labels, scores) if score >= threshold ] tags.sort(key=lambda x: -x["score"]) return tags def _tags_to_name(tags: list, max_tags: int = 8) -> str: content = [t["tag"] for t in tags if t["cat"] in (0, 4)][:max_tags] return " ".join(content).replace("_", " ") def _apply_transparency(png_bytes: bytes) -> bytes: """Use rembg to remove background and return PNG bytes with Alpha channel.""" try: from rembg import remove import io from PIL import Image img = Image.open(io.BytesIO(png_bytes)) # rembg works best on RGB if img.mode != "RGB": img = img.convert("RGB") out = remove(img) buf = io.BytesIO() out.save(buf, format="PNG") return buf.getvalue() except Exception as e: print(f"Error in transparency post-processing: {e}") return png_bytes # --- faceswapper (insightface + inswapper_128) -------------------------------- # Setup: pip install insightface onnxruntime-gpu opencv-python-headless # Download: place inswapper_128.onnx at ~/.insightface/models/inswapper_128.onnx # Source: https://huggingface.co/deepinsight/inswapper _faceswapper = None _faceswapper_lock = threading.Lock() # Dedicated single-worker pool for face-crop extraction. Running it here # instead of via FastAPI BackgroundTasks keeps the heavy insightface inference # (and its one-time model load) off the shared request threadpool, so quick # endpoints like /order stay responsive right after a "set preferred" click. # A single worker also serializes face jobs so a burst can't thrash the GPU. from concurrent.futures import ThreadPoolExecutor as _ThreadPoolExecutor _face_executor = _ThreadPoolExecutor(max_workers=1, thread_name_prefix="face") _gfpgan = None _gfpgan_lock = threading.Lock() def _load_faceswapper(): global _faceswapper if _faceswapper is not None: return _faceswapper with _faceswapper_lock: if _faceswapper is not None: return _faceswapper try: import insightface from insightface.app import FaceAnalysis except ImportError: raise RuntimeError("insightface not installed. Run: pip install insightface onnxruntime-gpu") providers = ['CUDAExecutionProvider', 'CPUExecutionProvider'] app = FaceAnalysis(name='buffalo_l', providers=providers) app.prepare(ctx_id=0, det_size=(640, 640)) model_path = _load_faceswap_model_path() if not os.path.exists(model_path): # Try HuggingFace download as fallback try: import huggingface_hub model_path = huggingface_hub.hf_hub_download( 'deepinsight/inswapper', 'inswapper_128.onnx', local_dir=os.path.dirname(model_path), local_files_only=True ) print(f"[faceswap] Downloaded inswapper_128.onnx to {model_path}") except Exception as de: raise RuntimeError( f"inswapper_128.onnx not found at {model_path}. " f"Download from https://huggingface.co/deepinsight/inswapper and place it there. " f"Download error: {de}" ) swapper = insightface.model_zoo.get_model(model_path, providers=providers) _faceswapper = (app, swapper) print(f"[faceswap] loaded insightface buffalo_l + inswapper_128 from {model_path}") return _faceswapper def _load_gfpgan(): """Lazy-load GFPGAN face restorer. Returns restorer or False if unavailable.""" global _gfpgan if _gfpgan is not None: return _gfpgan with _gfpgan_lock: if _gfpgan is not None: return _gfpgan try: from gfpgan import GFPGANer # Main GFPGAN model model_path = os.path.expanduser('~/.gfpgan/weights/GFPGANv1.4.pth') os.makedirs(os.path.dirname(model_path), exist_ok=True) if not os.path.exists(model_path): import urllib.request print('[gfpgan] Downloading GFPGANv1.4.pth (~333 MB)...') tmp = model_path + '.tmp' urllib.request.urlretrieve( 'https://github.com/TencentARC/GFPGAN/releases/download/v1.3.4/GFPGANv1.4.pth', tmp ) os.rename(tmp, model_path) # GFPGANer hardcodes facexlib download path to CWD/gfpgan/weights/ # → change CWD to ~ so models land at ~/gfpgan/weights/ (stable across runs) home = os.path.expanduser('~') os.makedirs(os.path.join(home, 'gfpgan', 'weights'), exist_ok=True) _orig_cwd = os.getcwd() os.chdir(home) try: restorer = GFPGANer(model_path=model_path, upscale=1, arch='clean', channel_multiplier=2, bg_upsampler=None) finally: os.chdir(_orig_cwd) _gfpgan = restorer print('[gfpgan] GFPGANv1.4 loaded') except Exception as e: print(f'[gfpgan] not available: {e}') _gfpgan = False return _gfpgan def _make_video_poster(video_path: str) -> str | None: """Extract a poster JPG (sibling `.jpg`) so the gallery can show a thumbnail for a video via a plain (file:// can't render