feat: vigilo refactoring

.github/workflows/docker-image.yml

@@ -6,7 +6,7 @@ on:
       - "Dockerfile"
       - "requirements.txt"
 env:
-  JUST_VERSION: "1.23.0"
+  JUST_VERSION: "1.51.0"
 
 jobs:
   push_to_registry:

.github/workflows/release-datasets.yml

@@ -12,7 +12,7 @@ permissions:
   contents: write
 
 env:
-  JUST_VERSION: "1.23.0"
+  JUST_VERSION: "1.51.0"
 
 jobs:
   release:

importer/release_cloudflare.sh

@@ -1,162 +1,27 @@
 #!/usr/bin/env bash
-# Upload parquets to Cloudflare R2 (S3-compatible) with parallel uploads.
+# Upload parquets to Cloudflare R2 via wdalib.uploadToR2().
 # Usage: DATAS_LIST="sirene,geonames" just release-cloudflare
 #
 # Required env vars:
 #   CF_ACCOUNT_ID           — Cloudflare account ID
 #   CF_R2_ACCESS_KEY_ID     — R2 API token Access Key ID
 #   CF_R2_SECRET_ACCESS_KEY — R2 API token Secret Access Key
-#
-# Optional:
-#   R2_PARALLEL             — number of parallel uploads (default: 8)
-#
-# Naming convention in R2: {dataset}/{subdir}__{path separators → __}{= → .}
-# Example: sirene/observable__ape.46.38Z__data_0.parquet
 
 set -e
 
-R2_BUCKET="world-datas-analysis"
-R2_PARALLEL="${R2_PARALLEL:-8}"
-
-if [ -z "$CF_ACCOUNT_ID" ] || [ -z "$CF_R2_ACCESS_KEY_ID" ] || [ -z "$CF_R2_SECRET_ACCESS_KEY" ]; then
-    echo "[release-r2] ERROR: missing credentials — set CF_ACCOUNT_ID, CF_R2_ACCESS_KEY_ID, CF_R2_SECRET_ACCESS_KEY"
-    exit 1
-fi
-
-if ! command -v aws &>/dev/null; then
-    echo "[release-r2] ERROR: aws CLI not found (pip install awscli)"
-    exit 1
-fi
-
 datasets=${DATAS_LIST:-""}
