diff --git a/frigate/api/export.py b/frigate/api/export.py index f0d93bc105..5873a565c8 100644 --- a/frigate/api/export.py +++ b/frigate/api/export.py @@ -12,6 +12,7 @@ from collections import deque from collections.abc import Iterator from pathlib import Path from urllib.parse import quote +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import psutil from fastapi import APIRouter, Depends, Query, Request @@ -81,7 +82,7 @@ from frigate.record.export import ( ) from frigate.util.path import sanitize_contained_path from frigate.util.recording_coverage import recording_overlap -from frigate.util.time import is_current_hour +from frigate.util.time import get_normalized_tz_name, is_current_hour logger = logging.getLogger(__name__) @@ -442,7 +443,29 @@ def _unique_archive_name(export: Export, used: set[str]) -> str: return candidate -def _stream_case_archive(exports: list[Export]) -> Iterator[bytes]: +def _archive_entry( + export: Export, arcname: str, tz_name: str | None +) -> zipfile.ZipInfo: + """Zip entry stamped with the export's recording start time. + + Zip times carry no zone and extractors read them as local time, so the + timestamp is converted with the UI timezone rather than the container's. + """ + tz = None + if tz_name: + try: + tz = ZoneInfo(get_normalized_tz_name(tz_name)) + except (ValueError, ZoneInfoNotFoundError): + pass + + start = datetime.datetime.fromtimestamp(float(export.date), tz=tz) + + # zip can't represent dates before 1980 + date_time = max(start.timetuple()[:6], (1980, 1, 1, 0, 0, 0)) + return zipfile.ZipInfo(arcname, date_time=date_time) + + +def _stream_case_archive(exports: list[Export], tz_name: str | None) -> Iterator[bytes]: """Yield bytes of a zip archive built from the given exports' mp4 files.""" buffer = _StreamingZipBuffer() used_names: set[str] = set() @@ -460,9 +483,10 @@ def _stream_case_archive(exports: list[Export]) -> Iterator[bytes]: continue arcname = _unique_archive_name(export, used_names) + zinfo = _archive_entry(export, arcname, tz_name) with ( - archive.open(arcname, mode="w", force_zip64=True) as entry, + archive.open(zinfo, mode="w", force_zip64=True) as entry, source.open("rb") as src, ): while True: @@ -501,6 +525,7 @@ def _content_disposition(filename: str, ascii_fallback: str) -> str: description="Streams a zip archive containing every completed export's mp4 for the given case.", ) def download_export_case( + request: Request, case_id: str, allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter), ): @@ -533,7 +558,7 @@ def download_export_case( archive_base = case_id return StreamingResponse( - _stream_case_archive(exports), + _stream_case_archive(exports, request.app.frigate_config.ui.timezone), media_type="application/zip", headers={ "Content-Disposition": _content_disposition( diff --git a/frigate/test/test_export.py b/frigate/test/test_export.py index 12b13477bb..8d4c688b19 100644 --- a/frigate/test/test_export.py +++ b/frigate/test/test_export.py @@ -4,7 +4,7 @@ import unittest from pathlib import Path from unittest.mock import MagicMock, patch -from frigate.api.export import _unique_archive_name +from frigate.api.export import _archive_entry, _unique_archive_name from frigate.const import MAX_PLAYLIST_SECONDS from frigate.models import Export from frigate.record.export import ( @@ -220,6 +220,38 @@ class TestUniqueArchiveName(unittest.TestCase): ) +class TestArchiveEntry(unittest.TestCase): + """Tests for the modified time stamped on case download zip entries.""" + + def build_export(self) -> Export: + return Export( + id="front_door_abc123", + camera="front_door", + name="front door", + date=1756000000.0, + video_path="/media/frigate/exports/front_door_abc123.mp4", + thumb_path="/media/frigate/exports/front_door_abc123.webp", + in_progress=False, + ) + + def test_uses_start_time_in_ui_timezone(self): + zinfo = _archive_entry(self.build_export(), "a.mp4", "America/New_York") + self.assertEqual(zinfo.date_time, (2025, 8, 23, 21, 46, 40)) + + def test_clamps_dates_before_1980(self): + export = self.build_export() + export.date = 100 + zinfo = _archive_entry(export, "a.mp4", "UTC") + self.assertEqual(zinfo.date_time, (1980, 1, 1, 0, 0, 0)) + + def test_invalid_timezone_falls_back_to_local_time(self): + zinfo = _archive_entry(self.build_export(), "a.mp4", "Not/A_Real_Zone") + self.assertEqual( + zinfo.date_time, + _archive_entry(self.build_export(), "a.mp4", None).date_time, + ) + + if __name__ == "__main__": unittest.main()