bria-fibo-image

Bria FIBO 图像生成

除非用户明确要求使用本地权重，否则请使用 Bria 第一方托管的 v2 API。将本地权重许可与托管 API 条款分开处理。第一方请求绝不可经由 fal.ai、Replicate 或其他网关转发。

证据用语

呈现重要事实时请使用以下标签：

• 文档记载的事实 - 第一方 API 架构、模型卡、论文、价格页面或法律文件中明确说明的内容。
• 提供方声明 - Bria 对性能、训练数据、安全性、权利、合规性或赔偿的陈述；不得将其复述为经过独立验证的事实。
• 操作性经验法则 - 本技能的保守建议，不代表 Bria 的限制或保证。
• 未知 - 在已审阅的第一方公开材料中未予明确。依赖此类信息前，请咨询 Bria 或查阅用户已签署的协议。

本技能中的易变事实和来源已于 2026-07-10 核查。在做出生产或采购决策前，请重新核查架构、定价、计划限额、模型卡和适用条款。

先记录决策

在编写代码或提交工作前，记录以下内容：

1. 意图：generate、基于一张参考图进行 inspire、优化已保存的结构化提示词 refine，或使用结构化提示词加种子进行 recreate。
2. 交付方式：托管 Standard、托管 Lite，或本地／本地部署评估。
3. 输入：提示词、一张参考图、已保存的 structured_prompt、种子，以及权利／同意的来源记录。
4. 输出：宽高比、png 或 jpeg、仅 Standard 支持的 1MP 或 4MP、数量、截止时间和验收标准。
5. 治理要求：账户／计划、允许用途、个人／机密数据分类、地区、保留要求，以及当前协议／DPA。
6. 费用：免费额度消耗量或预期费用、最大调用次数、美元支出上限，以及批准提交的人员。

如果权利、同意或数据分类不明确，请保持离线并询问。公开的图像 URL 并不能证明已获得使用许可。

选择准确的路由

文档记载的事实（v2 OpenAPI）： 基础 URL 为 https://engine.prod.bria-api.com/v2；身份验证使用 api_token 请求头。Bria 的 LLM 文档还要求 Agent 客户端发送 User-Agent: BriaPlatform/APIdocs/LLMsAgent，尽管端点 OpenAPI 中未体现此要求。请包含该请求头。

需求	路由	桥接模型行为	重要请求字段
最高保真度图像	POST /image/generate	Bria 记载使用 Gemini 2.5 Flash 作为 VLM 桥接模型，然后使用完整 FIBO	可用 resolution、negative_prompt、steps_num
更快／更低成本的迭代	POST /image/generate/lite	Bria 记载使用其开源 FIBO-VLM 桥接模型和 FIBO Lite	当前架构中没有 resolution、negative_prompt 或 steps_num
仅生成结构化提示词	POST /structured_prompt/generate	Standard VLM 桥接模型；不返回图像	无图像输出控制项
仅生成 Lite 结构化提示词	POST /structured_prompt/generate/lite	Lite VLM 桥接模型；不返回图像	无图像输出控制项

提供方声明： Standard 定位为提供最高质量的意图理解；FIBO Lite 定位为速度更快、成本更低。FIBO Lite 模型卡称其为经过蒸馏的少步数变体，并指出存在质量上的权衡。选择生产环境默认配置前，请使用用户自己的提示词集进行基准测试。

隐私影响： Standard 文档称 Gemini 2.5 Flash 会参与提示词／图像到 JSON 的转换。已审阅的公开页面未明确托管数据的具体路径、Google 子处理方／条款、地区、保留期限，或是否适用零数据保留。有关已签署的合同／DPA 和架构是否已回答这些问题之前，请勿发送敏感、受监管、个人或机密输入。Lite 托管路由仍会将数据发送给 Bria；“本地／本地部署”需要单独取得许可。

当前请求契约

对于两个图像生成路由，请严格使用以下互斥组合之一：

• prompt
• images（数组，最多包含一项）
• images 加 prompt
• structured_prompt
• structured_prompt 加 prompt

对于两个结构化提示词路由，当前第一方架构允许使用 prompt、images、images 加 prompt，或 structured_prompt 加用于细化的 prompt；不允许仅使用 structured_prompt。请遵循所选路由的架构，不要将四个端点合并为一个契约。

如需复现或细化，请提供已保存的 seed 和已保存的结构化提示词。structured_prompt 是一个包含 JSON 的字符串，而不是嵌套请求对象。请验证该字符串可解析为 JSON 对象，同时保留返回的原始字符串，以作来源凭证。

常见的文档化图像生成字段：

• model_version：当前枚举仅包含 FIBO。请发送此字段，以免字段被静默省略，但请注意，它不会固定不可变的后端版本。
• aspect_ratio：1:1、2:3、3:2、3:4、4:3、4:5、5:4、9:16 或 16:9。
• seed：整数；公开 OpenAPI 未给出最小值或最大值。请勿自行编造范围。
• sync：默认为 false；生产环境中请使用异步操作。
• webhook_url：可选的 HTTPS 异步任务回调地址。
• output_type：png 或 jpeg；默认为 png。
• ip_signal：默认为 false；当 IP 警告对工作流有帮助时，请设为 true。
• prompt_content_moderation、visual_input_content_moderation、visual_output_content_moderation：当前架构默认值均为 true。本技能始终发送 true，且不提供禁用开关。

仅用于标准图像的字段：

• resolution：1MP 或 4MP，默认为 1MP。Bria 表示 4MP 会增加约 30 秒；请将延迟视为服务提供方的估算值。
• steps_num：整数，范围为 35-50，默认为 50。
• negative_prompt：可选字符串。

