diff --git a/frigate/analytics/state.py b/frigate/analytics/state.py new file mode 100644 index 0000000000..a321285ed5 --- /dev/null +++ b/frigate/analytics/state.py @@ -0,0 +1,85 @@ +"""The install's analytics identity and last attempt time, kept under /config.""" + +import json +import logging +import os +import re +from dataclasses import dataclass +from uuid import uuid4 + +from frigate.const import CONFIG_DIR + +logger = logging.getLogger(__name__) + +STATE_PATH = os.path.join(CONFIG_DIR, ".analytics.json") +INSTALL_ID_PATTERN = re.compile(r"[0-9a-f]{32}") + + +@dataclass(frozen=True) +class AnalyticsState: + install_id: str + last_attempt_at: float + + +def new_state() -> AnalyticsState: + return AnalyticsState(install_id=uuid4().hex, last_attempt_at=0.0) + + +def load_state(path: str = STATE_PATH) -> AnalyticsState | None: + """The saved state, or None when the file is missing or unusable.""" + try: + with open(path) as f: + data = json.load(f) + except FileNotFoundError: + return None + except (OSError, ValueError): + logger.warning("Ignoring unreadable analytics state at %s", path) + return None + + if not isinstance(data, dict): + return None + + install_id = data.get("install_id") + last_attempt_at = data.get("last_attempt_at") + + if ( + not isinstance(install_id, str) + or not INSTALL_ID_PATTERN.fullmatch(install_id) + or isinstance(last_attempt_at, bool) + or not isinstance(last_attempt_at, int | float) + ): + logger.warning("Ignoring invalid analytics state at %s", path) + return None + + return AnalyticsState(install_id=install_id, last_attempt_at=float(last_attempt_at)) + + +def save_state(state: AnalyticsState, path: str = STATE_PATH) -> bool: + """Write the state atomically, returning False when it can't be written.""" + temp_path = f"{path}.tmp" + + try: + with open(temp_path, "w") as f: + json.dump( + { + "install_id": state.install_id, + "last_attempt_at": state.last_attempt_at, + }, + f, + ) + + os.replace(temp_path, path) + except OSError: + return False + + return True + + +def delete_state(path: str = STATE_PATH) -> None: + """Forget the install ID, so a later opt-in starts a fresh identity.""" + try: + os.remove(path) + except FileNotFoundError: + pass + except OSError: + logger.warning("Unable to delete analytics state at %s", path) diff --git a/frigate/test/test_analytics_state.py b/frigate/test/test_analytics_state.py new file mode 100644 index 0000000000..24e45d3619 --- /dev/null +++ b/frigate/test/test_analytics_state.py @@ -0,0 +1,64 @@ +"""Tests for the analytics state file.""" + +import os +import tempfile +import unittest + +from frigate.analytics.state import ( + AnalyticsState, + delete_state, + load_state, + new_state, + save_state, +) + + +class TestAnalyticsState(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + self.addCleanup(self.dir.cleanup) + self.path = os.path.join(self.dir.name, ".analytics.json") + + def write(self, content: str) -> None: + with open(self.path, "w") as f: + f.write(content) + + def test_missing_file_loads_as_none(self): + self.assertIsNone(load_state(self.path)) + + def test_save_then_load_round_trips(self): + state = AnalyticsState(install_id="a" * 32, last_attempt_at=1790000000.5) + + self.assertTrue(save_state(state, self.path)) + self.assertEqual(load_state(self.path), state) + + def test_unusable_files_load_as_none(self): + for content in ( + "{not json", + '["a"]', + '{"install_id": "short", "last_attempt_at": 1}', + '{"install_id": "' + "a" * 32 + '"}', + '{"install_id": "' + "a" * 32 + '", "last_attempt_at": true}', + ): + with self.subTest(content=content): + self.write(content) + self.assertIsNone(load_state(self.path)) + + def test_save_reports_failure_instead_of_raising(self): + missing = os.path.join(self.dir.name, "missing", ".analytics.json") + + self.assertFalse(save_state(new_state(), missing)) + + def test_new_state_has_a_hex_id_and_no_attempt(self): + state = new_state() + + self.assertRegex(state.install_id, r"^[0-9a-f]{32}$") + self.assertEqual(state.last_attempt_at, 0.0) + + def test_delete_removes_the_file_and_tolerates_a_missing_one(self): + save_state(new_state(), self.path) + + delete_state(self.path) + delete_state(self.path) + + self.assertFalse(os.path.exists(self.path))