diff --git a/.gitignore b/.gitignore index 23e3384..fa132c1 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ /configs/secrets.yaml +__pycache__/ +.idea/ +.venv/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..f058a6c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,49 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## What this is + +FastAPI service that polls the Ivano-Frankivsk energy company (oe.if.ua) for power outages at a configured address/account and pushes notifications to Telegram and (optionally) Matrix. Two independent data sources: + +- **PO** (planned outages) — HTML table scraped from `https://oe.if.ua/uk/shutdowns_table`, parsed with lxml XPath per row (`div[@class='table-row flex']`). +- **GPV** (hourly outage schedule) — JSON from `https://be-svitlo.oe.if.ua/schedule-by-search` by account number (requested with `verify=False`). + +Messages and log text are in Ukrainian; keep that. + +## Commands + +```bash +.venv/bin/pip install -r requirements.txt +.venv/bin/python main.py # runs uvicorn with host/port from configs/config.yaml +``` + +There is no test suite or linter configured. `test/` holds fixtures only (`data_po_2days_*.html`, `data_gpv.json`). To exercise the app without contacting real sites or messengers, set `application.debug: "true"` in `configs/config.yaml`: fetches then read the fixtures, and senders `print()` instead of sending. Note fixtures are dated April 2026, so entries are pruned as expired unless dates are shifted. + +## Configuration + +- `configs/config.yaml` is loaded at import of `main.py` via `cfg.init_config()`; values use `!secret ` resolved from `configs/secrets.yaml` (gitignored, must exist for a real run). The `!secret` tag is registered on a private loader subclass in `config_file.read_yaml`. +- `app_config.py` exposes settings as **module-level globals** (`cfg.TG_TOKEN`, `cfg.IS_DEBUG`, …) populated by `parse_config()`. Always access them as `cfg.X` at call time — `from app_config import X` would capture the pre-config default. +- Booleans in YAML may be strings (`"true"`) — parse with `as_bool`. +- `configs/log.yaml` is the uvicorn log config. All modules log through the single `cfg.LOG` logger, which also feeds the in-memory `LOG_LIST` served by `/api/v1/logs` (unauthenticated — never log secrets; use `utils.safe_error(e)` for request exceptions, since TG URLs contain the token). + +## Architecture + +- `main.py` — FastAPI `app` with a sub-app `api` mounted at `/api/v1` (the root HTML page in `templates/info.html` lists `api` routes). `lifespan` creates an `AsyncIOScheduler` with one cron job per (hour in `[start_hour, stop_hour)`, minute in `PROCESSING_MINUTES`) — every 2 minutes in debug — and runs an initial check as a background task. `check_and_notify()` runs `process_gpv` then `process_po` under `PROCESS_LOCK` so scheduled runs and `/update` never overlap (prevents duplicate notifications). +- `utils.py` — fetching, parsing, state and notification. State is in-memory only (`PARSED_PO_DATAS`, `PARSED_GPV_DATAS`); lost on restart, so everything current is re-notified after a restart. Blocking `requests` calls are run via `asyncio.to_thread`; every request has `cfg.HTTP_TIMEOUT`; `process_*` catch network/parse errors and return status codes instead of raising. +- `model.py` — `POData` / `GPVData` dataclasses. Service fields (`notified_at`, `updated_at`, `need_to_send`, `start_ts`) are `compare=False`, so `==` means "same outage content"; this drives change detection. + +### Notification lifecycle (per poll) + +1. Parse fresh entries → `update_*_data` merges into the list by `key` (PO: start ts + address; GPV: event date + queue). + - not yet sent → replaced with fresh data; + - already sent and content changed (PO `!=`, GPV `is_same_schedule`) → replaced, so it is re-sent; + - PO only: also re-sent by `is_reminder_needed` (a day since last notification, or the first processing slot of the day passed). +2. `remove_old_*_data` prunes expired entries (PO: 1h after start; GPV: 30 min after the last power-on / end of day). GPV `start_ts` holds power-**on** timestamps despite the name; `00:00` end rolls to the next day. +3. `check_and_send_*` sends entries with `need_to_send` via `notify()`, which sends to TG and, if enabled, Matrix (v3 `PUT` with txn id); an entry is marked sent if at least one channel succeeded. + +Messages use Telegram HTML `parse_mode` — escape any scraped text with `html.escape`. + +## Constraints + +- Don't change the public API (paths, params, response shapes): `/`, `/api/v1/status`, `/update`, `/health`, `/test_message/{tg|mx}`, `/logs`. Dataclasses in `/status` are serialized as-is, so renaming their fields changes the API. diff --git a/app_config.py b/app_config.py index 10b8f93..8363ad9 100644 --- a/app_config.py +++ b/app_config.py @@ -7,13 +7,18 @@ import uvicorn from config_file import read_yaml -APP_VERSION="1.0" -SCRIPT_PATH = os.path.dirname(__file__) -CONFIGS_DIR = SCRIPT_PATH + "/configs" -CONFIG_FILE_NAME = CONFIGS_DIR + "/config.yaml" +APP_VERSION = "1.0" +SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) +CONFIGS_DIR = os.path.join(SCRIPT_PATH, "configs") +CONFIG_FILE_NAME = os.path.join(CONFIGS_DIR, "config.yaml") +TEMPLATES_DIR = os.path.join(SCRIPT_PATH, "templates") +STATIC_DIR = os.path.join(SCRIPT_PATH, "static") +TEST_DIR = os.path.join(SCRIPT_PATH, "test") SECRETS_NAME = 'secrets.yaml' CONFIG: Dict[str, Any] = {} +LOG_FORMAT = '%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s' + API_KEY = '' IN_MEMORY_LOGS = False IN_MEMORY_LOGS_LEN = 100 @@ -35,6 +40,8 @@ APARTMENT = "" PROCESS_START_HOUR = 6 PROCESS_STOP_HOUR = 23 PROCESSING_MINUTES: List[int] = [5, 15, 30, 45, 55] +HTTP_TIMEOUT = 30 # seconds, for all outgoing requests + def read_app_config(): global CONFIG @@ -46,8 +53,38 @@ def read_app_config(): CONFIG = read_yaml(CONFIG_FILE_NAME, SECRETS_NAME) -def get_config_value(cfg, section, key, default): - return cfg[section][key] if section in cfg else default + +def get_section(cfg: Dict[str, Any], *path: str) -> Dict[str, Any]: + """Returns nested section of config or empty dict if it is absent or not a mapping""" + section = cfg + for key in path: + section = section.get(key) if isinstance(section, dict) else None + if not isinstance(section, dict): + return {} + return section + + +def as_bool(value: Any) -> bool: + if isinstance(value, bool): + return value + return str(value).strip().lower() in ('true', 'yes', 'on', '1') + + +def as_int(value: Any, default: int, min_value: int = None, max_value: int = None) -> int: + try: + res = int(value) + except (TypeError, ValueError): + LOG.warning(f'Wrong integer value in config: {value!r}, default {default} is used') + return default + if (min_value is not None and res < min_value) or (max_value is not None and res > max_value): + LOG.warning(f'Value {res} is out of range [{min_value}, {max_value}], default {default} is used') + return default + return res + + +def as_str(value: Any) -> str: + return '' if value is None else str(value) + def parse_config(): global API_KEY, IN_MEMORY_LOGS, IN_MEMORY_LOGS_LEN, HOST, PORT, IS_DEBUG, \ @@ -55,42 +92,51 @@ def parse_config(): USE_MATRIX, MX_SERVER, MX_ROOM, MX_TOKEN, \ ACCOUNT, SETTLEMENT, STREET, HOUSE, BUILDING_PART_NUMBER, APARTMENT, \ PROCESS_START_HOUR, PROCESS_STOP_HOUR - HOST = CONFIG.get('application', {}).get('host', '') - if not HOST: - HOST = '0.0.0.0' - PORT = CONFIG.get('application', {}).get('port', 8000) - API_KEY = CONFIG.get('application').get('api-key', '') - IN_MEMORY_LOGS = str(CONFIG.get('application', {}).get('logs_ep', {}).get('enabled', False)).lower() == 'true' - IN_MEMORY_LOGS_LEN = CONFIG.get('application', {}).get('logs_ep', {}).get('max_records', 100) - IS_DEBUG = CONFIG.get('application').get('debug', False) == 'true' - TG_TOKEN = CONFIG.get('notifications', {}).get('tg', {}).get('token', '') - TG_CHAT = CONFIG.get('notifications', {}).get('tg', {}).get('chat_id', '') - USE_MATRIX = str(CONFIG.get('notifications', {}).get('matrix', {}).get('use_matrix', False)).lower() == 'true' - MX_SERVER = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_server', '') - MX_ROOM = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_room_id', '') - MX_TOKEN = CONFIG.get('notifications', {}).get('matrix', {}).get('mx_access_token', '') - ACCOUNT = CONFIG.get('data', {}).get('account', {}) - SETTLEMENT = CONFIG.get('data', {}).get('settlement', {}) - STREET = CONFIG.get('data', {}).get('street', {}) - HOUSE = CONFIG.get('data', {}).get('house', {}) - BUILDING_PART_NUMBER = CONFIG.get('data', {}).get('building_part_number', {}) - APARTMENT = CONFIG.get('data', {}).get('apartment', {}) - PROCESS_START_HOUR = CONFIG.get('application').get('process', {}).get('start_hour', 6) - PROCESS_STOP_HOUR = CONFIG.get('application').get('process', {}).get('stop_hour', 23) - if PROCESS_START_HOUR > PROCESS_STOP_HOUR: - PROCESS_STOP_HOUR = 23 + app_section = get_section(CONFIG, 'application') + logs_section = get_section(app_section, 'logs_ep') + process_section = get_section(app_section, 'process') + tg_section = get_section(CONFIG, 'notifications', 'tg') + mx_section = get_section(CONFIG, 'notifications', 'matrix') + data_section = get_section(CONFIG, 'data') + + HOST = as_str(app_section.get('host')) or '0.0.0.0' + PORT = as_int(app_section.get('port', 8000), 8000, 1, 65535) + API_KEY = as_str(app_section.get('api-key')) + IN_MEMORY_LOGS = as_bool(logs_section.get('enabled', False)) + IN_MEMORY_LOGS_LEN = as_int(logs_section.get('max_records', 100), 100, 1) + IS_DEBUG = as_bool(app_section.get('debug', False)) + TG_TOKEN = as_str(tg_section.get('token')) + TG_CHAT = as_str(tg_section.get('chat_id')) + USE_MATRIX = as_bool(mx_section.get('use_matrix', False)) + MX_SERVER = as_str(mx_section.get('mx_server')).rstrip('/') + MX_ROOM = as_str(mx_section.get('mx_room_id')) + MX_TOKEN = as_str(mx_section.get('mx_access_token')) + ACCOUNT = as_str(data_section.get('account')) + SETTLEMENT = as_str(data_section.get('settlement')) + STREET = as_str(data_section.get('street')) + HOUSE = as_str(data_section.get('house')) + BUILDING_PART_NUMBER = as_str(data_section.get('building_part_number')) + APARTMENT = as_str(data_section.get('apartment')) + PROCESS_START_HOUR = as_int(process_section.get('start_hour', 6), 6, 0, 23) + PROCESS_STOP_HOUR = as_int(process_section.get('stop_hour', 23), 23, 1, 24) + if PROCESS_START_HOUR >= PROCESS_STOP_HOUR: + LOG.warning(f'start_hour {PROCESS_START_HOUR} >= stop_hour {PROCESS_STOP_HOUR}, stop_hour 24 is used') + PROCESS_STOP_HOUR = 24 + + if not TG_TOKEN or not TG_CHAT: + LOG.warning('Telegram token or chat_id is not configured') + if USE_MATRIX and not (MX_SERVER and MX_ROOM and MX_TOKEN): + LOG.warning('Matrix is enabled but not fully configured') def get_uvicorn_config(): - # global HOST, PORT if HOST and PORT: cfg = { "host": f"{HOST}", "port": PORT, "access_log": True, "log_level": "info", - "log_config": f"{CONFIGS_DIR}/log.yaml" - # "docs_url": "/api-docs" + "log_config": os.path.join(CONFIGS_DIR, "log.yaml") } else: cfg = { @@ -99,35 +145,41 @@ def get_uvicorn_config(): "reload": True, "access_log": False, "log_level": "info", - "log_config": f"{CONFIGS_DIR}/log.yaml" - # "docs_url": "/api-docs" + "log_config": os.path.join(CONFIGS_DIR, "log.yaml") } LOG.debug(f'got uvicorn_config: {cfg}') return cfg -def log_config(): # Not used due to not working +def log_config(): # Not used due to not working cfg = uvicorn.config.LOGGING_CONFIG - cfg["formatters"]["access"]["fmt"] = ( - '%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s' - ) + cfg["formatters"]["access"]["fmt"] = LOG_FORMAT return cfg -LOG_LIST = [] + +LOG_LIST: List[str] = [] + class ListHandler(logging.Handler): def emit(self, record): - if IN_MEMORY_LOGS: + if not IN_MEMORY_LOGS: + return + try: LOG_LIST.append(self.format(record)) - while len(LOG_LIST) > IN_MEMORY_LOGS_LEN: - LOG_LIST.pop(0) + if len(LOG_LIST) > IN_MEMORY_LOGS_LEN: + del LOG_LIST[:len(LOG_LIST) - IN_MEMORY_LOGS_LEN] + except Exception: + self.handleError(record) + + ### LOG = logging.getLogger(__name__) handler = ListHandler() -handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s')) +handler.setFormatter(logging.Formatter(LOG_FORMAT)) LOG.addHandler(handler) + def init_config(): read_app_config() parse_config() diff --git a/config_file.py b/config_file.py index 0654096..cf37f0b 100644 --- a/config_file.py +++ b/config_file.py @@ -1,43 +1,43 @@ import json import os + import yaml + def read_config(name): if not os.path.isfile(name): - raise Exception(f"File {name} doesn't exists") - filename, ext = os.path.splitext(name) - if 'json' in ext: + raise FileNotFoundError(f"File {name} doesn't exist") + _, ext = os.path.splitext(name) + ext = ext.lower() + if ext == '.json': return read_json(name) - elif 'properties' in ext: + elif ext == '.properties': return read_prop(name) - elif 'yaml' in ext or 'yml' in ext: + elif ext in ('.yaml', '.yml'): return read_yaml(name) else: - raise Exception("Wrong file type") + raise ValueError(f"Wrong file type: {ext}") + def read_json(name): with open(name, 'r', encoding='utf-8') as f: - j_conf = json.load(f) - conf = {} - for key, value in j_conf.items(): - conf[key] = value - return conf + return dict(json.load(f)) + def read_prop(filepath, sep='=', comment_char='#'): """ Read the file passed as parameter as a properties file. """ conf = {} - with open(filepath, "rt") as f: + with open(filepath, "rt", encoding='utf-8') as f: for line in f: l = line.strip() if l and not l.startswith(comment_char): - key_value = l.split(sep) - key = key_value[0].strip() - value = sep.join(key_value[1:]).strip().strip('"') - conf[key] = value + key, _, value = l.partition(sep) + conf[key.strip()] = value.strip().strip('"') return conf + def read_yaml(name, secrets_file='secrets.yaml'): # Load secrets first secrets = {} @@ -47,28 +47,24 @@ def read_yaml(name, secrets_file='secrets.yaml'): with open(secrets_path, 'r', encoding='utf-8') as f: secrets = yaml.safe_load(f) or {} - # Define a custom constructor for !secret tag def secret_constructor(loader, node): secret_key = loader.construct_scalar(node) if secret_key not in secrets: raise ValueError(f"Secret '{secret_key}' not found in {secrets_file}") return secrets[secret_key] - # Register the custom constructor - yaml.add_constructor('!secret', secret_constructor, Loader=yaml.SafeLoader) + # Own loader class, so the global yaml.SafeLoader is not modified + class SecretLoader(yaml.SafeLoader): + pass + + SecretLoader.add_constructor('!secret', secret_constructor) # Load the main configuration - conf = {} with open(name, 'r', encoding='utf-8') as f: - y_conf = yaml.safe_load(f) - if y_conf: - for key, value in y_conf.items(): - conf[key] = value + y_conf = yaml.load(f, Loader=SecretLoader) - return conf - -def main(): - pass - -if __name__ == "__main__": - main() \ No newline at end of file + if y_conf is None: + return {} + if not isinstance(y_conf, dict): + raise ValueError(f"Wrong config structure in {name}: top level must be a mapping") + return dict(y_conf) diff --git a/main.py b/main.py index 8249b74..9fe6844 100644 --- a/main.py +++ b/main.py @@ -1,16 +1,19 @@ +import asyncio import logging +import os +from contextlib import asynccontextmanager, suppress from datetime import datetime, UTC from typing import Optional, List import uvicorn from apscheduler.schedulers.asyncio import AsyncIOScheduler -from fastapi import FastAPI, BackgroundTasks, Request +from fastapi import FastAPI, Request from starlette.responses import FileResponse, HTMLResponse from starlette.routing import Mount from starlette.templating import Jinja2Templates import app_config as cfg -from model import POData, JobData +from model import JobData from utils import process_gpv, process_po, get_parsed_po_data, get_parsed_gpv_data, send_message_to_tg_get, \ send_matrix_notification @@ -26,38 +29,33 @@ Created by -=:dAs:=- ''' ######################## -# cfg.init_config() ToDo: TEST IT! +cfg.init_config() SCHEDULER: Optional[AsyncIOScheduler] = None LOG = cfg.LOG -TEMPLATES = Jinja2Templates(directory="templates") -######################## -TAGS_METADATA = [] -app = FastAPI( - title="Energy Outage Inform Service by -=:dAs:=-", - description=DESCRIPTION, - version=cfg.APP_VERSION, - license_info={ - "name": "Apache 2.0", - "url": "https://www.apache.org/licenses/LICENSE-2.0.html", - }, - openapi_tags=TAGS_METADATA -) -api = FastAPI() -app.mount("/api/v1", api) +TEMPLATES = Jinja2Templates(directory=cfg.TEMPLATES_DIR) +# prevents simultaneous processing (scheduled job and /update) and duplicated notifications +PROCESS_LOCK = asyncio.Lock() +INITIAL_CHECK_TASK: Optional[asyncio.Task] = None ######################## + + async def check_and_notify(): - await process_gpv() - await process_po() + async with PROCESS_LOCK: + for process in (process_gpv, process_po): + try: + await process() + except Exception: + LOG.exception(f'Unexpected error in {process.__name__}') -@app.on_event("startup") -async def startup_event(): - global SCHEDULER +@asynccontextmanager +async def lifespan(_: FastAPI): + global SCHEDULER, INITIAL_CHECK_TASK # --- STARTUP LOGIC --- LOG.info("--- Starting EO Notifier...") - SCHEDULER = AsyncIOScheduler() + SCHEDULER = AsyncIOScheduler(job_defaults={'coalesce': True, 'max_instances': 1, 'misfire_grace_time': 60}) + mins = cfg.PROCESSING_MINUTES if not cfg.IS_DEBUG else list(range(0, 60, 2)) for h in range(cfg.PROCESS_START_HOUR, cfg.PROCESS_STOP_HOUR): - mins = cfg.PROCESSING_MINUTES if not cfg.IS_DEBUG else [x for x in range(0, 60) if x % 2 == 0] for m in mins: SCHEDULER.add_job( check_and_notify, @@ -71,27 +69,54 @@ async def startup_event(): SCHEDULER.start() LOG.info("Scheduler started") - # Run initial check - await check_and_notify() + # Run initial check in background, so service starts even if data sources are unavailable + INITIAL_CHECK_TASK = asyncio.create_task(check_and_notify()) + yield -@app.on_event("shutdown") -async def shutdown_event(): # --- SHUTDOWN LOGIC --- - if SCHEDULER: - SCHEDULER.shutdown() + if INITIAL_CHECK_TASK and not INITIAL_CHECK_TASK.done(): + INITIAL_CHECK_TASK.cancel() + with suppress(asyncio.CancelledError): + await INITIAL_CHECK_TASK + + if SCHEDULER and SCHEDULER.running: + SCHEDULER.shutdown(wait=False) LOG.info('Scheduler stopped') LOG.info('--- EO Notifier shutting down') +TAGS_METADATA = [] +app = FastAPI( + title="Energy Outage Inform Service by -=:dAs:=-", + description=DESCRIPTION, + version=cfg.APP_VERSION, + license_info={ + "name": "Apache 2.0", + "url": "https://www.apache.org/licenses/LICENSE-2.0.html", + }, + openapi_tags=TAGS_METADATA, + lifespan=lifespan +) +api = FastAPI() +app.mount("/api/v1", api) + + +######################## + + +def is_scheduler_running() -> bool: + return bool(SCHEDULER and SCHEDULER.running) + + @app.get("/favicon.ico", include_in_schema=False) async def get_favicon(): - return FileResponse('static/favicon.png') + return FileResponse(os.path.join(cfg.STATIC_DIR, 'favicon.png')) def list_endpoints(): - url_list = [r for r in api.routes if 'HEAD' not in r.methods] + url_list = [r for r in api.routes if 'HEAD' not in (getattr(r, 'methods', None) or {'HEAD'})] return url_list @@ -108,12 +133,11 @@ async def root(request: Request): "Service": "Energy Outage Inform Service by -=:dAs:=", "version": f"{cfg.APP_VERSION}", "status": "running", - "Scheduler running": SCHEDULER.running, + "Scheduler running": is_scheduler_running(), "timestamp": date_format(datetime.now(UTC)) } - return TEMPLATES.TemplateResponse("info.html", { - "request": request, + return TEMPLATES.TemplateResponse(request, "info.html", { "help_messages": help_messages, "title": "Energy Outage Inform Service by -=:dAs:=", "EPs": list_endpoints(), @@ -128,13 +152,12 @@ async def root(request: Request): description='Showing current status of the service') async def get_status(): """Detailed health check""" - global SCHEDULER return { "status": "healthy", "config_loaded": bool(cfg.CONFIG), "PO_data": get_parsed_po_data(), "GPV_data": get_parsed_gpv_data(), - "scheduler_running": SCHEDULER.running, + "scheduler_running": is_scheduler_running(), "scheduler_jobs": await get_scheduled_jobs() } @@ -149,17 +172,9 @@ async def update(): async def get_scheduled_jobs() -> List[JobData]: - global SCHEDULER - jobs: List[JobData] = [] - for j in SCHEDULER.get_jobs(): - jobs.append( - JobData( - name=j.id, - trigger=str(j.trigger), - next=j.next_run_time) - ) - - return jobs + if not SCHEDULER: + return [] + return [JobData(name=j.id, trigger=str(j.trigger), next=j.next_run_time) for j in SCHEDULER.get_jobs()] @api.get("/health", @@ -174,17 +189,21 @@ async def health_check(): @api.get("/test_message/{messanger}", summary='Send message to messanger', description='Send test message to messanger according to current configuration') -async def send_test_message(messanger: str = None): +async def send_test_message(messanger: str): messanger_up = messanger.upper() msg = f'Test message to {messanger_up}' if messanger_up == "TG": - send_message_to_tg_get(msg) + is_sent = await asyncio.to_thread(send_message_to_tg_get, msg) == 200 elif messanger_up == "MX": - send_matrix_notification(msg) + is_sent = await asyncio.to_thread(send_matrix_notification, msg) else: return { "status": f"Error: Wrong messanger '{messanger_up}'" } + if not is_sent: + return { + "status": f"Error: message to '{messanger_up}' was not sent" + } return { "status": "OK" } @@ -194,7 +213,7 @@ async def send_test_message(messanger: str = None): summary='Show LOGs', description='Showing Service''s LOGs if it configured') async def get_logs(): - return {'logs': cfg.LOG_LIST} + return {'logs': list(cfg.LOG_LIST)} def date_format(dt: datetime, fmt: str = '%Y-%m-%d %H:%M:%S'): @@ -204,7 +223,7 @@ def date_format(dt: datetime, fmt: str = '%Y-%m-%d %H:%M:%S'): def main(): logging.basicConfig( level=logging.INFO, - format='%(asctime)s [%(levelname)s] - %(name)s: %(funcName)s[%(lineno)d] - %(message)s' + format=cfg.LOG_FORMAT ) cfg.LOG.info('Application starting') uvicorn.run("main:app", **cfg.get_uvicorn_config()) diff --git a/model.py b/model.py index 6e74174..4a3e911 100644 --- a/model.py +++ b/model.py @@ -1,15 +1,28 @@ from dataclasses import dataclass, field -from datetime import datetime -from typing import List, Self +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: int = field(repr=False) + house: str = field(repr=False) po_type: str = field(repr=False) reason: str = field(repr=False) placement_date: str @@ -17,16 +30,23 @@ class POData: stop_date: str start_time: str stop_time: str - notified_at: datetime = field(repr=False, default=None) - updated_at: datetime = field(repr=False, default=datetime.now()) - need_to_send: bool = field(repr=False, default=True) - start_ts: float = field(init=False) + # 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): - if self.start_date and self.start_time: - self.start_ts = datetime.strptime(f'{self.start_date} {self.start_time}', '%Y.%m.%d %H:%M').timestamp() - else: + 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 @@ -45,17 +65,44 @@ class GPVData: hours_off: List hours_on: List outage_queue: str = field(repr=False) - notified_at: datetime = field(repr=False, default=None) - updated_at: datetime = field(repr=False, default=datetime.now()) - need_to_send: bool = field(repr=False, default=True) - start_ts: List[float] = field(default_factory=list, init=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): - if self.event_date and self.hours_off and self.hours_on: - for index, value in enumerate(self.hours_on): - self.start_ts.append(datetime.strptime(f'{self.event_date} {self.hours_on[index]}', '%d.%m.%Y %H:%M').timestamp()) - else: - LOG.error(f'wrong start_date or start_time in {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 @@ -66,7 +113,7 @@ class GPVData: class JobData: name: str trigger: str - next: str + next: Optional[datetime] if __name__ == '__main__': diff --git a/test/data_gpv.json b/test/data_gpv.json new file mode 100644 index 0000000..1966e23 --- /dev/null +++ b/test/data_gpv.json @@ -0,0 +1,66 @@ +{ + "current": { + "gav": { + "message": "Черга спеціальних аварійних відключень (СГАВ):", + "queue": null + }, + "hasQueue": "yes", + "note": "Станом на 21:05 14.01.2026 за вказаним особовим рахунком '30014180' споживач підпадає під чергу '5.1' Графіку погодинного відключення(ГПВ)", + "queue": 5, + "sgav": { + "message": "Черга спеціальних аварійних відключень (СГАВ):", + "queue": null + }, + "subQueue": 1 + }, + "schedule": [ + { + "createdAt": "14.01.2026 19:11", + "eventDate": "15.04.2026", + "queues": { + "5.1": [ + { + "from": "07:30", + "shutdownHours": "07:30-11:00", + "status": 1, + "to": "11:00" + }, + { + "from": "14:30", + "shutdownHours": "14:30-20:00", + "status": 1, + "to": "20:00" + }, + { + "from": "22:00", + "shutdownHours": "22:00-00:00", + "status": 1, + "to": "00:00" + } + ] + }, + "scheduleApprovedSince": "14.04.2026 19:11" + }, + { + "createdAt": "13.04.2026 20:28", + "eventDate": "14.04.2026", + "queues": { + "5.1": [ + { + "from": "03:00", + "shutdownHours": "03:00-07:00", + "status": 1, + "to": "07:00" + }, + { + "from": "11:00", + "shutdownHours": "11:00-17:30", + "status": 1, + "to": "17:30" + } + ] + }, + "scheduleApprovedSince": "14.04.2026 09:56" + } + ] +} \ No newline at end of file diff --git a/utils.py b/utils.py index 770f8bd..f30141f 100644 --- a/utils.py +++ b/utils.py @@ -1,15 +1,19 @@ +import asyncio +import html import json import os +import re +import uuid from datetime import datetime, timedelta -from io import StringIO -from typing import List +from typing import List, Optional +from urllib.parse import quote import requests import urllib3 from lxml import etree import app_config as cfg -from model import POData, GPVData +from model import POData, GPVData, GPV_DATE_FORMAT STATUS_SEND_OK = 0 STATUS_ERROR_SEND_TO_TG = 20 @@ -23,24 +27,59 @@ PARSED_GPV_DATAS: List[GPVData] = [] LOG = cfg.LOG -def send_message_to_tg_get(msg: str): - url = f"https://api.telegram.org/bot{cfg.TG_TOKEN}/sendMessage?chat_id={cfg.TG_CHAT}&parse_mode=html&text={msg}" - if not cfg.IS_DEBUG: - return requests.get(url).status_code - else: +# be-svitlo.oe.if.ua is requested without certificate verification +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +DIGIT_CHARS = [f'{d}\U000020E3' for d in range(10)] + + +def safe_error(e: Exception) -> str: + """Error description without secrets (request URLs may contain the TG token)""" + msg = f'{type(e).__name__}: {e}' + for secret in (cfg.TG_TOKEN, cfg.MX_TOKEN): + if secret: + msg = msg.replace(secret, '***') + return msg + + +def send_message_to_tg_get(msg: str) -> int: + """Returns HTTP status code of Telegram response or 0 if the request failed""" + if cfg.IS_DEBUG: print(f"Send to TG GET: {msg}") return 200 + if not cfg.TG_TOKEN or not cfg.TG_CHAT: + LOG.error('Telegram is not configured') + return 0 + try: + response = requests.get( + f"https://api.telegram.org/bot{cfg.TG_TOKEN}/sendMessage", + params={'chat_id': cfg.TG_CHAT, 'parse_mode': 'html', 'text': msg}, + timeout=cfg.HTTP_TIMEOUT) + except requests.RequestException as e: + LOG.error(f'Error sending TG message: {safe_error(e)}') + return 0 + if response.status_code != 200: + LOG.error(f'TG response code={response.status_code}: {response.text[:500]}') + return response.status_code -def send_message_to_tg_post(msg: str): - if not cfg.IS_DEBUG: - return requests.post( - url='https://api.telegram.org/bot{0}/{1}'.format(cfg.TG_TOKEN, 'sendMessage'), - data={'chat_id': {cfg.TG_CHAT}, 'text': {msg}, 'parse_mode': 'html'} - ).status_code - else: +def send_message_to_tg_post(msg: str) -> int: + """Returns HTTP status code of Telegram response or 0 if the request failed""" + if cfg.IS_DEBUG: print(f"Send to TG POST: {msg}") return 200 + if not cfg.TG_TOKEN or not cfg.TG_CHAT: + LOG.error('Telegram is not configured') + return 0 + try: + return requests.post( + url=f'https://api.telegram.org/bot{cfg.TG_TOKEN}/sendMessage', + data={'chat_id': cfg.TG_CHAT, 'text': msg, 'parse_mode': 'html'}, + timeout=cfg.HTTP_TIMEOUT + ).status_code + except requests.RequestException as e: + LOG.error(f'Error sending TG message: {safe_error(e)}') + return 0 def send_matrix_notification(message: str) -> bool: @@ -48,31 +87,52 @@ def send_matrix_notification(message: str) -> bool: if not cfg.USE_MATRIX: return False - if not cfg.IS_DEBUG: - try: - url = f"{cfg.MX_SERVER}/_matrix/client/r0/rooms/{cfg.MX_ROOM}/send/m.room.message" - - headers = { - 'Authorization': f'Bearer {cfg.MX_TOKEN}', - 'Content-Type': 'application/json' - } - - payload = { - 'msgtype': 'm.text', - 'body': message, - 'format': 'org.matrix.custom.html', - 'formatted_body': message.replace('\n', '
') - } - - response = requests.post(url, headers=headers, json=payload) - return response.status_code == 200 - except Exception as e: - # logger.error(f"Error sending Matrix notification: {e}") - return False - else: + if cfg.IS_DEBUG: print(f"Send to MX: {message}") return True + if not (cfg.MX_SERVER and cfg.MX_ROOM and cfg.MX_TOKEN): + LOG.error('Matrix is not configured') + return False + + url = (f"{cfg.MX_SERVER}/_matrix/client/v3/rooms/{quote(cfg.MX_ROOM, safe='')}" + f"/send/m.room.message/{uuid.uuid4().hex}") + headers = { + 'Authorization': f'Bearer {cfg.MX_TOKEN}', + 'Content-Type': 'application/json' + } + payload = { + 'msgtype': 'm.text', + 'body': html.unescape(re.sub(r'<[^>]+>', '', message)), + 'format': 'org.matrix.custom.html', + 'formatted_body': message.replace('\n', '
') + } + try: + response = requests.put(url, headers=headers, json=payload, timeout=cfg.HTTP_TIMEOUT) + except requests.RequestException as e: + LOG.error(f"Error sending Matrix notification: {safe_error(e)}") + return False + if response.status_code != 200: + LOG.error(f'MX response code={response.status_code}: {response.text[:500]}') + return response.status_code == 200 + + +def notify(message: str) -> bool: + """Sends message to all configured messengers. Returns True if it was delivered at least to one of them""" + delivered = False + if send_message_to_tg_get(message) == 200: + delivered = True + LOG.debug(f'TG message send {message}') + else: + LOG.error(f'TG send error code: {STATUS_ERROR_SEND_TO_TG}') + + if cfg.USE_MATRIX: + if send_matrix_notification(message): + delivered = True + else: + LOG.error(f'MX send error code: {STATUS_ERROR_SEND_TO_MX}') + return delivered + def get_GPV_response_from_oe(account): url = 'https://be-svitlo.oe.if.ua/schedule-by-search' @@ -95,8 +155,7 @@ def get_GPV_response_from_oe(account): "userSearchChoice": "pob", "address": "" } - urllib3.disable_warnings() - return requests.post(url, headers=headers, data=data, verify=False) + return requests.post(url, headers=headers, data=data, verify=False, timeout=cfg.HTTP_TIMEOUT) def get_PO_response(): # Power Outage @@ -115,8 +174,7 @@ def get_PO_response(): # Power Outage 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36' } - urllib3.disable_warnings() - return requests.get(url=url, headers=headers, params=params) + return requests.get(url=url, headers=headers, params=params, timeout=cfg.HTTP_TIMEOUT) def has_numbers(in_str): @@ -124,55 +182,27 @@ def has_numbers(in_str): def get_utf_chars(number, is_time=False): - digit_chars = ['\U00000030\U000020E3', '\U00000031\U000020E3', '\U00000032\U000020E3', '\U00000033\U000020E3', - '\U00000034\U000020E3', '\U00000035\U000020E3', '\U00000036\U000020E3', '\U00000037\U000020E3', - '\U00000038\U000020E3', '\U00000039\U000020E3'] - res = '' - if is_time: - h = number.split(':')[0] - m = number.split(':')[1] - if m != 0: - value = str(h) + ':' + str(m) - else: - value = str(h) + ':00' - else: - value = str(number) - for c in value: - if c.isdigit and c not in ['.', ':']: - res += digit_chars[int(c)] - else: - res += c - return res + """Replaces ASCII digits with keycap emoji digits. is_time is kept for compatibility""" + return ''.join(DIGIT_CHARS[int(c)] if c in '0123456789' else c for c in str(number)) def get_digit_message(arr, is_time=False): - first = 0 - if len(arr) > 0: - first = arr[0] - - last = 0 - if len(arr) > 1: - last = arr[1] - - msg = 'з {num0} до {num1}' - return msg.format(num0=get_utf_chars(first, is_time), num1=get_utf_chars(last, is_time)) + first = arr[0] if len(arr) > 0 else 0 + last = arr[1] if len(arr) > 1 else 0 + return f'з {get_utf_chars(first, is_time)} до {get_utf_chars(last, is_time)}' def get_GPV_message(approved_from, event_date, hours_off, hours_on, outage_queue): - queue = get_utf_chars(outage_queue) - message = f'\U000026A1 ГПВ на {event_date} для {queue} черги' + queue = get_utf_chars(html.escape(str(outage_queue))) + message = f'\U000026A1 ГПВ на {html.escape(str(event_date))} для {queue} черги' if len(hours_off) > 0: message += '\nВимкнення:' - for i in range(len(hours_off)): - arr = [hours_off[i], hours_on[i]] - msg = get_digit_message(arr, True) - message += '\n' + msg - - # message += (f'\n{MENTIONS}') + for hour_off, hour_on in zip(hours_off, hours_on): + message += '\n' + get_digit_message([html.escape(str(hour_off)), html.escape(str(hour_on))], True) else: message += '\nВимкнень немає \U0001F44C \U0001F483' - message += (f'\n{approved_from}') + message += f'\n{html.escape(str(approved_from))}' return message @@ -181,44 +211,57 @@ def get_address(city, street, house): def is_outage_date_before_or_same(stop_date): - parsed_date = datetime.strptime(stop_date, "%Y.%m.%d").date() - current_date = datetime.now().date() - return parsed_date + timedelta(hours=1) >= current_date + """True if the outage is not finished before today (or the date can't be parsed)""" + try: + parsed_date = datetime.strptime(stop_date, "%Y.%m.%d").date() + except (TypeError, ValueError): + LOG.warning(f'Wrong stop date: {stop_date!r}') + return True + return parsed_date >= datetime.now().date() + + +def get_row_text(row, path: str) -> str: + return ' '.join(t.strip() for t in row.xpath(path) if t and t.strip()) def parse_po_data(root) -> List[POData]: res: List[POData] = [] - for i in range(len(root.findall(".//div[@class='table-row flex']"))): + for row in root.xpath("//div[@class='table-row flex']"): po_data = POData( - city=root.xpath("//div[@class='table-row flex']/div[@class='city']/text()")[i], - street=root.xpath("//div[@class='table-row flex']/div[@class='street']/text()")[i], - house=root.xpath("//div[@class='table-row flex']/div[@class='house_number']/text()")[i], - po_type=root.xpath("//div[@class='table-row flex']/div[@class='type-shutdown']/text()")[i], - reason=root.xpath("//div[@class='table-row flex']/div[@class='reason-shutdown reason-text-container']/div[@class='reason-text-hide']/text()")[i], - placement_date=root.xpath("//div[@class='table-row flex']/div[@class='placement_date']/div[@class='date']/text()")[i], - start_date=root.xpath("//div[@class='table-row flex']/div[@class='shutdown_date']/div[@class='date']/text()")[i], - stop_date=root.xpath("//div[@class='table-row flex']/div[@class='turn_on_date']/div[@class='date']/text()")[i], - start_time=root.xpath("//div[@class='table-row flex']/div[@class='shutdown_date']/div[@class='time']/text()")[i], - stop_time=root.xpath("//div[@class='table-row flex']/div[@class='turn_on_date']/div[@class='time']/text()")[i]) + city=get_row_text(row, "./div[@class='city']//text()"), + street=get_row_text(row, "./div[@class='street']//text()"), + house=get_row_text(row, "./div[@class='house_number']//text()"), + po_type=get_row_text(row, "./div[@class='type-shutdown']//text()"), + reason=get_row_text(row, "./div[@class='reason-shutdown reason-text-container']/div[@class='reason-text-hide']//text()"), + placement_date=get_row_text(row, "./div[@class='placement_date']/div[@class='date']//text()"), + start_date=get_row_text(row, "./div[@class='shutdown_date']/div[@class='date']//text()"), + stop_date=get_row_text(row, "./div[@class='turn_on_date']/div[@class='date']//text()"), + start_time=get_row_text(row, "./div[@class='shutdown_date']/div[@class='time']//text()"), + stop_time=get_row_text(row, "./div[@class='turn_on_date']/div[@class='time']//text()")) - if is_outage_date_before_or_same(po_data.stop_date): - po_data.need_to_send = True + if not po_data.start_ts: + LOG.warning(f'PO row skipped due to wrong start date/time: {po_data}') + continue + + if not is_outage_date_before_or_same(po_data.stop_date): + continue if po_data not in res: res.append(po_data) - res.sort(key=lambda pd: pd.start_date) + res.sort(key=lambda pd: pd.start_ts) return res def load_last_datas(file_name): if os.path.isfile(file_name): - with open(file_name, 'r') as f: + with open(file_name, 'r', encoding='utf-8') as f: return f.read() else: return '' + def get_parsed_po_data(): return PARSED_PO_DATAS @@ -227,118 +270,50 @@ def get_parsed_gpv_data(): return PARSED_GPV_DATAS -async def process_gpv(): - if cfg.IS_DEBUG: - x = json.loads( - """{ -"current": - { - "gav": - { - "message": "Черга спеціальних аварійних відключень (СГАВ):", - "queue": null - }, - "hasQueue": "yes", - "note": "Станом на 21:05 14.01.2026 за вказаним особовим рахунком '30014180' споживач підпадає під чергу '5.1' Графіку погодинного відключення(ГПВ)", - "queue": 5, - "sgav": - { - "message": "Черга спеціальних аварійних відключень (СГАВ):", - "queue": null - }, - "subQueue": 1 - }, -"schedule": - [ - { - "createdAt": "14.01.2026 19:11", - "eventDate": "15.04.2026", - "queues": - { - "5.1": - [ - { - "from": "07:30", - "shutdownHours": "07:30-11:00", - "status": 1, - "to": "11:00" - }, - { - "from": "14:30", - "shutdownHours": "14:30-20:00", - "status": 1, - "to": "20:00" - }, - { - "from": "22:00", - "shutdownHours": "22:00-00:00", - "status": 1, - "to": "00:00" - } - ] - }, - "scheduleApprovedSince": "14.04.2026 19:11" - }, - { - "createdAt": "13.04.2026 20:28", - "eventDate": "14.04.2026", - "queues": - { - "5.1": - [ - { - "from": "03:00", - "shutdownHours": "03:00-07:00", - "status": 1, - "to": "07:00" - }, - { - "from": "11:00", - "shutdownHours": "11:00-17:30", - "status": 1, - "to": "17:30" - } - ] - }, - "scheduleApprovedSince": "14.04.2026 09:56" - } - ] -} - """ - ) - else: - response = get_GPV_response_from_oe(cfg.ACCOUNT) - if response.status_code != 200: - print("code=" + str(response.status_code)) - return STATUS_ERROR_GPV_RESPONSE +def gpv_date_key(entry: dict): + try: + return datetime.strptime(str(entry.get('eventDate')), GPV_DATE_FORMAT) + except ValueError: + return datetime.max - x = response.json() - outage_queue = str(x["current"]["queue"]) - if x["current"]["subQueue"]: - outage_queue += '.' + str(x["current"]["subQueue"]) +def parse_gpv_data(x) -> List[GPVData]: + if not isinstance(x, dict): + raise ValueError(f'Unexpected GPV response type: {type(x).__name__}') - datas = '' - if 'queues' in str(x["schedule"]): - datas = sorted(x["schedule"], key=lambda item: item['eventDate']) + current = x.get('current') or {} + if current.get('queue') is None: + LOG.warning(f'No outage queue in GPV response: {current}') + return [] + outage_queue = str(current['queue']) + if current.get('subQueue'): + outage_queue += '.' + str(current['subQueue']) - for entry in datas: - hours_off = [] - hours_on = [] - if len(entry) > 0: - approved_from = 'Запроваджено ' + entry['scheduleApprovedSince'] - event_date = entry['eventDate'] - hours_list = entry['queues'][outage_queue] - else: - hours_list = [] - approved_from = 'Hемає даних про дату запровадження' - event_date = datetime.today().strftime('%Y-%m-%d') + schedule = x.get('schedule') + if not isinstance(schedule, list): + return [] + res: List[GPVData] = [] + for entry in sorted((e for e in schedule if isinstance(e, dict)), key=gpv_date_key): + queues = entry.get('queues') + event_date = entry.get('eventDate') + if not isinstance(queues, dict) or not event_date: + continue + if outage_queue not in queues: + LOG.warning(f'No queue {outage_queue} in GPV schedule for {event_date}') + continue + + approved_since = entry.get('scheduleApprovedSince') + approved_from = f'Запроваджено {approved_since}' if approved_since else 'Hемає даних про дату запровадження' + + hours_list = queues.get(outage_queue) or [] if cfg.IS_DEBUG: print(f'hours_list{hours_list}, approved_from={approved_from}, event_date={event_date}') + hours_off = [] + hours_on = [] for h in hours_list: - if h["status"] == 1: + if isinstance(h, dict) and h.get("status") == 1 and h.get("from") and h.get("to"): hours_off.append(h["from"]) hours_on.append(h["to"]) @@ -346,145 +321,177 @@ async def process_gpv(): print(f'hours_off = {hours_off}') print(f'hours_on = {hours_on}') - update_gpv_data(PARSED_GPV_DATAS, GPVData(approved_from=approved_from, event_date=event_date, hours_off=hours_off, hours_on=hours_on, outage_queue=outage_queue)) + gpv_data = GPVData(approved_from=approved_from, event_date=event_date, hours_off=hours_off, + hours_on=hours_on, outage_queue=outage_queue) + if gpv_data.actual_till_ts is None: + LOG.warning(f'GPV entry skipped due to wrong event date: {gpv_data}') + continue + res.append(gpv_data) + return res - check_and_send_gpv_message() + +def fetch_gpv_json(): + if cfg.IS_DEBUG: + with open(os.path.join(cfg.TEST_DIR, 'data_gpv.json'), 'r', encoding='utf-8') as f: + return json.load(f) + + if not cfg.ACCOUNT: + LOG.error('Account number is not configured') + return None + response = get_GPV_response_from_oe(cfg.ACCOUNT) + if response.status_code != 200: + LOG.error(f'GPV response code={response.status_code}') + return None + return response.json() + + +async def process_gpv(): + try: + x = await asyncio.to_thread(fetch_gpv_json) + if x is None: + return STATUS_ERROR_GPV_RESPONSE + new_data = parse_gpv_data(x) + except (requests.RequestException, ValueError, OSError) as e: + LOG.error(f'Error getting GPV data: {safe_error(e)}') + return STATUS_ERROR_GPV_RESPONSE + + for el in new_data: + update_gpv_data(PARSED_GPV_DATAS, el) + remove_old_gpv_data(PARSED_GPV_DATAS) + + await check_and_send_gpv_message() return STATUS_SEND_OK -def check_and_send_gpv_message(): - for el in PARSED_GPV_DATAS: - message = get_GPV_message(el.approved_from, el.event_date, el.hours_off, el.hours_on, el.outage_queue) - if el.need_to_send and (el.start_ts[-1] + 3600.0 >= datetime.now().timestamp()): - tg_response = send_message_to_tg_get(message) - if tg_response == 200: +async def check_and_send_gpv_message(): + now_ts = datetime.now().timestamp() + for el in list(PARSED_GPV_DATAS): + if el.need_to_send and (el.actual_till_ts + 3600.0 >= now_ts): + message = get_GPV_message(el.approved_from, el.event_date, el.hours_off, el.hours_on, el.outage_queue) + if await asyncio.to_thread(notify, message): el.set_send() - LOG.debug(f'TG message send {message}') - else: - LOG.error(f'TG send error code: {STATUS_ERROR_SEND_TO_TG}') - - if cfg.USE_MATRIX and not cfg.IS_DEBUG: - res = send_matrix_notification(message) - if not res: - LOG.error(f'MX send error code: {STATUS_ERROR_SEND_TO_MX}') def update_gpv_data(data_list: List[GPVData], new_data: GPVData): - # find by TS - obj = next((obj for obj in data_list if obj.start_ts == new_data.start_ts), None) - if obj: - # if exists - update: - old_data = data_list[data_list.index(obj)] - if not old_data.need_to_send: # notification already send - if old_data.start_ts != new_data.start_ts: - data_list[data_list.index(obj)] = new_data - LOG.debug(f'Updated element {old_data} in list with new data {new_data}') - else: - LOG.debug(f'Element {old_data} was not updated due to the same') - else: - data_list[data_list.index(obj)] = new_data - else: - # if not exists - add + # find by event date and queue + old_data = next((obj for obj in data_list if obj.key == new_data.key), None) + if old_data is None: data_list.append(new_data) LOG.debug(f'New element appended to list {new_data}') - # data_list SORT + elif old_data.need_to_send: + # notification is not sent yet - just take the latest data + data_list[data_list.index(old_data)] = new_data + elif not old_data.is_same_schedule(new_data): + # schedule is changed after notification - it will be sent again + data_list[data_list.index(old_data)] = new_data + LOG.debug(f'Updated element {old_data} in list with new data {new_data}') + else: + LOG.debug(f'Element {old_data} was not updated due to the same') - # remove OLD entries + +def remove_old_gpv_data(data_list: List[GPVData]): ts = datetime.now().timestamp() for el in data_list[:]: - if el.start_ts and (el.start_ts[-1] + 1800) < ts: + if el.actual_till_ts is None or (el.actual_till_ts + 1800) < ts: LOG.debug(f'Old element removed from list {el}') data_list.remove(el) def get_message_with_po(el: POData): - address = get_address(el.city, el.street, el.house) + address = html.escape(get_address(el.city, el.street, el.house)) if el.need_to_send: message = ( - f'\U000026A0 Увага!\nНа {el.start_date} за адресою {address} заплановано {el.po_type} відключення (заявка від {el.placement_date}).\n' - f'Причина відключення: {el.reason}\n' - f'Початок вимкнення {el.start_date} об {el.start_time}\n' - f'Кінець вимкнення {el.stop_date} об {el.stop_time}') + f'\U000026A0 Увага!\nНа {html.escape(el.start_date)} за адресою {address} заплановано ' + f'{html.escape(el.po_type)} відключення (заявка від {html.escape(el.placement_date)}).\n' + f'Причина відключення: {html.escape(el.reason)}\n' + f'Початок вимкнення {html.escape(el.start_date)} об {html.escape(el.start_time)}\n' + f'Кінець вимкнення {html.escape(el.stop_date)} об {html.escape(el.stop_time)}') else: - message = f'На {el.start_date} Планових вимкнень немає \U0001F44C \U0001F483' + message = f'На {html.escape(el.start_date)} Планових вимкнень немає \U0001F44C \U0001F483' return message -async def process_po(): - if not cfg.IS_DEBUG: +def fetch_po_root(): + if cfg.IS_DEBUG: + with open(os.path.join(cfg.TEST_DIR, 'data_po_2days_02.html'), 'r', encoding='utf-8') as f: + page = f.read() + else: response = get_PO_response() if response.status_code != 200: + LOG.error(f'PO response code={response.status_code}') + return None + page = response.text + if not page.strip(): + LOG.error('Empty PO page') + return None + return etree.fromstring(page, etree.HTMLParser()) + + +async def process_po(): + try: + root = await asyncio.to_thread(fetch_po_root) + if root is None: return STATUS_ERROR_PO_RESPONSE - html = response.text - parser = etree.HTMLParser() - root = etree.parse(StringIO(html), parser) - else: - # html = "
Район
Населений пункт
Вулиця
№ буд
" \ - # "
Корпус
Диспетчерська назва об'єкту
Вид робіт
Причина робіт
Дата розміщення
" \ - # "
Дата і час вимкнення
Дата і час включення
Наступна дата робіт
Піднянський район
" \ - # "
Гірна
Тевського
9
ДСП 110/Пр."ЗБВIК"/30-413/Л-6
Планове
" \ - # "
Поточний ремонт
2025.03.24
11:50
2026.04.10
" \ - # "
09:00
2026.04.10
18:00
" - with open('test/data_po_2days_02.html', 'r') as file: - html = file.read() - root = etree.fromstring(html) + new_data = parse_po_data(root) + except (requests.RequestException, etree.LxmlError, ValueError, OSError) as e: + LOG.error(f'Error getting PO data: {safe_error(e)}') + return STATUS_ERROR_PO_RESPONSE - global PARSED_PO_DATAS - new_data = parse_po_data(root) for el in new_data: - await update_po_data(PARSED_PO_DATAS, el) + update_po_data(PARSED_PO_DATAS, el) + remove_old_po_data(PARSED_PO_DATAS) - check_and_send_po_message() + await check_and_send_po_message() return STATUS_SEND_OK -def check_and_send_po_message(): - for el in PARSED_PO_DATAS: +async def check_and_send_po_message(): + for el in list(PARSED_PO_DATAS): if el.need_to_send: message = get_message_with_po(el) - tg_response = send_message_to_tg_get(message) - if tg_response == 200: + if await asyncio.to_thread(notify, message): el.set_sent() el.set_notified_now() - LOG.debug(f'TG message send {message}') - else: - LOG.error(f'TG send error code: {STATUS_ERROR_SEND_TO_TG}') - - if cfg.USE_MATRIX: - res = send_matrix_notification(message) - if not res: - LOG.error(f'MX send error code: {STATUS_ERROR_SEND_TO_MX}') -async def update_po_data(data_list:List[POData], new_data: POData): - # find by TS - obj = next((obj for obj in data_list if obj.start_ts == new_data.start_ts), None) - if obj: - # if exists - update - old_data = data_list[data_list.index(obj)] - if not old_data.need_to_send: # notification already send - if old_data != new_data: - # update 'need_to_send' to re-send notification every day - if ((not old_data.notified_at + timedelta(days=1) <= datetime.now()) or - # or if current hour - is the start data processing hour with firs minute in cfg.PROCESSING_MINUTES list - (cfg.PROCESS_START_HOUR and datetime.now().strftime('%H') == cfg.PROCESS_START_HOUR and datetime.now().strftime('%M') == cfg.PROCESSING_MINUTES[0])): - new_data.set_sent() - new_data.update_notified_at(old_data) - data_list[data_list.index(obj)] = new_data - LOG.debug(f'Updated element {old_data} in list with new data {new_data}') - else: - LOG.debug(f'Element {old_data} was not updated due to the same') - else: - data_list[data_list.index(obj)] = new_data - else: - # if not exists - add +def get_daily_reminder_time(now: datetime) -> datetime: + """Today's moment of the first data processing - notifications are repeated after it""" + first_minute = min(cfg.PROCESSING_MINUTES) if cfg.PROCESSING_MINUTES else 0 + return now.replace(hour=cfg.PROCESS_START_HOUR, minute=first_minute, second=0, microsecond=0) + + +def is_reminder_needed(old_data: POData, now: Optional[datetime] = None) -> bool: + now = now or datetime.now() + if old_data.notified_at is None: + return True + # re-send notification every day + if old_data.notified_at + timedelta(days=1) <= now: + return True + # or if the first data processing of the day is passed after the last notification + reminder_time = get_daily_reminder_time(now) + return old_data.notified_at < reminder_time <= now + + +def update_po_data(data_list: List[POData], new_data: POData): + # find by start TS and address + old_data = next((obj for obj in data_list if obj.key == new_data.key), None) + if old_data is None: data_list.append(new_data) LOG.debug(f'New element appended to list {new_data}') - # data_list SORT + elif old_data.need_to_send: + # notification is not sent yet - just take the latest data + data_list[data_list.index(old_data)] = new_data + elif old_data != new_data or is_reminder_needed(old_data): + # data changed or it is time to remind - notification will be sent again + data_list[data_list.index(old_data)] = new_data + LOG.debug(f'Updated element {old_data} in list with new data {new_data}') + else: + LOG.debug(f'Element {old_data} was not updated due to the same') - # remove OLD entries + +def remove_old_po_data(data_list: List[POData]): ts = datetime.now().timestamp() for el in data_list[:]: if el.start_ts and (el.start_ts + 3600) <= ts: