#!/usr/bin/env python3
"""Mise à jour incrémentale de l'extrait France des data centers par diffs OSM.

Au lieu de refiltrer le PBF France (5 Go) chaque jour, on applique les diffs
quotidiens Geofabrik (~4 Mo) à l'extrait data center (quelques centaines de Ko) :

  1. télécharge les diffs manquants depuis la dernière séquence appliquée ;
  2. les fusionne (osmium merge-changes) puis les applique à dc-raw.osm.pbf ;
  3. refiltre sur les tags data center (les objets étrangers au thème
     apportés par le diff disparaissent, les tags retirés aussi) ;
  4. complète via l'API OSM les ways/relations nouvellement taguées dont des
     nœuds non modifiés manquent (osmium check-refs) ;
  5. exporte en GeoJSON, journalise les créations, modifications et suppressions du thème ;
  6. relance build.py.

osmium tourne en priorité basse (nice/ionice) et sur un seul thread
(OSMIUM_POOL_THREADS, réglable par DC_THREADS).

Initialisation : un premier extract_osm.sh sur le PBF France ; la séquence et
l'URL de réplication sont lues dans l'en-tête du PBF Geofabrik.
Usage : update_daily.py [--zone france] [--max-days 30] [--no-build]
"""
import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
DATA = ROOT / "data"
API = "https://api.openstreetmap.org/api/0.6"
PBF = Path("/home/poule/encrypted/osm-pbf/france-latest.osm.pbf")
FILTERS = ["nwr/telecom=data_center", "nwr/building=data_center", "nwr/industrial=data_center",
           "nwr/man_made=data_center", "nwr/data_center"]
UA = "cipherbliss-data-center-map/1.0 (+https://cipherbliss.com)"
CHANGELOG_DAYS = 90
LOW_PRIO = ["nice", "-n", "19", "ionice", "-c", "3"]
ENV = {**os.environ, "OSMIUM_POOL_THREADS": os.environ.get("DC_THREADS", "1")}


def log(*a):
    print(time.strftime("%H:%M:%S"), *a, flush=True)


def run(*cmd, **kw):
    return subprocess.run(LOW_PRIO + [str(c) for c in cmd], env=ENV, **kw)


def osmium(*args):
    run("osmium", *args, "--no-progress", check=True)


def fetch(url, dest=None):
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=300) as r:
        if dest is None:
            return r.read().decode()
        with open(dest, "wb") as f:
            shutil.copyfileobj(r, f)


def header(pbf, opt):
    return run("osmium", "fileinfo", "-g", f"header.option.{opt}", pbf,
               capture_output=True, text=True, check=True).stdout.strip()


def seq_path(base, n):
    s = f"{n:09d}"
    return f"{base}/{s[:3]}/{s[3:6]}/{s[6:]}"


def parse_state(txt):
    seq = int(re.search(r"sequenceNumber=(\d+)", txt).group(1))
    ts = re.search(r"timestamp=(\S+)", txt).group(1).replace("\\:", ":")
    return seq, datetime.fromisoformat(ts.replace("Z", "+00:00"))


def export(pbf, out):
    osmium("export", pbf, "-a", "type,id,version,changeset,timestamp,user", "-f", "geojsonseq",
           "--geometry-types=point,polygon", "-o", out, "--overwrite")


def index_export(path):
    """{id: version, auteur, tags, géométrie} des objets d'un export."""
    out = {}
    if not path.exists():
        return out
    with open(path) as f:
        for line in f:
            line = line.strip("\x1e\n ")
            if not line:
                continue
            ft = json.loads(line)
            p = ft["properties"]
            key = f'{p["@type"]}/{p["@id"]}'
            tags = {k: v for k, v in p.items() if not k.startswith("@")}
            out[key] = {"version": p.get("@version"), "user": p.get("@user"),
                        "changeset": p.get("@changeset"), "timestamp": p.get("@timestamp"),
                        "tags": tags, "geom": ft["geometry"]}
    return out


def is_dc(tags):
    return (tags.get("telecom") == "data_center" or tags.get("building") == "data_center"
            or tags.get("industrial") == "data_center" or tags.get("man_made") == "data_center"
            or "data_center" in tags)


def lonlat(geom):
    if geom["type"] == "Point":
        return geom["coordinates"]
    ring = geom["coordinates"][0] if geom["type"] == "Polygon" else geom["coordinates"][0][0]
    return [round(sum(p[i] for p in ring[:-1]) / (len(ring) - 1), 6) for i in (0, 1)]


def diff_theme(old, new, seq_from, seq_to):
    """Changements du thème entre deux exports."""
    old = {k: v for k, v in old.items() if is_dc(v["tags"])}
    new = {k: v for k, v in new.items() if is_dc(v["tags"])}
    changes = []
    for k in new.keys() - old.keys():
        changes.append(("cree", k, None, new[k]))
    for k in old.keys() - new.keys():
        changes.append(("supprime", k, old[k], None))
    for k in old.keys() & new.keys():
        a, b = old[k], new[k]
        if a["tags"] != b["tags"] or a["geom"] != b["geom"]:
            changes.append(("modifie", k, a, b))
    out = []
    for kind, k, a, b in changes:
        ref = b or a
        entry = {"type": kind, "id": k, "name": ref["tags"].get("name"),
                 "operator": ref["tags"].get("operator"), "lonlat": lonlat(ref["geom"]),
                 "seq": [seq_from, seq_to]}
        if b:
            entry.update(version=b["version"], user=b["user"], changeset=b["changeset"],
                         timestamp=b["timestamp"])
        if a and b:
            ta, tb = a["tags"], b["tags"]
            entry["tags_added"] = {t: tb[t] for t in tb.keys() - ta.keys()}
            entry["tags_removed"] = sorted(ta.keys() - tb.keys())
            entry["tags_changed"] = {t: [ta[t], tb[t]] for t in ta.keys() & tb.keys() if ta[t] != tb[t]}
            entry["geom_changed"] = a["geom"] != b["geom"]
        elif kind == "supprime":
            entry["tags_removed"] = sorted(a["tags"])
        out.append(entry)
    return sorted(out, key=lambda e: (e["type"], e["id"]))


