110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""
|
|
backend/routers/videos.py - Marketing video catalog for Sentinel live sessions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter
|
|
|
|
router = APIRouter()
|
|
|
|
VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".webm"}
|
|
DEFAULT_COLORS = ["#3b82f6", "#06b6d4", "#8b5cf6", "#10b981", "#f59e0b", "#ef4444"]
|
|
|
|
|
|
def _slugify(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
|
|
|
|
|
|
def _humanize(value: str) -> str:
|
|
base = re.sub(r"[-_]+", " ", value).strip()
|
|
return re.sub(r"\s+", " ", base).title()
|
|
|
|
|
|
def _derive_unit(name: str) -> str:
|
|
parts = re.findall(r"[A-Za-z0-9]+", name)
|
|
if not parts:
|
|
return "N/A"
|
|
if len(parts) >= 2:
|
|
return f"{parts[0]}-{parts[1]}"
|
|
return parts[0]
|
|
|
|
|
|
def _build_record(
|
|
*,
|
|
file_path: Path,
|
|
public_root: str,
|
|
metadata: dict[str, Any] | None,
|
|
color_index: int,
|
|
) -> dict[str, Any]:
|
|
rel_path = file_path.relative_to(public_root).as_posix()
|
|
name = metadata.get("title") if metadata else None
|
|
title = name or _humanize(file_path.stem.replace("video", "").replace("Video", ""))
|
|
property_name = metadata.get("property_name") if metadata else None
|
|
property_name = property_name or _humanize(file_path.parent.name if file_path.parent.name != "videos" else file_path.stem)
|
|
slug = metadata.get("id") if metadata else None
|
|
slug = slug or _slugify(file_path.stem)
|
|
|
|
return {
|
|
"id": slug,
|
|
"title": title,
|
|
"property_name": property_name,
|
|
"unit_number": (metadata or {}).get("unit_number") or _derive_unit(file_path.stem),
|
|
"type": (metadata or {}).get("type") or "Property Walkthrough",
|
|
"duration_seconds": int((metadata or {}).get("duration_seconds") or 0),
|
|
"video_url": f"/assets/{rel_path}",
|
|
"thumbnail_color": (metadata or {}).get("thumbnail_color") or DEFAULT_COLORS[color_index % len(DEFAULT_COLORS)],
|
|
}
|
|
|
|
|
|
@router.get("/marketing", summary="List marketing videos available for Sentinel live sessions")
|
|
async def list_marketing_videos() -> dict[str, Any]:
|
|
asset_root = Path(os.getenv("VELOCITY_ASSET_DIR", "/opt/dlami/nvme/assets"))
|
|
video_root = Path(os.getenv("VELOCITY_VIDEO_DIR", str(asset_root / "videos")))
|
|
catalog_path = video_root / "catalog.json"
|
|
|
|
catalog_entries: list[dict[str, Any]] = []
|
|
if catalog_path.exists():
|
|
catalog_entries = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
|
|
records: list[dict[str, Any]] = []
|
|
indexed: set[Path] = set()
|
|
for idx, entry in enumerate(catalog_entries):
|
|
file_path = video_root / entry["storage_path"]
|
|
if not file_path.exists():
|
|
continue
|
|
indexed.add(file_path.resolve())
|
|
records.append(
|
|
_build_record(
|
|
file_path=file_path,
|
|
public_root=str(asset_root),
|
|
metadata=entry,
|
|
color_index=idx,
|
|
)
|
|
)
|
|
|
|
unindexed_files = sorted(
|
|
[
|
|
path
|
|
for path in video_root.rglob("*")
|
|
if path.is_file() and path.suffix.lower() in VIDEO_EXTENSIONS and path.resolve() not in indexed
|
|
]
|
|
)
|
|
for idx, file_path in enumerate(unindexed_files, start=len(records)):
|
|
records.append(
|
|
_build_record(
|
|
file_path=file_path,
|
|
public_root=str(asset_root),
|
|
metadata=None,
|
|
color_index=idx,
|
|
)
|
|
)
|
|
|
|
return {"count": len(records), "videos": records}
|