Téléchargement IGN atomique (.part + rename) et pagination du catalogue STAC

lidar_pipeline/fetch_ign.py

@@ -90,38 +90,61 @@ def match_feature(features, col, row):
     return None
 
 
-def find_tile_url(col, row, timeout=20):
+def find_tile_url(col, row, timeout=20, max_pages=5):
     """Cherche l'URL de téléchargement de la dalle (col,row) dans le catalogue STAC.
 
+    Suit la pagination du catalogue (lien rel=next) : sans elle, une dalle
+    au-delà de la première page de résultats semblait « introuvable ».
+
     Returns:
         URL (str) ou None si la dalle n'est pas (encore) publiée par l'IGN.
     """
     w, s, e, n = _bbox_wgs84(col, row)
     query = urllib.parse.urlencode({"bbox": f"{w:.6f},{s:.6f},{e:.6f},{n:.6f}", "limit": 50})
-    req = urllib.request.Request(f"{_STAC_ITEMS_URL}?{query}", headers=_HEADERS)
-    with urllib.request.urlopen(req, timeout=timeout) as response:
-        data = json.loads(response.read().decode("utf-8"))
-    feature = match_feature(data.get("features", []), col, row)
-    if not feature:
-        return None
-    return feature.get("assets", {}).get("data", {}).get("href")
+    url = f"{_STAC_ITEMS_URL}?{query}"
+    for _ in range(max_pages):
+        req = urllib.request.Request(url, headers=_HEADERS)
+        with urllib.request.urlopen(req, timeout=timeout) as response:
+            data = json.loads(response.read().decode("utf-8"))
+        feature = match_feature(data.get("features", []), col, row)
+        if feature:
+            return feature.get("assets", {}).get("data", {}).get("href")
+        nxt = next((link.get("href") for link in data.get("links", [])
+                    if link.get("rel") == "next" and link.get("href")), None)
+        if not nxt:
+            return None
+        url = nxt
+    return None
 
 
 def download_file(url, dest_path, timeout=120, chunk=1024 * 1024):
-    """Télécharge url vers dest_path en streaming. Retourne la taille en octets."""
+    """Télécharge url vers dest_path en streaming (écriture atomique).
+
+    Écrit dans un .part puis renomme : un run interrompu (SIGKILL, panne)
+    ne laisse jamais un LAZ partiel qu'un run suivant considérerait comme
+    déjà téléchargé. Retourne la taille en octets.
+    """
+    import os
+    dest_path = Path(dest_path)
+    tmp = dest_path.with_name(dest_path.name + ".part")
     req = urllib.request.Request(url, headers=_HEADERS)
     t0 = time.time()
-    with urllib.request.urlopen(req, timeout=timeout) as response, open(dest_path, "wb") as out:
-        done = 0
-        while True:
-            block = response.read(chunk)
-            if not block:
-                break
-            out.write(block)
-            done += len(block)
-        elapsed = time.time() - t0
-        logger.info(f"    {done / 1e6:.0f} Mo en {elapsed:.0f}s"
-                    f" ({done / 1e6 / max(elapsed, 0.1):.1f} Mo/s)")
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as response, open(tmp, "wb") as out:
+            done = 0
+            while True:
+                block = response.read(chunk)
+                if not block:
+                    break
+                out.write(block)
+                done += len(block)
+            elapsed = time.time() - t0
+            logger.info(f"    {done / 1e6:.0f} Mo en {elapsed:.0f}s"
+                        f" ({done / 1e6 / max(elapsed, 0.1):.1f} Mo/s)")
+        os.replace(tmp, dest_path)
+    except BaseException:
+        tmp.unlink(missing_ok=True)
+        raise
     return done