from __future__ import annotations import hashlib import ipaddress import io import json import logging import os import secrets import socket import threading import time import warnings from contextlib import asynccontextmanager from dataclasses import dataclass, field from typing import Any, Callable, Literal from urllib.parse import SplitResult, urlsplit import httpx from fastapi import APIRouter, HTTPException, Request from fastapi import status as http_status from fastapi.responses import JSONResponse from PIL import Image, ImageOps, UnidentifiedImageError from pydantic import BaseModel, ConfigDict, Field, field_validator LOGGER = logging.getLogger(__name__) DNS_RESOLVER_SLOTS = threading.BoundedSemaphore(4) IMAGE_FETCH_SLOTS = threading.BoundedSemaphore(4) MIN_IMAGE_DIMENSION = 32 DEFAULT_MAX_IMAGE_BYTES = 8 * 1024 * 1024 MAX_CONFIGURABLE_IMAGE_BYTES = 20 * 1024 * 1024 DEFAULT_MAX_IMAGE_PIXELS = 4_194_304 MAX_CONFIGURABLE_IMAGE_PIXELS = 20_000_000 DEFAULT_MAX_IMAGE_DIMENSION = 4096 MAX_CONFIGURABLE_IMAGE_DIMENSION = 8192 DEFAULT_MAX_RECORDS = 256 MIN_MAX_RECORDS = 8 MAX_MAX_RECORDS = 4096 JobStatus = Literal["queued", "running", "succeeded", "failed"] SummarySeparator = Literal["comma", "newline", "space"] AnalysisExecutor = Callable[ ["ImageInversionJobRequest", Image.Image, str], dict[str, Any], ] ImageFetcher = Callable[[str, "ImageInversionJobSettings"], Image.Image] def _read_int_env(name: str, default: int) -> int: """读取整数环境变量。 Args: name: 环境变量名称。 default: 变量为空时采用的默认值。 Returns: 解析完成的整数。 """ raw_value = os.getenv(name) if raw_value is None or not raw_value.strip(): return default try: return int(raw_value) except ValueError as exc: raise RuntimeError(f"{name} must be an integer") from exc def _read_float_env(name: str, default: float) -> float: """读取有限浮点环境变量。 Args: name: 环境变量名称。 default: 变量为空时采用的默认值。 Returns: 解析完成的有限浮点数。 """ raw_value = os.getenv(name) if raw_value is None or not raw_value.strip(): return default try: parsed_value = float(raw_value) except ValueError as exc: raise RuntimeError(f"{name} must be a number") from exc if not float("-inf") < parsed_value < float("inf"): raise RuntimeError(f"{name} must be finite") return parsed_value def normalize_allowed_hosts(raw_hosts: str) -> frozenset[str]: """规范化远程图片精确主机白名单。 Args: raw_hosts: 以逗号分隔的主机名，不允许协议、端口、路径或通配符。 Returns: 完成小写与 IDNA 转换的不可变主机集合。 """ normalized_hosts: set[str] = set() for raw_host in raw_hosts.split(","): host = raw_host.strip().rstrip(".") if not host: continue if any(marker in host for marker in ("://", "/", "?", "#", "*", "@", ":")): raise RuntimeError( "JOB_IMAGE_ALLOWED_HOSTS must contain exact hostnames only" ) try: normalized_hosts.add(host.encode("idna").decode("ascii").lower()) except UnicodeError as exc: raise RuntimeError( "JOB_IMAGE_ALLOWED_HOSTS contains an invalid hostname" ) from exc if not normalized_hosts: raise RuntimeError("JOB_IMAGE_ALLOWED_HOSTS must contain at least one hostname") return frozenset(normalized_hosts) @dataclass(frozen=True, slots=True) class ImageInversionJobSettings: """保存 Image Inversion 自定义任务 API 的部署边界与资源限制。""" api_key: str allowed_hosts: frozenset[str] result_ttl_seconds: int = 1800 poll_after_seconds: int = 2 fetch_timeout_seconds: float = 15.0 max_image_bytes: int = DEFAULT_MAX_IMAGE_BYTES max_image_pixels: int = DEFAULT_MAX_IMAGE_PIXELS max_image_dimension: int = DEFAULT_MAX_IMAGE_DIMENSION max_records: int = DEFAULT_MAX_RECORDS space_host: str = "" @classmethod def from_env(cls) -> "ImageInversionJobSettings": """从 Space Secret 与 Variables 读取并校验任务 API 配置。 Args: 此方法不接收参数，配置统一从当前进程环境读取。 Returns: 已完成密钥、网络及图片资源边界校验的设置。 """ api_key = os.getenv("JOB_API_KEY", "").strip() if len(api_key) < 32: raise RuntimeError( "JOB_API_KEY must be configured as a Space Secret with at least 32 characters" ) result_ttl_seconds = _read_int_env("JOB_RESULT_TTL_SECONDS", 1800) poll_after_seconds = _read_int_env("JOB_POLL_AFTER_SECONDS", 2) fetch_timeout_seconds = _read_float_env( "JOB_IMAGE_FETCH_TIMEOUT_SECONDS", 15.0, ) max_image_bytes = _read_int_env( "JOB_IMAGE_MAX_BYTES", DEFAULT_MAX_IMAGE_BYTES, ) max_image_pixels = _read_int_env( "JOB_IMAGE_MAX_PIXELS", DEFAULT_MAX_IMAGE_PIXELS, ) max_image_dimension = _read_int_env( "JOB_IMAGE_MAX_DIMENSION", DEFAULT_MAX_IMAGE_DIMENSION, ) max_records = _read_int_env("JOB_MAX_RECORDS", DEFAULT_MAX_RECORDS) if not 60 <= result_ttl_seconds <= 86400: raise RuntimeError("JOB_RESULT_TTL_SECONDS must be between 60 and 86400") if not 1 <= poll_after_seconds <= 30: raise RuntimeError("JOB_POLL_AFTER_SECONDS must be between 1 and 30") if not 1.0 <= fetch_timeout_seconds <= 15.0: raise RuntimeError( "JOB_IMAGE_FETCH_TIMEOUT_SECONDS must be between 1 and 15" ) if not 1024 <= max_image_bytes <= MAX_CONFIGURABLE_IMAGE_BYTES: raise RuntimeError("JOB_IMAGE_MAX_BYTES must be between 1 KiB and 20 MiB") if not 1 <= max_image_pixels <= MAX_CONFIGURABLE_IMAGE_PIXELS: raise RuntimeError( "JOB_IMAGE_MAX_PIXELS must be between 1 and 20000000" ) if not MIN_IMAGE_DIMENSION <= max_image_dimension <= MAX_CONFIGURABLE_IMAGE_DIMENSION: raise RuntimeError( "JOB_IMAGE_MAX_DIMENSION must be between 32 and 8192" ) if not MIN_MAX_RECORDS <= max_records <= MAX_MAX_RECORDS: raise RuntimeError("JOB_MAX_RECORDS must be between 8 and 4096") return cls( api_key=api_key, allowed_hosts=normalize_allowed_hosts( os.getenv("JOB_IMAGE_ALLOWED_HOSTS", "") ), result_ttl_seconds=result_ttl_seconds, poll_after_seconds=poll_after_seconds, fetch_timeout_seconds=fetch_timeout_seconds, max_image_bytes=max_image_bytes, max_image_pixels=max_image_pixels, max_image_dimension=max_image_dimension, max_records=max_records, space_host=os.getenv("SPACE_HOST", "").strip().rstrip("/"), ) class ImageInversionJobRequest(BaseModel): """定义 Image Inversion 标签分析任务的公开请求参数。""" model_config = ConfigDict( extra="forbid", str_strip_whitespace=True, allow_inf_nan=False, ) input_image_url: str = Field(min_length=1, max_length=4096) general_threshold: float = Field(default=0.35, ge=0.0, le=1.0) character_threshold: float = Field(default=0.85, ge=0.0, le=1.0) show_confidence: bool = True show_general: bool = True show_character: bool = True show_ip: bool = True separator: SummarySeparator = "comma" show_chinese: bool = True @field_validator("input_image_url") @classmethod def require_https_image_url(cls, value: str) -> str: """在 Schema 阶段拒绝非 HTTPS、凭据、片段和非标准端口。 Args: value: 调用方提交的远程图片 URL。 Returns: 通过基础 HTTPS 语法检查的原始 URL。 """ try: parsed = urlsplit(value) port = parsed.port except ValueError as exc: raise ValueError("image URL is invalid") from exc if ( parsed.scheme.lower() != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.fragment or port not in {None, 443} ): raise ValueError( "image URL must use credential-free HTTPS on the default port" ) return value @dataclass(frozen=True, slots=True) class ValidatedImageURL: """保存通过协议、主机与公网地址检查的图片 URL。""" value: str host: str class ImageSourceError(RuntimeError): """表示图片来源不合法或暂时无法抓取。""" def __init__(self, status_code: int, public_message: str) -> None: """创建携带脱敏公开信息的图片来源错误。 Args: status_code: API 应返回的 HTTP 状态码。 public_message: 不包含 URL、凭据或底层网络细节的公开文案。 Returns: 此初始化方法不返回数据。 """ super().__init__(public_message) self.status_code = status_code self.public_message = public_message def _resolve_host_addresses(host: str, timeout_seconds: float) -> list[Any]: """在硬期限内解析白名单主机的地址。 Args: host: 已通过精确白名单校验的规范化主机名。 timeout_seconds: DNS 解析允许占用的最长墙钟秒数。 Returns: socket.getaddrinfo 返回的非空地址记录列表。 """ completed = threading.Event() result: dict[str, Any] = {} if not DNS_RESOLVER_SLOTS.acquire(blocking=False): raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image host resolution is temporarily busy.", ) def resolve() -> None: try: result["records"] = socket.getaddrinfo( host, 443, type=socket.SOCK_STREAM, ) except Exception as exc: result["error"] = exc finally: DNS_RESOLVER_SLOTS.release() completed.set() resolver = threading.Thread( target=resolve, name="image-inversion-dns", daemon=True, ) try: resolver.start() except Exception: DNS_RESOLVER_SLOTS.release() raise if not completed.wait(timeout=max(0.001, timeout_seconds)): raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image host resolution timed out.", ) resolution_error = result.get("error") if resolution_error is not None: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image host could not be resolved.", ) from resolution_error address_records = result.get("records") if not address_records: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image host did not return an address.", ) return address_records def validate_image_url( url: str, settings: ImageInversionJobSettings, ) -> ValidatedImageURL: """校验远程图片 URL 只指向白名单内的公网 HTTPS 主机。 Args: url: 调用方提交的完整图片 URL。 settings: 包含白名单与网络资源限制的任务 API 设置。 Returns: 已规范化主机名并可安全用于抓取的 URL 描述。 """ try: parsed: SplitResult = urlsplit(url) port = parsed.port except ValueError as exc: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL is invalid.", ) from exc if parsed.scheme.lower() != "https" or not parsed.hostname: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "Image URLs must use HTTPS.", ) if parsed.username is not None or parsed.password is not None or parsed.fragment: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL contains unsupported credentials or fragments.", ) if port not in {None, 443}: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL must use the default HTTPS port.", ) try: normalized_host = ( parsed.hostname.rstrip(".").encode("idna").decode("ascii").lower() ) except UnicodeError as exc: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL hostname is invalid.", ) from exc if normalized_host not in settings.allowed_hosts: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL host is not allowed.", ) address_records = _resolve_host_addresses( normalized_host, settings.fetch_timeout_seconds, ) # 白名单限定预期域名；逐个拒绝非公网解析结果可阻断错误配置和 DNS 绕过。 for address_record in address_records: raw_address = address_record[4][0].split("%", 1)[0] try: resolved_ip = ipaddress.ip_address(raw_address) except ValueError as exc: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image host resolved to an invalid address.", ) from exc if not resolved_ip.is_global: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image host must resolve only to public addresses.", ) return ValidatedImageURL(value=url, host=normalized_host) def _validate_connected_peer(response: httpx.Response) -> None: """确认 HTTP 客户端实际连接的对端仍是公网 IP。 Args: response: 已建立 TLS 连接并收到响应头的 httpx 响应。 Returns: 对端地址可验证且属于公网时不返回数据。 """ network_stream = response.extensions.get("network_stream") if network_stream is None: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image connection peer could not be verified.", ) try: peer_address = network_stream.get_extra_info("server_addr") raw_address = peer_address[0].split("%", 1)[0] peer_ip = ipaddress.ip_address(raw_address) except (AttributeError, IndexError, TypeError, ValueError) as exc: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image connection peer could not be verified.", ) from exc if not peer_ip.is_global: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image connection peer must be a public address.", ) def decode_image_bytes( image_bytes: bytes, content_type: str, settings: ImageInversionJobSettings, ) -> Image.Image: """校验远程内容并解码为脱离底层流的 RGB 图片。 Args: image_bytes: 在字节上限内完整读取的图片内容。 content_type: 远程响应声明的 MIME 类型，仅为接口兼容保留。 settings: 图片字节、像素及单边尺寸限制。 Returns: 完成 EXIF 方向修正并复制到内存的 RGB PIL 图片。 """ if not image_bytes or len(image_bytes) > settings.max_image_bytes: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image is empty or exceeds the configured size limit.", ) # 远端可合法返回 application/octet-stream；只信任 Pillow 识别的真实格式。 _ = content_type try: with warnings.catch_warnings(): warnings.simplefilter("error", Image.DecompressionBombWarning) with Image.open(io.BytesIO(image_bytes)) as source_image: if source_image.format not in {"JPEG", "PNG", "WEBP"}: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image format is not supported.", ) width, height = source_image.size if width < MIN_IMAGE_DIMENSION or height < MIN_IMAGE_DIMENSION: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image dimensions are not supported.", ) if ( width > settings.max_image_dimension or height > settings.max_image_dimension or width * height > settings.max_image_pixels ): raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image exceeds the configured dimension limit.", ) source_image.load() return ImageOps.exif_transpose(source_image).convert("RGB").copy() except ImageSourceError: raise except ( UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError, Image.DecompressionBombWarning, ) as exc: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image could not be decoded safely.", ) from exc def _map_image_request_error( exc: Exception, deadline_reached: threading.Event, ) -> ImageSourceError: """把底层 HTTP 错误映射为不含远程 URL 的公开错误。 Args: exc: httpx 或客户端关闭路径产生的底层异常。 deadline_reached: 硬期限看门狗是否已经触发。 Returns: 可由 API 层安全公开的图片来源错误。 """ if deadline_reached.is_set() or isinstance(exc, httpx.TimeoutException): return ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) return ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image host is temporarily unavailable.", ) def _fetch_remote_image_inner( url: str, settings: ImageInversionJobSettings, ) -> Image.Image: """从白名单公网主机流式抓取并安全解码图片。 Args: url: 调用方提交的图片 HTTPS URL。 settings: 图片白名单、超时、体积、像素与尺寸边界。 Returns: 可直接交给标签模型的 RGB PIL 图片。 """ started_at = time.monotonic() validated_url = validate_image_url(url, settings) remaining_seconds = settings.fetch_timeout_seconds - ( time.monotonic() - started_at ) if remaining_seconds <= 0: raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) timeout = httpx.Timeout( remaining_seconds, connect=min(5.0, remaining_seconds), ) deadline_reached = threading.Event() client = httpx.Client( timeout=timeout, follow_redirects=False, trust_env=False, ) def stop_at_deadline() -> None: deadline_reached.set() try: client.close() except Exception: pass deadline_timer = threading.Timer(remaining_seconds, stop_at_deadline) deadline_timer.daemon = True try: with client: deadline_timer.start() with client.stream( "GET", validated_url.value, headers={"Accept": "image/png,image/jpeg,image/webp"}, ) as response: if time.monotonic() - started_at > settings.fetch_timeout_seconds: raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) _validate_connected_peer(response) if 300 <= response.status_code < 400: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "Image URL redirects are not allowed.", ) if response.status_code == http_status.HTTP_408_REQUEST_TIMEOUT: raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) if response.status_code == 429 or response.status_code >= 500: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image host is temporarily unavailable.", ) if response.status_code != http_status.HTTP_200_OK: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image URL did not return a readable resource.", ) declared_length = response.headers.get("content-length") if declared_length: try: parsed_length = int(declared_length) if parsed_length < 0: raise ValueError("negative content length") if parsed_length > settings.max_image_bytes: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image exceeds the configured size limit.", ) except ValueError as exc: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The image host returned an invalid content length.", ) from exc image_buffer = bytearray() for chunk in response.iter_bytes(): if time.monotonic() - started_at > settings.fetch_timeout_seconds: raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) image_buffer.extend(chunk) if len(image_buffer) > settings.max_image_bytes: raise ImageSourceError( http_status.HTTP_400_BAD_REQUEST, "The remote image exceeds the configured size limit.", ) decoded_image = decode_image_bytes( bytes(image_buffer), response.headers.get("content-type", ""), settings, ) deadline_timer.cancel() if ( deadline_reached.is_set() or time.monotonic() - started_at > settings.fetch_timeout_seconds ): decoded_image.close() raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) return decoded_image except ImageSourceError: raise except (httpx.RequestError, RuntimeError) as exc: raise _map_image_request_error(exc, deadline_reached) from exc finally: deadline_timer.cancel() def fetch_remote_image( url: str, settings: ImageInversionJobSettings, ) -> Image.Image: """以有界后台抓取器执行下载并强制墙钟总期限。 Args: url: 调用方提交的图片 HTTPS URL。 settings: 白名单、硬超时、字节、像素与尺寸边界。 Returns: 在总期限内完成验证、EXIF 转正和 RGB 转换的 PIL 图片。 """ if not IMAGE_FETCH_SLOTS.acquire(blocking=False): raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image fetch service is temporarily busy.", ) completed = threading.Event() state_lock = threading.Lock() state: dict[str, Any] = {"abandoned": False} def fetch() -> None: fetched_image: Image.Image | None = None try: fetched_image = _fetch_remote_image_inner(url, settings) should_close = False with state_lock: if state["abandoned"]: should_close = True else: state["image"] = fetched_image fetched_image = None if should_close and fetched_image is not None: fetched_image.close() fetched_image = None except Exception as exc: with state_lock: if not state["abandoned"]: state["error"] = exc finally: if fetched_image is not None: try: fetched_image.close() except Exception as exc: LOGGER.warning( "Failed to close an abandoned fetched image: %s", type(exc).__name__, ) IMAGE_FETCH_SLOTS.release() completed.set() fetch_thread = threading.Thread( target=fetch, name="image-inversion-fetch", daemon=True, ) try: fetch_thread.start() except Exception as exc: IMAGE_FETCH_SLOTS.release() raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image fetch service could not start.", ) from exc if not completed.wait(timeout=settings.fetch_timeout_seconds): abandoned_image: Image.Image | None = None with state_lock: state["abandoned"] = True abandoned_image = state.pop("image", None) if abandoned_image is not None: try: abandoned_image.close() except Exception as exc: LOGGER.warning( "Failed to close a timed-out fetched image: %s", type(exc).__name__, ) raise ImageSourceError( http_status.HTTP_504_GATEWAY_TIMEOUT, "The image download timed out.", ) with state_lock: fetch_error = state.get("error") fetched_image = state.get("image") if fetch_error is not None: if isinstance(fetch_error, ImageSourceError): raise fetch_error raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image host is temporarily unavailable.", ) from fetch_error if fetched_image is None: raise ImageSourceError( http_status.HTTP_502_BAD_GATEWAY, "The image response could not be completed.", ) return fetched_image def _constant_time_text_equal(supplied: str, expected: str) -> bool: """以字节形式比较不可信文本。 Args: supplied: 请求携带的待校验文本。 expected: 服务端保存的期望文本。 Returns: 两个 UTF-8 字节序列完全一致时返回 True。 """ return secrets.compare_digest( supplied.encode("utf-8", errors="surrogatepass"), expected.encode("utf-8", errors="surrogatepass"), ) def fingerprint_request(payload: ImageInversionJobRequest) -> str: """生成与 JSON 字段顺序无关的任务请求指纹。 Args: payload: 已通过 Pydantic 校验的标签分析参数。 Returns: 用于幂等键复用校验的 SHA-256 摘要。 """ canonical_payload = json.dumps( payload.model_dump(mode="json"), ensure_ascii=False, separators=(",", ":"), sort_keys=True, ) return hashlib.sha256(canonical_payload.encode("utf-8")).hexdigest() @dataclass(slots=True) class ImageInversionJobRecord: """保存单个标签分析任务在当前 Space 进程内的生命周期数据。""" job_id: str payload: ImageInversionJobRequest request_fingerprint: str input_image: Image.Image | None idempotency_key: str | None = None status: JobStatus = "queued" created_at: float = field(default_factory=time.time) started_at: float | None = None completed_at: float | None = None result: dict[str, Any] | None = None error_type: str | None = None class ImageInversionJobAPI: """实现 Image Inversion 异步 JSON 任务 API 状态机。""" def __init__( self, settings: ImageInversionJobSettings, executor: AnalysisExecutor, inference_slot: threading.Lock, image_fetcher: ImageFetcher = fetch_remote_image, ) -> None: """创建任务服务并注册 Image Inversion 自定义路由。 Args: settings: 已校验的部署配置和资源限制。 executor: 复用现有标签分析链路的执行适配函数。 inference_slot: 与 UI 共用的非阻塞单推理门闩。 image_fetcher: 可替换的安全远程图片抓取器。 Returns: 此初始化方法不返回数据。 """ self.settings = settings self.executor = executor self.inference_slot = inference_slot self.image_fetcher = image_fetcher self.jobs: dict[str, ImageInversionJobRecord] = {} self.idempotency_jobs: dict[str, str] = {} self.jobs_lock = threading.RLock() self.active_job_id: str | None = None self.router = APIRouter(tags=["jobs"]) self.router.add_api_route( "/api/jobs", self.create_job, methods=["POST"], name="image_inversion_create_job", ) self.router.add_api_route( "/api/jobs/{job_id}", self.get_job_status, methods=["GET"], name="image_inversion_get_job_status", ) def install_on_app(self, app: Any) -> None: """把自定义任务路由安装到 Gradio FastAPI 应用前部。 Args: app: Gradio 在 launch 阶段创建的 FastAPI 应用。 Returns: 路由已存在或安装完成后不返回数据。 """ if getattr(app.state, "image_inversion_job_api_registered", False): return original_route_count = len(app.router.routes) app.include_router(self.router) added_routes = app.router.routes[original_route_count:] original_routes = app.router.routes[:original_route_count] # Gradio 含宽泛路由，自定义 API 必须排在其前面才能稳定命中。 app.router.routes[:] = [*added_routes, *original_routes] app.state.image_inversion_job_api_registered = True def _require_api_key(self, request: Request) -> None: supplied_key = request.headers.get("x-api-key", "") if not supplied_key or not _constant_time_text_equal( supplied_key, self.settings.api_key, ): raise HTTPException( status_code=http_status.HTTP_401_UNAUTHORIZED, detail="A valid X-API-Key header is required.", ) def _public_route_url( self, request: Request, route_name: str, **path_params: str, ) -> str: path = str(request.app.url_path_for(route_name, **path_params)) if self.settings.space_host: space_host = self.settings.space_host if not space_host.startswith(("http://", "https://")): space_host = f"https://{space_host}" return f"{space_host}{path}" return str(request.url_for(route_name, **path_params)) def _is_expired( self, record: ImageInversionJobRecord, now: float | None = None, ) -> bool: current_time = time.time() if now is None else now if record.completed_at is None: return False return current_time - record.completed_at > self.settings.result_ttl_seconds @staticmethod def _close_images(*images: Image.Image | None) -> None: """逐一关闭已知图片并隔离单个清理异常。 Args: images: 要释放的零个或多个明确图片对象。 Returns: 清理完成后不返回数据；关闭错误仅写入脱敏日志。 """ for image in images: if image is not None: try: image.close() except Exception as exc: LOGGER.warning( "Failed to close an Image Inversion job image: %s", type(exc).__name__, ) @staticmethod def _close_job_image(record: ImageInversionJobRecord | None) -> None: if record is None: return ImageInversionJobAPI._close_images(record.input_image) record.input_image = None def _remove_record_locked(self, record: ImageInversionJobRecord) -> None: self.jobs.pop(record.job_id, None) if record.idempotency_key is not None: if self.idempotency_jobs.get(record.idempotency_key) == record.job_id: self.idempotency_jobs.pop(record.idempotency_key, None) def remove_expired_jobs(self) -> None: """清理当前所有过期终态任务及其输入图片引用。 Args: 此方法不接收参数。 Returns: 清理完成后不返回数据。 """ expired_records: list[ImageInversionJobRecord] = [] now = time.time() with self.jobs_lock: for record in tuple(self.jobs.values()): if self._is_expired(record, now): self._remove_record_locked(record) expired_records.append(record) for expired_record in expired_records: self._close_job_image(expired_record) def _ensure_capacity_for_new_record(self) -> None: """为一个新任务腾出有限记录容量。 Args: 此方法不接收参数。 Returns: 容量已经可用时不返回数据；无法安全腾出容量时抛出 HTTP 503。 """ evicted_records: list[ImageInversionJobRecord] = [] with self.jobs_lock: now = time.time() for record in tuple(self.jobs.values()): if self._is_expired(record, now): self._remove_record_locked(record) evicted_records.append(record) while len(self.jobs) >= self.settings.max_records: terminal_records = [ record for record in self.jobs.values() if record.status in {"succeeded", "failed"} and record.completed_at is not None ] if not terminal_records: raise HTTPException( status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE, detail="The prediction job store is full.", headers={"Retry-After": "5"}, ) # 仅淘汰最老终态；绝不删除 queued/running 任务。 oldest_terminal = min( terminal_records, key=lambda record: record.completed_at or record.created_at, ) self._remove_record_locked(oldest_terminal) evicted_records.append(oldest_terminal) for evicted_record in evicted_records: self._close_job_image(evicted_record) def _job_or_404(self, job_id: str) -> ImageInversionJobRecord: expired_record: ImageInversionJobRecord | None = None with self.jobs_lock: record = self.jobs.get(job_id) if record is not None and self._is_expired(record): expired_record = record self._remove_record_locked(record) record = None self._close_job_image(expired_record) if record is None: raise HTTPException( status_code=http_status.HTTP_404_NOT_FOUND, detail="Job not found or result expired.", ) return record def _status_response( self, request: Request, record: ImageInversionJobRecord, ) -> JSONResponse: status_url = self._public_route_url( request, "image_inversion_get_job_status", job_id=record.job_id, ) return JSONResponse( status_code=http_status.HTTP_202_ACCEPTED, headers={ "Location": status_url, "Cache-Control": "no-store", }, content={ "job_id": record.job_id, "status_url": status_url, "poll_after_seconds": self.settings.poll_after_seconds, }, ) def _lookup_idempotent_job( self, idempotency_key: str | None, request_fingerprint: str, ) -> ImageInversionJobRecord | None: if idempotency_key is None: return None with self.jobs_lock: existing_job_id = self.idempotency_jobs.get(idempotency_key) record = self.jobs.get(existing_job_id or "") if record is not None and self._is_expired(record): self._remove_record_locked(record) record = None if record is None: if existing_job_id is not None: self.idempotency_jobs.pop(idempotency_key, None) return None if not secrets.compare_digest( record.request_fingerprint, request_fingerprint, ): raise HTTPException( status_code=http_status.HTTP_409_CONFLICT, detail=( "Idempotency-Key was already used with a different " "request body." ), ) return record def create_job( self, payload: ImageInversionJobRequest, request: Request, ) -> JSONResponse: """验证远程图片并创建立即执行的异步标签分析任务。 Args: payload: 已通过 Pydantic 校验的命名分析参数。 request: 包含共享密钥的 FastAPI 请求。 Returns: HTTP 202 任务标识、状态 URL 与建议轮询间隔。 """ self._require_api_key(request) self.remove_expired_jobs() idempotency_key = request.headers.get("idempotency-key", "").strip() or None if idempotency_key is not None and len(idempotency_key) > 200: raise HTTPException( status_code=http_status.HTTP_400_BAD_REQUEST, detail="Idempotency-Key must be 200 characters or fewer.", ) request_fingerprint = fingerprint_request(payload) existing_record = self._lookup_idempotent_job( idempotency_key, request_fingerprint, ) if existing_record is not None: return self._status_response(request, existing_record) self._ensure_capacity_for_new_record() with self.jobs_lock: active_record = self.jobs.get(self.active_job_id or "") api_job_active = active_record is not None and active_record.status in { "queued", "running", } # 抓图前先做快速忙检查；抓图后仍在状态锁内做权威二次判定。 if api_job_active or self.inference_slot.locked(): raise HTTPException( status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE, detail="The prediction service is busy.", headers={"Retry-After": "5"}, ) slot_owned_by_request = False record: ImageInversionJobRecord | None = None replay_record: ImageInversionJobRecord | None = None input_image: Image.Image | None = None try: try: input_image = self.image_fetcher( payload.input_image_url, self.settings, ) except ImageSourceError as exc: raise HTTPException( status_code=exc.status_code, detail=exc.public_message, ) from exc with self.jobs_lock: # 相同幂等键可并发抓图；只有二次判定有权创建唯一任务。 replay_record = self._lookup_idempotent_job( idempotency_key, request_fingerprint, ) if replay_record is None: self._ensure_capacity_for_new_record() active_record = self.jobs.get(self.active_job_id or "") if active_record is not None and active_record.status in { "queued", "running", }: raise HTTPException( status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE, detail="The prediction service is busy.", headers={"Retry-After": "5"}, ) if not self.inference_slot.acquire(blocking=False): raise HTTPException( status_code=http_status.HTTP_503_SERVICE_UNAVAILABLE, detail="The prediction service is busy.", headers={"Retry-After": "5"}, ) slot_owned_by_request = True job_id = secrets.token_urlsafe(24) record = ImageInversionJobRecord( job_id=job_id, payload=payload, request_fingerprint=request_fingerprint, input_image=input_image, idempotency_key=idempotency_key, ) self.jobs[job_id] = record self.active_job_id = job_id if idempotency_key is not None: self.idempotency_jobs[idempotency_key] = job_id if replay_record is not None: self._close_images(input_image) input_image = None return self._status_response(request, replay_record) if record is None: raise RuntimeError("Job initialization did not produce a record.") response = self._status_response(request, record) worker = threading.Thread( target=self._execute_job, args=(record,), name=f"image-inversion-job-{record.job_id[:8]}", daemon=True, ) worker.start() # 线程启动成功后，槽位所有权转交给 worker 最外层 finally。 slot_owned_by_request = False return response except Exception: with self.jobs_lock: if record is not None: self._remove_record_locked(record) if record is not None and self.active_job_id == record.job_id: self.active_job_id = None self._close_job_image(record) if record is None: self._close_images(input_image) if slot_owned_by_request: self.inference_slot.release() raise def _execute_job(self, record: ImageInversionJobRecord) -> None: try: with self.jobs_lock: current_record = self.jobs.get(record.job_id) if current_record is not record or record.status != "queued": return record.status = "running" record.started_at = time.time() if record.input_image is None: raise RuntimeError("The job input image is unavailable.") result = self.executor( record.payload, record.input_image, record.job_id, ) if not isinstance(result, dict): raise RuntimeError("The prediction executor did not return an object.") # 任务结果必须在后台线程内完成 JSON 校验，避免轮询阶段才暴露序列化错误。 normalized_result = json.loads( json.dumps(result, ensure_ascii=False, allow_nan=False) ) with self.jobs_lock: current_record = self.jobs.get(record.job_id) if current_record is record and record.status == "running": record.status = "succeeded" record.result = normalized_result record.completed_at = time.time() except Exception as exc: LOGGER.error( "Image Inversion job %s failed with %s", record.job_id[:8], type(exc).__name__, ) with self.jobs_lock: current_record = self.jobs.get(record.job_id) if current_record is record and record.status in {"queued", "running"}: record.status = "failed" record.error_type = type(exc).__name__ record.completed_at = time.time() finally: with self.jobs_lock: if self.active_job_id == record.job_id: self.active_job_id = None self._close_job_image(record) self.inference_slot.release() self.remove_expired_jobs() def get_job_status(self, job_id: str, request: Request) -> JSONResponse: """读取标签分析任务状态并在成功后返回 JSON 结果。 Args: job_id: POST 创建任务时返回的自定义任务标识。 request: 用于鉴权的 FastAPI 请求。 Returns: 运行中返回 202，成功返回 result，失败返回脱敏 500。 """ self._require_api_key(request) record = self._job_or_404(job_id) with self.jobs_lock: status_value = record.status result_value = record.result if status_value in {"queued", "running"}: return JSONResponse( status_code=http_status.HTTP_202_ACCEPTED, headers={ "Retry-After": str(self.settings.poll_after_seconds), "Cache-Control": "no-store", }, content={"status": status_value}, ) if status_value == "failed": return JSONResponse( status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR, headers={"Cache-Control": "no-store"}, content={"error": {"code": "PREDICTION_FAILED"}}, ) if result_value is None: raise RuntimeError("A succeeded prediction job has no result.") return JSONResponse( content=result_value, headers={"Cache-Control": "no-store"}, ) def create_job_api_lifespan(job_api: ImageInversionJobAPI): """创建供 Gradio launch 组合使用的自定义路由 lifespan。 Args: job_api: 已配置执行器和共享推理锁的标签分析任务服务。 Returns: 可传给 Gradio app_kwargs 的异步 lifespan 上下文管理器。 """ @asynccontextmanager async def job_api_lifespan(app: Any): """在 Gradio 应用接收请求前安装自定义任务路由。 Args: app: Gradio 创建并传入 lifespan 的 FastAPI 应用。 Returns: lifespan 启动与关闭阶段不返回业务数据。 """ job_api.install_on_app(app) yield return job_api_lifespan