from __future__ import annotations from pathlib import Path from typing import Any, Iterable from .timeline import DEFAULT_TIMEZONE, format_beijing_time def seconds_to_timecode(seconds: float | int | None) -> str | None: if seconds is None: return None total_seconds = int(float(seconds)) hours = total_seconds // 3600 minutes = (total_seconds % 3600) // 60 remaining_seconds = total_seconds % 60 return f"{hours:02d}:{minutes:02d}:{remaining_seconds:02d}" def build_frame_records( video_id: str, output_dir: str | Path, frame_paths: Iterable[str | Path], *, frame_fps: float, timeline_start_epoch: float | int | str | None = None, timezone_name: str = DEFAULT_TIMEZONE, ) -> list[dict[str, Any]]: base_dir = Path(output_dir).expanduser().resolve(strict=False) records = [] for index, frame_path in enumerate(sorted(Path(path) for path in frame_paths), start=1): offset_seconds = round((index - 1) / frame_fps, 6) record = { "video_id": video_id, "frame_id": f"{video_id}_f{index:06d}", "frame_path": _relative_frame_path(frame_path, base_dir), "offset_seconds": offset_seconds, "timecode": seconds_to_timecode(offset_seconds), "pts_time": offset_seconds, "status": "sampled", "retry_count": 0, "last_error": None, } beijing_time = format_beijing_time( timeline_start_epoch, offset_seconds=offset_seconds, timezone_name=timezone_name, ) if beijing_time is not None: record["beijing_time"] = beijing_time records.append(record) return records def _relative_frame_path(frame_path: Path, base_dir: Path) -> str: resolved = frame_path.expanduser().resolve(strict=False) try: return resolved.relative_to(base_dir).as_posix() except ValueError: return resolved.as_posix()