Anry DasandClaude Opus 5.5
22fc8b38bf
- Actually load configs/config.yaml at startup (init_config was commented out) - Add timeouts and error handling to all HTTP calls; run blocking requests in threads; serialize processing with a lock; run initial check in background - URL-encode Telegram messages and HTML-escape scraped data - Fix GPV crash on empty schedule, 00:00 end time, date sorting and fallback format - Parse PO table per row; fix daily reminder check and change detection - Use per-instance timestamps in dataclasses; robust config value parsing - Matrix: use v3 PUT send endpoint; strip secrets from logged errors - Resolve templates/static/test paths relative to the project; use lifespan - Move debug GPV data to test/data_gpv.json; add CLAUDE.md Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Optional, Self
|
|
|
|
import app_config as cfg
|
|
|
|
LOG = cfg.LOG
|
|
|
|
PO_DATETIME_FORMAT = '%Y.%m.%d %H:%M'
|
|
GPV_DATETIME_FORMAT = '%d.%m.%Y %H:%M'
|
|
GPV_DATE_FORMAT = '%d.%m.%Y'
|
|
|
|
|
|
def parse_ts(value: str, fmt: str) -> Optional[float]:
|
|
try:
|
|
return datetime.strptime(value, fmt).timestamp()
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class POData:
|
|
city: str = field(repr=False)
|
|
street: str = field(repr=False)
|
|
house: str = field(repr=False)
|
|
po_type: str = field(repr=False)
|
|
reason: str = field(repr=False)
|
|
placement_date: str
|
|
start_date: str
|
|
stop_date: str
|
|
start_time: str
|
|
stop_time: str
|
|
# service fields are not compared: equality means the same outage content
|
|
notified_at: Optional[datetime] = field(repr=False, default=None, compare=False)
|
|
updated_at: datetime = field(repr=False, default_factory=datetime.now, compare=False)
|
|
need_to_send: bool = field(repr=False, default=True, compare=False)
|
|
start_ts: float = field(init=False, default=0.0, compare=False)
|
|
|
|
def __post_init__(self):
|
|
ts = parse_ts(f'{self.start_date} {self.start_time}', PO_DATETIME_FORMAT)
|
|
if ts is None:
|
|
LOG.error(f'wrong start_date or start_time in {self}')
|
|
else:
|
|
self.start_ts = ts
|
|
|
|
@property
|
|
def key(self):
|
|
"""Identity of the outage: the same address and start time"""
|
|
return self.start_ts, self.city, self.street, self.house
|
|
|
|
def set_sent(self):
|
|
self.need_to_send = False
|
|
|
|
def set_notified_now(self):
|
|
self.notified_at = datetime.now()
|
|
|
|
def update_notified_at(self, old_data: Self):
|
|
self.notified_at = old_data.notified_at
|
|
|
|
|
|
@dataclass
|
|
class GPVData:
|
|
approved_from: str
|
|
event_date: str
|
|
hours_off: List
|
|
hours_on: List
|
|
outage_queue: str = field(repr=False)
|
|
notified_at: Optional[datetime] = field(repr=False, default=None, compare=False)
|
|
updated_at: datetime = field(repr=False, default_factory=datetime.now, compare=False)
|
|
need_to_send: bool = field(repr=False, default=True, compare=False)
|
|
# timestamps of power-on moments (ends of outages)
|
|
start_ts: List[float] = field(default_factory=list, init=False, compare=False)
|
|
|
|
def __post_init__(self):
|
|
day_start = parse_ts(f'{self.event_date} 00:00', GPV_DATETIME_FORMAT)
|
|
if day_start is None:
|
|
LOG.error(f'wrong event_date in {self}')
|
|
return
|
|
for hour_off, hour_on in zip(self.hours_off, self.hours_on):
|
|
ts_off = parse_ts(f'{self.event_date} {hour_off}', GPV_DATETIME_FORMAT)
|
|
ts_on = parse_ts(f'{self.event_date} {hour_on}', GPV_DATETIME_FORMAT)
|
|
if ts_on is None:
|
|
LOG.error(f'wrong time {hour_on!r} in {self}')
|
|
continue
|
|
if ts_off is not None and ts_on <= ts_off:
|
|
# e.g. '22:00-00:00': power is on at the next day
|
|
ts_on = (datetime.fromtimestamp(ts_on) + timedelta(days=1)).timestamp()
|
|
self.start_ts.append(ts_on)
|
|
|
|
@property
|
|
def key(self):
|
|
return self.event_date, self.outage_queue
|
|
|
|
@property
|
|
def actual_till_ts(self) -> Optional[float]:
|
|
"""Moment after which the schedule is not actual: the last power-on or end of the event day"""
|
|
if self.start_ts:
|
|
return max(self.start_ts)
|
|
day_start = parse_ts(f'{self.event_date} 00:00', GPV_DATETIME_FORMAT)
|
|
if day_start is None:
|
|
return None
|
|
return (datetime.fromtimestamp(day_start) + timedelta(days=1)).timestamp()
|
|
|
|
def is_same_schedule(self, other: Self) -> bool:
|
|
return self.hours_off == other.hours_off and self.hours_on == other.hours_on
|
|
|
|
def set_send(self):
|
|
self.need_to_send = False
|
|
self.notified_at = datetime.now()
|
|
|
|
|
|
@dataclass
|
|
class JobData:
|
|
name: str
|
|
trigger: str
|
|
next: Optional[datetime]
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pass
|