Add timestamp_logger/src/timestamp_logger/core.py

timestamp_logger/src/timestamp_logger/core.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+import json
+from dataclasses import dataclass, asdict
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+__all__ = ["TimestampEntry", "log_timestamps"]
+
+
+class TimestampValidationError(ValueError):
+    """Wird ausgelöst, wenn ein ungültiger Zeitstempel übergeben wird."""
+
+
+@dataclass
+class TimestampEntry:
+    t_publish: datetime
+    t_gate_read: datetime
+    t_index_visible: datetime
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "t_publish": self.t_publish.isoformat(),
+            "t_gate_read": self.t_gate_read.isoformat(),
+            "t_index_visible": self.t_index_visible.isoformat(),
+        }
+
+
+LOG_PATH = Path("output/mess_log.jsonl")
+
+
+def _validate_datetime(value: Any) -> datetime:
+    if isinstance(value, datetime):
+        return value
+    if isinstance(value, str):
+        try:
+            # Python 3.11+: fromisoformat kann 'Z' nicht parsen, alternative 'replace'
+            if value.endswith("Z"):
+                value = value[:-1]
+            return datetime.fromisoformat(value)
+        except Exception as exc:
+            raise TimestampValidationError(f"Ungültiges Datumsformat: {value}") from exc
+    raise TimestampValidationError(f"Unerwarteter Typ für Zeitstempel: {type(value)}")
+
+
+def log_timestamps(t_publish: datetime, t_gate_read: datetime, t_index_visible: datetime) -> bool:
+    """Erfasst drei Zeitstempel und schreibt sie als JSONL-Eintrag."""
+    try:
+        t_publish_dt = _validate_datetime(t_publish)
+        t_gate_read_dt = _validate_datetime(t_gate_read)
+        t_index_visible_dt = _validate_datetime(t_index_visible)
+
+        entry = TimestampEntry(
+            t_publish=t_publish_dt,
+            t_gate_read=t_gate_read_dt,
+            t_index_visible=t_index_visible_dt,
+        )
+
+        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with LOG_PATH.open("a", encoding="utf-8") as f:
+            json_line = json.dumps(entry.to_dict(), ensure_ascii=False)
+            f.write(json_line + "\n")
+        assert LOG_PATH.exists(), "Log-Datei wurde nicht erstellt"
+        return True
+    except (OSError, TimestampValidationError, AssertionError) as err:
+        # Fehler protokollieren: stdout statt logging für minimale Komplexität
+        print(f"[ERROR] Timestamp logging failed: {err}")
+        return False