scripts/update_data.py - master - dgpolar/volga-908

scripts/update_data.py
176 строк6 KB
DGPolar
Harden daily updates against temporary source gaps and refresh data.
08 июл 2026, 10:26
08 июл 2026, 10:264a29de3
#!/usr/bin/env python3
"""Обновляет data.json для PWA. Запуск: python3 scripts/update_data.py"""
import json
import os
import ssl
import urllib.request
from collections import defaultdict
from datetime import datetime, timedelta

try:
    import xlrd
except ImportError:
    raise SystemExit("pip install xlrd")

POST_ZERO = 62.0
BULLETIN_BASE = "https://xn--80adbch2buek4ak3i.xn--p1ai/uploads/"
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(SCRIPT_DIR, "..", "data.json")
CACHE = os.path.join(SCRIPT_DIR, ".bulletins")
RECENT_BULLETIN_DAYS = 7  # всегда перекачивать свежие бюллетени


def fetch_json(url):
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
    ctx = ssl.create_default_context()
    with urllib.request.urlopen(req, context=ctx, timeout=30) as r:
        return json.loads(r.read().decode())


def parse_bulletin(path):
    try:
        wb = xlrd.open_workbook(path)
        sh = wb.sheet_by_name("уровни воды")
        for r in range(sh.nrows):
            km = sh.cell_value(r, 0)
            if isinstance(km, (int, float)) and abs(km - 908) < 0.01:
                level = sh.cell_value(r, 3)
                if level in ("", None):
                    return None
                level = float(level)
                if level > 72:
                    return None
                daily = sh.cell_value(r, 4)
                daily_cm = int(float(daily)) if daily not in ("", None) else 0
                fname = os.path.basename(path).replace(".xls", "")
                iso = "-".join(reversed(fname.split(".")))
                return {"date": iso, "level_m": round(level, 2), "daily_cm": daily_cm}
    except Exception:
        pass
    return None


def download_bulletin(cur, ctx):
    fname = cur.strftime("%d.%m.%Y") + ".xls"
    path = os.path.join(CACHE, fname)
    url = BULLETIN_BASE + fname
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
        with urllib.request.urlopen(req, context=ctx, timeout=15) as r:
            data = r.read()
        if data[:4] == b"\xd0\xcf\x11\xe0":
            with open(path, "wb") as f:
                f.write(data)
            return True
    except Exception:
        if os.path.exists(path):
            os.remove(path)
    return False


def build_aggregates(daily):
    monthly_map = defaultdict(list)
    for d in daily:
        monthly_map[d["date"][:7]].append(d["level_m"])
    monthly = []
    for ym in sorted(monthly_map):
        levels = monthly_map[ym]
        y, m = ym.split("-")
        monthly.append({
            "label": f"{m}.{y[2:]}",
            "avg": round(sum(levels) / len(levels), 2),
            "min": round(min(levels), 2),
            "max": round(max(levels), 2),
            "days": len(levels),
        })
    yearly_map = defaultdict(list)
    for d in daily:
        yearly_map[d["date"][:4]].append(d["level_m"])
    yearly = [
        {
            "year": y,
            "avg": round(sum(lv) / len(lv), 2),
            "max": round(max(lv), 2),
            "min": round(min(lv), 2),
            "days": len(lv),
        }
        for y, lv in sorted(yearly_map.items())
    ]
    return monthly, yearly


def main():
    records = {}
    for year in [2023, 2024, 2025, 2026]:
        data = fetch_json(f"https://www.urovenvody.ru/voda2_ajax.php?year={year}&platform=39905")
        for y, m, days in data.get("months", []):
            for day, val in enumerate(days, 1):
                if val and val != "-":
                    try:
                        lv = float(str(val).replace(",", "."))
                        if 0 < lv <= 72:
                            iso = f"{y:04d}-{m:02d}-{day:02d}"
                            records[iso] = {"date": iso, "level_m": round(lv, 2)}
                    except ValueError:
                        pass

    os.makedirs(CACHE, exist_ok=True)
    ctx = ssl.create_default_context()
    end = datetime.now()
    start = datetime(2024, 8, 23)
    today_iso = end.strftime("%Y-%m-%d")
    today_bulletin = False

    cur = start
    while cur <= end:
        fname = cur.strftime("%d.%m.%Y") + ".xls"
        path = os.path.join(CACHE, fname)
        age_days = (end.date() - cur.date()).days
        if age_days < RECENT_BULLETIN_DAYS or not os.path.exists(path):
            download_bulletin(cur, ctx)
        rec = parse_bulletin(path) if os.path.exists(path) else None
        if rec:
            records[rec["date"]] = rec
            if rec["date"] == today_iso:
                today_bulletin = True
        cur += timedelta(days=1)

    daily = sorted(records.values(), key=lambda x: x["date"])
    monthly, yearly = build_aggregates(daily)

    out = {
        "updated": end.strftime("%Y-%m-%d"),
        "daily": daily,
        "monthly": monthly,
        "yearly": yearly,
    }

    if os.path.exists(OUT):
        with open(OUT, encoding="utf-8") as f:
            prev = json.load(f)
        prev_daily = prev.get("daily", [])
        if prev_daily:
            # Подстраховка: если один из источников временно пустой,
            # не теряем исторические записи, а объединяем старые и новые.
            merged = {d["date"]: d for d in prev_daily}
            merged.update({d["date"]: d for d in out["daily"]})
            if len(merged) > len(out["daily"]):
                out["daily"] = sorted(merged.values(), key=lambda x: x["date"])
                out["monthly"], out["yearly"] = build_aggregates(out["daily"])
        if prev.get("daily") == out["daily"]:
            print(
                f"No changes ({len(out['daily'])} records, latest {out['daily'][-1]['date']}, "
                f"today bulletin={'yes' if today_bulletin else 'not yet'})"
            )
            return

    with open(OUT, "w", encoding="utf-8") as f:
        json.dump(out, f, ensure_ascii=False, indent=2)
    print(
        f"Wrote {len(out['daily'])} records to {OUT}, latest {out['daily'][-1]['date']}, "
        f"today bulletin={'yes' if today_bulletin else 'not yet'}"
    )


if __name__ == "__main__":
    main()