当前 Lite 图像架构不接受这三个仅用于标准图像的字段。当前结构化提示词路由不接受图像输出字段。请在本地拒绝未知字段，不要指望服务会忽略它们。

OpenAPI 接受公共 URL 或不带 data:image/...;base64, 前缀的原始 base64 图像，并列出 JPEG/JPG/PNG/WEBP 格式。文档未说明 FIBO 特有的字节数、像素、URL 获取、动画、元数据或解压缩限制。客户端限制属于操作层面的经验做法，并非服务提供方限制。

编写可控输入

对于初始文本生成，请将提示词写成可审计的创意规范：

Purpose: ecommerce hero image for a ceramic pour-over set.
Subject: one matte cobalt dripper and matching carafe; accurate circular rim; no people.
Composition: centered three-quarter view, object fills 65% of frame, clear negative space at upper right.
Environment: warm gray seamless studio sweep.
Lighting: large softbox camera-left, gentle rim light, grounded soft shadow.
Camera: eye-level, 70 mm product-photography perspective, all product edges sharp.
Palette/material: cobalt #164B8C ceramic, neutral background, physically plausible glaze.
Text: none. Logos/brands: none.
Acceptance: one product set, no duplicate handles, no floating parts, no clipped silhouette.

对于文字内容，请将准确的可见字符放在引号中，并在验收标准中重复这些字符。生成的文本仍需人工校对。

对于参考图像：

• 确认用户拥有该图像，或获得了提交许可；若图像中有人物，还需取得隐私权和肖像权方面的许可；
• 每次只使用一张图像；在政策允许时移除不必要的 EXIF/GPS 信息，同时单独保留未经修改的源文件；
• 明确说明哪些属性可以迁移（色彩、光线、整体构图），哪些绝不可迁移（身份、徽标、具有鲜明特征且受保护的角色、水印）；
• 不要将“汲取灵感”描述为已获得版权许可，也不要保证结果不会相似。

对于迭代，请保存返回的结构化提示词和 seed。一次只提出一项独立改动，例如“只将相机角度改为低机位三分之四视角；保留主体数量、色彩、光线和布局。”为便于审查，请对解析后的 JSON 进行差异比较，同时也保留逐字节一致的原始字符串。**服务提供方声明：**Bria 将结构化提示词加 seed 描述为可确定性地精确复现并进行独立细化。**操作层面的经验做法：**请将其视为追求可重复性的意图，而非保证逐位一致，因为托管路由未公开不可变的模型版本；请在自己的回归测试集中衡量感知和语义偏差。

审批与费用政策

可以准备提示词、载荷、试运行、验证器或本地测试夹具。任何托管生成或结构化提示词提交都会消耗配额，并可能产生费用。

发起实时请求前，必须满足以下全部四项：

1. SEND_BRIA_REQUEST=1
2. APPROVE_BRIA_PAID_CALL=YES（免费试用配额也必须设置）
3. 数值型 BRIA_MAX_COST_USD，且其值至少等于带日期的单次调用估算费用
4. BRIA_APPROVED_REQUEST_SHA256，从经审核的试运行中复制；该值将审批绑定到确切的路由、载荷／参考内容字节、输出主机允许列表、受控输出目录、费用、上限和客户端作业密钥

还要说明路由、调用次数、估算最高费用，以及重试是否可能创建第二个计费作业。不要尝试推测性变体。不要将客户端连接失败视为 Bria 未创建作业的证据。

标注日期的定价页面信息（2026-07-10）： Development 列出的价格为 Fibo $0.03/image、Fibo Lite $0.02/image、Fibo Structured Prompt $0.02/call；Business 和 Enterprise 为定制价格。页面还宣传提供 100 次免费生成。税费、抵扣额、协商价格、失败作业的计费方式、4MP 的处理方式，以及结构化提示词 Lite 定价，均未在公开架构／页面中得到完整说明；以账户控制台和已签署协议为准。

在设置 BRIA_ESTIMATED_CALL_USD（该值须根据实时账户／合同确定）之前，示例会阻止实时 4MP Standard 调用或 Lite 结构化提示词调用。随后，该操作员提供的估算值必须不超过 BRIA_MAX_COST_USD；这并非 Bria 公布的价格。

公开的速率限制信息存在冲突：API 概览列出的每分钟请求数为 Free Trial 10、Starter 60、Pro/Enterprise 1000，按端点计算；而价格对比页面显示每分钟操作数为 10、60 和 150，企业吞吐量则为定制值。不要将其中任何一项硬编码为账户权益。若响应中包含速率限制标头，请读取；遵守 429 响应，并确认账户的实时限制。

完整托管示例：先试运行

这个单文件 Python 3.11+ 客户端仅使用标准库调用 API，并使用 Pillow 在本地验证图像（python -m pip install Pillow）。它支持全部四个第一方路由、提示词／参考图／优化输入、异步轮询、严格的 URL 策略、有界读取、原子写入和来源清单。默认情况下，它绝不会发送请求。

将其复制到 bria_fibo.py：

from __future__ import annotations

import base64
import hashlib
import ipaddress
import json
import math
import os
import random
import socket
import tempfile
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, Request, build_opener

