126 lines
4.4 KiB
Python
126 lines
4.4 KiB
Python
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)
|