Implemented the full clean rewrite for Gitea

src/blackout_notifier/models.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass
+from datetime import date, datetime, time, timedelta
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import jdatetime
+
+TEHRAN = ZoneInfo("Asia/Tehran")
+
+
+class OutageDataError(ValueError):
+    """Raised when an outage payload cannot be normalized safely."""
+
+
+@dataclass(frozen=True, slots=True)
+class Outage:
+    outage_number: str
+    outage_date: str
+    start_time: str
+    stop_time: str
+    address: str
+    reason: str
+    is_planned: bool
+
+    @classmethod
+    def from_api(cls, payload: Any) -> Outage:
+        if not isinstance(payload, dict):
+            raise OutageDataError("outage entry must be an object")
+
+        outage_number = _required_text(payload, "outage_number")
+        outage_date = _required_text(payload, "outage_date")
+        start_time = _required_text(payload, "outage_start_time")
+        stop_time = _required_text(payload, "outage_stop_time")
+        address = str(payload.get("outage_address") or payload.get("address") or "").strip()
+        reason = str(payload.get("reason_outage") or "").strip()
+        is_planned = payload.get("is_planned", True)
+        if not isinstance(is_planned, bool):
+            raise OutageDataError("is_planned must be a boolean")
+
+        outage = cls(
+            outage_number=outage_number,
+            outage_date=outage_date,
+            start_time=start_time,
+            stop_time=stop_time,
+            address=address,
+            reason=reason,
+            is_planned=is_planned,
+        )
+        outage.start_datetime()
+        outage.stop_datetime()
+        return outage
+
+    @classmethod
+    def from_dict(cls, payload: Any) -> Outage:
+        if not isinstance(payload, dict):
+            raise OutageDataError("stored outage snapshot must be an object")
+        try:
+            outage = cls(
+                outage_number=str(payload["outage_number"]),
+                outage_date=str(payload["outage_date"]),
+                start_time=str(payload["start_time"]),
+                stop_time=str(payload["stop_time"]),
+                address=str(payload["address"]),
+                reason=str(payload["reason"]),
+                is_planned=payload["is_planned"],
+            )
+        except (KeyError, TypeError) as error:
+            raise OutageDataError("stored outage snapshot has an invalid schema") from error
+        if not isinstance(outage.is_planned, bool):
+            raise OutageDataError("stored is_planned must be a boolean")
+        outage.start_datetime()
+        outage.stop_datetime()
+        return outage
+
+    def to_dict(self) -> dict[str, str | bool]:
+        return asdict(self)
+
+    def fingerprint(self) -> str:
+        serialized = json.dumps(
+            self.to_dict(), ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+        return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+
+    def start_datetime(self) -> datetime:
+        return _jalali_datetime(self.outage_date, self.start_time)
+
+    def stop_datetime(self) -> datetime:
+        start = self.start_datetime()
+        stop = _jalali_datetime(self.outage_date, self.stop_time)
+        if stop <= start:
+            stop += timedelta(days=1)
+        return stop
+
+
+def jalali_query_window(now: datetime, lookahead_days: int) -> tuple[str, str]:
+    local_now = now.astimezone(TEHRAN)
+    start = jdatetime.date.fromgregorian(date=local_now.date())
+    end = jdatetime.date.fromgregorian(date=local_now.date() + timedelta(days=lookahead_days))
+    return start.strftime("%Y/%m/%d"), end.strftime("%Y/%m/%d")
+
+
+def tehran_now() -> datetime:
+    return datetime.now(TEHRAN)
+
+
+def _required_text(payload: dict[str, Any], field: str) -> str:
+    value = payload.get(field)
+    if value is None or not str(value).strip():
+        raise OutageDataError(f"{field} is required")
+    return str(value).strip()
+
+
+def _jalali_datetime(date_text: str, time_text: str) -> datetime:
+    try:
+        year, month, day = (int(part) for part in date_text.split("/"))
+        hour, minute = (int(part) for part in time_text.split(":"))
+        gregorian_date: date = jdatetime.date(year, month, day).togregorian()
+        parsed_time = time(hour, minute)
+    except (TypeError, ValueError) as error:
+        raise OutageDataError(f"invalid Jalali date/time: {date_text} {time_text}") from error
+    return datetime.combine(gregorian_date, parsed_time, tzinfo=TEHRAN)