| 1 | """Resolve browser-reachable video URLs from Echo / algorithm payloads.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from typing import Any |
| 6 | |
| 7 | _HTTP_PREFIXES = ("http://", "https://") |
| 8 | _STORAGE_ENTRY_KEYS = ("url", "public_url") |
| 9 | _STORAGE_CONTAINER_KEYS = ("asset_urls", "story_asset_urls", "oss_urls") |
| 10 | _DIRECT_URL_KEYS = ( |
| 11 | "result_url", |
| 12 | "artifact_url", |
| 13 | "video_url", |
| 14 | "download_url", |
| 15 | "output_url", |
| 16 | "url", |
| 17 | "final_output_url", |
| 18 | "story_mp4", |
| 19 | ) |
| 20 | |
| 21 | |
| 22 | def _is_public_http_url(value: Any) -> str | None: |
| 23 | if isinstance(value, str): |
| 24 | stripped = value.strip() |
| 25 | if stripped.startswith(_HTTP_PREFIXES): |
| 26 | return stripped |
| 27 | return None |
| 28 | |
| 29 | |
| 30 | def _storage_entry_public_url(entry: Any) -> str | None: |
| 31 | if isinstance(entry, dict): |
| 32 | for key in _STORAGE_ENTRY_KEYS: |
| 33 | url = _is_public_http_url(entry.get(key)) |
| 34 | if url: |
| 35 | return url |
| 36 | return None |
| 37 | return _is_public_http_url(entry) |
| 38 | |
| 39 | |
| 40 | def _iter_sources(*sources: dict[str, Any] | None) -> list[dict[str, Any]]: |
| 41 | ordered: list[dict[str, Any]] = [] |
| 42 | for source in sources: |
| 43 | if isinstance(source, dict): |
| 44 | ordered.append(source) |
| 45 | return ordered |
| 46 | |
| 47 | |
| 48 | def resolve_storage_video_url(*sources: dict[str, Any] | None) -> str | None: |
| 49 | """Return the first public URL from a provider-neutral storage mapping.""" |
| 50 | for source in _iter_sources(*sources): |
| 51 | for container_key in _STORAGE_CONTAINER_KEYS: |
| 52 | entries = source.get(container_key) |
| 53 | if not isinstance(entries, dict): |
| 54 | continue |
| 55 | for entry in entries.values(): |
| 56 | url = _storage_entry_public_url(entry) |
| 57 | if url: |
| 58 | return url |
| 59 | return None |
| 60 | |
| 61 | |
| 62 | def resolve_direct_video_url(*sources: dict[str, Any] | None) -> str | None: |
| 63 | """Return the first HTTP(S) URL from direct result fields.""" |
| 64 | for source in _iter_sources(*sources): |
| 65 | for key in _DIRECT_URL_KEYS: |
| 66 | url = _is_public_http_url(source.get(key)) |
| 67 | if url: |
| 68 | return url |
| 69 | return None |
| 70 | |
| 71 | |
| 72 | def resolve_public_video_url(*sources: dict[str, Any] | None) -> str | None: |
| 73 | """Pick a public video URL, preferring configured storage result mappings. |
| 74 | |
| 75 | Never returns PFS or other local filesystem paths — those are not playable |
| 76 | in the browser and should not be stored as ``artifact_url``. |
| 77 | """ |
| 78 | return resolve_storage_video_url(*sources) or resolve_direct_video_url(*sources) |
| 79 |