-datasets=${datasets//,/ }
-
 if [ -z "$datasets" ]; then
-    echo "[release-r2] DATAS_LIST is empty, nothing to do"
+    echo "[uploadToR2] DATAS_LIST is empty, nothing to do"
     exit 0
 fi
 
-R2_ENDPOINT="https://${CF_ACCOUNT_ID}.r2.cloudflarestorage.com"
-
-# Export credentials + config once — inherited by all xargs subshells
-export AWS_ACCESS_KEY_ID="$CF_R2_ACCESS_KEY_ID"
-export AWS_SECRET_ACCESS_KEY="$CF_R2_SECRET_ACCESS_KEY"
-export R2_ENDPOINT R2_BUCKET
-
-for dataset in $datasets; do
-    declare -A local_keys  # r2_key → local_file_size
-    declare -A r2_sizes    # r2_key → r2_object_size
-
-    # Quick local-hash check — skip entirely if parquets unchanged since last sync
-    sync_cache="dataset/${dataset}/.r2-sync-hash"
-    current_hash=$(find "dataset/${dataset}" \( -name "*.parquet" \) -printf '%s %T@\n' 2>/dev/null | sort | sha256sum | cut -d' ' -f1)
-    if [ -f "$sync_cache" ] && [ "$(cat "$sync_cache" 2>/dev/null)" = "$current_hash" ]; then
-        echo "[release-r2] ${dataset}: parquets unchanged since last sync, skipping"
-        continue
-    fi
-
-    # Fetch current R2 state once — used for both skip-upload and stale-delete
-    echo "[release-r2] ${dataset}: listing R2 objects..."
-    while IFS= read -r line; do
-        [ -z "$line" ] && continue
-        r2_size=$(echo "$line" | awk '{print $3}')
-        r2_key=$(echo "$line"  | awk '{print $4}')
-        r2_sizes["$r2_key"]=$r2_size
-    done < <(
-        aws s3 ls "s3://${R2_BUCKET}/${dataset}/" \
-            --endpoint-url "$R2_ENDPOINT" \
-            --recursive
-    )
-
-    task_file=$(mktemp)
-    skipped=0
-    uploaded=0
-
-    for subdir in raw observable; do
-        src_dir="dataset/${dataset}/${subdir}"
-        [ -d "$src_dir" ] || continue
-
-        while IFS= read -r f; do
-            rel=$(realpath --relative-to="dataset/${dataset}" "$f")
-            flat=$(echo "$rel" | sed 's|/|__|g')
-            r2_key="${dataset}/$(echo "$flat" | sed 's|=|\.|g')"
-            local_size=$(wc -c <"$f" | tr -d ' ')
-            local_keys["$r2_key"]=$local_size
-
-            # Skip upload if R2 already has the same size
-            if [ "${r2_sizes[$r2_key]+_}" ] && [ "${r2_sizes[$r2_key]}" -eq "$local_size" ]; then
-                skipped=$((skipped + 1))
-                continue
-            fi
-
-            printf '%s\0%s\0' "$f" "$r2_key" >>"$task_file"
-            uploaded=$((uploaded + 1))
-        done < <(find "$src_dir" -name "*.parquet" | sort)
-    done
-
-    if [ "$skipped" -gt 0 ]; then
-        echo "[release-r2] ${dataset}: $skipped file(s) unchanged, skipped"
-    fi
-
-    if [ "$uploaded" -eq 0 ] && [ "$skipped" -eq 0 ]; then
-        echo "[release-r2] ${dataset}: no raw/ or observable/ directory found, skipping"
-    elif [ "$uploaded" -gt 0 ]; then
-        echo "[release-r2] ${dataset}: uploading $uploaded new/changed file(s) (parallel=${R2_PARALLEL})..."
-        progress_file=$(mktemp)
-        export PROGRESS_FILE="$progress_file"
-
-        (
-            while true; do
-                done=$(wc -l <"$progress_file" 2>/dev/null | tr -d ' ')
-                printf "\r[release-r2]   %d / %d" "$done" "$uploaded"
-                [ "$done" -ge "$uploaded" ] && break
-                sleep 0.2
-            done
-        ) &
-        progress_pid=$!
-
-        xargs -0 -P "$R2_PARALLEL" -n 2 bash -c \
-            'aws s3 cp "$1" "s3://${R2_BUCKET}/$2" --endpoint-url "$R2_ENDPOINT" --quiet && echo >> "$PROGRESS_FILE"' _ \
-            <"$task_file"
-
-        wait "$progress_pid" 2>/dev/null
-        printf "\r[release-r2] ${dataset}: done (%d files uploaded)\n" "$uploaded"
-        rm -f "$progress_file"
-    fi
-    rm -f "$task_file"
-
-    # Delete R2 keys that no longer exist locally (batch DeleteObjects, 1000/req)
-    echo "[release-r2] ${dataset}: checking for stale R2 files..."
-    stale_keys=()
-    for r2_key in "${!r2_sizes[@]}"; do
-        if [ -z "${local_keys[$r2_key]+_}" ]; then
-            stale_keys+=("$r2_key")
-        fi
-    done
-
-    deleted=${#stale_keys[@]}
-    if [ "$deleted" -gt 0 ]; then
-        echo "[release-r2] ${dataset}: deleting $deleted stale file(s) in batches of 1000..."
-        i=0
-        while [ $i -lt $deleted ]; do
-            batch=("${stale_keys[@]:$i:1000}")
-            json=$(printf '{"Key":"%s"},' "${batch[@]}")
-            json="{\"Objects\":[${json%,}],\"Quiet\":true}"
-            aws s3api delete-objects \
-                --bucket "$R2_BUCKET" \
-                --endpoint-url "$R2_ENDPOINT" \
-                --delete "$json" \
-                --output text > /dev/null
-            i=$((i + 1000))
-            echo "[release-r2]   deleted $((i < deleted ? i : deleted)) / $deleted"
-        done
-        echo "[release-r2] ${dataset}: $deleted stale file(s) deleted"
-    else
-        echo "[release-r2] ${dataset}: no stale files"
-    fi
-
-    # Save hash so next run skips if nothing changed
-    echo "$current_hash" > "$sync_cache"
-
-    unset local_keys r2_sizes
-done
+PYTHON="${PYTHON:-.venv/bin/python3}"
+"$PYTHON" - "$datasets" <<'PYEOF'
+import sys
+sys.path.insert(0, 'importer')
+import wdalib
+for d in sys.argv[1].split(','):
+    d = d.strip()
+    if d:
+        wdalib.uploadToR2(d)
+PYEOF

importer/vigilo/README.md

@@ -0,0 +1,67 @@
+# Vigilo dataset
+
+## Overview
+
+Vigilo is a citizen reporting platform for cycling infrastructure issues in
+French cities. Each city runs its own Vigilo instance (called a _scope_).
+
+## Workflow
+
+```
+just download   →   just update   →   just release-cloudflare
+```
+
+### 1. `just download`
+
+Fetches live data from the Vigilo API (runs inside Docker):
+
+- `downloaded/vigilo/categories.json` — issue categories
+- `downloaded/vigilo/scopes.json` — active city scopes
+- `downloaded/vigilo/observations_<scope>.json` — observations per scope
+
+Only runs if the downloaded folder is older than 7 days.
+
+### 2. `just update`
+
+Merges new API data with historical data from R2 (runs inside Docker):
+
+1. **`downloadFromR2`** — downloads the latest `scopes.parquet` and
+   `observations.parquet` from R2 into `downloaded/vigilo/from_r2/raw/` using
+   the manifest. This ensures the merge always starts from the most recent
+   published state.
+2. **`_data2duckdb.sql`** — merges new observations with historical data:
+   - Active scopes: `first_seen_at` preserved from R2 history
+   - Inactive scopes (disappeared from API): kept with `is_active = false`
+   - Observations from inactive scopes: preserved from R2 history
+3. Outputs to `dataset/vigilo/raw/`
+
+### 3. `just release-cloudflare`
+
+Uploads the generated parquets to Cloudflare R2 (runs **locally**, needs R2
+credentials). Also generates `vigilo/manifest-vigilo.json` on R2, which is used
+by the next `just update` to download the historical data.
+
+Required env vars:
+
+Stored on `.env` file
+
+```
+CF_ACCOUNT_ID
+CF_R2_ACCESS_KEY_ID
+CF_R2_SECRET_ACCESS_KEY
+```
+
+## R2 naming convention
+
+Local path → R2 key: `/` replaced by `__`, `=` replaced by `.`
+
+```
+dataset/vigilo/raw/scopes.parquet  →  vigilo/raw__scopes.parquet
+```
+
+## First run (no prior R2 data)
+
+If no manifest exists on R2 yet, `downloadFromR2` skips gracefully and the SQL
+runs without historical data (all scopes treated as new,
+`first_seen_at = today`). Run `just release-cloudflare` after the first
+`just update` to publish data and create the manifest.

importer/vigilo/__init__.py

@@ -131,4 +131,5 @@ def download():
 
 def update():
     wdalib.show_title(f"Update {provider}")
+    wdalib.downloadFromR2(provider, subdir="raw")
     wdalib.data2duckdb(provider)

importer/vigilo/_data2duckdb.sql

@@ -60,7 +60,7 @@ INSERT INTO vigilo_scopes BY NAME
         true                                        AS is_active,
         COALESCE(
             (SELECT first_seen_at
-             FROM read_parquet('./dataset/vigilo/raw/scopes.parquet')
+             FROM read_parquet('./downloaded/vigilo/from_r2/raw/scopes.parquet')
              WHERE id = n.id LIMIT 1),
             CURRENT_DATE
         )                                           AS first_seen_at,
@@ -72,7 +72,7 @@ INSERT INTO vigilo_scopes BY NAME
 -- On force is_active = false quel que soit leur statut précédent (gère la première disparition)
 INSERT INTO vigilo_scopes BY NAME
     SELECT * REPLACE (false AS is_active)
-    FROM read_parquet('./dataset/vigilo/raw/scopes.parquet')
+    FROM read_parquet('./downloaded/vigilo/from_r2/raw/scopes.parquet')
     WHERE id NOT IN (SELECT id FROM vigilo_scopes_new);
 
 -------------------------------------------------------------------------------
@@ -109,7 +109,7 @@ INSERT INTO vigilo_observations
     SELECT scopeid, token, ts, latitude, longitude, address, "comment", explanation,
            catid, approved, cityname, geonames_districtid, geonames_district,
            geonames_cityid, geonames_city
-    FROM read_parquet('./dataset/vigilo/raw/observations.parquet')
+    FROM read_parquet('./downloaded/vigilo/from_r2/raw/observations.parquet')
     WHERE scopeid NOT IN (SELECT DISTINCT scopeid FROM vigilo_observations);
 
 -------------------------------------------------------------------------------