BASE = "https://engine.prod.bria-api.com/v2"
API_HOST = "engine.prod.bria-api.com"
USER_AGENT = "BriaPlatform/APIdocs/LLMsAgent"
ROUTES: dict[str, float | None] = {
    "/image/generate": 0.03,
    "/image/generate/lite": 0.02,
    "/structured_prompt/generate": 0.02,
    "/structured_prompt/generate/lite": None,  # public price page is not explicit
}
RATIOS = {"1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9"}
MAX_JSON_BYTES = 16 * 1024 * 1024       # client heuristic, not a Bria limit
MAX_INPUT_BYTES = 12 * 1024 * 1024      # client heuristic, not a FIBO limit
MAX_OUTPUT_BYTES = 64 * 1024 * 1024     # client heuristic
MAX_PIXELS = 40_000_000                  # client heuristic
POLL_TIMEOUT_S = 600


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RuntimeError(f"redirect refused ({code}); validate the new URL out of band")


class BriaHTTPError(RuntimeError):
    def __init__(self, status_code: int, message: str, retry_after: str | None = None):
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after


class AmbiguousTransportError(RuntimeError):
    pass


OPENER = build_opener(NoRedirect)


def env_bool(name: str) -> bool:
    return os.getenv(name, "") == "1"


def exact_https(url: str, allowed_hosts: set[str], path_prefix: str | None = None) -> str:
    p = urlsplit(url)
    if p.scheme != "https" or not p.hostname or p.username or p.password:
        raise ValueError("URL must be credential-free HTTPS")
    if p.port not in (None, 443) or p.hostname.lower() not in allowed_hosts:
        raise ValueError("URL host/port is not explicitly allowed")
    if path_prefix is not None and not p.path.startswith(path_prefix):
        raise ValueError("URL path is outside the allowed prefix")
    for item in socket.getaddrinfo(p.hostname, 443, type=socket.SOCK_STREAM):
        addr = ipaddress.ip_address(item[4][0])
        if not addr.is_global:
            raise ValueError("URL resolved to a non-public address")
    return url


def read_json_response(resp, cap: int = MAX_JSON_BYTES) -> dict[str, Any]:
    length = resp.headers.get("Content-Length")
    if length and int(length) > cap:
        raise RuntimeError("JSON response exceeds client cap")
    raw = resp.read(cap + 1)
    if len(raw) > cap:
        raise RuntimeError("JSON response exceeds client cap")
    ctype = resp.headers.get_content_type()
    if ctype != "application/json":
        raise RuntimeError(f"unexpected response Content-Type: {ctype}")
    value = json.loads(raw)
    if not isinstance(value, dict):
        raise RuntimeError("expected a JSON object")
    return value


def api_json(
    method: str,
    url: str,
    token: str,
    body: dict[str, Any] | None = None,
    timeout: float = 60,
) -> dict[str, Any]:
    exact_https(url, {API_HOST}, "/v2/")
    raw = None if body is None else json.dumps(body, separators=(",", ":")).encode()
    if raw is not None and len(raw) > MAX_JSON_BYTES:
        raise ValueError("request exceeds client cap")
    headers = {"api_token": token, "Accept": "application/json", "User-Agent": USER_AGENT}
    if raw is not None:
        headers["Content-Type"] = "application/json"
    req = Request(url, data=raw, headers=headers, method=method)
    try:
        with OPENER.open(req, timeout=timeout) as resp:
            return read_json_response(resp)
    except HTTPError as exc:
        retry_after = exc.headers.get("Retry-After")
        try:
            raw_error = exc.read(16_385)[:16_384]
        finally:
            exc.close()
        try:
            error_data = json.loads(raw_error)
        except (UnicodeDecodeError, json.JSONDecodeError):
            error_data = {}
        if not isinstance(error_data, dict):
            error_data = {}
        error_obj = error_data.get("error")
        error_obj = error_obj if isinstance(error_obj, dict) else {}
        message = str(error_obj.get("message", ""))
        raise BriaHTTPError(
            exc.code,
            f"Bria HTTP {exc.code}; request_id={error_data.get('request_id')!r}; "
            f"error_code={error_obj.get('code')!r}; "
            f"message_sha256={hashlib.sha256(message.encode()).hexdigest() if message else None}",
            retry_after,
        ) from None
    except (TimeoutError, URLError) as exc:
        raise AmbiguousTransportError(
            "transport outcome is ambiguous; do not resubmit a create request automatically; "
            "reconcile by request_id/account logs first"
        ) from exc


