Spaces:

scikit-plots
/

ai

Running

ai / _utils /_video_generation.py

Upload 144 files

cef9869 verified 15 days ago

24.4 kB

# scikitplot/_externals/_sphinx_ext/_sphinx_ai_assistant/_hf_spaces_proxy/_utils/_video_generation.py
#
# flake8: noqa: D213
#
# Authors: The scikit-plots developers
# SPDX-License-Identifier: BSD-3-Clause

"""
Video-generation contracts and a deterministic test backend.
The production proxy owns transport/authentication while this module owns the
bounded public contract, validation, and an in-memory stub lifecycle used only
when ``VIDEO_GENERATION_MODE=stub``.  The stub never publishes media and never
claims to have uploaded to YouTube; it exists so the browser lifecycle can be
exercised end to end before a real renderer/publisher is configured.
"""

from __future__ import annotations

import copy
import ipaddress
import json
import re
import time
import uuid
from collections import OrderedDict
from typing import Any
from urllib.parse import urlsplit, urlunsplit

VIDEO_GENERATION_REQUEST_CONTRACT = "learn.video-generation-request.v1"
VIDEO_GENERATION_JOB_CONTRACT = "learn.video-generation-job.v1"
VIDEO_GENERATION_CAPABILITY_VERSION = 2
VIDEO_GENERATION_MAX_BODY_BYTES = 128 * 1024
VIDEO_GENERATION_MAX_JOBS = 256
VIDEO_GENERATION_RESULT_PROVIDERS = ("youtube", "hls", "file", "external")
VIDEO_GENERATION_ACTIONS = ("cancel", "retry", "archive", "restore")
VIDEO_GENERATION_STATUSES = (
    "draft",
    "submitted",
    "queued",
    "running",
    "ready",
    "failed",
    "cancelled",
    "archived",
)

_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,160}$")
_MODE_SET = frozenset({"topic", "source", "url", "prompt"})
_STYLE_SET = frozenset({"explainer", "lecture", "storyboard", "comparison"})
_LENGTH_SET = frozenset({"short", "standard", "deep"})
_ASPECT_SET = frozenset({"16:9", "1:1", "9:16"})


class VideoGenerationError(ValueError):
    """
    Bounded contract/configuration error with a stable public code.
    ``message`` is the authored, client-safe sentence passed at the raise
    site. Responses expose ``code`` and ``message`` only - never ``str(exc)``,
    whose content is whatever the exception machinery makes of the instance
    (arguments, notes, a chained cause) rather than what the author approved.
    """

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


def _bounded_text(value: Any, limit: int, *, required: bool = False) -> str:
    text = str(value or "").strip()
    if required and not text:
        raise VideoGenerationError(
            "VIDEO_FIELD_REQUIRED",
            "A required video generation field is empty.",
        )
    if len(text) > limit:
        raise VideoGenerationError(
            "VIDEO_FIELD_TOO_LONG",
            "A video generation field exceeds its size limit.",
        )
    return text


def _public_https_url(
    value: Any,
    *,
    allow_empty: bool = False,
) -> str:
    raw = _bounded_text(value, 2048)
    if not raw and allow_empty:
        return ""
    try:
        parsed = urlsplit(raw)
    except ValueError as exc:
        raise VideoGenerationError(
            "VIDEO_URL_INVALID",
            "The video generation URL is invalid.",
        ) from exc
    if parsed.scheme.lower() != "https" or not parsed.hostname:
        raise VideoGenerationError(
            "VIDEO_URL_INVALID",
            "Video generation URLs must use public HTTPS.",
        )
    if parsed.username or parsed.password or parsed.fragment:
        raise VideoGenerationError(
            "VIDEO_URL_INVALID",
            "Video generation URLs cannot contain credentials or fragments.",
        )
    hostname = parsed.hostname.rstrip(".").lower()
    if hostname == "localhost" or hostname.endswith((".localhost", ".local")):
        raise VideoGenerationError(
            "VIDEO_URL_PRIVATE",
            "Local/private video source URLs are not allowed.",
        )
    try:
        literal_ip = ipaddress.ip_address(hostname)
    except ValueError:
        literal_ip = None
    if literal_ip is not None and not literal_ip.is_global:
        raise VideoGenerationError(
            "VIDEO_URL_PRIVATE",
            "Local/private video source URLs are not allowed.",
        )
    # DNS rebinding/private-address resolution remains the upstream fetcher's
    # responsibility; this boundary rejects obvious local literals early.
    return urlunsplit(("https", parsed.netloc, parsed.path or "/", parsed.query, ""))


def normalize_video_upstream_url(value: str) -> str:
    """Validate an operator-configured exact upstream collection endpoint."""
    raw = str(value or "").strip()
    if not raw:
        return ""
    parsed = urlsplit(raw)
    if parsed.scheme.lower() != "https" or not parsed.hostname:
        raise VideoGenerationError(
            "VIDEO_UPSTREAM_INVALID",
            "VIDEO_GENERATION_UPSTREAM_URL must be an absolute HTTPS URL.",
        )
    if parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise VideoGenerationError(
            "VIDEO_UPSTREAM_INVALID",
            "VIDEO_GENERATION_UPSTREAM_URL cannot contain credentials, query parameters, or fragments.",
        )
    path = (parsed.path or "").rstrip("/")
    if not path:
        raise VideoGenerationError(
            "VIDEO_UPSTREAM_INVALID",
            "VIDEO_GENERATION_UPSTREAM_URL must name the video-generation collection endpoint.",
        )
    return urlunsplit(("https", parsed.netloc, path, "", ""))


def validate_generation_id(value: Any) -> str:
    generation_id = _bounded_text(value, 160, required=True)
    if not _ID_RE.fullmatch(generation_id):
        raise VideoGenerationError(
            "VIDEO_JOB_ID_INVALID",
            "Video generation id contains unsupported characters.",
        )
    return generation_id


