diff --git a/artifact_2/src/artifact_2/core.py b/artifact_2/src/artifact_2/core.py new file mode 100644 index 0000000..ffd0d8f --- /dev/null +++ b/artifact_2/src/artifact_2/core.py @@ -0,0 +1,85 @@ +from __future__ import annotations +import logging +from dataclasses import dataclass +from datetime import datetime, timezone, timedelta +from typing import List + + +logger = logging.getLogger(__name__) + + +@dataclass +class LogEntry: + """Repräsentiert einen einzelnen Log-Eintrag mit Zeitinformationen.""" + + epoch_ms: int + monotonic_ns: int + tz_offset_minutes: int + run_id: str + step_id: str + + def __post_init__(self) -> None: + # Eingabevalidierung (CI-Ready & Input Validation Required) + assert isinstance(self.epoch_ms, int) and self.epoch_ms >= 0, 'epoch_ms muss eine positive Ganzzahl sein.' + assert isinstance(self.monotonic_ns, int) and self.monotonic_ns >= 0, 'monotonic_ns muss eine positive Ganzzahl sein.' + assert isinstance(self.tz_offset_minutes, int), 'tz_offset_minutes muss eine Ganzzahl sein.' + assert isinstance(self.run_id, str) and self.run_id.strip(), 'run_id muss ein nicht-leerer String sein.' + assert isinstance(self.step_id, str) and self.step_id.strip(), 'step_id muss ein nicht-leerer String sein.' + + def to_datetime(self) -> datetime: + tz = timezone(timedelta(minutes=self.tz_offset_minutes)) + return datetime.fromtimestamp(self.epoch_ms / 1000.0, tz=tz) + + +def check_timestamp_consistency(log_entries: List[LogEntry]) -> bool: + """Prüft die Konsistenz von Zeitstempeln. + + Stellt sicher: + - epochale Zeit und monotone Zeit steigen monoton. + - höchstens ein Zeitzonenwechsel tritt auf. + - keine negativen Deltas. + + Rückgabe: + True, wenn konsistent; sonst False. + """ + + if not isinstance(log_entries, list) or not all(isinstance(e, LogEntry) for e in log_entries): + raise TypeError('log_entries muss eine Liste von LogEntry-Instanzen sein.') + + if not log_entries: + logger.info('Leere Eingabeliste – gilt als konsistent.') + return True + + # Nach Zeit sortieren zur Sicherheit + log_entries_sorted = sorted(log_entries, key=lambda e: e.epoch_ms) + tz_switches = set() + + prev_entry = log_entries_sorted[0] + previous_dt = prev_entry.to_datetime() + previous_mono = prev_entry.monotonic_ns + tz_switches.add(prev_entry.tz_offset_minutes) + + for entry in log_entries_sorted[1:]: + current_dt = entry.to_datetime() + current_mono = entry.monotonic_ns + + # Zeitzonenwechsel speichern + tz_switches.add(entry.tz_offset_minutes) + + # Zeitdifferenzen prüfen + delta_epoch = (current_dt - previous_dt).total_seconds() + delta_mono = current_mono - previous_mono + + if delta_epoch < 0 or delta_mono < 0: + logger.warning('Negative Zeitdifferenz erkannt zwischen %s und %s', previous_dt, current_dt) + return False + + previous_dt = current_dt + previous_mono = current_mono + + # Gültig: höchstens ein Zeitzonenwechsel + if len(tz_switches) > 2: + logger.warning('Mehrere Zeitzonenwechsel erkannt: %s', tz_switches) + return False + + return True