102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
import json
|
||
import logging
|
||
import requests
|
||
from pathlib import Path
|
||
from urllib3.exceptions import InsecureRequestWarning
|
||
import urllib3
|
||
|
||
import config
|
||
|
||
urllib3.disable_warnings(InsecureRequestWarning)
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
filename="logs/parser.log",
|
||
format="%(asctime)s - %(module)s - %(levelname)s - %(funcName)s: %(lineno)d - %(message)s",
|
||
datefmt='%H:%M:%S',
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PLAN_FILE = Path("plan.txt")
|
||
VN_FILE = Path("vnereglament.txt")
|
||
AVAR_FILE = Path("avar.txt")
|
||
|
||
def clean_street(street: str) -> str:
|
||
"""Очищает и форматирует строку с адресами."""
|
||
street = street.replace('Белгород г; ', '').replace('\n\n', '\n').strip()
|
||
lines = [line.strip() for line in street.split('\n') if line.strip()]
|
||
return '\n'.join(lines)
|
||
|
||
def format_time(dt_str: str) -> str:
|
||
"""Преобразует '2025-10-02T09:00:00' → '02.10.2025 09:00'."""
|
||
try:
|
||
return dt_str.replace('T', ' ')
|
||
except:
|
||
return dt_str
|
||
|
||
def parse_and_save(data, file_path: Path, mode: str):
|
||
"""
|
||
mode: 'plan', 'vnereglament', 'avar'
|
||
"""
|
||
if not data or data == 0:
|
||
file_path.write_text("[Нет отключений]", encoding='utf-8')
|
||
return
|
||
|
||
records = data if isinstance(data, list) else [data]
|
||
lines = []
|
||
|
||
for item in records:
|
||
if mode in ('plan', 'vnereglament'):
|
||
street = clean_street(item.get("DisconnectionObject", ""))
|
||
time_down = format_time(item.get("DisconnectionDateTime", ""))
|
||
time_up = format_time(item.get("EnergyOnPlanningDateTime", ""))
|
||
lines.append(f"{street}\nОтключение: {time_down} — {time_up}\n{'─' * 30}")
|
||
elif mode == 'avar':
|
||
street = clean_street(item.get("StreetHome", ""))
|
||
time_up = format_time(item.get("ScheduledTimeRemoval", ""))
|
||
lines.append(f"{street}\nВосстановление: ~{time_up}\n{'─' * 30}")
|
||
|
||
file_path.write_text('\n\n'.join(lines), encoding='utf-8')
|
||
|
||
def get_plan():
|
||
try:
|
||
r = requests.get(url=config.url_plan, verify=False, timeout=10)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
parse_and_save(data, PLAN_FILE, 'plan')
|
||
logger.info("[Плановые отключения] обновлены")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении плановых отключений: {e}")
|
||
PLAN_FILE.write_text("[Ошибка загрузки данных]", encoding='utf-8')
|
||
|
||
def get_vnereglament():
|
||
try:
|
||
r = requests.get(url=config.url_vnereglament, verify=False, timeout=10)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
parse_and_save(data, VN_FILE, 'vnereglament')
|
||
logger.info("[Внерегламентные отключения] обновлены")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении внерегламентных отключений: {e}")
|
||
VN_FILE.write_text("[Ошибка загрузки данных]", encoding='utf-8')
|
||
|
||
def get_avar():
|
||
try:
|
||
r = requests.get(url=config.url_avar, verify=False, timeout=10)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
parse_and_save(data, AVAR_FILE, 'avar')
|
||
logger.info("[Аварийные отключения] обновлены")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении аварийных отключений: {e}")
|
||
AVAR_FILE.write_text("[Ошибка загрузки данных]", encoding='utf-8')
|
||
|
||
def start_parser():
|
||
get_plan()
|
||
get_vnereglament()
|
||
get_avar()
|
||
logger.info("Парсер отработал")
|
||
|
||
if __name__ == '__main__':
|
||
start_parser() |