"""Private adapters for Huobao: Dify, Qwen Image, local H3, and Grok CLI."""
from __future__ import annotations

import asyncio
import base64
from contextlib import closing
from email import policy
from email.parser import BytesParser
import hashlib
import hmac
import json
import math
import os
from pathlib import Path
import re
import secrets
import sqlite3
import struct
import tempfile
import time
from urllib.parse import quote

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse


def chat_query(body):
    messages = body.get("messages")
    if not isinstance(messages, list) or not messages:
        raise HTTPException(400, "messages must be a non-empty array")
    if body.get("tool_choice") == "required" or isinstance(body.get("tool_choice"), dict):
        raise HTTPException(400, "This chatflow does not support required tool calls; use DeepSeek")
    parts = []
    for item in messages:
        content = item.get("content", "")
        if isinstance(content, list):
            if any(p.get("type") != "text" for p in content):
                raise HTTPException(400, "This adapter accepts text messages only")
            content = "\n".join(p.get("text", "") for p in content)
        if content:
            parts.append(f"[{item.get('role', 'user')}]\n{content}")
    fmt = body.get("response_format") or {}
    if fmt.get("type") == "json_schema":
        parts.append("Return only JSON matching this schema:\n" + json.dumps(fmt.get("json_schema", {}).get("schema", {}), ensure_ascii=False))
    elif fmt.get("type") == "json_object":
        parts.append("Return only a valid JSON object, without Markdown fences.")
    if not parts:
        raise HTTPException(400, "Empty messages")
    return "\n\n".join(parts)


def image_dimensions(data_url):
    try:
        data = base64.b64decode(data_url.split(",", 1)[1], validate=True)
        if data.startswith(b"\x89PNG\r\n\x1a\n"):
            return struct.unpack(">II", data[16:24])
        if data.startswith((b"GIF87a", b"GIF89a")):
            return struct.unpack("<HH", data[6:10])
        if data[:2] == b"\xff\xd8":
            offset = 2
            while offset + 4 < len(data):
                if data[offset] != 255:
                    offset += 1
                    continue
                while data[offset] == 255:
                    offset += 1
                marker = data[offset]
                offset += 1
                if marker in {0xD8, 0xD9} or 0xD0 <= marker <= 0xD7:
                    continue
                size = int.from_bytes(data[offset:offset + 2], "big")
                if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}:
                    height, width = struct.unpack(">HH", data[offset + 3:offset + 7])
                    return width, height
                if size < 2:
                    break
                offset += size
        if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
            if data[12:16] == b"VP8X":
                return 1 + int.from_bytes(data[24:27], "little"), 1 + int.from_bytes(data[27:30], "little")
            if data[12:16] == b"VP8L" and data[20] == 0x2F:
                bits = int.from_bytes(data[21:25], "little")
                return (bits & 0x3FFF) + 1, ((bits >> 14) & 0x3FFF) + 1
            if data[12:16] == b"VP8 " and data[23:26] == b"\x9d\x01\x2a":
                width, height = struct.unpack("<HH", data[26:30])
                return width & 0x3FFF, height & 0x3FFF
    except (ValueError, IndexError, struct.error):
        pass
    raise HTTPException(400, "Unable to read first-frame dimensions; use PNG, JPEG or WebP")


VIDEO_MODELS = {
    "MiniMax-H3-VDN-H3": {"reference": "minimax_h3_vdn_hd", "min_images": 1, "max_images": 9},
    "MiniMax-H3": {"text": "minimax_h3_t2v_official", "first": "minimax_h3_i2v", "reference": "minimax_h3_r2v", "min_images": 0, "max_images": 9},
    "MiniMax-H3-Hybrid": {"text": "minimax_h3_hybrid_v0331_t2v", "first": "minimax_h3_hybrid_v0331_i2v", "reference": "minimax_h3_hybrid_v0331_i2v", "min_images": 0, "max_images": 9},
    "MiniMax-H3-R2V-4Step": {"reference": "minimax_h3_r2v_4step", "min_images": 1, "max_images": 9},
    "LTX-2.3": {"text": "ltx23_t2v", "first": "ltx23_t2v", "reference": "ltx23_t2v", "min_images": 0, "max_images": 1},
    "LTX-2.3-Text-Preserve": {"text": "ltx23_t2v_text_preserve", "first": "ltx23_t2v_text_preserve", "reference": "ltx23_t2v_text_preserve", "min_images": 0, "max_images": 1},
    "Wan2.2-I2V": {"first": "wan22_simple", "reference": "wan22_simple", "min_images": 1, "max_images": 1},
}