def validate_idempotency_key(value: Any) -> str:
    key = _bounded_text(value, 160, required=True)
    if not _ID_RE.fullmatch(key):
        raise VideoGenerationError(
            "VIDEO_IDEMPOTENCY_INVALID",
            "Idempotency-Key contains unsupported characters.",
        )
    return key


def _source_snapshot(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise VideoGenerationError(
            "VIDEO_SOURCE_INVALID",
            "The Source snapshot is invalid.",
        )
    out = {
        "id": _bounded_text(value.get("id"), 160, required=True),
        "title": _bounded_text(value.get("title"), 240, required=True),
        "summary": _bounded_text(value.get("summary"), 2000),
        "publisher": _bounded_text(value.get("publisher"), 160),
        "format": _bounded_text(value.get("format"), 120),
        "url": _public_https_url(value.get("url"), allow_empty=True),
    }
    return out  # ruff: ignore[unnecessary-assign]


def _topic_snapshot(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise VideoGenerationError(
            "VIDEO_TOPIC_INVALID",
            "The Topic snapshot is invalid.",
        )
    domains = value.get("domains") if isinstance(value.get("domains"), list) else []
    sources = value.get("sources") if isinstance(value.get("sources"), list) else []
    return {
        "id": _bounded_text(value.get("id"), 160, required=True),
        "title": _bounded_text(value.get("title"), 240, required=True),
        "summary": _bounded_text(value.get("summary"), 2000),
        "domains": [
            _bounded_text(item, 80) for item in domains[:24] if str(item or "").strip()
        ],
        "sources": [_source_snapshot(item) for item in sources[:24]],
    }


def _model_snapshot(value: Any) -> dict[str, Any] | None:
    if value in (None, {}):
        return None
    if not isinstance(value, dict):
        raise VideoGenerationError(
            "VIDEO_MODEL_INVALID",
            "The requested model selection is invalid.",
        )
    effort = value.get("effort") if isinstance(value.get("effort"), dict) else {}
    return {
        "source": _bounded_text(value.get("source"), 80) or "ai-assistant",
        "id": _bounded_text(value.get("id"), 160, required=True),
        "label": _bounded_text(value.get("label"), 240),
        "provider": _bounded_text(value.get("provider"), 80),
        "model": _bounded_text(value.get("model"), 240),
        "effort": {
            "id": _bounded_text(effort.get("id"), 80) or "default",
            "label": _bounded_text(effort.get("label"), 120) or "Default",
            "supported": bool(effort.get("supported")),
        },
    }


def parse_video_generation_request(  # ruff: ignore[too-many-branches]
    raw: bytes,
) -> dict[str, Any]:
    if not raw:
        raise VideoGenerationError(
            "VIDEO_REQUEST_EMPTY",
            "Video generation request body is empty.",
        )
    if len(raw) > VIDEO_GENERATION_MAX_BODY_BYTES:
        raise VideoGenerationError(
            "VIDEO_REQUEST_TOO_LARGE",
            "Video generation request body is too large.",
        )
    try:
        value = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise VideoGenerationError(
            "VIDEO_REQUEST_JSON",
            "Video generation request must be valid UTF-8 JSON.",
        ) from exc
    if not isinstance(value, dict):
        raise VideoGenerationError(
            "VIDEO_REQUEST_INVALID",
            "Video generation request must be a JSON object.",
        )
    if value.get("contract") != VIDEO_GENERATION_REQUEST_CONTRACT:
        raise VideoGenerationError(
            "VIDEO_CONTRACT_UNSUPPORTED",
            "Unsupported video generation request contract.",
        )

    client_request_id = _bounded_text(
        value.get("client_request_id"), 160, required=True
    )
    if not _ID_RE.fullmatch(client_request_id):
        raise VideoGenerationError(
            "VIDEO_REQUEST_ID_INVALID",
            "Video generation request id is invalid.",
        )

    raw_input = value.get("input")
    if not isinstance(raw_input, dict):
        raise VideoGenerationError(
            "VIDEO_INPUT_INVALID",
            "Video generation input is invalid.",
        )
    mode = _bounded_text(raw_input.get("mode"), 32, required=True).lower()
    if mode not in _MODE_SET:
        raise VideoGenerationError(
            "VIDEO_INPUT_MODE",
            "Unsupported video generation input mode.",
        )
    input_doc: dict[str, Any] = {"mode": mode}
    if mode == "topic":
        input_doc["topic"] = _topic_snapshot(raw_input.get("topic"))
    elif mode == "source":
        input_doc["source"] = _source_snapshot(raw_input.get("source"))
    elif mode == "url":
        input_doc["url"] = _public_https_url(raw_input.get("url"))
    else:
        input_doc["prompt"] = _bounded_text(
            raw_input.get("prompt"), 6000, required=True
        )

    presentation = (
        value.get("presentation") if isinstance(value.get("presentation"), dict) else {}
    )
    style = _bounded_text(presentation.get("style"), 40) or "explainer"
    length = _bounded_text(presentation.get("length"), 40) or "standard"
    aspect = _bounded_text(presentation.get("aspect_ratio"), 20) or "16:9"
    if (
        style not in _STYLE_SET
        or length not in _LENGTH_SET
        or aspect not in _ASPECT_SET
    ):
        raise VideoGenerationError(
            "VIDEO_PRESENTATION_INVALID",
            "Unsupported video presentation option.",
        )
    presentation_doc = {
        "style": style,
        "length": length,
        "language": _bounded_text(presentation.get("language"), 32) or "en",
        "voice": _bounded_text(presentation.get("voice"), 160) or "default",
        "aspect_ratio": aspect,
        "captions": bool(presentation.get("captions", True)),
        "branding": bool(presentation.get("branding", True)),
    }

    provenance = (
        value.get("provenance") if isinstance(value.get("provenance"), dict) else {}
    )
    source_page = _bounded_text(provenance.get("source_page"), 2048)
    if source_page:
        source_page = _public_https_url(source_page)
    provenance_doc = {
        "site_id": _bounded_text(provenance.get("site_id"), 160),
        "catalog_revision": _bounded_text(provenance.get("catalog_revision"), 160),
        "source_page": source_page,
    }

    out: dict[str, Any] = {
        "contract": VIDEO_GENERATION_REQUEST_CONTRACT,
        "client_request_id": client_request_id,
        "input": input_doc,
        "instructions": _bounded_text(value.get("instructions"), 4000),
        "presentation": presentation_doc,
        "model_selection": _model_snapshot(value.get("model_selection")),
        "provenance": provenance_doc,
    }
    derived = _bounded_text(value.get("derived_from_video_id"), 160)
    if derived:
        if not _ID_RE.fullmatch(derived):
            raise VideoGenerationError(
                "VIDEO_DERIVED_ID_INVALID",
                "Derived video id is invalid.",
            )
        out["derived_from_video_id"] = derived
    return out


def normalize_video_job(value: Any) -> dict[str, Any]:
    """Validate and privacy-minimize one backend job receipt."""
    if not isinstance(value, dict):
        raise VideoGenerationError(
            "VIDEO_JOB_INVALID",
            "Video generation backend returned an invalid job.",
        )
    generation_id = _bounded_text(
        value.get("generation_id") or value.get("id"), 160, required=True
    )
    if not _ID_RE.fullmatch(generation_id):
        raise VideoGenerationError(
            "VIDEO_JOB_INVALID",
            "Video generation backend returned an invalid job id.",
        )
    status = _bounded_text(value.get("status"), 40, required=True)
    if status not in VIDEO_GENERATION_STATUSES:
        raise VideoGenerationError(
            "VIDEO_JOB_INVALID",
            "Video generation backend returned an unsupported status.",
        )
    try:
        progress_raw = value.get("progress")
        progress = (
            None if progress_raw is None else max(0.0, min(1.0, float(progress_raw)))
        )
    except (TypeError, ValueError):
        progress = None
    out: dict[str, Any] = {
        "contract": VIDEO_GENERATION_JOB_CONTRACT,
        "generation_id": generation_id,
        "status": status,
        "stage": _bounded_text(value.get("stage"), 80) or status,
        "progress": progress,
        "title": _bounded_text(value.get("title"), 240) or "Generated video",
        "created_at": _bounded_text(value.get("created_at"), 80),
        "updated_at": _bounded_text(value.get("updated_at"), 80),
    }
    error = value.get("error")
    if isinstance(error, dict):
        out["error"] = {
            "code": _bounded_text(error.get("code"), 120),
            "message": (
                _bounded_text(error.get("message"), 600)
                or "Generation could not complete."
            ),
            "retryable": error.get("retryable") is not False,
        }
    result = value.get("result")
    if isinstance(result, dict):
        provider = _bounded_text(result.get("provider"), 40)
        if provider and provider not in VIDEO_GENERATION_RESULT_PROVIDERS:
            raise VideoGenerationError(
                "VIDEO_JOB_INVALID",
                "Video generation backend returned an unsupported result provider.",
            )
        duration = result.get("duration_seconds")
        try:
            duration_number = max(0, min(24 * 3600, int(float(duration or 0))))
        except (TypeError, ValueError):
            duration_number = 0
        out["result"] = {
            "provider": provider,
            "provider_id": _bounded_text(result.get("provider_id"), 240),
            "url": _public_https_url(result.get("url"), allow_empty=True),
            "thumbnail_url": _public_https_url(
                result.get("thumbnail_url") or result.get("poster"), allow_empty=True
            ),
            "duration_seconds": duration_number,
            "test_mode": bool(result.get("test_mode")),
        }
    execution = value.get("execution")
    if isinstance(execution, dict):
        # Deliberately bounded provenance only; never pass provider credentials,
        # raw prompts, internal traces, or arbitrary backend objects.
        out["execution"] = {
            "pipeline_version": _bounded_text(execution.get("pipeline_version"), 160),
            "planner_model": _bounded_text(execution.get("planner_model"), 240),
            "renderer": _bounded_text(execution.get("renderer"), 160),
            "publish_provider": _bounded_text(execution.get("publish_provider"), 80),
            "fallback_used": bool(execution.get("fallback_used")),
        }
    return out


def request_title(request_doc: dict[str, Any]) -> str:
    input_doc = request_doc.get("input") or {}
    if isinstance(input_doc.get("topic"), dict):
        return str(input_doc["topic"].get("title") or "Generated video")[:200]
    if isinstance(input_doc.get("source"), dict):
        return str(input_doc["source"].get("title") or "Generated video")[:200]
    if input_doc.get("url"):
        return str(input_doc["url"])[:200]
    return str(input_doc.get("prompt") or "Generated video")[:200]


def _iso(now: float | None = None) -> str:
    # UTC ISO-8601 without importing datetime keeps this module stdlib-light.
    stamp = time.gmtime(time.time() if now is None else now)
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", stamp)


class StubVideoGenerationStore:
    """Bounded deterministic lifecycle for browser/proxy integration tests."""

    _RUN_STAGES = (
        ("queued", 0.05),
        ("preparing", 0.14),
        ("grounding", 0.28),
        ("outline", 0.42),
        ("script", 0.58),
        ("visuals", 0.74),
        ("rendering", 0.90),
        ("verifying", 0.97),
    )

    def __init__(
        self, *, max_jobs: int = VIDEO_GENERATION_MAX_JOBS, ready_seconds: float = 3.0
    ) -> None:
        self.max_jobs = max(8, min(4096, int(max_jobs)))
        self.ready_seconds = max(0.0, min(300.0, float(ready_seconds)))
        self._jobs: OrderedDict[str, dict[str, Any]] = OrderedDict()
        self._idempotency: dict[str, str] = {}

    def _trim(self) -> None:
        while len(self._jobs) > self.max_jobs:
            generation_id, _job = self._jobs.popitem(last=False)
            stale = [
                key
                for key, value in self._idempotency.items()
                if value == generation_id
            ]
            for key in stale:
                self._idempotency.pop(key, None)

    def create(
        self, request_doc: dict[str, Any], idempotency_key: str
    ) -> dict[str, Any]:
        existing = self._idempotency.get(idempotency_key)
        if existing and existing in self._jobs:
            return self.get(existing)
        now = time.time()
        generation_id = "vg_stub_" + uuid.uuid4().hex[:20]
        job = {
            "contract": VIDEO_GENERATION_JOB_CONTRACT,
            "generation_id": generation_id,
            "status": "queued",
            "stage": "queued",
            "progress": 0.05,
            "title": request_title(request_doc),
            "created_at": _iso(now),
            "updated_at": _iso(now),
            "_started_epoch": now,
            "_request": copy.deepcopy(request_doc),
            "_archived_from": "",
        }
        self._jobs[generation_id] = job
        self._idempotency[idempotency_key] = generation_id
        self._trim()
        return self.get(generation_id)

    def _materialize(self, job: dict[str, Any]) -> dict[str, Any]:
        if job.get("status") in {"cancelled", "failed", "archived"}:
            return job
        elapsed = max(
            0.0, time.time() - float(job.get("_started_epoch") or time.time())
        )
        if self.ready_seconds <= 0 or elapsed >= self.ready_seconds:
            job.update(
                status="ready",
                stage="ready",
                progress=1.0,
                updated_at=_iso(),
                result={
                    "provider": "external",
                    "provider_id": "stub:" + str(job["generation_id"]),
                    "url": "",
                    "thumbnail_url": "",
                    "duration_seconds": 0,
                    "test_mode": True,
                },
            )
            return job
        fraction = elapsed / self.ready_seconds
        index = min(len(self._RUN_STAGES) - 1, int(fraction * len(self._RUN_STAGES)))
        stage, floor = self._RUN_STAGES[index]
        next_floor = self._RUN_STAGES[min(index + 1, len(self._RUN_STAGES) - 1)][1]
        local = (fraction * len(self._RUN_STAGES)) - index
        progress = min(
            0.99, floor + max(0.0, next_floor - floor) * max(0.0, min(1.0, local))
        )
        job.update(
            status=("queued" if stage == "queued" else "running"),
            stage=stage,
            progress=progress,
            updated_at=_iso(),
        )
        return job

    @staticmethod
    def _public(job: dict[str, Any]) -> dict[str, Any]:
        return {
            key: copy.deepcopy(value)
            for key, value in job.items()
            if not key.startswith("_")
        }

    def get(self, generation_id: str) -> dict[str, Any]:
        job = self._jobs.get(generation_id)
        if not job:
            raise VideoGenerationError(
                "VIDEO_JOB_NOT_FOUND",
                "Video generation was not found.",
            )
        self._materialize(job)
        self._jobs.move_to_end(generation_id)
        return self._public(job)

    def list(self, *, include_archived: bool = True) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []
        for generation_id in reversed(list(self._jobs.keys())):
            item = self.get(generation_id)
            if not include_archived and item.get("status") == "archived":
                continue
            out.append(item)
        return out

    def action(self, generation_id: str, action: str) -> dict[str, Any]:
        if action not in VIDEO_GENERATION_ACTIONS:
            raise VideoGenerationError(
                "VIDEO_ACTION_UNSUPPORTED",
                "Unsupported video generation lifecycle action.",
            )
        job = self._jobs.get(generation_id)
        if not job:
            raise VideoGenerationError(
                "VIDEO_JOB_NOT_FOUND",
                "Video generation was not found.",
            )
        self._materialize(job)
        status = str(job.get("status") or "")
        if action == "cancel":
            if status not in {"submitted", "queued", "running"}:
                raise VideoGenerationError(
                    "VIDEO_ACTION_INVALID_STATE",
                    "Only an active generation can be cancelled.",
                )
            job.update(
                status="cancelled",
                stage="cancelled",
                progress=job.get("progress"),
                updated_at=_iso(),
            )
        elif action == "retry":
            if status not in {"failed", "cancelled"}:
                raise VideoGenerationError(
                    "VIDEO_ACTION_INVALID_STATE",
                    "Only a failed or cancelled generation can be retried.",
                )
            job.pop("error", None)
            job.pop("result", None)
            job.update(
                status="queued",
                stage="queued",
                progress=0.05,
                updated_at=_iso(),
                _started_epoch=time.time(),
            )
        elif action == "archive":
            if status in {"submitted", "queued", "running", "archived"}:
                raise VideoGenerationError(
                    "VIDEO_ACTION_INVALID_STATE",
                    "Active or already archived generations cannot be archived.",
                )
            job["_archived_from"] = status
            job.update(status="archived", stage="archived", updated_at=_iso())
        elif action == "restore":
            if status != "archived":
                raise VideoGenerationError(
                    "VIDEO_ACTION_INVALID_STATE",
                    "Only an archived generation can be restored.",
                )
            restored = str(
                job.get("_archived_from")
                or ("ready" if job.get("result") else "cancelled")
            )
            job.update(status=restored, stage=restored, updated_at=_iso())
        return self._public(job)

	# scikitplot/_externals/_sphinx_ext/_sphinx_ai_assistant/_hf_spaces_proxy/_utils/_video_generation.py
	#
	# flake8: noqa: D213
	#
	# Authors: The scikit-plots developers
	# SPDX-License-Identifier: BSD-3-Clause

	"""
	Video-generation contracts and a deterministic test backend.

	The production proxy owns transport/authentication while this module owns the
	bounded public contract, validation, and an in-memory stub lifecycle used only
	when ``VIDEO_GENERATION_MODE=stub``. The stub never publishes media and never
	claims to have uploaded to YouTube; it exists so the browser lifecycle can be
	exercised end to end before a real renderer/publisher is configured.
	"""

	from __future__ import annotations

	import copy
	import ipaddress
	import json
	import re
	import time
	import uuid
	from collections import OrderedDict
	from typing import Any
	from urllib.parse import urlsplit, urlunsplit

	VIDEO_GENERATION_REQUEST_CONTRACT = "learn.video-generation-request.v1"
	VIDEO_GENERATION_JOB_CONTRACT = "learn.video-generation-job.v1"
	VIDEO_GENERATION_CAPABILITY_VERSION = 2
	VIDEO_GENERATION_MAX_BODY_BYTES = 128 * 1024
	VIDEO_GENERATION_MAX_JOBS = 256
	VIDEO_GENERATION_RESULT_PROVIDERS = ("youtube", "hls", "file", "external")
	VIDEO_GENERATION_ACTIONS = ("cancel", "retry", "archive", "restore")
	VIDEO_GENERATION_STATUSES = (
	"draft",
	"submitted",
	"queued",
	"running",
	"ready",
	"failed",
	"cancelled",
	"archived",
	)

	_ID_RE = re.compile(r"^[A-Za-z0-9._:-]{1,160}$")
	_MODE_SET = frozenset({"topic", "source", "url", "prompt"})
	_STYLE_SET = frozenset({"explainer", "lecture", "storyboard", "comparison"})
	_LENGTH_SET = frozenset({"short", "standard", "deep"})
	_ASPECT_SET = frozenset({"16:9", "1:1", "9:16"})


	class VideoGenerationError(ValueError):
	"""
	Bounded contract/configuration error with a stable public code.

	``message`` is the authored, client-safe sentence passed at the raise
	site. Responses expose ``code`` and ``message`` only - never ``str(exc)``,
	whose content is whatever the exception machinery makes of the instance
	(arguments, notes, a chained cause) rather than what the author approved.
	"""

	def __init__(self, code: str, message: str) -> None:
	super().__init__(message)
	self.code = code
	self.message = message


	def _bounded_text(value: Any, limit: int, *, required: bool = False) -> str:
	text = str(value or "").strip()
	if required and not text:
	raise VideoGenerationError(
	"VIDEO_FIELD_REQUIRED",
	"A required video generation field is empty.",
	)
	if len(text) > limit:
	raise VideoGenerationError(
	"VIDEO_FIELD_TOO_LONG",
	"A video generation field exceeds its size limit.",
	)
	return text


	def _public_https_url(
	value: Any,
	*,
	allow_empty: bool = False,
	) -> str:
	raw = _bounded_text(value, 2048)
	if not raw and allow_empty:
	return ""
	try:
	parsed = urlsplit(raw)
	except ValueError as exc:
	raise VideoGenerationError(
	"VIDEO_URL_INVALID",
	"The video generation URL is invalid.",
	) from exc
	if parsed.scheme.lower() != "https" or not parsed.hostname:
	raise VideoGenerationError(
	"VIDEO_URL_INVALID",
	"Video generation URLs must use public HTTPS.",
	)
	if parsed.username or parsed.password or parsed.fragment:
	raise VideoGenerationError(
	"VIDEO_URL_INVALID",
	"Video generation URLs cannot contain credentials or fragments.",
	)
	hostname = parsed.hostname.rstrip(".").lower()
	if hostname == "localhost" or hostname.endswith((".localhost", ".local")):
	raise VideoGenerationError(
	"VIDEO_URL_PRIVATE",
	"Local/private video source URLs are not allowed.",
	)
	try:
	literal_ip = ipaddress.ip_address(hostname)
	except ValueError:
	literal_ip = None
	if literal_ip is not None and not literal_ip.is_global:
	raise VideoGenerationError(
	"VIDEO_URL_PRIVATE",
	"Local/private video source URLs are not allowed.",
	)
	# DNS rebinding/private-address resolution remains the upstream fetcher's
	# responsibility; this boundary rejects obvious local literals early.
	return urlunsplit(("https", parsed.netloc, parsed.path or "/", parsed.query, ""))


	def normalize_video_upstream_url(value: str) -> str:
	"""Validate an operator-configured exact upstream collection endpoint."""
	raw = str(value or "").strip()
	if not raw:
	return ""
	parsed = urlsplit(raw)
	if parsed.scheme.lower() != "https" or not parsed.hostname:
	raise VideoGenerationError(
	"VIDEO_UPSTREAM_INVALID",
	"VIDEO_GENERATION_UPSTREAM_URL must be an absolute HTTPS URL.",
	)
	if parsed.username or parsed.password or parsed.query or parsed.fragment:
	raise VideoGenerationError(
	"VIDEO_UPSTREAM_INVALID",
	"VIDEO_GENERATION_UPSTREAM_URL cannot contain credentials, query parameters, or fragments.",
	)
	path = (parsed.path or "").rstrip("/")
	if not path:
	raise VideoGenerationError(
	"VIDEO_UPSTREAM_INVALID",
	"VIDEO_GENERATION_UPSTREAM_URL must name the video-generation collection endpoint.",
	)
	return urlunsplit(("https", parsed.netloc, path, "", ""))


	def validate_generation_id(value: Any) -> str:
	generation_id = _bounded_text(value, 160, required=True)
	if not _ID_RE.fullmatch(generation_id):
	raise VideoGenerationError(
	"VIDEO_JOB_ID_INVALID",
	"Video generation id contains unsupported characters.",
	)
	return generation_id


	def validate_idempotency_key(value: Any) -> str:
	key = _bounded_text(value, 160, required=True)
	if not _ID_RE.fullmatch(key):
	raise VideoGenerationError(
	"VIDEO_IDEMPOTENCY_INVALID",
	"Idempotency-Key contains unsupported characters.",
	)
	return key


	def _source_snapshot(value: Any) -> dict[str, Any]:
	if not isinstance(value, dict):
	raise VideoGenerationError(
	"VIDEO_SOURCE_INVALID",
	"The Source snapshot is invalid.",
	)
	out = {
	"id": _bounded_text(value.get("id"), 160, required=True),
	"title": _bounded_text(value.get("title"), 240, required=True),
	"summary": _bounded_text(value.get("summary"), 2000),
	"publisher": _bounded_text(value.get("publisher"), 160),
	"format": _bounded_text(value.get("format"), 120),
	"url": _public_https_url(value.get("url"), allow_empty=True),
	}
	return out # ruff: ignore[unnecessary-assign]


	def _topic_snapshot(value: Any) -> dict[str, Any]:
	if not isinstance(value, dict):
	raise VideoGenerationError(
	"VIDEO_TOPIC_INVALID",
	"The Topic snapshot is invalid.",
	)
	domains = value.get("domains") if isinstance(value.get("domains"), list) else []
	sources = value.get("sources") if isinstance(value.get("sources"), list) else []
	return {
	"id": _bounded_text(value.get("id"), 160, required=True),
	"title": _bounded_text(value.get("title"), 240, required=True),
	"summary": _bounded_text(value.get("summary"), 2000),
	"domains": [
	_bounded_text(item, 80) for item in domains[:24] if str(item or "").strip()
	],
	"sources": [_source_snapshot(item) for item in sources[:24]],
	}


	def _model_snapshot(value: Any) -> dict[str, Any] \| None:
	if value in (None, {}):
	return None
	if not isinstance(value, dict):
	raise VideoGenerationError(
	"VIDEO_MODEL_INVALID",
	"The requested model selection is invalid.",
	)
	effort = value.get("effort") if isinstance(value.get("effort"), dict) else {}
	return {
	"source": _bounded_text(value.get("source"), 80) or "ai-assistant",
	"id": _bounded_text(value.get("id"), 160, required=True),
	"label": _bounded_text(value.get("label"), 240),
	"provider": _bounded_text(value.get("provider"), 80),
	"model": _bounded_text(value.get("model"), 240),
	"effort": {
	"id": _bounded_text(effort.get("id"), 80) or "default",
	"label": _bounded_text(effort.get("label"), 120) or "Default",
	"supported": bool(effort.get("supported")),
	},
	}


	def parse_video_generation_request( # ruff: ignore[too-many-branches]
	raw: bytes,
	) -> dict[str, Any]:
	if not raw:
	raise VideoGenerationError(
	"VIDEO_REQUEST_EMPTY",
	"Video generation request body is empty.",
	)
	if len(raw) > VIDEO_GENERATION_MAX_BODY_BYTES:
	raise VideoGenerationError(
	"VIDEO_REQUEST_TOO_LARGE",
	"Video generation request body is too large.",
	)
	try:
	value = json.loads(raw.decode("utf-8"))
	except (UnicodeDecodeError, json.JSONDecodeError) as exc:
	raise VideoGenerationError(
	"VIDEO_REQUEST_JSON",
	"Video generation request must be valid UTF-8 JSON.",
	) from exc
	if not isinstance(value, dict):
	raise VideoGenerationError(
	"VIDEO_REQUEST_INVALID",
	"Video generation request must be a JSON object.",
	)
	if value.get("contract") != VIDEO_GENERATION_REQUEST_CONTRACT:
	raise VideoGenerationError(
	"VIDEO_CONTRACT_UNSUPPORTED",
	"Unsupported video generation request contract.",
	)

	client_request_id = _bounded_text(
	value.get("client_request_id"), 160, required=True
	)
	if not _ID_RE.fullmatch(client_request_id):
	raise VideoGenerationError(
	"VIDEO_REQUEST_ID_INVALID",
	"Video generation request id is invalid.",
	)

	raw_input = value.get("input")
	if not isinstance(raw_input, dict):
	raise VideoGenerationError(
	"VIDEO_INPUT_INVALID",
	"Video generation input is invalid.",
	)
	mode = _bounded_text(raw_input.get("mode"), 32, required=True).lower()
	if mode not in _MODE_SET:
	raise VideoGenerationError(
	"VIDEO_INPUT_MODE",
	"Unsupported video generation input mode.",
	)
	input_doc: dict[str, Any] = {"mode": mode}
	if mode == "topic":
	input_doc["topic"] = _topic_snapshot(raw_input.get("topic"))
	elif mode == "source":
	input_doc["source"] = _source_snapshot(raw_input.get("source"))
	elif mode == "url":
	input_doc["url"] = _public_https_url(raw_input.get("url"))
	else:
	input_doc["prompt"] = _bounded_text(
	raw_input.get("prompt"), 6000, required=True
	)

	presentation = (
	value.get("presentation") if isinstance(value.get("presentation"), dict) else {}
	)
	style = _bounded_text(presentation.get("style"), 40) or "explainer"
	length = _bounded_text(presentation.get("length"), 40) or "standard"
	aspect = _bounded_text(presentation.get("aspect_ratio"), 20) or "16:9"
	if (
	style not in _STYLE_SET
	or length not in _LENGTH_SET
	or aspect not in _ASPECT_SET
	):
	raise VideoGenerationError(
	"VIDEO_PRESENTATION_INVALID",
	"Unsupported video presentation option.",
	)
	presentation_doc = {
	"style": style,
	"length": length,
	"language": _bounded_text(presentation.get("language"), 32) or "en",
	"voice": _bounded_text(presentation.get("voice"), 160) or "default",
	"aspect_ratio": aspect,
	"captions": bool(presentation.get("captions", True)),
	"branding": bool(presentation.get("branding", True)),
	}

	provenance = (
	value.get("provenance") if isinstance(value.get("provenance"), dict) else {}
	)
	source_page = _bounded_text(provenance.get("source_page"), 2048)
	if source_page:
	source_page = _public_https_url(source_page)
	provenance_doc = {
	"site_id": _bounded_text(provenance.get("site_id"), 160),
	"catalog_revision": _bounded_text(provenance.get("catalog_revision"), 160),
	"source_page": source_page,
	}

	out: dict[str, Any] = {
	"contract": VIDEO_GENERATION_REQUEST_CONTRACT,
	"client_request_id": client_request_id,
	"input": input_doc,
	"instructions": _bounded_text(value.get("instructions"), 4000),
	"presentation": presentation_doc,
	"model_selection": _model_snapshot(value.get("model_selection")),
	"provenance": provenance_doc,
	}
	derived = _bounded_text(value.get("derived_from_video_id"), 160)
	if derived:
	if not _ID_RE.fullmatch(derived):
	raise VideoGenerationError(
	"VIDEO_DERIVED_ID_INVALID",
	"Derived video id is invalid.",
	)
	out["derived_from_video_id"] = derived
	return out


	def normalize_video_job(value: Any) -> dict[str, Any]:
	"""Validate and privacy-minimize one backend job receipt."""
	if not isinstance(value, dict):
	raise VideoGenerationError(
	"VIDEO_JOB_INVALID",
	"Video generation backend returned an invalid job.",
	)
	generation_id = _bounded_text(
	value.get("generation_id") or value.get("id"), 160, required=True
	)
	if not _ID_RE.fullmatch(generation_id):
	raise VideoGenerationError(
	"VIDEO_JOB_INVALID",
	"Video generation backend returned an invalid job id.",
	)
	status = _bounded_text(value.get("status"), 40, required=True)
	if status not in VIDEO_GENERATION_STATUSES:
	raise VideoGenerationError(
	"VIDEO_JOB_INVALID",
	"Video generation backend returned an unsupported status.",
	)
	try:
	progress_raw = value.get("progress")
	progress = (
	None if progress_raw is None else max(0.0, min(1.0, float(progress_raw)))
	)
	except (TypeError, ValueError):
	progress = None
	out: dict[str, Any] = {
	"contract": VIDEO_GENERATION_JOB_CONTRACT,
	"generation_id": generation_id,
	"status": status,
	"stage": _bounded_text(value.get("stage"), 80) or status,
	"progress": progress,
	"title": _bounded_text(value.get("title"), 240) or "Generated video",
	"created_at": _bounded_text(value.get("created_at"), 80),
	"updated_at": _bounded_text(value.get("updated_at"), 80),
	}
	error = value.get("error")
	if isinstance(error, dict):
	out["error"] = {
	"code": _bounded_text(error.get("code"), 120),
	"message": (
	_bounded_text(error.get("message"), 600)
	or "Generation could not complete."
	),
	"retryable": error.get("retryable") is not False,
	}
	result = value.get("result")
	if isinstance(result, dict):
	provider = _bounded_text(result.get("provider"), 40)
	if provider and provider not in VIDEO_GENERATION_RESULT_PROVIDERS:
	raise VideoGenerationError(
	"VIDEO_JOB_INVALID",
	"Video generation backend returned an unsupported result provider.",
	)
	duration = result.get("duration_seconds")
	try:
	duration_number = max(0, min(24 * 3600, int(float(duration or 0))))
	except (TypeError, ValueError):
	duration_number = 0
	out["result"] = {
	"provider": provider,
	"provider_id": _bounded_text(result.get("provider_id"), 240),
	"url": _public_https_url(result.get("url"), allow_empty=True),
	"thumbnail_url": _public_https_url(
	result.get("thumbnail_url") or result.get("poster"), allow_empty=True
	),
	"duration_seconds": duration_number,
	"test_mode": bool(result.get("test_mode")),
	}
	execution = value.get("execution")
	if isinstance(execution, dict):
	# Deliberately bounded provenance only; never pass provider credentials,
	# raw prompts, internal traces, or arbitrary backend objects.
	out["execution"] = {
	"pipeline_version": _bounded_text(execution.get("pipeline_version"), 160),
	"planner_model": _bounded_text(execution.get("planner_model"), 240),
	"renderer": _bounded_text(execution.get("renderer"), 160),
	"publish_provider": _bounded_text(execution.get("publish_provider"), 80),
	"fallback_used": bool(execution.get("fallback_used")),
	}
	return out


	def request_title(request_doc: dict[str, Any]) -> str:
	input_doc = request_doc.get("input") or {}
	if isinstance(input_doc.get("topic"), dict):
	return str(input_doc["topic"].get("title") or "Generated video")[:200]
	if isinstance(input_doc.get("source"), dict):
	return str(input_doc["source"].get("title") or "Generated video")[:200]
	if input_doc.get("url"):
	return str(input_doc["url"])[:200]
	return str(input_doc.get("prompt") or "Generated video")[:200]


	def _iso(now: float \| None = None) -> str:
	# UTC ISO-8601 without importing datetime keeps this module stdlib-light.
	stamp = time.gmtime(time.time() if now is None else now)
	return time.strftime("%Y-%m-%dT%H:%M:%SZ", stamp)


	class StubVideoGenerationStore:
	"""Bounded deterministic lifecycle for browser/proxy integration tests."""

	_RUN_STAGES = (
	("queued", 0.05),
	("preparing", 0.14),
	("grounding", 0.28),
	("outline", 0.42),
	("script", 0.58),
	("visuals", 0.74),
	("rendering", 0.90),
	("verifying", 0.97),
	)

	def __init__(
	self, *, max_jobs: int = VIDEO_GENERATION_MAX_JOBS, ready_seconds: float = 3.0
	) -> None:
	self.max_jobs = max(8, min(4096, int(max_jobs)))
	self.ready_seconds = max(0.0, min(300.0, float(ready_seconds)))
	self._jobs: OrderedDict[str, dict[str, Any]] = OrderedDict()
	self._idempotency: dict[str, str] = {}

	def _trim(self) -> None:
	while len(self._jobs) > self.max_jobs:
	generation_id, _job = self._jobs.popitem(last=False)
	stale = [
	key
	for key, value in self._idempotency.items()
	if value == generation_id
	]
	for key in stale:
	self._idempotency.pop(key, None)

	def create(
	self, request_doc: dict[str, Any], idempotency_key: str
	) -> dict[str, Any]:
	existing = self._idempotency.get(idempotency_key)
	if existing and existing in self._jobs:
	return self.get(existing)
	now = time.time()
	generation_id = "vg_stub_" + uuid.uuid4().hex[:20]
	job = {
	"contract": VIDEO_GENERATION_JOB_CONTRACT,
	"generation_id": generation_id,
	"status": "queued",
	"stage": "queued",
	"progress": 0.05,
	"title": request_title(request_doc),
	"created_at": _iso(now),
	"updated_at": _iso(now),
	"_started_epoch": now,
	"_request": copy.deepcopy(request_doc),
	"_archived_from": "",
	}
	self._jobs[generation_id] = job
	self._idempotency[idempotency_key] = generation_id
	self._trim()
	return self.get(generation_id)

	def _materialize(self, job: dict[str, Any]) -> dict[str, Any]:
	if job.get("status") in {"cancelled", "failed", "archived"}:
	return job
	elapsed = max(
	0.0, time.time() - float(job.get("_started_epoch") or time.time())
	)
	if self.ready_seconds <= 0 or elapsed >= self.ready_seconds:
	job.update(
	status="ready",
	stage="ready",
	progress=1.0,
	updated_at=_iso(),
	result={
	"provider": "external",
	"provider_id": "stub:" + str(job["generation_id"]),
	"url": "",
	"thumbnail_url": "",
	"duration_seconds": 0,
	"test_mode": True,
	},
	)
	return job
	fraction = elapsed / self.ready_seconds
	index = min(len(self._RUN_STAGES) - 1, int(fraction * len(self._RUN_STAGES)))
	stage, floor = self._RUN_STAGES[index]
	next_floor = self._RUN_STAGES[min(index + 1, len(self._RUN_STAGES) - 1)][1]
	local = (fraction * len(self._RUN_STAGES)) - index
	progress = min(
	0.99, floor + max(0.0, next_floor - floor) * max(0.0, min(1.0, local))
	)
	job.update(
	status=("queued" if stage == "queued" else "running"),
	stage=stage,
	progress=progress,
	updated_at=_iso(),
	)
	return job

	@staticmethod
	def _public(job: dict[str, Any]) -> dict[str, Any]:
	return {
	key: copy.deepcopy(value)
	for key, value in job.items()
	if not key.startswith("_")
	}

	def get(self, generation_id: str) -> dict[str, Any]:
	job = self._jobs.get(generation_id)
	if not job:
	raise VideoGenerationError(
	"VIDEO_JOB_NOT_FOUND",
	"Video generation was not found.",
	)
	self._materialize(job)
	self._jobs.move_to_end(generation_id)
	return self._public(job)

	def list(self, *, include_archived: bool = True) -> list[dict[str, Any]]:
	out: list[dict[str, Any]] = []
	for generation_id in reversed(list(self._jobs.keys())):
	item = self.get(generation_id)
	if not include_archived and item.get("status") == "archived":
	continue
	out.append(item)
	return out

	def action(self, generation_id: str, action: str) -> dict[str, Any]:
	if action not in VIDEO_GENERATION_ACTIONS:
	raise VideoGenerationError(
	"VIDEO_ACTION_UNSUPPORTED",
	"Unsupported video generation lifecycle action.",
	)
	job = self._jobs.get(generation_id)
	if not job:
	raise VideoGenerationError(
	"VIDEO_JOB_NOT_FOUND",
	"Video generation was not found.",
	)
	self._materialize(job)
	status = str(job.get("status") or "")
	if action == "cancel":
	if status not in {"submitted", "queued", "running"}:
	raise VideoGenerationError(
	"VIDEO_ACTION_INVALID_STATE",
	"Only an active generation can be cancelled.",
	)
	job.update(
	status="cancelled",
	stage="cancelled",
	progress=job.get("progress"),
	updated_at=_iso(),
	)
	elif action == "retry":
	if status not in {"failed", "cancelled"}:
	raise VideoGenerationError(
	"VIDEO_ACTION_INVALID_STATE",
	"Only a failed or cancelled generation can be retried.",
	)
	job.pop("error", None)
	job.pop("result", None)
	job.update(
	status="queued",
	stage="queued",
	progress=0.05,
	updated_at=_iso(),
	_started_epoch=time.time(),
	)
	elif action == "archive":
	if status in {"submitted", "queued", "running", "archived"}:
	raise VideoGenerationError(
	"VIDEO_ACTION_INVALID_STATE",
	"Active or already archived generations cannot be archived.",
	)
	job["_archived_from"] = status
	job.update(status="archived", stage="archived", updated_at=_iso())
	elif action == "restore":
	if status != "archived":
	raise VideoGenerationError(
	"VIDEO_ACTION_INVALID_STATE",
	"Only an archived generation can be restored.",
	)
	restored = str(
	job.get("_archived_from")
	or ("ready" if job.get("result") else "cancelled")
	)
	job.update(status=restored, stage=restored, updated_at=_iso())
	return self._public(job)