def retry_after_seconds(value: str | None) -> float:
    if not value:
        return 0.0
    try:
        seconds = float(value)
    except ValueError:
        try:
            moment = parsedate_to_datetime(value)
            if moment.tzinfo is None:
                moment = moment.replace(tzinfo=timezone.utc)
            seconds = (moment - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0.0
    return seconds if math.isfinite(seconds) and seconds > 0 else 0.0


def validate_local_image(path: Path, byte_cap: int = MAX_INPUT_BYTES) -> bytes:
    if not path.is_file() or path.stat().st_size > byte_cap:
        raise ValueError("reference is missing or exceeds the client cap")
    raw = path.read_bytes()
    if not (raw.startswith(b"\x89PNG\r\n\x1a\n") or raw.startswith(b"\xff\xd8\xff") or raw.startswith(b"RIFF") and raw[8:12] == b"WEBP"):
        raise ValueError("reference magic is not PNG/JPEG/WEBP")
    try:
        from PIL import Image, UnidentifiedImageError
        Image.MAX_IMAGE_PIXELS = MAX_PIXELS
        with Image.open(path) as im:
            if im.width * im.height > MAX_PIXELS or getattr(im, "n_frames", 1) != 1:
                raise ValueError("reference pixels/frames exceed client policy")
            im.verify()
        with Image.open(path) as im:
            im.load()
    except ImportError as exc:
        raise RuntimeError("install Pillow to validate image inputs") from exc
    except (UnidentifiedImageError, OSError) as exc:
        raise ValueError("reference image failed full decode") from exc
    return raw


def load_structured_prompt(path: Path) -> str:
    text = path.read_text(encoding="utf-8")
    parsed = json.loads(text)
    if not isinstance(parsed, dict):
        raise ValueError("structured_prompt must encode a JSON object")
    return text


def build_payload(route: str) -> dict[str, Any]:
    prompt = os.getenv("BRIA_PROMPT", "").strip()
    ref_path = os.getenv("BRIA_REFERENCE_PATH", "").strip()
    struct_path = os.getenv("BRIA_STRUCTURED_PROMPT_PATH", "").strip()
    if ref_path and struct_path:
        raise ValueError("images and structured_prompt are mutually exclusive")
    payload: dict[str, Any] = {}
    if prompt:
        payload["prompt"] = prompt
    if ref_path:
        payload["images"] = [base64.b64encode(validate_local_image(Path(ref_path))).decode("ascii")]
    if struct_path:
        payload["structured_prompt"] = load_structured_prompt(Path(struct_path))
    if not any(k in payload for k in ("prompt", "images", "structured_prompt")):
        raise ValueError("set BRIA_PROMPT, BRIA_REFERENCE_PATH, or BRIA_STRUCTURED_PROMPT_PATH")
    if route.startswith("/structured_prompt/") and "structured_prompt" in payload and "prompt" not in payload:
        raise ValueError("structured-prompt routes require a prompt when refining a structured_prompt")

    seed_text = os.getenv("BRIA_SEED", "").strip()
    if struct_path and not seed_text:
        raise ValueError("recreation/refinement requires the saved BRIA_SEED")
    if seed_text:
        payload["seed"] = int(seed_text)

    payload.update({
        "sync": False,
        "ip_signal": True,
        "prompt_content_moderation": True,
        "visual_input_content_moderation": True,
    })
    if route.startswith("/image/"):
        ratio = os.getenv("BRIA_ASPECT_RATIO", "1:1")
        if ratio not in RATIOS:
            raise ValueError("unsupported aspect ratio")
        payload.update({
            "model_version": "FIBO",
            "aspect_ratio": ratio,
            "output_type": os.getenv("BRIA_OUTPUT_TYPE", "png"),
            "visual_output_content_moderation": True,
        })
        if payload["output_type"] not in {"png", "jpeg"}:
            raise ValueError("output type must be png or jpeg")
    if route == "/image/generate":
        resolution = os.getenv("BRIA_RESOLUTION", "1MP")
        steps = int(os.getenv("BRIA_STEPS", "50"))
        if resolution not in {"1MP", "4MP"} or not 35 <= steps <= 50:
            raise ValueError("invalid Standard resolution/steps")
        payload.update({"resolution": resolution, "steps_num": steps})
        if os.getenv("BRIA_NEGATIVE_PROMPT"):
            payload["negative_prompt"] = os.environ["BRIA_NEGATIVE_PROMPT"]
    return payload


def poll(request_id: str, token: str) -> dict[str, Any]:
    # Construct from the trusted request_id; never forward api_token to returned arbitrary URLs.
    if not request_id or "/" in request_id or len(request_id) > 256:
        raise RuntimeError("invalid request_id")
    url = f"{BASE}/status/{request_id}"
    deadline = time.monotonic() + POLL_TIMEOUT_S
    delay = 1.0
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError(
                f"poll timeout for {request_id}; this does not cancel the job or prove it was not billed"
            )
        try:
            data = api_json("GET", url, token, timeout=max(0.001, min(60.0, remaining)))
        except BriaHTTPError as exc:
            if exc.status_code == 404:
                raise RuntimeError(f"status record is missing or expired for {request_id}") from exc
            if exc.status_code != 429 and exc.status_code < 500:
                raise
            print(f"status retry after HTTP {exc.status_code}")
            delay = max(min(delay * 1.7, 15.0), retry_after_seconds(exc.retry_after))
        except AmbiguousTransportError:
            # GET status is safe to retry; still bound the attempts by the deadline.
            print("status retry after transport failure")
            delay = min(delay * 1.7, 15.0)
        else:
            if data.get("request_id") not in (None, request_id):
                raise RuntimeError("status request_id mismatch")
            status = data.get("status")
            if status == "IN_PROGRESS":
                pass
            elif status == "COMPLETED":
                return data
            elif status in {"ERROR", "UNKNOWN"}:
                err = data.get("error") or {}
                if not isinstance(err, dict):
                    raise RuntimeError(f"Bria job {status} returned an invalid error object")
                message = str(err.get("message", ""))
                raise RuntimeError(
                    f"Bria job {status}: code={err.get('code')!r} "
                    f"message_sha256={hashlib.sha256(message.encode()).hexdigest() if message else None}"
                )
            else:
                raise RuntimeError(f"unknown Bria job status: {status!r}")
            delay = min(delay * 1.7, 15.0)
        remaining = deadline - time.monotonic()
        sleep_for = delay + random.uniform(0, min(delay * 0.2, 1.0))
        if sleep_for >= remaining:
            raise TimeoutError(
                f"poll deadline cannot accommodate retry delay for {request_id}; "
                "this does not cancel the job or prove it was not billed"
            )
        time.sleep(sleep_for)


def atomic_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
            fd = None
            f.write(text)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp_name, path)
    finally:
        if fd is not None:
            os.close(fd)
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