def video_workflows(spec):
    return {spec[mode] for mode in ("text", "first", "reference") if mode in spec}


def gpu_routing(body):
    gpu = body.get("gpu", "auto")
    if gpu not in ("auto", "gpu1", "gpu2", "gpu3"):
        raise HTTPException(400, "gpu must be auto, gpu1, gpu2 or gpu3")
    if gpu == "auto":
        return {"provider_id": "default", "huobao_panels": True}
    return {"provider_id": {"gpu1": "gpu1_8189", "gpu2": "default", "gpu3": "gpu3"}[gpu], "strict_provider": True, "huobao_panels": True}


def worker_routing(provider_id):
    worker = {"default": ("gpu2", 8188), "gpu1_8189": ("gpu1", 8189), "gpu3": ("gpu3", 8191), "gpu1": ("gpu1", None)}.get(provider_id)
    if not worker:
        return None
    gpu, port = worker
    return {"gpu": gpu, "port": port, "providerId": provider_id, "legacy": provider_id == "gpu1"}


def h3_request(body):
    """Translate the local video protocol. 'default' means upstream auto-balance."""
    model = body.get("model", "MiniMax-H3")
    spec = VIDEO_MODELS.get(model)
    if spec is None:
        raise HTTPException(400, "Unknown local video model; query /v1/models")
    content = body.get("content") or []
    if not isinstance(content, list):
        raise HTTPException(400, "content must be an array")
    prompt = "\n".join(str(c.get("text", "")) for c in content if c.get("type") == "text").strip()
    if not prompt or len(prompt) > 7000:
        raise HTTPException(400, "A prompt of 1-7000 characters is required")
    images, roles = [], []
    for item in content:
        kind = item.get("type")
        if kind == "text":
            continue
        if kind != "image_url" or item.get("role") == "last_frame":
            raise HTTPException(400, "Local video supports first-frame/reference images; last frames, reference video and voice samples are unsupported")
        raw = item.get("image_url") or {}
        url = raw.get("url", "") if isinstance(raw, dict) else raw
        if not isinstance(url, str) or not url.startswith("data:image/"):
            raise HTTPException(400, "Local video requires inline image data; enable Huobao's local-image conversion")
        images.append(url)
        roles.append(item.get("role", "reference_image"))
    if not spec["min_images"] <= len(images) <= spec["max_images"]:
        raise HTTPException(400, f"{model} requires {spec['min_images']}-{spec['max_images']} images")
    if "first_frame" in roles and len(images) != 1:
        raise HTTPException(400, "Use either one fixed first frame or reference images; these modes cannot be combined")
    mode = "text" if not images else "first" if roles == ["first_frame"] else "reference"
    if mode not in spec:
        raise HTTPException(400, f"{model} does not support {mode} mode; use reference images")
    try:
        duration = float(body.get("duration", 5))
    except (ValueError, TypeError):
        raise HTTPException(400, "Invalid duration")
    if not math.isfinite(duration) or not 4 <= duration <= 15:
        raise HTTPException(400, "Duration must be 4-15 seconds")
    ratio = body.get("ratio", "16:9")
    if ratio not in {"16:9", "9:16", "1:1", "4:3", "3:4", "21:9"}:
        raise HTTPException(400, "Unsupported aspect ratio")
    resolution = str(body.get("resolution", "768P")).upper()
    resolutions = {"480P": 480, "720P": 720, "1080P": 1080, "768P": 768, "2K": 1152}
    if model == "MiniMax-H3-VDN-H3":
        if ratio not in {"16:9", "9:16"}:
            raise HTTPException(400, "VDN HD supports 16:9 or 9:16")
        resolutions = {"2K": 1440, "4K": 2160}
    if resolution not in resolutions:
        raise HTTPException(400, "Unsupported video resolution")
    short = resolutions[resolution]
    a, b = map(int, ratio.split(":"))
    if roles == ["first_frame"] and "ratio" not in body:
        a, b = image_dimensions(images[0])
        if a <= 0 or b <= 0:
            raise HTTPException(400, "Invalid first-frame dimensions")
    width = round(short * max(1, a / b) / 64) * 64
    height = round(short * max(1, b / a) / 64) * 64
    if model == "MiniMax-H3-VDN-H3":
        width, height = (short * 16 // 9, short) if a > b else (short, short * 16 // 9)
    workflow = spec[mode]
    request = {**gpu_routing(body), "workflow_key": workflow, "prompt": prompt,
            "input_images": images, "duration": duration, "width": width, "height": height,
            "metadata": {"source": "huobao-drama", "model": model}}
    if body.get("seed") is not None:
        if not isinstance(body["seed"], int) or not 0 <= body["seed"] < 2**63:
            raise HTTPException(400, "Invalid seed")
        request["seed"] = body["seed"]
    return request


def qwen_image_request(body, images=None):
    if body.get("model") != "qwen_image_2_1":
        raise HTTPException(400, "Use model qwen_image_2_1")
    prompt = body.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 32000:
        raise HTTPException(400, "A prompt of 1-32000 characters is required")
    if str(body.get("n", 1)) != "1":
        raise HTTPException(400, "This image adapter supports n=1 per task")
    if body.get("response_format", "url") != "url":
        raise HTTPException(400, "Asynchronous image tasks require response_format=url")
    size = str(body.get("size", "1024x1024")).lower()
    if size == "auto":
        size = "1024x1024"
    match = re.fullmatch(r"(\d{3,4})x(\d{3,4})", size)
    if not match or not all(256 <= int(d) <= 4096 for d in match.groups()):
        raise HTTPException(400, "size must be WIDTHxHEIGHT with dimensions 256-4096")
    width, height = map(int, match.groups())
    refs = images or []
    if len(refs) > 6:
        raise HTTPException(400, "Qwen Image 2.1 supports at most 6 reference images")
    return {**gpu_routing(body), "workflow_key": "qwen_image_2_1", "prompt": prompt.strip(),
            "width": width, "height": height, "batch_size": 1, "input_images": refs,
            "metadata": {"source": "huobao-drama", "output_type": "image"}}


async def parse_image_edit(request):
    content_type = request.headers.get("content-type", "")
    if not content_type.lower().startswith("multipart/form-data;"):
        raise HTTPException(415, "Image edits require multipart/form-data")
    raw = bytearray()
    async for chunk in request.stream():
        raw.extend(chunk)
        if len(raw) > 64 * 1024 * 1024:
            raise HTTPException(413, "Reference uploads exceed 64 MiB")
    message = BytesParser(policy=policy.default).parsebytes(
        ("Content-Type: " + content_type + "\r\nMIME-Version: 1.0\r\n\r\n").encode() + raw)
    if not message.is_multipart() or message.defects:
        raise HTTPException(400, "Invalid multipart image request")
    fields, images = {}, []
    for part in message.iter_parts():
        name = part.get_param("name", header="content-disposition")
        data = part.get_payload(decode=True)
        if not isinstance(data, bytes):
            raise HTTPException(400, "Invalid multipart field")
        if name in {"image", "image[]"}:
            mime = part.get_content_type()
            if mime not in {"image/png", "image/jpeg", "image/webp"} or not data or len(data) > 10 * 1024 * 1024:
                raise HTTPException(400, "Each reference must be PNG, JPEG or WebP, up to 10 MiB")
            url = "data:" + mime + ";base64," + base64.b64encode(data).decode()
            width, height = image_dimensions(url)
            if not (0 < width <= 16384 and 0 < height <= 16384):
                raise HTTPException(400, "Invalid reference image dimensions")
            images.append(url)
            if len(images) > 6:
                raise HTTPException(400, "Qwen Image 2.1 supports at most 6 reference images")
        elif name in {"model", "prompt", "size", "n", "response_format", "gpu"}:
            if name in fields or len(data) > 128000:
                raise HTTPException(400, "Duplicate or oversized form field")
            try:
                fields[name] = data.decode("utf-8")
            except UnicodeDecodeError:
                raise HTTPException(400, "Form text must be UTF-8")
        else:
            raise HTTPException(400, "Unsupported image edit field: " + str(name)[:60])
    if not images:
        raise HTTPException(400, "At least one reference image is required")
    return qwen_image_request(fields, images)


def create_app(config):
    app = FastAPI(title="Huobao private model bridge", docs_url=None, redoc_url=None)
    state_dir = Path(config.get("state_dir", "./state"))
    state_dir.mkdir(parents=True, exist_ok=True)
    db_path = state_dir / "jobs.sqlite3"
    with closing(sqlite3.connect(db_path)) as db, db:
        db.execute("CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, duration REAL NOT NULL, created REAL NOT NULL)")
        db.execute("CREATE TABLE IF NOT EXISTS image_jobs (id TEXT PRIMARY KEY, created REAL NOT NULL)")

    def auth(authorization: str | None = Header(default=None)):
        expected = config.get("token", "")
        if not expected or not secrets.compare_digest(authorization or "", "Bearer " + expected):
            raise HTTPException(401, "Invalid bridge token")

    @app.exception_handler(HTTPException)
    async def http_error(request, exc):
        return JSONResponse({"error": {"message": str(exc.detail), "type": "bridge_error", "code": exc.status_code}}, status_code=exc.status_code)

    @app.exception_handler(httpx.HTTPError)
    async def upstream_error(request, exc):
        return JSONResponse({"error": {"message": "Upstream connection failed: " + type(exc).__name__, "type": "upstream_error"}}, status_code=502)

    def upstream_headers():
        return {"Authorization": "Bearer " + config["h3"]["token"]}

    async def upstream(method, path, body=None):
        if not config.get("h3"):
            raise HTTPException(404, "H3 is not configured on this bridge")
        async with httpx.AsyncClient(timeout=httpx.Timeout(180, connect=10), trust_env=False) as client:
            r = await client.request(method, config["h3"]["url"] + path, headers=upstream_headers(), json=body)
        if r.is_error:
            try:
                detail = r.json().get("detail", "Model service request failed")
            except ValueError:
                detail = "Model service request failed"
            raise HTTPException(r.status_code, str(detail)[:600])
        return r.json()

    @app.get("/health")
    async def health():
        return {"status": "ok", "service": "huobao-model-bridge"}

    @app.get("/v1/models", dependencies=[Depends(auth)])
    async def models():
        names = list(config.get("dify", {}))
        if config.get("h3"):
            # Verify the real service, instead of making a local catalogue look healthy.
            registry = await upstream("GET", "/v1/models")
            available = {m["id"] for m in registry.get("data", [])}
            names.extend(name for name, spec in VIDEO_MODELS.items() if video_workflows(spec) <= available)
            if any(m.get("id") == "qwen_image_2_1" and m.get("model_type") == "image" for m in registry.get("data", [])):
                names.append("qwen_image_2_1")
        if config.get("grok"):
            names.append("grok-local")
        return {"object": "list", "data": [{"id": n, "object": "model", "owned_by": "private"} for n in names]}

    async def dify_events(model, query):
        upstream_config = config["dify"][model]
        payload = {"inputs": {}, "query": query, "response_mode": "streaming",
                   "user": "huobao-" + secrets.token_hex(8), "auto_generate_name": False}
        finished, usage = False, {}
        async with httpx.AsyncClient(timeout=httpx.Timeout(240, connect=10), trust_env=False) as client:
            async with client.stream("POST", upstream_config["url"] + "/chat-messages",
                                     headers={"Authorization": "Bearer " + upstream_config["key"]}, json=payload) as response:
                if response.is_error:
                    await response.aread()
                    try:
                        message = response.json().get("message", "Dify request failed")
                    except ValueError:
                        message = "Dify request failed"
                    raise HTTPException(response.status_code, str(message)[:400])
                async for line in response.aiter_lines():
                    if not line.startswith("data:"):
                        continue
                    event = json.loads(line[5:].strip())
                    kind = event.get("event")
                    if kind in {"message", "agent_message"}:
                        yield "text", event.get("answer", "")
                    elif kind == "error":
                        raise HTTPException(502, str(event.get("message", "Dify generation failed"))[:400])
                    elif kind == "message_end":
                        finished = True
                        usage = event.get("metadata", {}).get("usage", {})
                    elif kind == "workflow_finished":
                        data = event.get("data") or {}
                        if data.get("status") not in {None, "succeeded"}:
                            raise HTTPException(502, str(data.get("error") or "Dify workflow failed")[:400])
                if not finished:
                    raise HTTPException(502, "Dify stream ended before message_end")
        yield "usage", usage

    async def grok_events(query):
        item = config["grok"]
        args = [item["exe"], "--cwd", item["cwd"], "--output-format", "json", "--verbatim",
                "--no-subagents", "--no-memory", "--disable-web-search", "--max-turns", "1",
                "--tools", "", "--deny", "MCPTool", "--permission-mode", "dontAsk",
                "--system-prompt-override", "Answer the supplied writing request using text only. Do not use tools."]
        options = {"creationflags": 0x08000000} if os.name == "nt" else {}
        # A file avoids the Windows command-line length limit for full scripts.
        with tempfile.TemporaryDirectory(prefix="huobao-grok-") as scratch:
            prompt_file = Path(scratch) / "prompt.txt"
            prompt_file.write_text(query, encoding="utf-8")
            args += ["--prompt-file", str(prompt_file)]
            process = await asyncio.create_subprocess_exec(*args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, **options)
            try:
                out, err = await asyncio.wait_for(process.communicate(), timeout=240)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                process.kill()
                await process.wait()
                raise
        try:
            data = json.loads(out.decode("utf-8"))
        except (ValueError, UnicodeDecodeError):
            raise HTTPException(502, "Grok CLI did not return JSON; run Grok login on this computer")
        if process.returncode or not data.get("text"):
            raise HTTPException(502, "Grok CLI failed; check local login and available quota")
        yield "text", data["text"]
        yield "usage", data.get("usage", {})

    @app.post("/v1/chat/completions", dependencies=[Depends(auth)])
    async def chat(body: dict):
        model = body.get("model", "")
        if model not in config.get("dify", {}) and not (model == "grok-local" and config.get("grok")):
            raise HTTPException(404, "Unknown text model")
        query = chat_query(body)
        events = grok_events(query) if model == "grok-local" else dify_events(model, query)
        cid, created = "chatcmpl-" + secrets.token_hex(12), int(time.time())

        def chunk(delta, finish=None):
            return {"id": cid, "object": "chat.completion.chunk", "created": created, "model": model,
                    "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]}

        if body.get("stream"):
            async def stream():
                def encode(value):
                    return "data: " + json.dumps(value, ensure_ascii=False) + "\n\n"
                yield encode(chunk({"role": "assistant", "content": ""}))
                try:
                    async for kind, value in events:
                        if kind == "text":
                            yield encode(chunk({"content": value}))
                    yield encode(chunk({}, "stop"))
                except (HTTPException, httpx.HTTPError) as error:
                    message = str(error.detail) if isinstance(error, HTTPException) else "Upstream connection failed"
                    yield encode({"error": {"message": message, "type": "upstream_error"}})
                yield "data: [DONE]\n\n"
            return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"})
        text, usage = "", {}
        async for kind, value in events:
            if kind == "text":
                text += value
            else:
                usage = value
        return {"id": cid, "object": "chat.completion", "created": created, "model": model,
                "choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}],
                "usage": {"prompt_tokens": usage.get("prompt_tokens", usage.get("input_tokens", 0)),
                          "completion_tokens": usage.get("completion_tokens", usage.get("output_tokens", 0)),
                          "total_tokens": usage.get("total_tokens", 0)}}

    async def require_qwen_image():
        registry = await upstream("GET", "/v1/models")
        if not any(m.get("id") == "qwen_image_2_1" and m.get("model_type") == "image" for m in registry.get("data", [])):
            raise HTTPException(503, "Qwen Image 2.1 workflow is unavailable")

    @app.get("/v1/images/pool", dependencies=[Depends(auth)])
    async def image_pool():
        await require_qwen_image()
        health, catalogue = await asyncio.gather(upstream("GET", "/health"), upstream("GET", "/v1/providers"))
        return {"model": "qwen_image_2_1", "auto_balance": health.get("autoBalance") is True,
                "routing": "shortest-queue-with-reservations" if health.get("autoBalance") is True else "single-provider",
                "configured_workers": [{"id": p["id"], "name": p.get("name", p["id"])} for p in catalogue.get("providers", [])]}

    async def submit_image(payload):
        await require_qwen_image()
        result = await upstream("POST", "/v1/jobs", payload)
        jid = result.get("id")
        if not isinstance(jid, str) or not jid:
            raise HTTPException(502, "Image service did not return a task ID")
        with closing(sqlite3.connect(db_path)) as db, db:
            db.execute("INSERT INTO image_jobs VALUES(?,?)", (jid, time.time()))
        return {"task_id": jid, "status": "processing", "provider_id": result.get("providerId")}

    @app.post("/v1/images/validate", dependencies=[Depends(auth)])
    async def validate_image(body: dict):
        payload = qwen_image_request(body)
        await require_qwen_image()
        return {"valid": True, "workflow": payload["workflow_key"], "requested_size": [payload["width"], payload["height"]], "max_reference_images": 6,
                "pool": await image_pool()}

    @app.post("/v1/images/generations", dependencies=[Depends(auth)])
    async def generate_image(body: dict):
        return await submit_image(qwen_image_request(body))

    @app.post("/v1/images/edits", dependencies=[Depends(auth)])
    async def edit_image(request: Request):
        return await submit_image(await parse_image_edit(request))

    @app.post("/v1/images/task/{jid}/cancel", dependencies=[Depends(auth)])
    async def cancel_image(jid: str):
        with closing(sqlite3.connect(db_path)) as db:
            found = db.execute("SELECT 1 FROM image_jobs WHERE id=?", (jid,)).fetchone()
        if not found:
            raise HTTPException(404, "Unknown image task")
        result = await upstream("POST", "/v1/jobs/" + quote(jid, safe="") + "/cancel", {})
        return {"task": {"id": jid, "status": result.get("status", "running"),
                         "cancelRequested": result.get("cancelRequested", False)}}

    @app.get("/v1/images/task/{jid}", dependencies=[Depends(auth)])
    async def poll_image(jid: str):
        with closing(sqlite3.connect(db_path)) as db:
            found = db.execute("SELECT 1 FROM image_jobs WHERE id=?", (jid,)).fetchone()
        if not found:
            raise HTTPException(404, "Unknown image task")
        result = await upstream("GET", "/v1/huobao/jobs/" + quote(jid, safe=""))
        state = result.get("status", "running")
        if state in {"failed", "cancelled"}:
            return {"task_id": jid, "status": "failed", "cancelled": state == "cancelled", "error": {"message": str(result.get("error") or "Image generation failed")[:600]}}
        if state != "succeeded":
            return {"task_id": jid, "status": "processing", "provider_id": result.get("providerId"), "worker_state": result.get("worker_state"), "queue_position": result.get("queue_position")}
        output = next((o for o in result.get("outputs", []) if o.get("kind") == "image"), None)
        if output is None:
            return {"task_id": jid, "status": "failed", "error": {"message": "Task completed without an image output"}}
        index, expiry = int(output["index"]), int(time.time()) + 86400
        extension = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp"}.get(output.get("mimeType"), "png")
        # Huobao determines the saved extension from the URL, not Content-Type.
        url = f"{config['public_url']}/media/{quote(jid, safe='')}/{index}/image.{extension}?expires={expiry}&sig={signature(jid, index, expiry)}"
        return {"task_id": jid, "status": "completed", "image_url": url, "data": [{"url": url}], "provider_id": result.get("providerId")}

    @app.post("/v2/video_generation/validate", dependencies=[Depends(auth)])
    async def validate_video(body: dict):
        request = h3_request(body)
        registry = await upstream("GET", "/v1/models")
        if request["workflow_key"] not in {m["id"] for m in registry.get("data", [])}:
            raise HTTPException(503, "Selected video workflow is unavailable")
        plan = await upstream("POST", "/v1/audio-plan", request)
        route = await upstream("POST", "/v1/huobao/route", request)
        return {"valid": plan.get("status", "ready") == "ready", "workflow": request["workflow_key"],
                "width": request["width"], "height": request["height"], "duration": request["duration"], "audio_plan": plan,
                "routing": worker_routing(route.get("providerId"))}

    @app.post("/v2/video_generation", dependencies=[Depends(auth)])
    async def video(body: dict):
        # Huobao probes video providers with an empty body. Check the real
        # upstream without submitting a GPU task, then report readiness.
        if not body:
            registry = await upstream("GET", "/v1/models")
            available = {m["id"] for m in registry.get("data", [])}
            if not {"minimax_h3_t2v_official", "minimax_h3_i2v", "minimax_h3_r2v"} <= available:
                raise HTTPException(503, "Required H3 workflows are unavailable")
            return {"status": "ready", "probe": True, "model": "MiniMax-H3"}
        request = h3_request(body)
        registry = await upstream("GET", "/v1/models")
        if request["workflow_key"] not in {m["id"] for m in registry.get("data", [])}:
            raise HTTPException(503, "Selected video workflow is unavailable")
        result = await upstream("POST", "/v1/jobs", request)
        jid = result.get("id")
        if not isinstance(jid, str) or not jid:
            raise HTTPException(502, "Video service did not return a job ID")
        with closing(sqlite3.connect(db_path)) as db, db:
            db.execute("INSERT INTO jobs VALUES(?,?,?)", (jid, request["duration"], time.time()))
        return {"task_id": jid, "status": "submitted", "provider_id": result.get("providerId"), "routing": worker_routing(result.get("providerId"))}

    def get_job(jid):
        with closing(sqlite3.connect(db_path)) as db:
            row = db.execute("SELECT duration FROM jobs WHERE id=?", (jid,)).fetchone()
        if not row:
            raise HTTPException(404, "Unknown bridge job")
        return row[0]

    def signature(jid, index, expiry):
        data = f"{jid}:{index}:{expiry}".encode()
        return hmac.new(config["token"].encode(), data, hashlib.sha256).hexdigest()

    @app.post("/v2/video_generation/{jid}/cancel", dependencies=[Depends(auth)])
    async def cancel_video(jid: str):
        # Only a video submitted through this authenticated bridge may be cancelled.
        get_job(jid)
        result = await upstream("POST", "/v1/jobs/" + quote(jid, safe="") + "/cancel", {})
        return {"task": {"id": jid, "status": result.get("status", "running"),
                         "cancelRequested": result.get("cancelRequested", False)}}

    @app.get("/v2/query/video_generation/{jid}", dependencies=[Depends(auth)])
    async def poll(jid: str):
        duration = get_job(jid)
        result = await upstream("GET", "/v1/huobao/jobs/" + quote(jid, safe=""))
        state = result.get("status", "running")
        task = {"id": jid, "status": state, "duration": duration, "provider_id": result.get("providerId"), "routing": worker_routing(result.get("providerId")), "worker_state": result.get("worker_state"), "queue_position": result.get("queue_position")}
        task["progress"] = result.get("progress")
        if state == "succeeded":
            output = next((o for o in result.get("outputs", []) if o.get("kind") == "video"), None)
            if not output:
                task.update(status="failed", error="Video task completed without a video output")
            else:
                index, expiry = int(output["index"]), int(time.time()) + 86400
                url = f"{config['public_url']}/media/{quote(jid, safe='')}/{index}/video.mp4?expires={expiry}&sig={signature(jid, index, expiry)}"
                task["content"] = {"url": url}
        elif state in {"failed", "cancelled"}:
            task["error"] = str(result.get("error") or "Video task failed")[:600]
        return {"task": task}

    @app.get("/media/{jid}/{index}")
    @app.get("/media/{jid}/{index}/{filename}")
    async def media(jid: str, index: int, expires: int, sig: str, filename: str | None = None):
        if filename is not None and not re.fullmatch(r"image\.(png|jpg|webp)|video\.mp4", filename):
            raise HTTPException(404, "Unknown media filename")
        with closing(sqlite3.connect(db_path)) as db:
            found = db.execute("SELECT 1 FROM jobs WHERE id=? UNION ALL SELECT 1 FROM image_jobs WHERE id=?", (jid, jid)).fetchone()
        if not found:
            raise HTTPException(404, "Unknown bridge task")
        if expires < time.time() or not hmac.compare_digest(sig, signature(jid, index, expires)):
            raise HTTPException(403, "Invalid or expired media link; poll the task again")
        client = httpx.AsyncClient(timeout=httpx.Timeout(180, connect=10), trust_env=False)
        url = config["h3"]["url"] + f"/v1/jobs/{quote(jid, safe='')}/outputs/{index}"
        try:
            response = await client.send(client.build_request("GET", url, headers=upstream_headers()), stream=True)
            if response.is_error:
                await response.aclose()
                await client.aclose()
                raise HTTPException(response.status_code, "Model output is not available")
        except BaseException:
            await client.aclose()
            raise
        async def download():
            try:
                async for part in response.aiter_bytes():
                    yield part
            finally:
                await response.aclose()
                await client.aclose()
        return StreamingResponse(download(), media_type=response.headers.get("content-type", "video/mp4"))

    return app


def create_from_file():
    path = Path(os.environ.get("HUOBAO_BRIDGE_CONFIG", "/bridge/config.json"))
    return create_app(json.loads(path.read_text(encoding="utf-8")))