MEGAINFRA-WEBAPP/scripts/backfill-fuel-log-previous-reading.py

#!/usr/bin/env python3
"""Backfill the `previous_reading` self-relation on FUEL_LOGS.

For every vehicle, walk its fuel_log rows ordered by `date_logged` (then
`id` to break ties deterministically) and set each row's
`previous_reading` to the row immediately before it. The oldest row per
vehicle is left with `previous_reading` unset (null) — by design, there
is no predecessor.

Uses the existing `idx_fuel_logs_vehicle_date` index for the sort+groupBy
so the per-vehicle scan is sequential. Updates are batched as PATCH
calls.

Idempotent: re-running is safe; rows that already have a `previous_reading`
are skipped.

Usage:
    python3 scripts/backfill-fuel-log-previous-reading.py [--site <site_id>]
"""
import argparse
import sys
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

import pb_env
from pb_env import http, superuser_token

PB = pb_env.PB
TOKEN = superuser_token()


def list_sites() -> list[dict]:
    code, body = http("GET", "/api/collections/sites/records?perPage=500", token=TOKEN)
    if code != 200:
        sys.exit(f"list sites failed: {code} {body}")
    return body.get("items", [])


def list_fuel_logs_for_site(site_id: str) -> list[dict]:
    """Return all fuel_log ids+date_logged+vehicle for the site, paginated."""
    items: list[dict] = []
    page = 1
    while True:
        code, body = http(
            "GET",
            f"/api/collections/fuel_logs/records"
            f"?filter=site='{site_id}'&fields=id,date_logged,vehicle&perPage=500&page={page}",
            token=TOKEN,
        )
        if code != 200:
            sys.exit(f"list fuel_logs page {page} failed: {code} {body}")
        page_items = body.get("items", [])
        items.extend(page_items)
        if len(page_items) < 500:
            break
        page += 1
    return items


def update_previous(rid: str, prev_id: str) -> tuple[str, int]:
    body = {"previous_reading": prev_id}.__str__().replace("'", '"').encode()
    req = urllib.request.Request(
        f"{PB}/api/collections/fuel_logs/records/{rid}",
        data=body,
        method="PATCH",
        headers={
            "Authorization": TOKEN,
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(req) as r:
            return rid, r.status
    except urllib.error.HTTPError as e:
        return rid, e.code


def backfill_site(site_id: str) -> tuple[int, int, list]:
    rows = list_fuel_logs_for_site(site_id)
    print(f"  fetched {len(rows)} fuel_logs for site {site_id}")

    # Already linked? Skip them — idempotent.
    code, body = http(
        "GET",
        f"/api/collections/fuel_logs/records"
        f"?filter=site='{site_id}'~and~previous_reading!=null&fields=id&perPage=1",
        token=TOKEN,
    )
    already_linked = body.get("totalItems", 0) if code == 200 else 0
    if already_linked:
        print(f"  {already_linked} rows already linked; will skip those rows")
        code, body = http(
            "GET",
            f"/api/collections/fuel_logs/records"
            f"?filter=site='{site_id}'~and~previous_reading!=null&fields=id&perPage=500",
            token=TOKEN,
        )
        already_set = {it["id"] for it in body.get("items", [])}
    else:
        already_set = set()

    # Group by vehicle, sort by date_logged then id (stable tiebreak).
    by_vehicle: dict[str, list[dict]] = {}
    for r in rows:
        if r["id"] in already_set:
            continue
        by_vehicle.setdefault(r.get("vehicle") or "", []).append(r)

    updates: list[tuple[str, str]] = []  # (row_id, prev_row_id)
    for vid, vrows in by_vehicle.items():
        vrows.sort(key=lambda r: (r.get("date_logged") or "", r["id"]))
        for i in range(1, len(vrows)):
            updates.append((vrows[i]["id"], vrows[i - 1]["id"]))

    print(f"  vehicles touched: {len(by_vehicle)}, updates queued: {len(updates)}")
    if not updates:
        return 0, 0, []

    successes = 0
    errors = []
    with ThreadPoolExecutor(max_workers=16) as ex:
        futures = {ex.submit(update_previous, rid, prev): rid for rid, prev in updates}
        for fut in as_completed(futures):
            rid, code = fut.result()
            if code == 200:
                successes += 1
            else:
                errors.append({"id": rid, "code": code})
    return successes, len(updates), errors


def main() -> int:
    p = argparse.ArgumentParser()
    p.add_argument(
        "--site",
        help="Specific site_id to backfill. If omitted, backfills every site.",
    )
    args = p.parse_args()

    sites = list_sites()
    if args.site:
        sites = [s for s in sites if s["id"] == args.site]
        if not sites:
            sys.exit(f"site {args.site!r} not found")

    total_ok = 0
    total_q = 0
    for s in sites:
        print(f"\n=== site {s['id']} ({s.get('name')}) ===")
        ok, q, errs = backfill_site(s["id"])
        total_ok += ok
        total_q += q
        if errs:
            print(f"  ERRORS ({len(errs)}): {errs[:5]}")

    print(f"\n=== TOTAL ===")
    print(f"  updated {total_ok}/{total_q} rows across {len(sites)} site(s)")
    return 0


if __name__ == "__main__":
    sys.exit(main())