def download_image(
    url: str,
    destination: Path,
    allowed_hosts: set[str],
    expected_content_type: str,
) -> tuple[str, int, str]:
    exact_https(url, allowed_hosts)
    req = Request(url, headers={"Accept": "image/png,image/jpeg", "User-Agent": USER_AGENT}, method="GET")
    destination.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(prefix=destination.name + ".", dir=destination.parent)
    digest = hashlib.sha256()
    size = 0
    ctype = ""
    try:
        with os.fdopen(fd, "wb") as out:
            fd = None
            with OPENER.open(req, timeout=60) as resp:
                ctype = resp.headers.get_content_type()
                if ctype != expected_content_type:
                    raise RuntimeError(f"unexpected image Content-Type: {ctype}")
                length = resp.headers.get("Content-Length")
                if length and int(length) > MAX_OUTPUT_BYTES:
                    raise RuntimeError("image exceeds client cap")
                while True:
                    chunk = resp.read(1024 * 1024)
                    if not chunk:
                        break
                    size += len(chunk)
                    if size > MAX_OUTPUT_BYTES:
                        raise RuntimeError("image exceeds client cap")
                    digest.update(chunk)
                    out.write(chunk)
            out.flush()
            os.fsync(out.fileno())
        with Path(tmp_name).open("rb") as check:
            raw_head = check.read(12)
        if ctype == "image/png" and not raw_head.startswith(b"\x89PNG\r\n\x1a\n"):
            raise RuntimeError("PNG magic mismatch")
        if ctype == "image/jpeg" and not raw_head.startswith(b"\xff\xd8\xff"):
            raise RuntimeError("JPEG magic mismatch")
        validate_local_image(Path(tmp_name), MAX_OUTPUT_BYTES)
        os.replace(tmp_name, destination)
        return digest.hexdigest(), size, ctype
    finally:
        if fd is not None:
            os.close(fd)
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


def read_local_json(path: Path) -> dict[str, Any]:
    if path.stat().st_size > MAX_JSON_BYTES:
        raise RuntimeError("local job record exceeds JSON cap")
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise RuntimeError("local job record is not a JSON object")
    return value