def complete_missing(pbf, work):
    """Récupère via l'API les ways/relations dont des membres manquent."""
    res = run("osmium", "check-refs", "-i", "-r", pbf, capture_output=True, text=True)
    parents = set(re.findall(r"\bin ([wr]\d+)", res.stdout + res.stderr))
    if not parents:
        return False
    log(f"{len(parents)} objets incomplets, complétion via l'API")
    parts = []
    for ref in sorted(parents):
        kind = "way" if ref[0] == "w" else "relation"
        dest = work / f"api-{ref}.osm"
        try:
            fetch(f"{API}/{kind}/{ref[1:]}/full", dest)
            parts.append(dest)
        except Exception as e:  # objet supprimé depuis, ou API indisponible
            log("  échec", ref, e)
        time.sleep(0.5)
    if not parts:
        return False
    sorted_api = work / "api.osm.pbf"
    osmium("sort", *parts, "-o", sorted_api, "--overwrite")
    merged = work / "merged.osm.pbf"
    osmium("merge", pbf, sorted_api, "-o", merged, "--overwrite")
    shutil.move(merged, pbf)
    return True


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--zone", default="france")
    ap.add_argument("--pbf", type=Path, default=PBF, help="PBF source (en-tête de réplication)")
    ap.add_argument("--max-days", type=int, default=30, help="au-delà, mieux vaut refiltrer le PBF")
    ap.add_argument("--no-build", action="store_true")
    args = ap.parse_args()

    zdir = DATA / args.zone
    raw = zdir / "dc-raw.osm.pbf"
    state_file = zdir / "replication.state"
    work = zdir / "work"
    work.mkdir(parents=True, exist_ok=True)
    if not raw.exists():
        sys.exit(f"{raw} absent : lancer d'abord scripts/extract_osm.sh {args.pbf} {args.zone}")

    if state_file.exists():
        seq, base = state_file.read_text().split()[:2]
        seq = int(seq)
    else:
        seq = int(header(args.pbf, "osmosis_replication_sequence_number"))
        base = header(args.pbf, "osmosis_replication_base_url")
        log("séquence initiale lue dans le PBF :", seq, base)
    latest, latest_ts = parse_state(fetch(f"{base}/state.txt"))
    if latest <= seq:
        log("déjà à jour, séquence", seq)
        if not state_file.exists():
            state_file.write_text(f"{seq} {base} {latest_ts.isoformat()}\n")
        return
    todo = list(range(seq + 1, latest + 1))
    if len(todo) > args.max_days:
        sys.exit(f"{len(todo)} jours de retard (> {args.max_days}) : refiltrer le PBF")

    diffs = []
    for n in todo:
        dest = work / f"{n}.osc.gz"
        if not dest.exists():
            log("diff", n)
            fetch(seq_path(base, n) + ".osc.gz", dest.with_suffix(".part"))
            dest.with_suffix(".part").rename(dest)
        diffs.append(dest)

    merged = work / "merged.osc.gz"
    osmium("merge-changes", "-s", *diffs, "-o", merged, "--overwrite")
    applied = work / "applied.osm.pbf"
    log("application des diffs")
    osmium("apply-changes", raw, merged, "-o", applied, "--overwrite")
    new_raw = work / "dc-raw.new.osm.pbf"
    osmium("tags-filter", applied, *FILTERS, "-o", new_raw, "--overwrite")
    if complete_missing(new_raw, work):
        osmium("tags-filter", new_raw, *FILTERS, "-o", applied, "--overwrite")
        shutil.move(applied, new_raw)

    old_export = zdir / "dc-osm.geojsonseq"
    new_export = work / "dc-osm.new.geojsonseq"
    export(new_raw, new_export)
    changes = diff_theme(index_export(old_export), index_export(new_export), seq + 1, latest)
    log(f"{len(changes)} changements dans le thème data center",
        {k: sum(1 for c in changes if c["type"] == k) for k in ("cree", "modifie", "supprime")})

    shutil.move(new_raw, raw)
    shutil.move(new_export, old_export)
    (zdir / "extract-date").write_text(latest_ts.date().isoformat() + "\n")
    state_file.write_text(f"{latest} {base} {latest_ts.isoformat()}\n")

    # journal des changements (fenêtre glissante)
    log_file = zdir / "changes.json"
    history = json.loads(log_file.read_text()) if log_file.exists() else []
    history.append({"date": latest_ts.date().isoformat(), "seq": [seq + 1, latest], "changes": changes})
    cutoff = datetime.now(timezone.utc).timestamp() - CHANGELOG_DAYS * 86400
    history = [h for h in history
               if datetime.fromisoformat(h["date"]).replace(tzinfo=timezone.utc).timestamp() >= cutoff]
    log_file.write_text(json.dumps(history, ensure_ascii=False, indent=1))

    for f in diffs + [merged, applied]:
        f.unlink(missing_ok=True)
    for f in work.glob("api*"):
        f.unlink()

    if not args.no_build:
        run(sys.executable, "-I", ROOT / "scripts" / "build.py", args.zone, check=True)


if __name__ == "__main__":
    main()