"""Aliyun Bailian / DashScope **generation** client — images, video, speech. Sibling of :mod:`app.domains.bailian.llm_client`, which owns the *text* surfaces (native text-generation, compatible-mode chat, embeddings). This module owns the three generation families that live on entirely different DashScope paths and answer in entirely different envelope shapes: * **multimodal-generation** (synchronous) — ``/api/v1/services/aigc/multimodal-generation/generation``. Text-to-image (``qwen-image-3.0-pro``), image editing (``qwen-image-edit-max``), text-to-speech (``qwen3-tts-instruct-flash``) and speech-to-text (``qwen3-asr-flash``) all post here and answer in one round trip. Measured 2026-09-03 against the live Singapore endpoint: a 1024x1024 image came back in **9.0s**. * **async task submit** — ``/api/v1/services/aigc/video-generation/video-synthesis`` and ``.../text2image/image-synthesis``, both with the ``X-DashScope-Async: enable`` header. These answer immediately with a ``task_id`` and nothing else. * **task poll** — ``/api/v1/tasks/{task_id}``. Measured: a 5-second 1080p ``wan3.0-video-prime`` clip went PENDING → RUNNING → SUCCEEDED in **51s**, and the resulting MP4 was **24 MB**. Every result URL DashScope hands back is a **short-lived vendor OSS link** (measured: ``Expires`` exactly 24h out). They are therefore useless as durable references — :meth:`fetch_result` exists so the caller can pull the bytes and re-home them in Helix's own bucket before answering an employee. Nothing in Helix should ever persist or render a vendor URL directly. Region ------ Same split as the LLM client, and the same hard constraint: **an sk- key is bound to the region that issued it.** Verified 2026-09-03 — the production key answers on ``dashscope-intl`` and is rejected as ``invalid_api_key`` at the mainland host. ``region`` is resolved through the shared helpers in :mod:`app.domains.bailian.llm_client` so there is exactly one place that knows the two hostnames. Free-tier gate (the one non-obvious failure mode) ------------------------------------------------- DashScope answers ``403 AllocationQuota.FreeTierOnly`` when a model's per-model "免费额度用完即停" switch is ON in the Bailian console and that model has no free quota left. The message text talks about *payment information*, which points every reader at the account's billing setup — the real cause is a console toggle, and pay-as-you-go was working the whole time. :func:`_raise_for_payload` therefore maps that one vendor code to its own :class:`AppException` code with the actual remedy in the message, so a future outage is diagnosed from the log line instead of from the billing console. (All 123 affected models were switched off on 2026-09-03; a model added to the catalogue later can reintroduce it.) """ from __future__ import annotations import logging from typing import Any import httpx from app.core.exceptions import AppException from app.domains.bailian.llm_client import ( DEFAULT_REGION, REGION_CN, REGION_INTL, ) logger = logging.getLogger(__name__) #: API roots. The generation families all hang off ``/api/v1`` rather than #: the ``/compatible-mode/v1`` prefix the chat surface uses. _NATIVE_ROOTS = { REGION_CN: "https://dashscope.aliyuncs.com/api/v1", REGION_INTL: "https://dashscope-intl.aliyuncs.com/api/v1", } #: Synchronous multimodal generation — images, image edits, TTS, ASR. PATH_MULTIMODAL = "services/aigc/multimodal-generation/generation" #: Asynchronous text-to-image (the ``wan*-image`` family). PATH_TEXT2IMAGE = "services/aigc/text2image/image-synthesis" #: Asynchronous video synthesis (the ``wan*-video`` / ``wan*-t2v`` family). PATH_VIDEO = "services/aigc/video-generation/video-synthesis" #: Default per-request timeout. Generous because the *synchronous* image #: call genuinely takes ~9s and a busy endpoint takes longer; still far #: below nginx's 300s ``proxy_read_timeout`` so an abort is always ours. DEFAULT_TIMEOUT_SECONDS = 120.0 #: Ceiling on one downloaded artifact. A 5s/1080p clip measured 24 MB, so #: this leaves room for the longest clip the catalogue offers while still #: refusing something pathological before it reaches the event loop's #: memory. ``fetch_result`` streams and aborts mid-transfer once exceeded. MAX_RESULT_BYTES = 192 * 1024 * 1024 #: Redirect hops allowed when downloading an artifact. The vendor's OSS #: links normally serve directly; a couple of hops covers a CDN in front #: of them without turning this into an open fetcher. MAX_RESULT_REDIRECTS = 3 #: Statuses treated as a redirect when following one by hand. _REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) #: The only host suffix an artifact may be downloaded from. #: #: Measured 2026-09-03 against the live key rather than assumed: an image #: came back on ``dashscope-463f.oss-accelerate.aliyuncs.com`` and a speech #: clip on ``dashscope-463f.oss-ap-southeast-1.aliyuncs.com``. The bucket #: id and the OSS region both vary, so the suffix is what can be pinned. #: #: An allowlist rather than a private-IP denylist on purpose. A denylist #: has to enumerate loopback, link-local, every RFC 1918 range, IPv6 #: equivalents, and DNS names that resolve into them — and it still lets a #: redirect reach any public host on the internet. This is a URL the vendor #: chose, fetched by a server inside the VPC; "somewhere at the vendor" is #: the whole legitimate range. _ARTIFACT_HOST_SUFFIX = ".aliyuncs.com" def _assert_downloadable(url: str, *, redirected: bool = False) -> None: """Refuse any artifact URL that is not https on a vendor host. Applied to the URL from the envelope **and** to every redirect target, because a check that runs once is a check the second hop bypasses. """ where = "redirected the download to" if redirected else "returned" try: parsed = httpx.URL(url) scheme = parsed.scheme host = (parsed.host or "").lower() except (httpx.InvalidURL, ValueError) as exc: # A malformed host raises before any check can look at it, and the # type is not always httpx's: a bad punycode label comes out of the # ``idna`` package as ``InvalidCodepoint`` (a ``UnicodeError``, so a # ``ValueError``), which nothing downstream was catching — it # escaped as an unhandled exception on the first-URL path. raise AppException( code="BAILIAN_MEDIA_BAD_RESULT_URL", message=f"The vendor {where} an address that is not a valid URL.", status_code=502, ) from exc if scheme != "https": raise AppException( code="BAILIAN_MEDIA_BAD_RESULT_URL", message=f"The vendor {where} a non-HTTPS location.", status_code=502, ) # A single trailing dot is the fully-qualified form of the same name and # resolves identically, so refusing it would reject a legitimate host. # Stripped rather than allowed as a suffix, so ``evil.test.`` cannot # match either. host = host.rstrip(".") if not (host == _ARTIFACT_HOST_SUFFIX[1:] or host.endswith(_ARTIFACT_HOST_SUFFIX)): # The host is deliberately not echoed: on the redirect path it is # attacker-chosen text that would otherwise land in a persisted # ``error_message`` the employee reads. logger.warning( "bailian_media: refusing artifact download from host %r " "(redirected=%s)", host, redirected, ) raise AppException( code="BAILIAN_MEDIA_BAD_RESULT_URL", message=( f"The vendor {where} a location outside its own storage, " "so it was not downloaded." ), status_code=502, ) #: Vendor task states, normalised. DashScope's ``UNKNOWN`` means "this task #: id is not one of ours (any more)" — terminal, not retryable. TASK_RUNNING = "running" TASK_SUCCEEDED = "succeeded" TASK_FAILED = "failed" _VENDOR_TASK_STATES = { "PENDING": TASK_RUNNING, "RUNNING": TASK_RUNNING, "SUSPENDED": TASK_RUNNING, "SUCCEEDED": TASK_SUCCEEDED, "FAILED": TASK_FAILED, "CANCELED": TASK_FAILED, "CANCELLED": TASK_FAILED, "UNKNOWN": TASK_FAILED, } #: The vendor code that means "a console switch is blocking this model", #: not "this account cannot pay". See the module docstring. _FREE_TIER_CODE = "AllocationQuota.FreeTierOnly" _FREE_TIER_REMEDY = ( "DashScope refused this model with AllocationQuota.FreeTierOnly. This is " "NOT a billing-setup problem: it means the model's 「免费额度用完即停」 " "switch is ON in the Bailian console (用量 & 费用 → 免费额度) while the " "model has no free quota left, so every call is refused and none ever " "falls through to pay-as-you-go. Switch it off for that model." ) def native_root_for_region(region: str | None) -> str: """Resolve a region code → ``/api/v1`` root for the generation paths.""" code = (region or "").strip().lower() or DEFAULT_REGION root = _NATIVE_ROOTS.get(code) if root is None: logger.warning( "BailianMediaClient: unknown region %r — defaulting to %s", region, DEFAULT_REGION, ) return _NATIVE_ROOTS[DEFAULT_REGION] return root def _error_from_payload(payload: dict[str, Any]) -> tuple[str, str] | None: """``(code, message)`` if this envelope carries a vendor error. DashScope reports failures in three places depending on the family: top-level ``code``/``message`` (sync + submit), the same pair nested under ``output`` (a task that failed after being accepted), and — for batch image results — per-item ``code``/``message`` inside ``output.results``. All three are checked; the first wins. """ code = payload.get("code") if code: return str(code), str(payload.get("message") or "") output = payload.get("output") or {} if isinstance(output, dict): code = output.get("code") if code: return str(code), str(output.get("message") or "") results = output.get("results") if isinstance(results, list): for item in results: if isinstance(item, dict) and item.get("code"): return str(item["code"]), str(item.get("message") or "") return None def _raise_for_payload(payload: dict[str, Any], *, model: str) -> None: """Translate a vendor error envelope into an ``AppException``. A 200 response can still be a failure — DashScope soft-fails with ``code`` set and HTTP 200 — so this runs on every payload, not only on non-2xx ones. """ found = _error_from_payload(payload) if found is None: return code, message = found if code == _FREE_TIER_CODE: raise AppException( code="BAILIAN_MEDIA_FREE_TIER_GATE", message=f"{_FREE_TIER_REMEDY} (model={model})", status_code=502, ) # The vendor's prose is NOT relayed. It reaches the employee verbatim # through ``generations.error_message``, and vendor rejections routinely # quote the offending input back — which would put a fragment of the # employee's own prompt into a field, and into whatever renders it, on # the one surface that promises nobody else reads it. It is also English # developer text in a trilingual UI. The vendor *code* is short, stable # and safe to show; the prose goes to the log, where an operator can # still find it by trace. logger.warning( "bailian_media: vendor rejected model=%s code=%s: %s", model, code, message[:400], ) raise AppException( code="BAILIAN_MEDIA_VENDOR_ERROR", message=( f"The model provider rejected this request (code {code[:64]})." ), status_code=502, ) class BailianMediaClient: """Stateless generation caller; instantiate per request. Construct with the sk-xxx DashScope key (the same one the LLM client uses — Helix keeps a single AI/LLM connector). ``region`` must match the region that issued the key. """ def __init__( self, api_key: str, *, region: str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, ) -> None: api_key = (api_key or "").strip() if not api_key: raise AppException( code="BAILIAN_MEDIA_NO_KEY", message=( "No Bailian (DashScope) API key configured. Open Admin → " "Connectors → AI / LLM and paste an sk-xxx key issued " "from the Bailian console (业务空间 → API-KEY 管理)." ), status_code=503, ) self._api_key = api_key self._region = (region or DEFAULT_REGION).strip().lower() self._root = native_root_for_region(self._region) self._timeout = timeout @property def region(self) -> str: return self._region # ── plumbing ─────────────────────────────────────────────── def _headers(self, *, async_task: bool = False) -> dict[str, str]: headers = { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } if async_task: # Without this header the video/text2image paths answer # synchronously — and then time out, because they take minutes. headers["X-DashScope-Async"] = "enable" return headers async def _post( self, path: str, body: dict[str, Any], *, model: str, async_task: bool = False, ) -> dict[str, Any]: url = f"{self._root}/{path}" try: async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, follow_redirects=False, ) as client: resp = await client.post( url, headers=self._headers(async_task=async_task), json=body, ) except httpx.HTTPError as exc: raise AppException( code="BAILIAN_MEDIA_NETWORK", message=f"DashScope network error calling {path}: {exc}", status_code=502, ) from exc payload = self._payload(resp, path=path) _raise_for_payload(payload, model=model) if resp.status_code != 200: logger.warning( "bailian_media: HTTP %s calling %s: %s", resp.status_code, path, resp.text[:400], ) raise AppException( code="BAILIAN_MEDIA_HTTP_ERROR", # Body deliberately not relayed — same reasoning as # ``_raise_for_payload``; it is logged immediately above. message=( "The model provider returned an unexpected response " f"(HTTP {resp.status_code})." ), status_code=502, ) return payload @staticmethod def _payload(resp: httpx.Response, *, path: str) -> dict[str, Any]: """Parse a response body, tolerating a non-JSON error page. A gateway 502 from in front of DashScope answers HTML; letting ``json()`` raise there would report a JSON-decode error and hide the status code that actually explains the failure. """ try: data = resp.json() except ValueError: if resp.status_code == 200: raise AppException( code="BAILIAN_MEDIA_BAD_RESPONSE", message=( f"DashScope returned non-JSON 200 for {path}: " f"{resp.text[:200]}" ), status_code=502, ) from None return {} return data if isinstance(data, dict) else {} # ── synchronous multimodal generation ────────────────────── async def multimodal_generate( self, *, model: str, input_: dict[str, Any], parameters: dict[str, Any] | None = None, ) -> dict[str, Any]: """One synchronous ``multimodal-generation`` round trip. Returns the raw envelope. Use :meth:`extract_media_urls` / :meth:`extract_audio_url` / :meth:`extract_text` to read it — the shape differs per family and those helpers hold the differences. """ body: dict[str, Any] = {"model": model, "input": input_} if parameters: body["parameters"] = parameters return await self._post(PATH_MULTIMODAL, body, model=model) # ── asynchronous task family ─────────────────────────────── async def submit_task( self, *, path: str, model: str, input_: dict[str, Any], parameters: dict[str, Any] | None = None, ) -> str: """Submit an async generation task; return its vendor ``task_id``. ``path`` is :data:`PATH_VIDEO` or :data:`PATH_TEXT2IMAGE`. """ body: dict[str, Any] = {"model": model, "input": input_} if parameters: body["parameters"] = parameters payload = await self._post(path, body, model=model, async_task=True) task_id = ((payload.get("output") or {}).get("task_id") or "").strip() if not task_id: raise AppException( code="BAILIAN_MEDIA_NO_TASK_ID", message=( f"DashScope accepted the {model} task but returned no " "task_id, so its result can never be collected." ), status_code=502, ) return task_id async def get_task(self, task_id: str, *, model: str = "") -> dict[str, Any]: """Poll one async task. Returns a **normalised** verdict. ``{"status": running|succeeded|failed, "media_urls": [...], "error_code": str|None, "error_message": str|None, "usage": {...}}`` A vendor error inside a terminal task is reported through the return value, NOT raised: a task that failed on the vendor's side is a finished job with a reason, and every caller needs to persist that reason rather than treat it as a transport fault. Transport faults (network, HTTP, unparseable body) still raise. """ url = f"{self._root}/tasks/{task_id}" try: async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, follow_redirects=False, ) as client: resp = await client.get(url, headers=self._headers()) except httpx.HTTPError as exc: raise AppException( code="BAILIAN_MEDIA_NETWORK", message=f"DashScope network error polling task {task_id}: {exc}", status_code=502, ) from exc payload = self._payload(resp, path=f"tasks/{task_id}") if resp.status_code != 200: logger.warning( "bailian_media: HTTP %s polling task %s: %s", resp.status_code, task_id, resp.text[:300], ) # A 5xx or a 429 from the *polling* endpoint says nothing about # the task — the vendor is still rendering work already paid # for, and the task id is on our row, so another poll costs # nothing. That earns a retryable code. A 4xx is a rejection of # the poll itself (a task id the key cannot see) and is not # worth repeating. retryable = resp.status_code >= 500 or resp.status_code == 429 raise AppException( code=( "BAILIAN_MEDIA_RESULT_UNAVAILABLE" if retryable else "BAILIAN_MEDIA_HTTP_ERROR" ), # Vendor prose withheld — see ``_raise_for_payload``. message=( "The model provider could not report on this job " f"(HTTP {resp.status_code})." ), status_code=502, ) output = payload.get("output") or {} raw_state = str(output.get("task_status") or "").strip().upper() status = _VENDOR_TASK_STATES.get(raw_state) if status is None: # An unrecognised state must not be read as "still working" — # that is how a job pins a poller open forever. Treat it as # terminal and name the state, so the row records the truth. logger.warning( "DashScope task %s reported unknown state %r", task_id, raw_state, ) return { "status": TASK_FAILED, "media_urls": [], "error_code": "UNRECOGNISED_TASK_STATE", "error_message": f"DashScope reported task_status={raw_state!r}", "usage": output.get("usage") or payload.get("usage") or {}, } error = _error_from_payload(payload) error_code, error_message = error if error else (None, None) if error_code == _FREE_TIER_CODE: error_message = f"{_FREE_TIER_REMEDY} (model={model or 'unknown'})" media_urls = self.extract_media_urls(payload) if status == TASK_SUCCEEDED else [] if status == TASK_SUCCEEDED and not media_urls: status = TASK_FAILED error_code = error_code or "EMPTY_RESULT" error_message = error_message or ( "DashScope reported SUCCEEDED but the envelope carried no " "media URL, so there is nothing to store." ) return { "status": status, "media_urls": media_urls, "error_code": error_code, "error_message": error_message, "usage": output.get("usage") or payload.get("usage") or {}, } # ── envelope readers ─────────────────────────────────────── @staticmethod def extract_media_urls(payload: dict[str, Any]) -> list[str]: """Every image/video URL in a generation envelope, in order. Covers the four shapes DashScope actually uses, measured: * ``output.video_url`` — video synthesis. * ``output.results[].url`` — async text-to-image. * ``output.choices[].message.content[].image`` — synchronous multimodal image generation and image editing. * ``output.results[].video_url`` — the image-to-video variants. """ output = payload.get("output") if not isinstance(output, dict): return [] urls: list[str] = [] def _add(value: Any) -> None: if isinstance(value, str) and value.strip(): urls.append(value.strip()) _add(output.get("video_url")) results = output.get("results") if isinstance(results, list): for item in results: if isinstance(item, dict): _add(item.get("url")) _add(item.get("video_url")) _add(item.get("image")) choices = output.get("choices") if isinstance(choices, list): for choice in choices: if not isinstance(choice, dict): continue content = (choice.get("message") or {}).get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict): _add(part.get("image")) _add(part.get("video")) # Order-preserving de-duplication: a couple of families echo the # same URL in two places and an employee should not be handed the # same picture twice. seen: set[str] = set() unique: list[str] = [] for url in urls: if url not in seen: seen.add(url) unique.append(url) return unique @staticmethod def extract_audio_url(payload: dict[str, Any]) -> str | None: """The synthesised-speech URL from a TTS envelope.""" output = payload.get("output") if not isinstance(output, dict): return None audio = output.get("audio") if isinstance(audio, dict): url = audio.get("url") if isinstance(url, str) and url.strip(): return url.strip() return None @staticmethod def extract_text(payload: dict[str, Any]) -> str: """Concatenated text from a multimodal envelope (ASR transcripts). ASR answers in the chat-completion shape, where ``content`` is a list of parts and the transcript sits on a ``text`` key. """ output = payload.get("output") if not isinstance(output, dict): return "" pieces: list[str] = [] for choice in output.get("choices") or []: if not isinstance(choice, dict): continue content = (choice.get("message") or {}).get("content") if isinstance(content, str): pieces.append(content) elif isinstance(content, list): for part in content: if isinstance(part, dict): text = part.get("text") if isinstance(text, str): pieces.append(text) elif isinstance(part, str): pieces.append(part) if not pieces: text = output.get("text") if isinstance(text, str): pieces.append(text) return "".join(pieces).strip() # ── result download ──────────────────────────────────────── async def fetch_result(self, url: str) -> tuple[bytes, str]: """Download one vendor artifact. Returns ``(bytes, content_type)``. Streamed with a hard :data:`MAX_RESULT_BYTES` ceiling enforced as the body arrives, so an unexpectedly huge artifact is abandoned mid-flight instead of after it has already been buffered. The vendor's OSS links carry their own signature and expire ~24h after the job finished, so this must run *before* the row is handed to an employee — never lazily at render time. """ _assert_downloadable(url) try: # Redirects are followed **manually**, on ONE client so a hop # that sets a cookie is still authenticated at the next one. # # ``follow_redirects=True`` would apply the checks above to the # first hop only: a 302 to ``http://``, to ``169.254.169.254``, # or to anything on the VPC is then fetched by a server-side # client with no further check and up to 192 MB of the response # is stored as the employee's artifact and shown to them. That # is the classic SSRF shape, and an earlier version of this # comment claimed a private-address check the code did not # have. Every hop now goes through ``_assert_downloadable``. async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, follow_redirects=False, ) as client: for _hop in range(MAX_RESULT_REDIRECTS + 1): async with client.stream("GET", url) as resp: if resp.status_code not in _REDIRECT_STATUSES: return await self._read_result_body(resp) location = (resp.headers.get("location") or "").strip() if not location: # ``join("")`` returns the same URL, so without # this the loop re-fetched the identical # request until the hop budget ran out and then # reported a redirect loop that never happened. raise AppException( code="BAILIAN_MEDIA_BAD_RESULT_URL", message=( "The vendor redirected the download " "without saying where." ), status_code=502, ) url = str(httpx.URL(url).join(location)) _assert_downloadable(url, redirected=True) raise AppException( code="BAILIAN_MEDIA_BAD_RESULT_URL", message="The vendor download redirected too many times.", status_code=502, ) except AppException: raise except (httpx.HTTPError, httpx.InvalidURL) as exc: # ``InvalidURL`` is NOT an ``HTTPError`` subclass, so a # malformed Location used to escape this handler entirely and # reach the executor's generic handler, which persists # ``str(exc)[:500]`` — putting the crafted URL into a row the # employee reads. raise AppException( code="BAILIAN_MEDIA_NETWORK", message="The generated file could not be downloaded.", status_code=502, ) from exc async def _read_result_body( self, resp: httpx.Response, ) -> tuple[bytes, str]: """Stream one artifact body under the size ceiling.""" chunks: list[bytes] = [] total = 0 if resp.status_code != 200: raise AppException( code="BAILIAN_MEDIA_RESULT_UNAVAILABLE", message=( "The generated file could not be downloaded from the " f"vendor (HTTP {resp.status_code}). Its link may have " "expired." ), status_code=502, ) content_type = ( resp.headers.get("content-type") or "" ).split(";")[0].strip() async for chunk in resp.aiter_bytes(): total += len(chunk) if total > MAX_RESULT_BYTES: raise AppException( code="BAILIAN_MEDIA_RESULT_TOO_LARGE", message=( "The generated file exceeds the " f"{MAX_RESULT_BYTES // (1024 * 1024)} MB ceiling " "and was not stored." ), status_code=502, ) chunks.append(chunk) return b"".join(chunks), content_type or "application/octet-stream" # ── convenience ──────────────────────────────────────────── async def probe_models(self) -> set[str]: """Model ids this key can see, via compatible-mode ``/models``. Used by the admin console to show which catalogue entries the live key actually offers. Returns an empty set (never raises) so a connector hiccup degrades the badge rather than the page. """ base = ( "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/models" if self._region == REGION_INTL else "https://dashscope.aliyuncs.com/compatible-mode/v1/models" ) try: async with httpx.AsyncClient(timeout=20.0, trust_env=False) as client: resp = await client.get(base, headers=self._headers()) if resp.status_code != 200: return set() data = resp.json() except (TimeoutError, httpx.HTTPError, ValueError): return set() rows = data.get("data") if isinstance(data, dict) else None if not isinstance(rows, list): return set() return { str(row["id"]).strip() for row in rows if isinstance(row, dict) and row.get("id") } __all__ = [ "MAX_RESULT_BYTES", "PATH_MULTIMODAL", "PATH_TEXT2IMAGE", "PATH_VIDEO", "TASK_FAILED", "TASK_RUNNING", "TASK_SUCCEEDED", "BailianMediaClient", "native_root_for_region", ]