def claim_job(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as output:
            fd = None
            json.dump(value, output, indent=2, sort_keys=True)
            output.flush(); os.fsync(output.fileno())
    finally:
        if fd is not None: os.close(fd)


def save_job(path: Path, value: dict[str, Any]) -> None:
    atomic_text(path, json.dumps(value, indent=2, sort_keys=True) + "\n")


def main() -> None:
    route = os.getenv("BRIA_ROUTE", "/image/generate")
    if route not in ROUTES:
        raise ValueError("BRIA_ROUTE is not one of the four first-party FIBO routes")
    payload = build_payload(route)
    public_cost = ROUTES[route]
    override = os.getenv("BRIA_ESTIMATED_CALL_USD", "").strip()
    try: cost = float(override) if override else public_cost
    except ValueError as exc: raise RuntimeError("BRIA_ESTIMATED_CALL_USD must be numeric") from exc
    if route == "/image/generate" and payload.get("resolution") == "4MP" and not override:
        cost = None  # public page does not state whether 4MP changes billing
    if cost is not None and (not math.isfinite(cost) or cost <= 0):
        raise RuntimeError("estimated call cost must be finite and positive")
    if cost is not None and public_cost is not None and cost < public_cost:
        raise RuntimeError("operator estimate cannot undercut the dated public route price")
    try: max_cost = float(os.getenv("BRIA_MAX_COST_USD", "0"))
    except ValueError as exc: raise RuntimeError("BRIA_MAX_COST_USD must be numeric") from exc
    if not math.isfinite(max_cost) or max_cost < 0:
        raise RuntimeError("BRIA_MAX_COST_USD must be finite and non-negative")
    request_sha256 = hashlib.sha256(json.dumps(
        payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    client_job_key = os.getenv("BRIA_CLIENT_JOB_KEY", f"bria-{request_sha256[:20]}")
    output_hosts = {h.strip().lower() for h in os.getenv("BRIA_OUTPUT_HOSTS", "").split(",") if h.strip()}
    out_dir = Path(os.getenv("BRIA_OUTPUT_DIR", "bria-output")).resolve()
    approval_value = {"route": route, "request_sha256": request_sha256,
                      "estimated_call_usd": cost, "max_cost_usd": max_cost,
                      "client_job_key": client_job_key,
                      "output_hosts": sorted(output_hosts), "output_dir": str(out_dir)}
    approval_sha256 = hashlib.sha256(json.dumps(
        approval_value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
    input_mode = "+".join(k for k in ("prompt", "images", "structured_prompt") if k in payload)
    print(json.dumps({
        "dry_run": not env_bool("SEND_BRIA_REQUEST"),
        "url": BASE + route,
        "input_mode": input_mode,
        "aspect_ratio": payload.get("aspect_ratio"),
        "resolution": payload.get("resolution"),
        "estimated_max_usd_2026_07_10": cost,
        "maximum_approved_usd": max_cost,
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "approval_sha256": approval_sha256,
        "output_hosts": sorted(output_hosts),
        "output_dir": str(out_dir),
        "moderation": "all applicable stages enabled",
    }, indent=2))
    if not env_bool("SEND_BRIA_REQUEST"):
        print("DRY RUN ONLY: no network request made")
        return
    if os.getenv("APPROVE_BRIA_PAID_CALL") != "YES":
        raise RuntimeError("set APPROVE_BRIA_PAID_CALL=YES after reviewing this request")
    if cost is None:
        raise RuntimeError("set BRIA_ESTIMATED_CALL_USD from the live account/contract before approval")
    if max_cost <= 0 or max_cost < cost:
        raise RuntimeError("BRIA_MAX_COST_USD is below the dated per-call estimate")
    if os.getenv("BRIA_APPROVED_REQUEST_SHA256") != approval_sha256:
        raise RuntimeError("approval hash does not match this exact request, cost, and client job key")
    token = os.getenv("BRIA_API_TOKEN", "")
    if not token:
        raise RuntimeError("BRIA_API_TOKEN is required only for a live call")
    if route.startswith("/image/") and not output_hosts:
        raise RuntimeError("preconfigure exact BRIA_OUTPUT_HOSTS from your Bria deployment/egress policy")

    job_token = hashlib.sha256(client_job_key.encode()).hexdigest()[:20]
    job_dir = out_dir / job_token
    ledger_path = job_dir / "job.json"
    manifest_path = job_dir / "manifest.json"
    ledger_base = {"client_job_key": client_job_key, "route": route,
                   "request_sha256": request_sha256, "approval_sha256": approval_sha256,
                   "estimated_call_usd": cost, "approved_max_cost_usd": max_cost}
    if manifest_path.exists():
        existing_manifest = read_local_json(manifest_path)
        if any(existing_manifest.get(key) != value for key, value in ledger_base.items()):
            raise RuntimeError("existing manifest does not match this approved request")
        print(json.dumps(existing_manifest, indent=2))
        return
    warnings: list[str] = []
    warning_hashes: list[str] = []
    warning_present = False
    if ledger_path.exists():
        existing = read_local_json(ledger_path)
        if any(existing.get(key) != value for key, value in ledger_base.items()):
            raise RuntimeError("existing job ledger does not match this approved request")
        request_id = existing.get("request_id")
        if not isinstance(request_id, str):
            raise RuntimeError(
                "existing create has no confirmed request_id; treat it as ambiguous and do not replay. "
                "Use a new explicit BRIA_CLIENT_JOB_KEY only after operator reconciliation and approval."
            )
        saved_hashes = existing.get("provider_warning_sha256", [])
        if not isinstance(saved_hashes, list) or not all(isinstance(value, str) for value in saved_hashes):
            raise RuntimeError("existing job ledger has invalid warning hashes")
        warning_hashes = saved_hashes
        warning_present = bool(existing.get("provider_warning_present"))
        final = poll(request_id, token)
    else:
        try:
            claim_job(ledger_path, {**ledger_base, "state": "pre_submit",
                                   "request_id": None, "updated_unix": time.time()})
        except FileExistsError as exc:
            raise RuntimeError(
                "another worker claimed this client job key; do not submit; retry later to resume"
            ) from exc
        try:
            initial = api_json("POST", BASE + route, token, payload)  # never auto-retry this create
        except AmbiguousTransportError:
            save_job(ledger_path, {**ledger_base, "state": "ambiguous_create",
                                   "request_id": None, "updated_unix": time.time()})
            raise
        except BriaHTTPError as exc:
            state = "ambiguous_create" if exc.status_code >= 500 else "create_rejected"
            save_job(ledger_path, {**ledger_base, "state": state,
                                   "request_id": None, "http_status": exc.status_code,
                                   "updated_unix": time.time()})
            raise
        request_id = initial.get("request_id")
        if not isinstance(request_id, str) or not request_id or "/" in request_id or len(request_id) > 256:
            save_job(ledger_path, {**ledger_base, "state": "invalid_create_response",
                                   "request_id": None, "updated_unix": time.time()})
            raise RuntimeError("missing or invalid request_id")
        warnings = [initial["warning"]] if isinstance(initial.get("warning"), str) else []
        warning_hashes = [hashlib.sha256(value.encode()).hexdigest() for value in warnings]
        warning_present = bool(warnings)
        save_job(ledger_path, {**ledger_base, "state": "accepted",
                               "request_id": request_id,
                               "provider_warning_present": warning_present,
                               "provider_warning_sha256": warning_hashes,
                               "updated_unix": time.time()})
        if "status_url" in initial:
            expected = f"{BASE}/status/{request_id}"
            if initial["status_url"] != expected:
                raise RuntimeError("unexpected status_url; token was not forwarded")
            final = poll(request_id, token)
        else:
            # Defensive support for a synchronous-shaped response, though this client sends sync=false.
            final = initial
    if isinstance(final.get("warning"), str):
        final_warning_hash = hashlib.sha256(final["warning"].encode()).hexdigest()
        if final_warning_hash not in warning_hashes: warning_hashes.append(final_warning_hash)
        warning_present = True
    if warning_present:
        print(f"Bria returned {len(warning_hashes)} distinct warning(s) for {request_id}; review required")
    result = final.get("result")
    if not isinstance(result, dict):
        raise RuntimeError("completed response has no result object")

    structured = result.get("structured_prompt")
    if isinstance(structured, str):
        parsed = json.loads(structured)
        if not isinstance(parsed, dict):
            raise RuntimeError("returned structured_prompt is not a JSON object")
        atomic_text(job_dir / "structured_prompt.json", structured)
    manifest: dict[str, Any] = {
        "request_id": request_id,
        "route": route,
        "client_job_key": client_job_key,
        "request_sha256": request_sha256,
        "approval_sha256": approval_sha256,
        "seed": result.get("seed"),
        "structured_prompt_sha256": hashlib.sha256(structured.encode()).hexdigest() if isinstance(structured, str) else None,
        "source_url_recorded": False,
        "provider_warning_present": warning_present,
        "provider_warning_sha256": warning_hashes,
        "moderation_flags": {
            "prompt": True,
            "visual_input": True,
            "visual_output": True if route.startswith("/image/") else None,
        },
        "estimated_call_usd_2026_07_10": cost,
        "estimated_call_usd": cost,
        "approved_max_cost_usd": max_cost,
    }
    if route.startswith("/image/"):
        image_url = result.get("image_url")
        if not isinstance(image_url, str):
            raise RuntimeError("completed image response has no image_url")
        suffix = ".png" if payload["output_type"] == "png" else ".jpg"
        expected_type = "image/png" if payload["output_type"] == "png" else "image/jpeg"
        sha256, size, ctype = download_image(
            image_url,
            job_dir / ("image" + suffix),
            output_hosts,
            expected_type,
        )
        manifest.update({"image_sha256": sha256, "image_bytes": size, "content_type": ctype})
    atomic_text(manifest_path, json.dumps(manifest, indent=2, sort_keys=True) + "\n")
    save_job(ledger_path, {**ledger_base, "state": "completed",
                           "request_id": request_id,
                           "provider_warning_present": warning_present,
                           "provider_warning_sha256": warning_hashes,
                           "updated_unix": time.time()})
    print(json.dumps(manifest, indent=2))


if __name__ == "__main__":
    main()

试运行示例（PowerShell；均不会发送请求）：

$env:BRIA_PROMPT='A cobalt ceramic pour-over set on a warm gray studio sweep; no text or logos'
python .\bria_fibo.py

$env:BRIA_ROUTE='/image/generate/lite'
$env:BRIA_PROMPT='Use only the reference palette and soft side lighting; create a distinct unbranded product'
$env:BRIA_REFERENCE_PATH='C:\approved-inputs\reference.png'
python .\bria_fibo.py

$env:BRIA_ROUTE='/image/generate'
$env:BRIA_REFERENCE_PATH=''
$env:BRIA_STRUCTURED_PROMPT_PATH='C:\approved-inputs\structured_prompt.json'
$env:BRIA_SEED='123456'
$env:BRIA_PROMPT='Change only the background to pale warm gray; preserve every object and camera property'
python .\bria_fibo.py

只有在人类审核试运行摘要、确认当前账户价格／条款，并批准一次调用后，才可执行：

$env:BRIA_API_TOKEN='<set securely; never paste into logs or source>'
$env:BRIA_OUTPUT_HOSTS='<exact host approved for your Bria result URLs>'
$env:BRIA_MAX_COST_USD='0.03'
$env:BRIA_APPROVED_REQUEST_SHA256='<copy approval_sha256 from the dry run with this same cost ceiling>'
$env:APPROVE_BRIA_PAID_CALL='YES'
$env:SEND_BRIA_REQUEST='1'
python .\bria_fibo.py

获批调用完成后，立即取消设置 SEND_BRIA_REQUEST、APPROVE_BRIA_PAID_CALL、BRIA_APPROVED_REQUEST_SHA256 和 API 令牌。不要提交 .env、包含机密数据的提示词、base64 参考内容、签名结果 URL 或 API 响应。该示例特意不会记录令牌、base64、提示词文本、结构化提示词或签名 URL。

异步处理与 Webhook 正确性

文档明确说明：sync=false 会返回 HTTP 202，以及 request_id 和 status_url；使用 api_token 轮询 GET /status/{request_id}。终止状态为 COMPLETED、ERROR 和 UNKNOWN；IN_PROGRESS 表示任务仍在继续。任务级错误通常仍会返回 HTTP 200。请求缺失或已过期时可能返回 404。

状态 OpenAPI 的成功响应架构目前只列出了 COMPLETED，尽管其说明文字要求包含 IN_PROGRESS；请处理说明文字中定义的状态。切勿将 api_token 发送到未经验证的返回 URL。应像示例那样，使用可信基址和请求 ID 构造状态 URL。

发生超时、连接重置或结果不明确的 5xx 错误后，不要自动重试 POST：文档未说明幂等键字段，也未说明支持按客户端密钥查询任务。请通过已捕获的 request_id、账户日志或联系支持人员来核对。状态 GET 请求可以采用有界退避。轮询超时不会取消任务；未发现公开的取消端点。

对于生产环境中的回调，Bria 建议使用 Webhook。文档说明的契约如下：

• 原始 POST 请求体，以及 Bria-Webhook-Id、Bria-Webhook-Timestamp 和 Bria-Webhook-Signature（v1=<base64>）；
• HMAC-SHA256 签名密钥：HMAC-SHA256(api_token, "bria-webhook-signing-v1")；
• 签名消息：{webhook_id}.{timestamp}.{raw_body}；
• 在 10 秒内确认收到；
• 45 分钟内最多尝试投递五次，且可能重复；应按任务/请求 ID 去重。

在解析 JSON 之前验证原始字节，并在返回任何成功响应之前，使用 hmac.compare_digest 或等效的常量时间原语比较预期签名和收到的签名。验证通过并完成持久化去重后，将繁重处理加入队列，并在 10 秒内确认收到；在后台执行工件获取和下游工作。Bria 的公开页面未说明允许的时间戳偏差/重放窗口。请设置较短且受监控的时效窗口（例如五分钟），将已处理 ID 持久化保存，期限应长于 45 分钟的重试时限；谨慎轮换令牌，因为签名密钥由令牌派生，并确认时钟已同步。不要允许回调接收任意大小的负载。

工件与 URL 安全

即使看不到查询参数，也应将 image_url 和 status_url 视为携带凭据或敏感信息。公开文档未说明 FIBO 输出 URL 的有效期。请立即将文件下载到受控存储中，然后对持久化副本进行哈希处理并登记。

至少应做到：

• 要求使用 HTTPS、默认端口、精确的主机允许列表和公共 DNS；
• 禁用重定向；绝不将 api_token 转发到资源主机；
• 在网络层配置出口/DNS 固定，以增强防御 SSRF/DNS 重绑定的能力；
• 限制标头、正文大小和处理时长，检查 MIME 类型及文件魔数，完整解码，限制像素数/帧数，并拒绝压缩包/HTML/SVG；
• 写入同一目录下的临时文件，执行 fsync，再以原子方式替换；
• 保留原始字节，避免意外移除 C2PA 或其他来源元数据；如有需要，应单独验证来源信息；
• 存储请求 ID、路由、种子、结构化提示词哈希、输出哈希、审核/警告处理结果、审批、价格快照以及代码/配置版本；
• 普通日志中省略签名 URL、令牌、base64、个人数据和机密原始提示词。

质量与发布检查

根据书面验收标准进行评估，而不只是看视觉效果：

• 主体数量、解剖结构／几何形状、空间关系、裁切、文本拼写、品牌／知识产权泄漏；
• 与参考内容过于相似的情况，以及身份／公开权利风险；
• 输出尺寸／宽高比、解码完整性、色彩模式、Alpha 通道行为和哈希值；
• 下游优化后是否按要求保留元数据／来源信息；
• 使用已保存的提示词／种子／模型路由时能否重复生成，以及后端更新后是否出现漂移；
• 对高影响、公开、受监管、政治或涉及身份的用途进行人工审核。

审核通过不代表获得许可、内容准确、不侵权或已获品牌批准。请保持提供方的审核功能处于启用状态，并增加应用层面的策略检查。

权利、隐私与许可台账

分别记录以下事项；不要将它们统称为“商业用途安全”：

• **训练数据——提供方声明：**Bria 表示 FIBO 完全使用授权数据训练，并称其符合可追溯性／GDPR／《欧盟人工智能法案》要求。论文／模型卡由 Bria 撰写，属于 Bria 提供的证据，并非独立审计。
• **托管输出权利／赔偿——视合同而定：**价格页面宣传不同套餐提供标准或全面的知识产权／隐私赔偿，而 Legal Lobby 表示在线／免费条款的保证有限，企业条款则提供更全面的保护。适用条款、例外、上限、地区、使用限制，以及用户的提示词／参考内容都可能改变结果。如需依赖这些保障，请进行法律审核。
• **输入内容权利：**用户必须拥有提示词／参考内容的所有权或许可，并拥有相关隐私权／公开权。Bria 的内容指南禁止侵权、欺骗、冒充、违法、仇恨、色情、恶意和欺诈等用途。
• **输出独占性：**较早发布的条款声明，AI 输出可能并非独一无二。请核查当前协议；切勿承诺输出具有独占性或可注册。
• 权重：briaai/FIBO 需通过授权访问，标记为 bria-fibo；其模型卡元数据链接到 CC BY-NC 4.0，并注明仅限非商业用途。briaai/Fibo-lite 标记为 CC BY-NC 4.0，也注明仅限非商业用途。商业本地／本地部署用途须另行取得 Bria 许可／遵守其条款。“开源”是 Bria 使用的措辞；不要以此规避非商业限制或访问条件。下载／部署前，请审核确切的已接受条款及所有依赖项。
• 使用客户数据训练：Bria 的 Legal Lobby 声明，未经明确书面协议，不会使用试点客户的内容／输出训练或改进生成式模型；2025 年 3 月的企业文件也有类似说明，但存在微调例外。若无当前已签署条款，是否适用于所有当前自助服务套餐为未知。
• 保留／删除：2024 年 7 月的隐私政策规定，会根据所述目的及可能的法律需求保留个人数据。较早的公开条款称，除非另有记录，否则不会存储数据，并会在终止后 30 天内删除；但当前 FIBO 任务／结果的保留期限和 URL 有效期为未知。不要承诺零保留。
• 个人数据：较早的条款规定，客户内容不得包含个人数据。当前针对具体套餐的处理方式为未知。除非当前合同和 DPA 明确授权此类用途，否则应谨慎禁止此类内容。
• **数据驻留／子处理方／安全：**从已审核的公开页面中，无法确定 FIBO 的确切处理区域、客户可选的数据驻留区域、子处理方名单、加密详情、备份、事件响应承诺和状态日志保留期限。请获取 DPA／安全材料并确认架构。

本内容是操作指南，不构成法律建议。

当前第一方来源

检查日期：2026-07-10：

• Image Generation v2 API 和可下载的 OpenAPI
• Generate Image 端点
• Status Service
• Webhook 投递与签名
• API 概览：身份验证、速率限制、异步流程、输入编码
• Bria 定价
• Bria 法律文档中心
• Bria 法律文档中心链接的当前内容指南（2026 年 1 月）
• Bria 法律文档中心链接的当前在线条款（2026 年 3 月）
• Bria 内容指南（公开的第一方 PDF；2024 年 6 月索引版本）
• Bria 隐私政策，最后更新于 2024-07-08
• FIBO 模型卡
• FIBO Lite 模型卡
• FIBO 论文，arXiv:2511.06876

如果文档内容不一致，线缆格式以端点 OpenAPI 为准，商业和运营权益以用户的账户／合同为准；在 Bria 以书面形式解决冲突之前，采用更严格的安全与隐私立场。