Lazy-load torch so health starts without model import

app.py

@@ -1,580 +1,584 @@
-from __future__ import annotations
-
-import hashlib
-import io
-import os
-import threading
-import time
-from pathlib import Path
-from urllib.parse import urlparse
-
-import cv2
-import numpy as np
-import onnxruntime as ort
-import torch
-from fastapi import FastAPI, File, Form, HTTPException, UploadFile
-from fastapi.responses import Response
-from huggingface_hub import hf_hub_download
-from PIL import Image
-from torch.hub import download_url_to_file, get_dir
-
-
-LAMA_MODEL_REPO = os.getenv("LAMA_MODEL_REPO", "Carve/LaMa-ONNX")
-LAMA_MODEL_FILE = os.getenv("LAMA_MODEL_FILE", "lama_fp32.onnx")
-MIGAN_MODEL_URL = os.getenv(
-    "MIGAN_MODEL_URL",
-    "https://github.com/Sanster/models/releases/download/migan/migan_traced.pt",
-)
-MIGAN_MODEL_MD5 = os.getenv("MIGAN_MODEL_MD5", "76eb3b1a71c400ee3290524f7a11b89c")
-DEFAULT_MODE = os.getenv("DEFAULT_MODE", "quality")
-LOAD_MODEL_ON_STARTUP = os.getenv("LOAD_MODEL_ON_STARTUP", "1") == "1"
-PROVIDERS = ["CPUExecutionProvider"]
-MODES = {
-    "quality": "selfbuilt_lama_onnx",
-    "fast": "iopaint_migan_torchscript",
-}
-
-DOUBAO_ALPHA_ASSET = Path(__file__).with_name("doubao_alpha.png")
-DOUBAO_WIDTH_FRAC = 0.22
-DOUBAO_HEIGHT_FRAC = 0.075
-DOUBAO_MARGIN_RIGHT_FRAC = 0.004
-DOUBAO_MARGIN_BOTTOM_FRAC = 0.004
-DOUBAO_MAX_SATURATION = 55
-DOUBAO_LOGO_MIN_LUMA = 150
-DOUBAO_TOPHAT_DELTA = 12
-DOUBAO_DETECT_MIN_COVERAGE = 0.04
-DOUBAO_DETECT_NCC_THRESHOLD = 0.4
-DOUBAO_ALPHA_WIDTH_FRAC = 0.1636
-DOUBAO_ALPHA_HEIGHT_FRAC = 0.0405
-DOUBAO_ALPHA_MARGIN_RIGHT_FRAC = 0.0132
-DOUBAO_ALPHA_MARGIN_BOTTOM_FRAC = 0.0166
-DOUBAO_ALPHA_ALIGN_SEARCH = (0.88, 1.12, 25)
-DOUBAO_RESIDUAL_ALPHA_FLOOR = 0.05
-DOUBAO_TEMPLATE_DILATE = 4
-
-LAMA_SESSION: ort.InferenceSession | None = None
-MIGAN_MODEL: torch.jit.ScriptModule | None = None
-DOUBAO_ALPHA: np.ndarray | None = None
-DOUBAO_SILHOUETTE: np.ndarray | None = None
-MODEL_LOCK = threading.RLock()
-
-app = FastAPI(title="Image Services Inpaint")
-
-
-def _md5sum(path: Path) -> str:
-    digest = hashlib.md5()
-    with path.open("rb") as handle:
-        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
-
-
-def _cache_path_by_url(url: str) -> Path:
-    parts = urlparse(url)
-    model_dir = Path(get_dir()) / "checkpoints"
-    model_dir.mkdir(parents=True, exist_ok=True)
-    return model_dir / Path(parts.path).name
-
-
-def _download_torchscript(url: str, expected_md5: str) -> Path:
-    path = _cache_path_by_url(url)
-    if not path.exists():
-        download_url_to_file(url, str(path), None, progress=True)
-    actual_md5 = _md5sum(path)
-    if actual_md5 != expected_md5:
-        try:
-            path.unlink()
-        finally:
-            raise RuntimeError(f"Model md5 mismatch: {actual_md5} != {expected_md5}")
-    return path
-
-
-def _load_lama_session() -> ort.InferenceSession:
-    global LAMA_SESSION
-    if LAMA_SESSION is not None:
-        return LAMA_SESSION
-    with MODEL_LOCK:
-        if LAMA_SESSION is not None:
-            return LAMA_SESSION
-        model_path = hf_hub_download(repo_id=LAMA_MODEL_REPO, filename=LAMA_MODEL_FILE)
-        LAMA_SESSION = ort.InferenceSession(model_path, providers=PROVIDERS)
-        return LAMA_SESSION
-
-
-def _load_migan_model() -> torch.jit.ScriptModule:
-    global MIGAN_MODEL
-    if MIGAN_MODEL is not None:
-        return MIGAN_MODEL
-    with MODEL_LOCK:
-        if MIGAN_MODEL is not None:
-            return MIGAN_MODEL
-        model_path = _download_torchscript(MIGAN_MODEL_URL, MIGAN_MODEL_MD5)
-        MIGAN_MODEL = torch.jit.load(str(model_path), map_location="cpu").eval()
-        return MIGAN_MODEL
-
-
-def _prepare_mask(mask: Image.Image, size: tuple[int, int]) -> np.ndarray:
-    arr = np.array(mask.convert("L").resize(size, Image.Resampling.NEAREST), copy=True)
-    return np.where(arr > 127, 255, 0).astype(np.uint8)
-
-
-def _load_doubao_alpha() -> np.ndarray:
-    global DOUBAO_ALPHA
-    if DOUBAO_ALPHA is not None:
-        return DOUBAO_ALPHA
-    alpha = cv2.imread(str(DOUBAO_ALPHA_ASSET), cv2.IMREAD_GRAYSCALE)
-    if alpha is None:
-        raise RuntimeError(f"Missing Doubao alpha asset: {DOUBAO_ALPHA_ASSET}")
-    DOUBAO_ALPHA = alpha.astype(np.float32) / 255.0
-    return DOUBAO_ALPHA
-
-
-def _load_doubao_silhouette() -> np.ndarray:
-    global DOUBAO_SILHOUETTE
-    if DOUBAO_SILHOUETTE is not None:
-        return DOUBAO_SILHOUETTE
-    DOUBAO_SILHOUETTE = (_load_doubao_alpha() > 0.15).astype(np.uint8) * 255
-    return DOUBAO_SILHOUETTE
-
-
-def _doubao_locate(image_bgr: np.ndarray) -> tuple[int, int, int, int]:
-    height, width = image_bgr.shape[:2]
-    mark_w = max(40, int(width * DOUBAO_WIDTH_FRAC))
-    mark_h = max(16, int(width * DOUBAO_HEIGHT_FRAC))
-    margin_right = max(4, int(width * DOUBAO_MARGIN_RIGHT_FRAC))
-    margin_bottom = max(4, int(width * DOUBAO_MARGIN_BOTTOM_FRAC))
-    x = max(0, width - margin_right - mark_w)
-    y = max(0, height - margin_bottom - mark_h)
-    return x, y, min(mark_w, width - x), min(mark_h, height - y)
-
-
-def _doubao_extract_candidate_mask(image_bgr: np.ndarray, loc: tuple[int, int, int, int]) -> np.ndarray:
-    x, y, width, height = loc
-    if width < 16 or height < 16:
-        return np.zeros((height, width), np.uint8)
-    roi = image_bgr[y : y + height, x : x + width].astype(np.float32)
-    luma = roi.mean(axis=2)
-    saturation = roi.max(axis=2) - roi.min(axis=2)
-    grayish = saturation < DOUBAO_MAX_SATURATION
-    sigma = max(4.0, height * 0.4)
-    local_bg = cv2.GaussianBlur(luma, (0, 0), sigmaX=sigma, sigmaY=sigma)
-    tophat = luma - local_bg
-    glyph = (
-        grayish
-        & (tophat > DOUBAO_TOPHAT_DELTA)
-        & (luma > DOUBAO_LOGO_MIN_LUMA)
-    ).astype(np.uint8) * 255
-    glyph = cv2.morphologyEx(glyph, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))
-    return cv2.morphologyEx(glyph, cv2.MORPH_OPEN, np.ones((5, 5), np.uint8))
-
-
-def _doubao_template_match_score(box_mask: np.ndarray, image_width: int) -> float:
-    silhouette = _load_doubao_silhouette()
-    if box_mask.size == 0:
-        return 0.0
-    gw = min(box_mask.shape[1] - 1, max(8, int(DOUBAO_ALPHA_WIDTH_FRAC * image_width)))
-    gh = min(box_mask.shape[0] - 1, max(4, int(DOUBAO_ALPHA_HEIGHT_FRAC * image_width)))
-    if gw < 8 or gh < 4:
-        return 0.0
-    template = cv2.resize(silhouette, (gw, gh), interpolation=cv2.INTER_NEAREST)
-    return float(cv2.matchTemplate(box_mask, template, cv2.TM_CCOEFF_NORMED).max())
-
-
-def _doubao_aligned_alpha(image_bgr: np.ndarray) -> tuple[np.ndarray, tuple[int, int, int, int]] | None:
-    alpha = _load_doubao_alpha()
-    silhouette = _load_doubao_silhouette()
-    image_width = image_bgr.shape[1]
-    bx, by, bw, bh = _doubao_locate(image_bgr)
-    box_mask = _doubao_extract_candidate_mask(image_bgr, (bx, by, bw, bh))
-    expected = DOUBAO_ALPHA_WIDTH_FRAC * image_width
-    best: tuple[float, int, int, int, int] | None = None
-    for scale in np.linspace(*DOUBAO_ALPHA_ALIGN_SEARCH):
-        gw = int(expected * scale)
-        gh = int(DOUBAO_ALPHA_HEIGHT_FRAC * image_width * scale)
-        if gw < 8 or gh < 4 or gw >= bw or gh >= bh:
-            continue
-        template = cv2.resize(silhouette, (gw, gh), interpolation=cv2.INTER_NEAREST)
-        _, score, _, top_left = cv2.minMaxLoc(cv2.matchTemplate(box_mask, template, cv2.TM_CCOEFF_NORMED))
-        if best is None or score > best[0]:
-            best = (score, gw, gh, top_left[0], top_left[1])
-    if best is None:
-        return None
-    _, gw, gh, ox, oy = best
-    return cv2.resize(alpha, (gw, gh), interpolation=cv2.INTER_LINEAR), (bx + ox, by + oy, gw, gh)
-
-
-def _doubao_fixed_alpha(image_bgr: np.ndarray) -> tuple[np.ndarray, tuple[int, int, int, int]]:
-    alpha = _load_doubao_alpha()
-    image_h, image_w = image_bgr.shape[:2]
-    gw = min(image_w, max(1, int(DOUBAO_ALPHA_WIDTH_FRAC * image_w)))
-    gh = min(image_h, max(1, int(DOUBAO_ALPHA_HEIGHT_FRAC * image_w)))
-    ax = max(0, image_w - int(DOUBAO_ALPHA_MARGIN_RIGHT_FRAC * image_w) - gw)
-    ay = max(0, image_h - int(DOUBAO_ALPHA_MARGIN_BOTTOM_FRAC * image_w) - gh)
-    return cv2.resize(alpha, (gw, gh), interpolation=cv2.INTER_LINEAR), (ax, ay, gw, gh)
-
-
-def generate_doubao_template_bbox_mask(image_bgr: np.ndarray) -> tuple[np.ndarray, dict[str, object]]:
-    if min(image_bgr.shape[:2]) < 200:
-        raise ValueError("Image is too small for Doubao watermark detection.")
-    loc = _doubao_locate(image_bgr)
-    candidate_mask = _doubao_extract_candidate_mask(image_bgr, loc)
-    coverage = float((candidate_mask > 0).sum()) / float(max(1, loc[2] * loc[3]))
-    confidence = _doubao_template_match_score(candidate_mask, image_bgr.shape[1]) if coverage >= DOUBAO_DETECT_MIN_COVERAGE else 0.0
-    detected = confidence >= DOUBAO_DETECT_NCC_THRESHOLD
-    if not detected:
-        raise ValueError(f"Doubao watermark not detected. confidence={confidence:.3f} coverage={coverage:.3f}")
-
-    placed = _doubao_aligned_alpha(image_bgr) or _doubao_fixed_alpha(image_bgr)
-    alpha_block, (x, y, width, height) = placed
-    silhouette = (alpha_block > DOUBAO_RESIDUAL_ALPHA_FLOOR).astype(np.uint8) * 255
-    if int((silhouette > 0).sum()) == 0:
-        raise ValueError("Detected Doubao watermark but generated an empty mask.")
-    glyph_mask = np.zeros(image_bgr.shape[:2], np.uint8)
-    glyph_mask[y : y + height, x : x + width] = silhouette
-    if DOUBAO_TEMPLATE_DILATE > 0:
-        kernel = cv2.getStructuringElement(
-            cv2.MORPH_ELLIPSE,
-            (2 * DOUBAO_TEMPLATE_DILATE + 1, 2 * DOUBAO_TEMPLATE_DILATE + 1),
-        )
-        glyph_mask = cv2.dilate(glyph_mask, kernel)
-
-    ys, xs = np.where(glyph_mask > 0)
-    if len(xs) == 0 or len(ys) == 0:
-        raise ValueError("Detected Doubao watermark but bbox mask is empty.")
-    bbox_mask = np.zeros_like(glyph_mask)
-    x0, x1 = int(xs.min()), int(xs.max()) + 1
-    y0, y1 = int(ys.min()), int(ys.max()) + 1
-    bbox_mask[y0:y1, x0:x1] = 255
-    return bbox_mask, {
-        "confidence": round(confidence, 4),
-        "coverage": round(coverage, 4),
-        "bbox": [x0, y0, x1 - x0, y1 - y0],
-    }
-
-
-def _mask_bbox(mask: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
-    ys, xs = np.where(mask > 0)
-    if len(xs) == 0 or len(ys) == 0:
-        raise ValueError("Mask is empty.")
-    return ys, xs
-
-
-def _ceil_modulo(value: int, mod: int) -> int:
-    if value % mod == 0:
-        return value
-    return (value // mod + 1) * mod
-
-
-def _pad_img_to_modulo(
-    image: np.ndarray,
-    *,
-    mod: int,
-    square: bool = False,
-    min_size: int | None = None,
-) -> np.ndarray:
-    if image.ndim == 2:
-        image = image[:, :, None]
-    height, width = image.shape[:2]
-    out_height = _ceil_modulo(height, mod)
-    out_width = _ceil_modulo(width, mod)
-    if min_size is not None:
-        out_width = max(min_size, out_width)
-        out_height = max(min_size, out_height)
-    if square:
-        side = max(out_height, out_width)
-        out_height = side
-        out_width = side
-    return np.pad(
-        image,
-        ((0, out_height - height), (0, out_width - width), (0, 0)),
-        mode="symmetric",
-    )
-
-
-def _resize_max_size(image: np.ndarray, size_limit: int, interpolation: int = cv2.INTER_CUBIC) -> np.ndarray:
-    height, width = image.shape[:2]
-    if max(height, width) <= size_limit:
-        return image
-    ratio = size_limit / max(height, width)
-    return cv2.resize(image, (int(width * ratio + 0.5), int(height * ratio + 0.5)), interpolation=interpolation)
-
-
-def _norm_img(image: np.ndarray) -> np.ndarray:
-    if image.ndim == 2:
-        image = image[:, :, None]
-    image = np.transpose(image, (2, 0, 1))
-    return image.astype("float32") / 255.0
-
-
-def _boxes_from_mask(mask: np.ndarray) -> list[np.ndarray]:
-    _, thresh = cv2.threshold(mask, 127, 255, 0)
-    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
-    boxes: list[np.ndarray] = []
-    height, width = mask.shape[:2]
-    for contour in contours:
-        x, y, box_width, box_height = cv2.boundingRect(contour)
-        box = np.array([x, y, x + box_width, y + box_height]).astype(int)
-        box[::2] = np.clip(box[::2], 0, width)
-        box[1::2] = np.clip(box[1::2], 0, height)
-        boxes.append(box)
-    return boxes
-
-
-def _crop_box(
-    image: np.ndarray,
-    mask: np.ndarray,
-    box: np.ndarray,
-    *,
-    margin: int,
-) -> tuple[np.ndarray, np.ndarray, tuple[int, int, int, int]]:
-    box_h = int(box[3] - box[1])
-    box_w = int(box[2] - box[0])
-    cx = int((box[0] + box[2]) // 2)
-    cy = int((box[1] + box[3]) // 2)
-    img_h, img_w = image.shape[:2]
-
-    crop_w = box_w + margin * 2
-    crop_h = box_h + margin * 2
-    raw_l = cx - crop_w // 2
-    raw_r = cx + crop_w // 2
-    raw_t = cy - crop_h // 2
-    raw_b = cy + crop_h // 2
-
-    left = max(raw_l, 0)
-    right = min(raw_r, img_w)
-    top = max(raw_t, 0)
-    bottom = min(raw_b, img_h)
-
-    if raw_l < 0:
-        right += abs(raw_l)
-    if raw_r > img_w:
-        left -= raw_r - img_w
-    if raw_t < 0:
-        bottom += abs(raw_t)
-    if raw_b > img_h:
-        top -= raw_b - img_h
-
-    left = max(left, 0)
-    right = min(right, img_w)
-    top = max(top, 0)
-    bottom = min(bottom, img_h)
-    return image[top:bottom, left:right, :], mask[top:bottom, left:right], (left, top, right, bottom)
-
-
-def _erase_lama_onnx(image_bgr: np.ndarray, mask: np.ndarray) -> np.ndarray:
-    session = _load_lama_session()
-    inputs = session.get_inputs()
-    image_name = inputs[0].name
-    mask_name = inputs[1].name
-    dims = inputs[0].shape
-    size = next((dim for dim in reversed(dims) if isinstance(dim, int) and dim > 1), 512)
-
-    height, width = image_bgr.shape[:2]
-    ys, xs = _mask_bbox(mask)
-
-    pad = max(16, int(0.4 * max(xs.max() - xs.min() + 1, ys.max() - ys.min() + 1)))
-    cx0 = max(0, int(xs.min()) - pad)
-    cy0 = max(0, int(ys.min()) - pad)
-    cx1 = min(width, int(xs.max()) + 1 + pad)
-    cy1 = min(height, int(ys.max()) + 1 + pad)
-
-    crop = image_bgr[cy0:cy1, cx0:cx1]
-    crop_mask = mask[cy0:cy1, cx0:cx1]
-    crop_h, crop_w = crop.shape[:2]
-
-    crop_rs = cv2.resize(crop, (size, size), interpolation=cv2.INTER_AREA)
-    mask_rs = cv2.resize(crop_mask, (size, size), interpolation=cv2.INTER_NEAREST)
-
-    image_input = cv2.cvtColor(crop_rs, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
-    image_input = np.transpose(image_input, (2, 0, 1))[None]
-    mask_input = (mask_rs > 127).astype(np.float32)[None, None]
-
-    output = session.run(None, {image_name: image_input, mask_name: mask_input})[0]
-    output = np.asarray(output)[0]
-    output = np.transpose(output, (1, 2, 0))
-    if float(output.max()) <= 1.5:
-        output = output * 255.0
-    output = np.clip(output, 0, 255).astype(np.uint8)
-    output_bgr = cv2.cvtColor(output, cv2.COLOR_RGB2BGR)
-    output_crop = cv2.resize(output_bgr, (crop_w, crop_h), interpolation=cv2.INTER_LINEAR)
-
-    result = image_bgr.copy()
-    region = result[cy0:cy1, cx0:cx1]
-    paste = crop_mask > 127
-    region[paste] = output_crop[paste]
-    result[cy0:cy1, cx0:cx1] = region
-    return result
-
-
-@torch.inference_mode()
-def _migan_forward(model: torch.jit.ScriptModule, image_rgb: np.ndarray, mask: np.ndarray) -> np.ndarray:
-    image = _norm_img(image_rgb) * 2 - 1
-    mask_binary = (mask > 120).astype(np.uint8) * 255
-    mask_input = _norm_img(mask_binary)
-    image_tensor = torch.from_numpy(image).unsqueeze(0)
-    mask_tensor = torch.from_numpy(mask_input).unsqueeze(0)
-    erased = image_tensor * (1 - mask_tensor)
-    model_input = torch.cat([0.5 - mask_tensor, erased], dim=1)
-    output = model(model_input)
-    output = (
-        (output.permute(0, 2, 3, 1) * 127.5 + 127.5)
-        .round()
-        .clamp(0, 255)
-        .to(torch.uint8)
-    )
-    result = output[0].cpu().numpy()
-    return cv2.cvtColor(result, cv2.COLOR_RGB2BGR)
-
-
-def _migan_pad_forward(model: torch.jit.ScriptModule, image_rgb: np.ndarray, mask: np.ndarray) -> np.ndarray:
-    origin_h, origin_w = image_rgb.shape[:2]
-    padded_image = _pad_img_to_modulo(image_rgb, mod=512, square=True, min_size=512)
-    padded_mask = _pad_img_to_modulo(mask, mod=512, square=True, min_size=512)
-    return _migan_forward(model, padded_image, padded_mask)[:origin_h, :origin_w, :]
-
-
-def _erase_migan_torchscript(image_bgr: np.ndarray, mask: np.ndarray) -> np.ndarray:
-    model = _load_migan_model()
-    image_rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
-    if image_rgb.shape[:2] == (512, 512):
-        return _migan_pad_forward(model, image_rgb, mask)
-
-    result = image_rgb[:, :, ::-1].copy()
-    for box in _boxes_from_mask(mask):
-        crop_img, crop_mask, (left, top, right, bottom) = _crop_box(image_rgb, mask, box, margin=128)
-        origin_size = crop_img.shape[:2]
-        resized_img = _resize_max_size(crop_img, 512)
-        resized_mask = _resize_max_size(crop_mask, 512)
-        inpaint = _migan_pad_forward(model, resized_img, resized_mask)
-        inpaint = cv2.resize(inpaint, (origin_size[1], origin_size[0]), interpolation=cv2.INTER_CUBIC)
-        keep = crop_mask < 127
-        inpaint[keep] = crop_img[:, :, ::-1][keep]
-        result[top:bottom, left:right, :] = inpaint
-    return result
-
-
-def _normalize_mode(mode: str | None) -> str:
-    requested = (mode or DEFAULT_MODE).strip().lower()
-    aliases = {
-        "lama": "quality",
-        "high": "quality",
-        "high-quality": "quality",
-        "migan": "fast",
-        "quick": "fast",
-    }
-    normalized = aliases.get(requested, requested)
-    if normalized not in MODES:
-        raise ValueError(f"Unsupported mode: {mode}. Supported modes: {', '.join(MODES)}")
-    return normalized
-
-
-def run_inpaint(image: Image.Image, mask: Image.Image, *, mode: str | None = None) -> tuple[Image.Image, str, float]:
-    normalized_mode = _normalize_mode(mode)
-    rgb_image = np.array(image.convert("RGB"), copy=True)
-    image_bgr = cv2.cvtColor(rgb_image, cv2.COLOR_RGB2BGR)
-    prepared_mask = _prepare_mask(mask, image.size)
-
-    start = time.time()
-    with MODEL_LOCK:
-        if normalized_mode == "quality":
-            result_bgr = _erase_lama_onnx(image_bgr, prepared_mask)
-        elif normalized_mode == "fast":
-            result_bgr = _erase_migan_torchscript(image_bgr, prepared_mask)
-        else:
-            raise ValueError(f"Unsupported mode: {normalized_mode}")
-    elapsed = time.time() - start
-
-    result_rgb = cv2.cvtColor(result_bgr.astype(np.uint8), cv2.COLOR_BGR2RGB)
-    return Image.fromarray(result_rgb), normalized_mode, elapsed
-
-
-def run_remove_doubao(image: Image.Image, *, mode: str | None = None) -> tuple[Image.Image, str, float, dict[str, object]]:
-    rgb_image = np.array(image.convert("RGB"), copy=True)
-    image_bgr = cv2.cvtColor(rgb_image, cv2.COLOR_RGB2BGR)
-    mask, metadata = generate_doubao_template_bbox_mask(image_bgr)
-    mask_image = Image.fromarray(mask, mode="L")
-    result, normalized_mode, elapsed = run_inpaint(image, mask_image, mode=mode)
-    return result, normalized_mode, elapsed, metadata
-
-
-@app.on_event("startup")
-def startup() -> None:
-    if not LOAD_MODEL_ON_STARTUP:
-        return
-    if DEFAULT_MODE == "fast":
-        _load_migan_model()
-    else:
-        _load_lama_session()
-
-
-@app.get("/")
-def root() -> dict[str, object]:
-    return health()
-
-
-@app.get("/health")
-def health() -> dict[str, object]:
-    return {
-        "status": "ok",
-        "mode": "dual-selected-backends",
-        "default_mode": DEFAULT_MODE,
-        "supported_modes": MODES,
-        "loaded": {
-            "quality": LAMA_SESSION is not None,
-            "fast": MIGAN_MODEL is not None,
-        },
-    }
-
-
-@app.post("/inpaint")
-async def inpaint(
-    image: UploadFile = File(...),
-    mask: UploadFile = File(...),
-    mode: str = Form(DEFAULT_MODE),
-) -> Response:
-    try:
-        image_bytes = await image.read()
-        mask_bytes = await mask.read()
-        pil_image = Image.open(io.BytesIO(image_bytes))
-        pil_mask = Image.open(io.BytesIO(mask_bytes))
-        result, normalized_mode, elapsed = run_inpaint(pil_image, pil_mask, mode=mode)
-        output = io.BytesIO()
-        result.save(output, format="PNG")
-        return Response(
-            content=output.getvalue(),
-            media_type="image/png",
-            headers={
-                "X-Inpaint-Mode": normalized_mode,
-                "X-Inpaint-Backend": MODES[normalized_mode],
-                "X-Inpaint-Elapsed": f"{elapsed:.3f}",
-            },
-        )
-    except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
-
-
-@app.post("/remove-doubao")
-async def remove_doubao(
-    image: UploadFile = File(...),
-    mode: str = Form(DEFAULT_MODE),
-) -> Response:
-    try:
-        image_bytes = await image.read()
-        pil_image = Image.open(io.BytesIO(image_bytes))
-        result, normalized_mode, elapsed, metadata = run_remove_doubao(pil_image, mode=mode)
-        output = io.BytesIO()
-        result.save(output, format="PNG")
-        return Response(
-            content=output.getvalue(),
-            media_type="image/png",
-            headers={
-                "X-Inpaint-Mode": normalized_mode,
-                "X-Inpaint-Backend": MODES[normalized_mode],
-                "X-Inpaint-Elapsed": f"{elapsed:.3f}",
-                "X-Doubao-Confidence": str(metadata["confidence"]),
-                "X-Doubao-Coverage": str(metadata["coverage"]),
-                "X-Doubao-BBox": ",".join(str(v) for v in metadata["bbox"]),
-            },
-        )
-    except Exception as exc:
-        raise HTTPException(status_code=500, detail=str(exc)) from exc
+from __future__ import annotations
+
+import hashlib
+import io
+import os
+import threading
+import time
+from pathlib import Path
+from urllib.parse import urlparse
+
+import cv2
+import numpy as np
+import onnxruntime as ort
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.responses import Response
+from huggingface_hub import hf_hub_download
+from PIL import Image
+
+
+LAMA_MODEL_REPO = os.getenv("LAMA_MODEL_REPO", "Carve/LaMa-ONNX")
+LAMA_MODEL_FILE = os.getenv("LAMA_MODEL_FILE", "lama_fp32.onnx")
+MIGAN_MODEL_URL = os.getenv(
+    "MIGAN_MODEL_URL",
+    "https://github.com/Sanster/models/releases/download/migan/migan_traced.pt",
+)
+MIGAN_MODEL_MD5 = os.getenv("MIGAN_MODEL_MD5", "76eb3b1a71c400ee3290524f7a11b89c")
+DEFAULT_MODE = os.getenv("DEFAULT_MODE", "quality")
+LOAD_MODEL_ON_STARTUP = os.getenv("LOAD_MODEL_ON_STARTUP", "1") == "1"
+PROVIDERS = ["CPUExecutionProvider"]
+MODES = {
+    "quality": "selfbuilt_lama_onnx",
+    "fast": "iopaint_migan_torchscript",
+}
+
+DOUBAO_ALPHA_ASSET = Path(__file__).with_name("doubao_alpha.png")
+DOUBAO_WIDTH_FRAC = 0.22
+DOUBAO_HEIGHT_FRAC = 0.075
+DOUBAO_MARGIN_RIGHT_FRAC = 0.004
+DOUBAO_MARGIN_BOTTOM_FRAC = 0.004
+DOUBAO_MAX_SATURATION = 55
+DOUBAO_LOGO_MIN_LUMA = 150
+DOUBAO_TOPHAT_DELTA = 12
+DOUBAO_DETECT_MIN_COVERAGE = 0.04
+DOUBAO_DETECT_NCC_THRESHOLD = 0.4
+DOUBAO_ALPHA_WIDTH_FRAC = 0.1636
+DOUBAO_ALPHA_HEIGHT_FRAC = 0.0405
+DOUBAO_ALPHA_MARGIN_RIGHT_FRAC = 0.0132
+DOUBAO_ALPHA_MARGIN_BOTTOM_FRAC = 0.0166
+DOUBAO_ALPHA_ALIGN_SEARCH = (0.88, 1.12, 25)
+DOUBAO_RESIDUAL_ALPHA_FLOOR = 0.05
+DOUBAO_TEMPLATE_DILATE = 4
+
+LAMA_SESSION: ort.InferenceSession | None = None
+MIGAN_MODEL: object | None = None
+DOUBAO_ALPHA: np.ndarray | None = None
+DOUBAO_SILHOUETTE: np.ndarray | None = None
+MODEL_LOCK = threading.RLock()
+
+app = FastAPI(title="Image Services Inpaint")
+
+
+def _md5sum(path: Path) -> str:
+    digest = hashlib.md5()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _cache_path_by_url(url: str) -> Path:
+    parts = urlparse(url)
+    model_dir = Path(os.getenv("TORCH_HOME", "/tmp/torch")) / "checkpoints"
+    model_dir.mkdir(parents=True, exist_ok=True)
+    return model_dir / Path(parts.path).name
+
+
+def _download_torchscript(url: str, expected_md5: str) -> Path:
+    from torch.hub import download_url_to_file
+
+    path = _cache_path_by_url(url)
+    if not path.exists():
+        download_url_to_file(url, str(path), None, progress=True)
+    actual_md5 = _md5sum(path)
+    if actual_md5 != expected_md5:
+        try:
+            path.unlink()
+        finally:
+            raise RuntimeError(f"Model md5 mismatch: {actual_md5} != {expected_md5}")
+    return path
+
+
+def _load_lama_session() -> ort.InferenceSession:
+    global LAMA_SESSION
+    if LAMA_SESSION is not None:
+        return LAMA_SESSION
+    with MODEL_LOCK:
+        if LAMA_SESSION is not None:
+            return LAMA_SESSION
+        model_path = hf_hub_download(repo_id=LAMA_MODEL_REPO, filename=LAMA_MODEL_FILE)
+        LAMA_SESSION = ort.InferenceSession(model_path, providers=PROVIDERS)
+        return LAMA_SESSION
+
+
+def _load_migan_model() -> object:
+    global MIGAN_MODEL
+    if MIGAN_MODEL is not None:
+        return MIGAN_MODEL
+    with MODEL_LOCK:
+        if MIGAN_MODEL is not None:
+            return MIGAN_MODEL
+        model_path = _download_torchscript(MIGAN_MODEL_URL, MIGAN_MODEL_MD5)
+        import torch
+
+        MIGAN_MODEL = torch.jit.load(str(model_path), map_location="cpu").eval()
+        return MIGAN_MODEL
+
+
+def _prepare_mask(mask: Image.Image, size: tuple[int, int]) -> np.ndarray:
+    arr = np.array(mask.convert("L").resize(size, Image.Resampling.NEAREST), copy=True)
+    return np.where(arr > 127, 255, 0).astype(np.uint8)
+
+
+def _load_doubao_alpha() -> np.ndarray:
+    global DOUBAO_ALPHA
+    if DOUBAO_ALPHA is not None:
+        return DOUBAO_ALPHA
+    alpha = cv2.imread(str(DOUBAO_ALPHA_ASSET), cv2.IMREAD_GRAYSCALE)
+    if alpha is None:
+        raise RuntimeError(f"Missing Doubao alpha asset: {DOUBAO_ALPHA_ASSET}")
+    DOUBAO_ALPHA = alpha.astype(np.float32) / 255.0
+    return DOUBAO_ALPHA
+
+
+def _load_doubao_silhouette() -> np.ndarray:
+    global DOUBAO_SILHOUETTE
+    if DOUBAO_SILHOUETTE is not None:
+        return DOUBAO_SILHOUETTE
+    DOUBAO_SILHOUETTE = (_load_doubao_alpha() > 0.15).astype(np.uint8) * 255
+    return DOUBAO_SILHOUETTE
+
+
+def _doubao_locate(image_bgr: np.ndarray) -> tuple[int, int, int, int]:
+    height, width = image_bgr.shape[:2]
+    mark_w = max(40, int(width * DOUBAO_WIDTH_FRAC))
+    mark_h = max(16, int(width * DOUBAO_HEIGHT_FRAC))
+    margin_right = max(4, int(width * DOUBAO_MARGIN_RIGHT_FRAC))
+    margin_bottom = max(4, int(width * DOUBAO_MARGIN_BOTTOM_FRAC))
+    x = max(0, width - margin_right - mark_w)
+    y = max(0, height - margin_bottom - mark_h)
+    return x, y, min(mark_w, width - x), min(mark_h, height - y)
+
+
+def _doubao_extract_candidate_mask(image_bgr: np.ndarray, loc: tuple[int, int, int, int]) -> np.ndarray:
+    x, y, width, height = loc
+    if width < 16 or height < 16:
+        return np.zeros((height, width), np.uint8)
+    roi = image_bgr[y : y + height, x : x + width].astype(np.float32)
+    luma = roi.mean(axis=2)
+    saturation = roi.max(axis=2) - roi.min(axis=2)
+    grayish = saturation < DOUBAO_MAX_SATURATION
+    sigma = max(4.0, height * 0.4)
+    local_bg = cv2.GaussianBlur(luma, (0, 0), sigmaX=sigma, sigmaY=sigma)
+    tophat = luma - local_bg
+    glyph = (
+        grayish
+        & (tophat > DOUBAO_TOPHAT_DELTA)
+        & (luma > DOUBAO_LOGO_MIN_LUMA)
+    ).astype(np.uint8) * 255
+    glyph = cv2.morphologyEx(glyph, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))
+    return cv2.morphologyEx(glyph, cv2.MORPH_OPEN, np.ones((5, 5), np.uint8))
+
+
+def _doubao_template_match_score(box_mask: np.ndarray, image_width: int) -> float:
+    silhouette = _load_doubao_silhouette()
+    if box_mask.size == 0:
+        return 0.0
+    gw = min(box_mask.shape[1] - 1, max(8, int(DOUBAO_ALPHA_WIDTH_FRAC * image_width)))
+    gh = min(box_mask.shape[0] - 1, max(4, int(DOUBAO_ALPHA_HEIGHT_FRAC * image_width)))
+    if gw < 8 or gh < 4:
+        return 0.0
+    template = cv2.resize(silhouette, (gw, gh), interpolation=cv2.INTER_NEAREST)
+    return float(cv2.matchTemplate(box_mask, template, cv2.TM_CCOEFF_NORMED).max())
+
+
+def _doubao_aligned_alpha(image_bgr: np.ndarray) -> tuple[np.ndarray, tuple[int, int, int, int]] | None:
+    alpha = _load_doubao_alpha()
+    silhouette = _load_doubao_silhouette()
+    image_width = image_bgr.shape[1]
+    bx, by, bw, bh = _doubao_locate(image_bgr)
+    box_mask = _doubao_extract_candidate_mask(image_bgr, (bx, by, bw, bh))
+    expected = DOUBAO_ALPHA_WIDTH_FRAC * image_width
+    best: tuple[float, int, int, int, int] | None = None
+    for scale in np.linspace(*DOUBAO_ALPHA_ALIGN_SEARCH):
+        gw = int(expected * scale)
+        gh = int(DOUBAO_ALPHA_HEIGHT_FRAC * image_width * scale)
+        if gw < 8 or gh < 4 or gw >= bw or gh >= bh:
+            continue
+        template = cv2.resize(silhouette, (gw, gh), interpolation=cv2.INTER_NEAREST)
+        _, score, _, top_left = cv2.minMaxLoc(cv2.matchTemplate(box_mask, template, cv2.TM_CCOEFF_NORMED))
+        if best is None or score > best[0]:
+            best = (score, gw, gh, top_left[0], top_left[1])
+    if best is None:
+        return None
+    _, gw, gh, ox, oy = best
+    return cv2.resize(alpha, (gw, gh), interpolation=cv2.INTER_LINEAR), (bx + ox, by + oy, gw, gh)
+
+
+def _doubao_fixed_alpha(image_bgr: np.ndarray) -> tuple[np.ndarray, tuple[int, int, int, int]]:
+    alpha = _load_doubao_alpha()
+    image_h, image_w = image_bgr.shape[:2]
+    gw = min(image_w, max(1, int(DOUBAO_ALPHA_WIDTH_FRAC * image_w)))
+    gh = min(image_h, max(1, int(DOUBAO_ALPHA_HEIGHT_FRAC * image_w)))
+    ax = max(0, image_w - int(DOUBAO_ALPHA_MARGIN_RIGHT_FRAC * image_w) - gw)
+    ay = max(0, image_h - int(DOUBAO_ALPHA_MARGIN_BOTTOM_FRAC * image_w) - gh)
+    return cv2.resize(alpha, (gw, gh), interpolation=cv2.INTER_LINEAR), (ax, ay, gw, gh)
+
+
+def generate_doubao_template_bbox_mask(image_bgr: np.ndarray) -> tuple[np.ndarray, dict[str, object]]:
+    if min(image_bgr.shape[:2]) < 200:
+        raise ValueError("Image is too small for Doubao watermark detection.")
+    loc = _doubao_locate(image_bgr)
+    candidate_mask = _doubao_extract_candidate_mask(image_bgr, loc)
+    coverage = float((candidate_mask > 0).sum()) / float(max(1, loc[2] * loc[3]))
+    confidence = _doubao_template_match_score(candidate_mask, image_bgr.shape[1]) if coverage >= DOUBAO_DETECT_MIN_COVERAGE else 0.0
+    detected = confidence >= DOUBAO_DETECT_NCC_THRESHOLD
+    if not detected:
+        raise ValueError(f"Doubao watermark not detected. confidence={confidence:.3f} coverage={coverage:.3f}")
+
+    placed = _doubao_aligned_alpha(image_bgr) or _doubao_fixed_alpha(image_bgr)
+    alpha_block, (x, y, width, height) = placed
+    silhouette = (alpha_block > DOUBAO_RESIDUAL_ALPHA_FLOOR).astype(np.uint8) * 255
+    if int((silhouette > 0).sum()) == 0:
+        raise ValueError("Detected Doubao watermark but generated an empty mask.")
+    glyph_mask = np.zeros(image_bgr.shape[:2], np.uint8)
+    glyph_mask[y : y + height, x : x + width] = silhouette
+    if DOUBAO_TEMPLATE_DILATE > 0:
+        kernel = cv2.getStructuringElement(
+            cv2.MORPH_ELLIPSE,
+            (2 * DOUBAO_TEMPLATE_DILATE + 1, 2 * DOUBAO_TEMPLATE_DILATE + 1),
+        )
+        glyph_mask = cv2.dilate(glyph_mask, kernel)
+
+    ys, xs = np.where(glyph_mask > 0)
+    if len(xs) == 0 or len(ys) == 0:
+        raise ValueError("Detected Doubao watermark but bbox mask is empty.")
+    bbox_mask = np.zeros_like(glyph_mask)
+    x0, x1 = int(xs.min()), int(xs.max()) + 1
+    y0, y1 = int(ys.min()), int(ys.max()) + 1
+    bbox_mask[y0:y1, x0:x1] = 255
+    return bbox_mask, {
+        "confidence": round(confidence, 4),
+        "coverage": round(coverage, 4),
+        "bbox": [x0, y0, x1 - x0, y1 - y0],
+    }
+
+
+def _mask_bbox(mask: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
+    ys, xs = np.where(mask > 0)
+    if len(xs) == 0 or len(ys) == 0:
+        raise ValueError("Mask is empty.")
+    return ys, xs
+
+
+def _ceil_modulo(value: int, mod: int) -> int:
+    if value % mod == 0:
+        return value
+    return (value // mod + 1) * mod
+
+
+def _pad_img_to_modulo(
+    image: np.ndarray,
+    *,
+    mod: int,
+    square: bool = False,
+    min_size: int | None = None,
+) -> np.ndarray:
+    if image.ndim == 2:
+        image = image[:, :, None]
+    height, width = image.shape[:2]
+    out_height = _ceil_modulo(height, mod)
+    out_width = _ceil_modulo(width, mod)
+    if min_size is not None:
+        out_width = max(min_size, out_width)
+        out_height = max(min_size, out_height)
+    if square:
+        side = max(out_height, out_width)
+        out_height = side
+        out_width = side
+    return np.pad(
+        image,
+        ((0, out_height - height), (0, out_width - width), (0, 0)),
+        mode="symmetric",
+    )
+
+
+def _resize_max_size(image: np.ndarray, size_limit: int, interpolation: int = cv2.INTER_CUBIC) -> np.ndarray:
+    height, width = image.shape[:2]
+    if max(height, width) <= size_limit:
+        return image
+    ratio = size_limit / max(height, width)
+    return cv2.resize(image, (int(width * ratio + 0.5), int(height * ratio + 0.5)), interpolation=interpolation)
+
+
+def _norm_img(image: np.ndarray) -> np.ndarray:
+    if image.ndim == 2:
+        image = image[:, :, None]
+    image = np.transpose(image, (2, 0, 1))
+    return image.astype("float32") / 255.0
+
+
+def _boxes_from_mask(mask: np.ndarray) -> list[np.ndarray]:
+    _, thresh = cv2.threshold(mask, 127, 255, 0)
+    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+    boxes: list[np.ndarray] = []
+    height, width = mask.shape[:2]
+    for contour in contours:
+        x, y, box_width, box_height = cv2.boundingRect(contour)
+        box = np.array([x, y, x + box_width, y + box_height]).astype(int)
+        box[::2] = np.clip(box[::2], 0, width)
+        box[1::2] = np.clip(box[1::2], 0, height)
+        boxes.append(box)
+    return boxes
+
+
+def _crop_box(
+    image: np.ndarray,
+    mask: np.ndarray,
+    box: np.ndarray,
+    *,
+    margin: int,
+) -> tuple[np.ndarray, np.ndarray, tuple[int, int, int, int]]:
+    box_h = int(box[3] - box[1])
+    box_w = int(box[2] - box[0])
+    cx = int((box[0] + box[2]) // 2)
+    cy = int((box[1] + box[3]) // 2)
+    img_h, img_w = image.shape[:2]
+
+    crop_w = box_w + margin * 2
+    crop_h = box_h + margin * 2
+    raw_l = cx - crop_w // 2
+    raw_r = cx + crop_w // 2
+    raw_t = cy - crop_h // 2
+    raw_b = cy + crop_h // 2
+
+    left = max(raw_l, 0)
+    right = min(raw_r, img_w)
+    top = max(raw_t, 0)
+    bottom = min(raw_b, img_h)
+
+    if raw_l < 0:
+        right += abs(raw_l)
+    if raw_r > img_w:
+        left -= raw_r - img_w
+    if raw_t < 0:
+        bottom += abs(raw_t)
+    if raw_b > img_h:
+        top -= raw_b - img_h
+
+    left = max(left, 0)
+    right = min(right, img_w)
+    top = max(top, 0)
+    bottom = min(bottom, img_h)
+    return image[top:bottom, left:right, :], mask[top:bottom, left:right], (left, top, right, bottom)
+
+
+def _erase_lama_onnx(image_bgr: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    session = _load_lama_session()
+    inputs = session.get_inputs()
+    image_name = inputs[0].name
+    mask_name = inputs[1].name
+    dims = inputs[0].shape
+    size = next((dim for dim in reversed(dims) if isinstance(dim, int) and dim > 1), 512)
+
+    height, width = image_bgr.shape[:2]
+    ys, xs = _mask_bbox(mask)
+
+    pad = max(16, int(0.4 * max(xs.max() - xs.min() + 1, ys.max() - ys.min() + 1)))
+    cx0 = max(0, int(xs.min()) - pad)
+    cy0 = max(0, int(ys.min()) - pad)
+    cx1 = min(width, int(xs.max()) + 1 + pad)
+    cy1 = min(height, int(ys.max()) + 1 + pad)
+
+    crop = image_bgr[cy0:cy1, cx0:cx1]
+    crop_mask = mask[cy0:cy1, cx0:cx1]
+    crop_h, crop_w = crop.shape[:2]
+
+    crop_rs = cv2.resize(crop, (size, size), interpolation=cv2.INTER_AREA)
+    mask_rs = cv2.resize(crop_mask, (size, size), interpolation=cv2.INTER_NEAREST)
+
+    image_input = cv2.cvtColor(crop_rs, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
+    image_input = np.transpose(image_input, (2, 0, 1))[None]
+    mask_input = (mask_rs > 127).astype(np.float32)[None, None]
+
+    output = session.run(None, {image_name: image_input, mask_name: mask_input})[0]
+    output = np.asarray(output)[0]
+    output = np.transpose(output, (1, 2, 0))
+    if float(output.max()) <= 1.5:
+        output = output * 255.0
+    output = np.clip(output, 0, 255).astype(np.uint8)
+    output_bgr = cv2.cvtColor(output, cv2.COLOR_RGB2BGR)
+    output_crop = cv2.resize(output_bgr, (crop_w, crop_h), interpolation=cv2.INTER_LINEAR)
+
+    result = image_bgr.copy()
+    region = result[cy0:cy1, cx0:cx1]
+    paste = crop_mask > 127
+    region[paste] = output_crop[paste]
+    result[cy0:cy1, cx0:cx1] = region
+    return result
+
+
+def _migan_forward(model: object, image_rgb: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    import torch
+
+    image = _norm_img(image_rgb) * 2 - 1
+    mask_binary = (mask > 120).astype(np.uint8) * 255
+    mask_input = _norm_img(mask_binary)
+    image_tensor = torch.from_numpy(image).unsqueeze(0)
+    mask_tensor = torch.from_numpy(mask_input).unsqueeze(0)
+    erased = image_tensor * (1 - mask_tensor)
+    model_input = torch.cat([0.5 - mask_tensor, erased], dim=1)
+    with torch.inference_mode():
+        output = model(model_input)
+    output = (
+        (output.permute(0, 2, 3, 1) * 127.5 + 127.5)
+        .round()
+        .clamp(0, 255)
+        .to(torch.uint8)
+    )
+    result = output[0].cpu().numpy()
+    return cv2.cvtColor(result, cv2.COLOR_RGB2BGR)
+
+
+def _migan_pad_forward(model: object, image_rgb: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    origin_h, origin_w = image_rgb.shape[:2]
+    padded_image = _pad_img_to_modulo(image_rgb, mod=512, square=True, min_size=512)
+    padded_mask = _pad_img_to_modulo(mask, mod=512, square=True, min_size=512)
+    return _migan_forward(model, padded_image, padded_mask)[:origin_h, :origin_w, :]
+
+
+def _erase_migan_torchscript(image_bgr: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    model = _load_migan_model()
+    image_rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
+    if image_rgb.shape[:2] == (512, 512):
+        return _migan_pad_forward(model, image_rgb, mask)
+
+    result = image_rgb[:, :, ::-1].copy()
+    for box in _boxes_from_mask(mask):
+        crop_img, crop_mask, (left, top, right, bottom) = _crop_box(image_rgb, mask, box, margin=128)
+        origin_size = crop_img.shape[:2]
+        resized_img = _resize_max_size(crop_img, 512)
+        resized_mask = _resize_max_size(crop_mask, 512)
+        inpaint = _migan_pad_forward(model, resized_img, resized_mask)
+        inpaint = cv2.resize(inpaint, (origin_size[1], origin_size[0]), interpolation=cv2.INTER_CUBIC)
+        keep = crop_mask < 127
+        inpaint[keep] = crop_img[:, :, ::-1][keep]
+        result[top:bottom, left:right, :] = inpaint
+    return result
+
+
+def _normalize_mode(mode: str | None) -> str:
+    requested = (mode or DEFAULT_MODE).strip().lower()
+    aliases = {
+        "lama": "quality",
+        "high": "quality",
+        "high-quality": "quality",
+        "migan": "fast",
+        "quick": "fast",
+    }
+    normalized = aliases.get(requested, requested)
+    if normalized not in MODES:
+        raise ValueError(f"Unsupported mode: {mode}. Supported modes: {', '.join(MODES)}")
+    return normalized
+
+
+def run_inpaint(image: Image.Image, mask: Image.Image, *, mode: str | None = None) -> tuple[Image.Image, str, float]:
+    normalized_mode = _normalize_mode(mode)
+    rgb_image = np.array(image.convert("RGB"), copy=True)
+    image_bgr = cv2.cvtColor(rgb_image, cv2.COLOR_RGB2BGR)
+    prepared_mask = _prepare_mask(mask, image.size)
+
+    start = time.time()
+    with MODEL_LOCK:
+        if normalized_mode == "quality":
+            result_bgr = _erase_lama_onnx(image_bgr, prepared_mask)
+        elif normalized_mode == "fast":
+            result_bgr = _erase_migan_torchscript(image_bgr, prepared_mask)
+        else:
+            raise ValueError(f"Unsupported mode: {normalized_mode}")
+    elapsed = time.time() - start
+
+    result_rgb = cv2.cvtColor(result_bgr.astype(np.uint8), cv2.COLOR_BGR2RGB)
+    return Image.fromarray(result_rgb), normalized_mode, elapsed
+
+
+def run_remove_doubao(image: Image.Image, *, mode: str | None = None) -> tuple[Image.Image, str, float, dict[str, object]]:
+    rgb_image = np.array(image.convert("RGB"), copy=True)
+    image_bgr = cv2.cvtColor(rgb_image, cv2.COLOR_RGB2BGR)
+    mask, metadata = generate_doubao_template_bbox_mask(image_bgr)
+    mask_image = Image.fromarray(mask, mode="L")
+    result, normalized_mode, elapsed = run_inpaint(image, mask_image, mode=mode)
+    return result, normalized_mode, elapsed, metadata
+
+
+@app.on_event("startup")
+def startup() -> None:
+    if not LOAD_MODEL_ON_STARTUP:
+        return
+    if DEFAULT_MODE == "fast":
+        _load_migan_model()
+    else:
+        _load_lama_session()
+
+
+@app.get("/")
+def root() -> dict[str, object]:
+    return health()
+
+
+@app.get("/health")
+def health() -> dict[str, object]:
+    return {
+        "status": "ok",
+        "mode": "dual-selected-backends",
+        "default_mode": DEFAULT_MODE,
+        "supported_modes": MODES,
+        "loaded": {
+            "quality": LAMA_SESSION is not None,
+            "fast": MIGAN_MODEL is not None,
+        },
+    }
+
+
+@app.post("/inpaint")
+async def inpaint(
+    image: UploadFile = File(...),
+    mask: UploadFile = File(...),
+    mode: str = Form(DEFAULT_MODE),
+) -> Response:
+    try:
+        image_bytes = await image.read()
+        mask_bytes = await mask.read()
+        pil_image = Image.open(io.BytesIO(image_bytes))
+        pil_mask = Image.open(io.BytesIO(mask_bytes))
+        result, normalized_mode, elapsed = run_inpaint(pil_image, pil_mask, mode=mode)
+        output = io.BytesIO()
+        result.save(output, format="PNG")
+        return Response(
+            content=output.getvalue(),
+            media_type="image/png",
+            headers={
+                "X-Inpaint-Mode": normalized_mode,
+                "X-Inpaint-Backend": MODES[normalized_mode],
+                "X-Inpaint-Elapsed": f"{elapsed:.3f}",
+            },
+        )
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+
+@app.post("/remove-doubao")
+async def remove_doubao(
+    image: UploadFile = File(...),
+    mode: str = Form(DEFAULT_MODE),
+) -> Response:
+    try:
+        image_bytes = await image.read()
+        pil_image = Image.open(io.BytesIO(image_bytes))
+        result, normalized_mode, elapsed, metadata = run_remove_doubao(pil_image, mode=mode)
+        output = io.BytesIO()
+        result.save(output, format="PNG")
+        return Response(
+            content=output.getvalue(),
+            media_type="image/png",
+            headers={
+                "X-Inpaint-Mode": normalized_mode,
+                "X-Inpaint-Backend": MODES[normalized_mode],
+                "X-Inpaint-Elapsed": f"{elapsed:.3f}",
+                "X-Doubao-Confidence": str(metadata["confidence"]),
+                "X-Doubao-Coverage": str(metadata["coverage"]),
+                "X-Doubao-BBox": ",".join(str(v) for v in metadata["bbox"]),
+            },
+        )
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=str(exc)) from exc