commit a7dadaf0d71c0db3aedb1d0c2a161c2da7909d69 Author: Mika Date: Sat Feb 14 15:31:58 2026 +0000 Add timestamp_logger/src/timestamp_logger/core.py diff --git a/timestamp_logger/src/timestamp_logger/core.py b/timestamp_logger/src/timestamp_logger/core.py new file mode 100644 index 0000000..389e8ce --- /dev/null +++ b/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