diff --git a/.github/workflows/github-pages.yml b/.github/workflows/github-pages.yml index 40e49d7..112a9d4 100644 --- a/.github/workflows/github-pages.yml +++ b/.github/workflows/github-pages.yml @@ -11,7 +11,7 @@ permissions: concurrency: group: pages - cancel-in-progress: true + cancel-in-progress: false env: JUST_VERSION: "1.51.0" diff --git a/README.md b/README.md index 28285db..7400af9 100644 --- a/README.md +++ b/README.md @@ -98,13 +98,6 @@ just update # Convert to Parquet (DuckDB) just import # Load into PostgreSQL ``` -To process a specific dataset: - -```bash -DATAS_LIST=sirene just update -DATAS_LIST=sirene just import -``` - > **Note:** `DATAS_LIST` accepts multiple comma-separated dataset names: > > ```bash @@ -112,7 +105,60 @@ DATAS_LIST=sirene just import > DATAS_LIST=geonames,vigilo,nafrev2,sirene just import > ``` -Available datasets: `geonames`, `vigilo`, `sirene`, `nafrev2` +### Per-dataset workflows + +Each dataset has two workflows: **update** (full refresh from source → publish to R2) and **import** (consume published Parquet files from R2). + +#### GeoNames + +```bash +# Full update (download from geonames.org, convert, publish) +DATAS_LIST=geonames just download +DATAS_LIST=geonames just update +DATAS_LIST=geonames just release + +# Import only (download Parquet from R2, load into DuckDB) +DATAS_LIST=geonames just import +``` + +#### Vigilo + +```bash +# Full update (download from Vigilo API, convert, publish) +DATAS_LIST=vigilo just download +DATAS_LIST=vigilo just update +DATAS_LIST=vigilo just release + +# Import only (download Parquet from R2, load into DuckDB) +DATAS_LIST=vigilo just import +``` + +#### NAF Rev. 2 + +```bash +# Full update (download from INSEE, convert, publish) +DATAS_LIST=nafrev2 just download +DATAS_LIST=nafrev2 just update +DATAS_LIST=nafrev2 just release + +# Import only (download Parquet from R2, load into DuckDB) +DATAS_LIST=nafrev2 just import +``` + +#### SIRENE + +```bash +# Full update (download from data.gouv.fr ~4 GB, convert, publish) +DATAS_LIST=sirene just download +DATAS_LIST=sirene just update +DATAS_LIST=sirene just release + +# Import only (download Parquet from R2, load into DuckDB) +DATAS_LIST=sirene just import +just sirene-etab-parquets # generate Observable etab/ parquets from APE parquets +``` + +> **Note:** SIRENE download is ~4 GB. `just sirene-etab-parquets` must be run after import to generate the per-letter establishment parquets used by the Observable interface. --- diff --git a/importer/release_download.sh b/importer/release_download.sh index 080efde..c2c5f2e 100755 --- a/importer/release_download.sh +++ b/importer/release_download.sh @@ -56,7 +56,7 @@ for dataset in $datasets; do echo "[release-download] ${dataset}: syncing from ${tag}..." python3 - "$manifest_local" "$dataset" "$FORCE" "$DL_TMP" "$REPO" "$tag" << 'PYEOF' -import sys, json, os, subprocess +import sys, json, os, shutil, subprocess manifest_path, dataset, force, dl_tmp, repo, tag = sys.argv[1:] force = force == "1" @@ -79,7 +79,7 @@ for asset_name, rel_path in manifest.items(): print(f"[release-download] ERROR downloading {asset_name}: {result.stderr.strip()}") continue downloaded = os.path.join(dl_tmp, asset_name) - os.rename(downloaded, target) + shutil.move(downloaded, target) print(f"[release-download] {asset_name} → {target}") PYEOF diff --git a/importer/vigilo/__init__.py b/importer/vigilo/__init__.py index 12a6098..82c4a2c 100644 --- a/importer/vigilo/__init__.py +++ b/importer/vigilo/__init__.py @@ -3,9 +3,6 @@ import json import os -import shutil -import subprocess -import tempfile import requests import wdalib @@ -14,21 +11,7 @@ REQUEST_TIMEOUT = 30 -def _write_download_results(scopes, instances, failures): - """Write the successful results and the failures collected during download.""" - wdalib.writeContentToFile( - f"./downloaded/{provider}/scopes.json", json.dumps(scopes) - ) - wdalib.writeContentToFile( - f"./downloaded/{provider}/instances.json", json.dumps(instances) - ) - wdalib.writeContentToFile( - f"./downloaded/{provider}/download_failures.json", json.dumps(failures) - ) - - def _record_failure(failures, name, scope, error): - """Record and display a failed city download.""" failure = { "name": name, "scope": scope, @@ -41,70 +24,69 @@ def _record_failure(failures, name, scope, error): def download_categories(): filename = f"./downloaded/{provider}/categories.json" - resp = requests.get( "https://vigilo-bf7f2.firebaseio.com/categorieslist.json", timeout=REQUEST_TIMEOUT, ) resp.raise_for_status() - wdalib.writeContentToFile(filename, resp.text) +def download_observations(scope, api_path): + filename = f"./downloaded/{provider}/observations_{scope}.json" + response = requests.get( + f"{api_path}/get_issues.php?scope={scope}&format=json", + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + response.json() + wdalib.writeContentToFile(filename, response.text) + + def download_scopes(): scopes = [] instances = [] failures = [] + VIGILO_CONF_URL = "https://github.com/jesuisundesdeux/vigilo-conf/releases/latest/download/check_result.json" try: - citiesresp = requests.get( - "https://vigilo-bf7f2.firebaseio.com/citylist.json", - timeout=REQUEST_TIMEOUT, - ) + citiesresp = requests.get(VIGILO_CONF_URL, timeout=REQUEST_TIMEOUT) citiesresp.raise_for_status() citiesresult = citiesresp.json() - if not isinstance(citiesresult, dict): - raise ValueError("city list is not a JSON object") + if not isinstance(citiesresult.get("instances"), list): + raise ValueError("check_result.json has no instances list") except (requests.exceptions.RequestException, ValueError) as error: _record_failure(failures, "citylist", "citylist", error) - _write_download_results(scopes, instances, failures) + _write_results(scopes, instances, failures) return - # Download - for name, value in citiesresult.items(): + for value in citiesresult["instances"]: scope = "unknown" try: - if not value["prod"]: + if not value.get("prod"): continue - value["name"] = name - + name = value.get("name", "unknown") scope = value.get("scope", "unknown") api_path = value["api_path"] - # Get scope informations scoperesp = requests.get( f"{api_path}/get_scope.php?scope={scope}", timeout=REQUEST_TIMEOUT, ) scoperesp.raise_for_status() - - # Merge with scope informations scoperesult = scoperesp.json() if not isinstance(scoperesult, dict): raise ValueError("scope response is not a JSON object") if not isinstance(scoperesult.get("cities"), list): raise ValueError("scope response has no cities list") value = value | scoperesult - value.pop("cities", None) value.pop("prod", None) - download_observations(scope, name, api_path) - + download_observations(scope, api_path) scopes.append(value) - # Cities informations for city in scoperesult["cities"]: city["scope"] = scope instances.append(city) @@ -117,19 +99,19 @@ def download_scopes(): ) as error: _record_failure(failures, name, scope, error) - _write_download_results(scopes, instances, failures) - + _write_results(scopes, instances, failures) -def download_observations(scope, name, api_path): - filename = f"./downloaded/{provider}/observations_{scope}.json" - response = requests.get( - f"{api_path}/get_issues.php?scope={scope}&format=json", - timeout=REQUEST_TIMEOUT, +def _write_results(scopes, instances, failures): + wdalib.writeContentToFile( + f"./downloaded/{provider}/scopes.json", json.dumps(scopes) + ) + wdalib.writeContentToFile( + f"./downloaded/{provider}/instances.json", json.dumps(instances) + ) + wdalib.writeContentToFile( + f"./downloaded/{provider}/download_failures.json", json.dumps(failures) ) - response.raise_for_status() - response.json() - wdalib.writeContentToFile(filename, response.text) def download(): @@ -147,148 +129,6 @@ def download(): download_scopes() -MIGRATION_DATE = "2025-05-11" - -_HISTORICAL_SCHEMA = ( - "id TEXT, name TEXT, display_name TEXT, iso TEXT, country TEXT, department BIGINT," - " lat_min DOUBLE, lat_max DOUBLE, lon_min DOUBLE, lon_max DOUBLE," - " map_center_string TEXT, map_zoom BIGINT, api_path TEXT, map_url TEXT," - " nominatim_urlbase TEXT, contact_email TEXT, tweet_content TEXT, twitter TEXT," - " backend_version TEXT, geonames_admin_filter TEXT, geonames_countryid BIGINT," - " is_active BOOLEAN, first_seen_at DATE, last_seen_at DATE" -) - -_BASE_COLS = ( - "id, name, display_name, iso, country, department," - " lat_min, lat_max, lon_min, lon_max, map_center_string, map_zoom," - " api_path, map_url, nominatim_urlbase, contact_email, tweet_content," - " twitter, backend_version, geonames_admin_filter, geonames_countryid" -) - - -def _run_duckdb_sql(sql): - with tempfile.NamedTemporaryFile(mode="w", suffix=".sql", delete=False) as f: - f.write(sql) - tmp_path = f.name - try: - subprocess.run(f"duckdb :memory: < {tmp_path}", shell=True, check=True) - finally: - os.remove(tmp_path) - - -def _save_scopes_historical(): - """Back up current scopes.parquet (normalized to new schema) before the dataset dir is wiped.""" - current = f"./dataset/{provider}/scopes.parquet" - historical = f"./downloaded/{provider}/scopes_historical.parquet" - - if not os.path.exists(current) or os.path.getsize(current) == 0: - if os.path.exists(historical): - # Previous backup survives a failed run — keep it, just enrich with partitions - _add_partition_only_scopes(historical) - return - # First ever run — create empty historical parquet with correct schema - _run_duckdb_sql( - f"CREATE OR REPLACE TABLE _h ({_HISTORICAL_SCHEMA});" - f" COPY _h TO '{historical}' (FORMAT PARQUET, COMPRESSION ZSTD);" - ) - _add_partition_only_scopes(historical) - return - - # Check if current parquet already has the tracking columns (post-migration) - check = subprocess.run( - f"duckdb :memory: -c \"SELECT is_active FROM read_parquet('{current}') LIMIT 0\"", - shell=True, - capture_output=True, - ) - if check.returncode == 0: - tracking = "is_active, first_seen_at, last_seen_at" - else: - tracking = ( - f"false AS is_active," - f" DATE '{MIGRATION_DATE}' AS first_seen_at," - f" DATE '{MIGRATION_DATE}' AS last_seen_at" - ) - - _run_duckdb_sql( - f"COPY (SELECT {_BASE_COLS}, {tracking} FROM read_parquet('{current}'))" - f" TO '{historical}' (FORMAT PARQUET, COMPRESSION ZSTD);" - ) - _add_partition_only_scopes(historical) - - -def _add_partition_only_scopes(historical): - """Add minimal scope entries for scopes found in observation partitions but absent from scopes_historical.""" - obs_dir = f"./dataset/{provider}/observations.parquet" - if not os.path.exists(obs_dir): - obs_dir = f"./downloaded/{provider}/observations_historical" - if not os.path.exists(obs_dir): - return - - discovered = [ - d[len("scopeid="):] - for d in os.listdir(obs_dir) - if d.startswith("scopeid=") - ] - if not discovered: - return - - values = ", ".join(f"('{sid}')" for sid in discovered) - _run_duckdb_sql( - f"CREATE OR REPLACE TABLE _disc AS" - f" SELECT id," - f" NULL::TEXT AS name, NULL::TEXT AS display_name, NULL::TEXT AS iso," - f" NULL::TEXT AS country, NULL::BIGINT AS department," - f" NULL::DOUBLE AS lat_min, NULL::DOUBLE AS lat_max," - f" NULL::DOUBLE AS lon_min, NULL::DOUBLE AS lon_max," - f" NULL::TEXT AS map_center_string, NULL::BIGINT AS map_zoom," - f" NULL::TEXT AS api_path, NULL::TEXT AS map_url," - f" NULL::TEXT AS nominatim_urlbase, NULL::TEXT AS contact_email," - f" NULL::TEXT AS tweet_content, NULL::TEXT AS twitter," - f" NULL::TEXT AS backend_version, NULL::TEXT AS geonames_admin_filter," - f" NULL::BIGINT AS geonames_countryid," - f" false AS is_active," - f" DATE '{MIGRATION_DATE}' AS first_seen_at," - f" DATE '{MIGRATION_DATE}' AS last_seen_at" - f" FROM (VALUES {values}) t(id)" - f" WHERE id NOT IN (SELECT id FROM read_parquet('{historical}'));" - f"CREATE OR REPLACE TABLE _merged AS" - f" SELECT * FROM read_parquet('{historical}')" - f" UNION ALL SELECT * FROM _disc;" - f"COPY _merged TO '{historical}' (FORMAT PARQUET, COMPRESSION ZSTD);" - ) - - -def _save_observations_historical(): - """Back up all observation partitions before the dataset dir is wiped.""" - current = f"./dataset/{provider}/observations.parquet" - historical = f"./downloaded/{provider}/observations_historical" - - if not os.path.exists(current): - # No current data — keep existing historical backup from a previous run if any - return - if os.path.exists(historical): - shutil.rmtree(historical) - shutil.copytree(current, historical) - - -def _restore_historical_observations(): - """Restore observation partitions for inactive scopes after the SQL update.""" - historical = f"./downloaded/{provider}/observations_historical" - current = f"./dataset/{provider}/observations.parquet" - - if not os.path.exists(historical): - return - - for scope_dir in os.listdir(historical): - src = os.path.join(historical, scope_dir) - dst = os.path.join(current, scope_dir) - if not os.path.exists(dst): - shutil.copytree(src, dst) - - def update(): - _save_scopes_historical() - _save_observations_historical() - wdalib.init_dataset(provider) + wdalib.show_title(f"Update {provider}") wdalib.data2duckdb(provider) - _restore_historical_observations() diff --git a/importer/vigilo/_data2duckdb.sql b/importer/vigilo/_data2duckdb.sql index 3132a89..e39cbff 100644 --- a/importer/vigilo/_data2duckdb.sql +++ b/importer/vigilo/_data2duckdb.sql @@ -1,421 +1,199 @@ BEGIN TRANSACTION; ------------------------------------------------------------------------------- --- Import +-- Categories ------------------------------------------------------------------------------- --- categories -CREATE OR REPLACE TABLE vigilo_categories ( - id BIGINT, - name TEXT, - name_en TEXT, - color TEXT -) -; - -INSERT INTO vigilo_categories - SELECT catid,catname,catname_en_US,catcolor - FROM read_json('./downloaded/vigilo/categories.json') -; - --- scopes from current download (active) -CREATE OR REPLACE TABLE vigilo_scopes_new ( - id TEXT, - name TEXT, - display_name TEXT, - iso TEXT, - country TEXT, - department BIGINT, - lat_min DOUBLE, - lat_max DOUBLE, - lon_min DOUBLE, - lon_max DOUBLE, - map_center_string TEXT, - map_zoom BIGINT, - api_path TEXT, - map_url TEXT, - nominatim_urlbase TEXT, - contact_email TEXT, - tweet_content TEXT, - twitter TEXT, - backend_version TEXT, - geonames_admin_filter TEXT, - geonames_countryid BIGINT -) -; +CREATE OR REPLACE TABLE vigilo_categories AS + SELECT catid AS id, catname AS name, catname_en_US AS name_en, catcolor AS color + FROM read_json('./downloaded/vigilo/categories.json'); -INSERT INTO vigilo_scopes_new - SELECT scope, - name, - display_name, - NULL, - country, - department, - coordinate_lat_min, - coordinate_lat_max, - coordinate_lon_min, - coordinate_lon_max, - map_center_string, - map_zoom,api_path, - map_url, - nominatim_urlbase, - contact_email, - tweet_content, - twitter, - backend_version, - NULL, - NULL - FROM read_json('./downloaded/vigilo/scopes.json') -; +------------------------------------------------------------------------------- +-- Scopes actifs (depuis l'API Vigilo) +------------------------------------------------------------------------------- +CREATE OR REPLACE TABLE vigilo_scopes_new AS + SELECT + scope AS id, + name, + display_name, + NULL::TEXT AS iso, + 'France' AS country, + department, + TRY_CAST(coordinate_lat_min AS DOUBLE) AS lat_min, + TRY_CAST(coordinate_lat_max AS DOUBLE) AS lat_max, + TRY_CAST(coordinate_lon_min AS DOUBLE) AS lon_min, + TRY_CAST(coordinate_lon_max AS DOUBLE) AS lon_max, + map_center_string, + TRY_CAST(map_zoom AS BIGINT) AS map_zoom, + api_path, + map_url, + nominatim_urlbase, + contact_email, + tweet_content, + twitter, + backend_version, + NULL::TEXT AS geonames_admin_filter, + NULL::BIGINT AS geonames_countryid, + version, + "count" AS check_count, + issues_time, + TRY_CAST("last" AS DATE) AS last_check_date, + ok AS check_ok + FROM read_json('./downloaded/vigilo/scopes.json'); --- historical scopes: previous release parquet downloaded by download_last_dataset() --- always exists (created empty with correct schema if no release available) --- Explicit column list to handle schema evolution (old releases lack is_active, first_seen_at, etc.) -CREATE OR REPLACE TABLE vigilo_scopes_historical ( - id TEXT, name TEXT, display_name TEXT, iso TEXT, country TEXT, department BIGINT, +------------------------------------------------------------------------------- +-- Scopes fusionnés : actifs (API) + inactifs (parquet existant) +------------------------------------------------------------------------------- +CREATE OR REPLACE TABLE vigilo_scopes ( + id TEXT, name TEXT, display_name TEXT, iso TEXT, country TEXT, department TEXT, lat_min DOUBLE, lat_max DOUBLE, lon_min DOUBLE, lon_max DOUBLE, map_center_string TEXT, map_zoom BIGINT, api_path TEXT, map_url TEXT, nominatim_urlbase TEXT, contact_email TEXT, tweet_content TEXT, twitter TEXT, backend_version TEXT, geonames_admin_filter TEXT, geonames_countryid BIGINT, - is_active BOOLEAN, first_seen_at DATE, last_seen_at DATE, nb_observations BIGINT + is_active BOOLEAN, first_seen_at DATE, last_seen_at DATE, nb_observations BIGINT, + version TEXT, check_count BIGINT, issues_time DOUBLE, last_check_date DATE, check_ok BOOLEAN ); -INSERT INTO vigilo_scopes_historical +-- Scopes actifs : préserve first_seen_at depuis le parquet existant si connu +INSERT INTO vigilo_scopes BY NAME SELECT - id, name, display_name, iso, country, department, - lat_min, lat_max, lon_min, lon_max, - map_center_string, map_zoom, api_path, map_url, - nominatim_urlbase, contact_email, tweet_content, twitter, - backend_version, geonames_admin_filter, geonames_countryid, - NULL::BOOLEAN AS is_active, - NULL::DATE AS first_seen_at, - NULL::DATE AS last_seen_at, - NULL::BIGINT AS nb_observations - FROM read_parquet('./downloaded/vigilo/last_release/scopes.parquet') -; + n.*, + true AS is_active, + COALESCE( + (SELECT first_seen_at + FROM read_parquet('./dataset/vigilo/raw/scopes.parquet') + WHERE id = n.id LIMIT 1), + CURRENT_DATE + ) AS first_seen_at, + CURRENT_DATE AS last_seen_at, + 0::BIGINT AS nb_observations + FROM vigilo_scopes_new n; + +-- Scopes inactifs : présents dans le parquet existant mais absents du dernier download +-- 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') + WHERE id NOT IN (SELECT id FROM vigilo_scopes_new); --- merged scopes: active (current download) + inactive (historical only) -CREATE OR REPLACE TABLE vigilo_scopes ( - id TEXT, - name TEXT, - display_name TEXT, - iso TEXT, - country TEXT, - department BIGINT, - lat_min DOUBLE, - lat_max DOUBLE, - lon_min DOUBLE, - lon_max DOUBLE, - map_center_string TEXT, - map_zoom BIGINT, - api_path TEXT, - map_url TEXT, - nominatim_urlbase TEXT, - contact_email TEXT, - tweet_content TEXT, - twitter TEXT, - backend_version TEXT, - geonames_admin_filter TEXT, - geonames_countryid BIGINT, - is_active BOOLEAN, - first_seen_at DATE, - last_seen_at DATE, - nb_observations BIGINT -) -; - --- Active scopes: preserve first_seen_at from history when already known -INSERT INTO vigilo_scopes - SELECT - n.id, - n.name, - n.display_name, - n.iso, - n.country, - n.department, - n.lat_min, - n.lat_max, - n.lon_min, - n.lon_max, - n.map_center_string, - n.map_zoom, - n.api_path, - n.map_url, - n.nominatim_urlbase, - n.contact_email, - n.tweet_content, - n.twitter, - n.backend_version, - n.geonames_admin_filter, - n.geonames_countryid, - true AS is_active, - COALESCE(h.first_seen_at, CURRENT_DATE) AS first_seen_at, - CURRENT_DATE AS last_seen_at, - COALESCE(h.nb_observations, 0) AS nb_observations - FROM vigilo_scopes_new n - LEFT JOIN vigilo_scopes_historical h ON h.id = n.id -; - --- Inactive scopes: previously known but absent from current download --- nb_observations preserved from last_release (includes orphan scopes already enriched there) -INSERT INTO vigilo_scopes - SELECT - h.id, h.name, h.display_name, h.iso, h.country, h.department, - h.lat_min, h.lat_max, h.lon_min, h.lon_max, - h.map_center_string, h.map_zoom, h.api_path, h.map_url, - h.nominatim_urlbase, h.contact_email, h.tweet_content, h.twitter, - h.backend_version, h.geonames_admin_filter, h.geonames_countryid, - false AS is_active, - h.first_seen_at, - h.last_seen_at, - COALESCE(h.nb_observations, 0) AS nb_observations - FROM vigilo_scopes_historical h - WHERE h.id NOT IN (SELECT id FROM vigilo_scopes_new) -; - --- observations +------------------------------------------------------------------------------- +-- Observations actives (depuis les JSON téléchargés) +------------------------------------------------------------------------------- CREATE OR REPLACE TABLE vigilo_observations ( - scopeid TEXT, - token TEXT, - ts BIGINT, - latitude DOUBLE, - longitude DOUBLE, - address TEXT, - "comment" TEXT, - explanation TEXT, - catid BIGINT, - approved BIGINT, - cityname TEXT, - geonames_districtid BIGINT, - geonames_district TEXT, - geonames_cityid BIGINT, - geonames_city TEXT -) -; + scopeid TEXT, token TEXT, ts BIGINT, latitude DOUBLE, longitude DOUBLE, + address TEXT, "comment" TEXT, explanation TEXT, catid BIGINT, approved BIGINT, + cityname TEXT, geonames_districtid BIGINT, geonames_district TEXT, + geonames_cityid BIGINT, geonames_city TEXT +); INSERT INTO vigilo_observations - SELECT regexp_extract(filename,'.*([0-9][0-9]_.*)\..*',1), + SELECT + regexp_extract(filename, '.*observations_(.*)\.json', 1) AS scopeid, token, - "time", - coordinates_lat, - coordinates_lon, + "time" AS ts, + coordinates_lat AS latitude, + coordinates_lon AS longitude, address, "comment", explanation, - categorie, + categorie AS catid, approved, cityname, - NULL, - NULL, - NULL, - NULL - FROM read_json('./downloaded/vigilo/observations_*.json',filename = true) -; --- historical observations: inactive scopes not covered by fresh JSON download + NULL::BIGINT AS geonames_districtid, + NULL::TEXT AS geonames_district, + NULL::BIGINT AS geonames_cityid, + NULL::TEXT AS geonames_city + FROM read_json('./downloaded/vigilo/observations_*.json', filename=true); + +-- Observations des scopes inactifs (depuis le parquet existant) INSERT INTO vigilo_observations - SELECT * FROM read_parquet('./downloaded/vigilo/last_release/observations.parquet') - WHERE scopeid NOT IN (SELECT DISTINCT scopeid FROM vigilo_observations) -; -DROP INDEX IF EXISTS idx_vigilo_observations_lat_lon; -CREATE INDEX idx_vigilo_observations_lat_lon ON vigilo_observations (latitude,longitude); + 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') + WHERE scopeid NOT IN (SELECT DISTINCT scopeid FROM vigilo_observations); ------------------------------------------------------------------------------- --- Fix +-- Enrichissement geonames ------------------------------------------------------------------------------- --- Find geoname ISO code UPDATE vigilo_scopes vs - SET iso = ( - SELECT iso - FROM geonames_countries gc - WHERE vs.country = gc.country - ) -; + SET iso = (SELECT iso FROM geonames_countries gc WHERE vs.country = gc.country LIMIT 1); UPDATE vigilo_scopes vs - SET geonames_countryid = ( - SELECT geonameid - FROM geonames_countries gc - WHERE vs.iso = gc.iso - ) -; --- Define filter for getting a city values (used in SQL query) -UPDATE vigilo_scopes SET geonames_admin_filter = 'ADM4' WHERE iso = 'FR'; - --- clean² -UPDATE vigilo_observations set cityname=trim(cityname); --- DELETE FROM vigilo_observations WHERE cityname = 'undefined'; --- DELETE FROM vigilo_observations WHERE cityname IS NULL; --- DELETE FROM vigilo_observations WHERE cityname = 'Ille-et-Vilaine'; - --- UPDATE vigilo_observations SET cityname = 'Noyal-sur-Vilaine' WHERE cityname = 'Noyal-Sur-Vilaine'; --- UPDATE vigilo_observations SET cityname = 'Saint-Marcellin' WHERE cityname = 'Saint Marcellin'; --- UPDATE vigilo_observations SET cityname = 'Servon-sur-Vilaine' WHERE cityname = 'Servon-Sur-Vilaine'; --- UPDATE vigilo_observations SET cityname = 'Hœnheim' WHERE cityname = 'Hoenheim'; --- UPDATE vigilo_observations SET cityname = 'Brest' WHERE cityname = '29200 brest'; --- UPDATE vigilo_observations SET cityname = 'Brest' WHERE cityname = 'Brest.'; --- UPDATE vigilo_observations SET cityname = 'Saint-Aubin-de-Médoc' WHERE cityname = 'Saint-Aubin de Médoc'; --- UPDATE vigilo_observations SET cityname = 'Saint-Sauveur' WHERE cityname = 'Saint Sauveur'; --- UPDATE vigilo_observations SET cityname = 'La Teste-de-Buch' WHERE cityname = 'La Teste de Buch'; --- UPDATE vigilo_observations SET cityname = 'L''Isle-d''Abeau' WHERE cityname = 'L''Isle-dAbeau'; --- UPDATE vigilo_observations SET cityname = 'Lattes' WHERE cityname = 'Boirargues'; --- UPDATE vigilo_observations SET cityname = 'Saint-Vérand' WHERE cityname = 'Saint Vérand'; - --- DELETE FROM vigilo_observations WHERE cityname IN (SELECT cityname FROM vigilo_observations WHERE geonames_cityid IS NULL GROUP BY cityname having count()=1); --- DELETE FROM vigilo_observations WHERE cityname IN ('Heonheim'); + SET geonames_countryid = (SELECT geonameid FROM geonames_countries gc WHERE vs.iso = gc.iso LIMIT 1); --- find geoname cityid --- UPDATE vigilo_observations vo --- SET geonames_cityid = ( --- SELECT gc.id --- FROM vigilo_observations vo1 LEFT JOIN vigilo_scopes vs ON vo1.scopeid = vs.id --- LEFT JOIN geonames_allentries gc ON vs.iso = gc.country_code AND gc.feature_class='A' AND gc.feature_code=vs.geonames_admin_filter AND vo1.cityname = gc.name --- WHERE vo1.token = vo.token --- ) --- ; +UPDATE vigilo_scopes SET geonames_admin_filter = 'ADM4' WHERE iso = 'FR'; UPDATE vigilo_observations AS vo SET geonames_districtid = ( - SELECT g.id - FROM geonames_latlon_cache AS g - WHERE g.latlon = vo.latitude || '-' || vo.longitude + SELECT g.id FROM geonames_latlon_cache AS g + WHERE g.latlon = vo.latitude || '-' || vo.longitude LIMIT 1 ) WHERE EXISTS ( - SELECT 1 - FROM geonames_latlon_cache AS g + SELECT 1 FROM geonames_latlon_cache AS g WHERE g.latlon = vo.latitude || '-' || vo.longitude ); UPDATE vigilo_observations vo SET geonames_districtid = ( - SELECT id - FROM geonames_allentries - WHERE feature_class = 'P' and latitude > vo.latitude - 0.001 and latitude < vo.latitude + 0.001 and longitude > vo.longitude - 0.001 and longitude < vo.longitude + 0.001 - ORDER BY sqrt( power((latitude - vo.latitude), 2) + power((longitude - vo.longitude), 2) ) ASC - LIMIT 1 -) -WHERE geonames_districtid IS NULL -; - -UPDATE vigilo_observations vo -SET geonames_districtid = ( - SELECT id - FROM geonames_allentries - WHERE feature_class = 'P' and latitude > vo.latitude - 0.005 and latitude < vo.latitude + 0.005 and longitude > vo.longitude - 0.005 and longitude < vo.longitude + 0.005 - ORDER BY sqrt( power((latitude - vo.latitude), 2) + power((longitude - vo.longitude), 2) ) ASC - LIMIT 1 -) -WHERE geonames_districtid IS NULL -; - -UPDATE vigilo_observations vo -SET geonames_districtid = ( - SELECT id - FROM geonames_allentries - WHERE feature_class = 'P' and latitude > vo.latitude - 0.01 and latitude < vo.latitude + 0.01 and longitude > vo.longitude - 0.01 and longitude < vo.longitude + 0.01 - ORDER BY sqrt( power((latitude - vo.latitude), 2) + power((longitude - vo.longitude), 2) ) ASC - LIMIT 1 -) -WHERE geonames_districtid IS NULL -; - -UPDATE vigilo_observations vo -SET geonames_districtid = ( - SELECT id - FROM geonames_allentries - WHERE feature_class = 'P' and latitude > vo.latitude - 0.05 and latitude < vo.latitude + 0.05 and longitude > vo.longitude - 0.05 and longitude < vo.longitude + 0.05 - ORDER BY sqrt( power((latitude - vo.latitude), 2) + power((longitude - vo.longitude), 2) ) ASC + SELECT id FROM geonames_allentries + WHERE feature_class = 'P' + AND latitude BETWEEN vo.latitude - 0.05 AND vo.latitude + 0.05 + AND longitude BETWEEN vo.longitude - 0.05 AND vo.longitude + 0.05 + ORDER BY sqrt(power(latitude - vo.latitude, 2) + power(longitude - vo.longitude, 2)) LIMIT 1 ) -WHERE geonames_districtid IS NULL -; +WHERE geonames_districtid IS NULL; UPDATE vigilo_observations vo -SET geonames_district = ( - SELECT name - FROM geonames_allentries - WHERE id = vo.geonames_districtid -) +SET geonames_district = (SELECT name FROM geonames_allentries WHERE id = vo.geonames_districtid LIMIT 1) WHERE geonames_districtid IS NOT NULL; UPDATE vigilo_observations vo -SET geonames_cityid = ( - SELECT city_id - FROM geonames_allentries - WHERE id = vo.geonames_districtid -) +SET geonames_cityid = (SELECT city_id FROM geonames_allentries WHERE id = vo.geonames_districtid LIMIT 1) WHERE geonames_districtid IS NOT NULL; UPDATE vigilo_observations vo -SET geonames_city = ( - SELECT city_name - FROM geonames_allentries - WHERE id = vo.geonames_districtid -) +SET geonames_city = (SELECT city_name FROM geonames_allentries WHERE id = vo.geonames_districtid LIMIT 1) WHERE geonames_districtid IS NOT NULL; INSERT INTO geonames_latlon_cache - SELECT geonames_districtid,latitude || '-' || longitude as latlon FROM vigilo_observations WHERE latlon NOT IN (SELECT latlon FROM geonames_latlon_cache) -; -COPY geonames_latlon_cache TO './dataset/geonames/raw/latlon_cache.parquet' (FORMAT 'parquet', COMPRESSION 'zstd'); + SELECT geonames_districtid, latitude || '-' || longitude AS latlon + FROM vigilo_observations + WHERE geonames_districtid IS NOT NULL + AND (latitude || '-' || longitude) NOT IN (SELECT latlon FROM geonames_latlon_cache); + +COPY geonames_latlon_cache TO './dataset/geonames/raw/latlon_cache.parquet' (FORMAT PARQUET, COMPRESSION ZSTD); ------------------------------------------------------------------------------- --- Orphan scopes: scopeids in vigilo_observations but absent from active scopes --- (e.g. scopes removed from the Vigilo API but with historical observations) --- Stats (nb_observations, first_seen_at, last_seen_at) are filled by the UPDATE below. +-- Mise à jour des stats des scopes ------------------------------------------------------------------------------- -INSERT INTO vigilo_scopes - SELECT - o.scopeid AS id, - COALESCE(h.name, o.scopeid) AS name, - COALESCE(h.display_name, o.scopeid) AS display_name, - h.iso, h.country, h.department, - h.lat_min, h.lat_max, h.lon_min, h.lon_max, - h.map_center_string, h.map_zoom, h.api_path, h.map_url, - h.nominatim_urlbase, h.contact_email, h.tweet_content, h.twitter, - h.backend_version, h.geonames_admin_filter, h.geonames_countryid, - false AS is_active, - NULL AS first_seen_at, - NULL AS last_seen_at, - 0 AS nb_observations - FROM (SELECT DISTINCT scopeid FROM vigilo_observations) o - LEFT JOIN vigilo_scopes_historical h ON h.id = o.scopeid - WHERE o.scopeid NOT IN (SELECT id FROM vigilo_scopes) -; - --- Update stats for all scopes from vigilo_observations (fresh + historical) UPDATE vigilo_scopes vs -SET - nb_observations = stats.nb_observations, +SET nb_observations = stats.nb_observations, first_seen_at = stats.first_seen_at, last_seen_at = stats.last_seen_at FROM ( - SELECT - scopeid AS id, - COUNT(*) AS nb_observations, - MIN(to_timestamp(ts)::DATE) AS first_seen_at, - MAX(to_timestamp(ts)::DATE) AS last_seen_at + SELECT scopeid AS id, + COUNT(*) AS nb_observations, + MIN(to_timestamp(ts::DOUBLE)::DATE) AS first_seen_at, + MAX(to_timestamp(ts::DOUBLE)::DATE) AS last_seen_at FROM vigilo_observations GROUP BY scopeid ) stats -WHERE vs.id = stats.id -; +WHERE vs.id = stats.id; --- Fallback: scopes with no metadata get their id as display name UPDATE vigilo_scopes SET name = id, display_name = id WHERE name IS NULL; ------------------------------------------------------------------------------- --- Cleanup temporary tables +-- Nettoyage ------------------------------------------------------------------------------- DROP TABLE IF EXISTS vigilo_scopes_new; -DROP TABLE IF EXISTS vigilo_scopes_historical; ------------------------------------------------------------------------------- --- export to ./dataset +-- Export parquets ------------------------------------------------------------------------------- -COPY vigilo_categories TO './dataset/vigilo/raw/categories.parquet' (FORMAT 'parquet', COMPRESSION 'zstd', ROW_GROUP_SIZE 4096); -COPY vigilo_scopes TO './dataset/vigilo/raw/scopes.parquet' (FORMAT 'parquet', COMPRESSION 'zstd', ROW_GROUP_SIZE 4096); - +COPY vigilo_categories TO './dataset/vigilo/raw/categories.parquet' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 4096); +COPY vigilo_scopes TO './dataset/vigilo/raw/scopes.parquet' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 4096); COPY (SELECT * FROM vigilo_observations ORDER BY scopeid) -TO './dataset/vigilo/raw/observations.parquet' -(FORMAT 'parquet', COMPRESSION 'zstd', ROW_GROUP_SIZE 4096); + TO './dataset/vigilo/raw/observations.parquet' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 4096); COMMIT; diff --git a/justfile b/justfile index c559f3a..c892107 100644 --- a/justfile +++ b/justfile @@ -7,6 +7,7 @@ envname:=`basename $(pwd)` dockerimage:='badele/world-datas-analysis:latest' dockerimage_push:='badele/world-datas-analysis:latest' observabledir:='observable' +WDA_PUBLIC_DATASET_URL := env("WDA_PUBLIC_DATASET_URL", "https://pub-6526c18d68154746a16baf2f76a38544.r2.dev") # This help @help: @@ -76,7 +77,7 @@ docker-show-context-size: # Run the wda docker image [group('docker')] @docker-run CMD="": docker-build - docker run --net host -i --rm -e DATAS_LIST="$DATAS_LIST" -v $(pwd):/wda -v $(pwd)/dataset:/var/lib/postgresql/data/dataset -w /wda {{ dockerimage }} {{ CMD }} + docker run --net host -i --rm -e DATAS_LIST="$DATAS_LIST" -e WDA_PUBLIC_DATASET_URL="{{ WDA_PUBLIC_DATASET_URL }}" -v $(pwd):/wda -v $(pwd)/dataset:/var/lib/postgresql/data/dataset -w /wda {{ dockerimage }} {{ CMD }} ############################################################################### # DB diff --git a/observable/src/data/vigilo-stats.json.js b/observable/src/data/vigilo-stats.json.js index 351f035..23c62d1 100644 --- a/observable/src/data/vigilo-stats.json.js +++ b/observable/src/data/vigilo-stats.json.js @@ -5,7 +5,7 @@ const obs = parquetPath("vigilo", "observations.parquet"); await runDuckDBLoader( "vigilo-stats", - `SELECT s.id, s.display_name, s.is_active, + `SELECT s.id, s.display_name, s.iso, s.is_active, COUNT(o.token)::INTEGER AS count, MIN(o.ts) AS first_ts, MAX(o.ts) AS last_ts, (SELECT COUNT(*)::INTEGER FROM read_parquet(${JSON.stringify( @@ -19,7 +19,7 @@ await runDuckDBLoader( )})) AS grand_total_obs FROM read_parquet(${JSON.stringify(scopes)}) s LEFT JOIN read_parquet(${JSON.stringify(obs)}) o ON o.scopeid = s.id - GROUP BY s.id, s.display_name, s.is_active + GROUP BY s.id, s.display_name, s.iso, s.is_active HAVING COUNT(o.token) > 200 ORDER BY count DESC`, ); diff --git a/observable/src/vigilo.md b/observable/src/vigilo.md index d7c9f0a..f033d11 100644 --- a/observable/src/vigilo.md +++ b/observable/src/vigilo.md @@ -151,8 +151,20 @@ const fig3 = Plot.plot({ y: "display_name", fill: "status", r: 5, + channels: { + Instance: "display_name", + Date: (d) => fmtDate(d.x), + Statut: "status", + }, tip: { - format: { y: false, x: (d) => fmtDate(d.x) }, + format: { + Instance: true, + Date: true, + Statut: true, + x: false, + y: false, + fill: false, + }, fill: theme.surface, stroke: theme.border, }, @@ -249,17 +261,27 @@ const scopeOrder = [...totalsPerScope.entries()] .sort((a, b) => b.total - a.total) .map((d) => d.name); -// Ordre graphe 2 : par % des 2 premières catégories décroissant -const TOP2_CATS = [categoryOrder[0], categoryOrder[1]]; +// Ordre graphe 2 : tri multi-clé par % de chaque catégorie dans l'ordre CATEGORY_ORDER const scopeOrder2 = [...totalsPerScope.entries()] .map(([name, rows]) => { const total = rows.reduce((s, r) => s + r.count, 0); - const top2 = rows - .filter((r) => TOP2_CATS.includes(r.category)) - .reduce((s, r) => s + r.count, 0); - return { name, top2Pct: total > 0 ? top2 / total : 0 }; + const pct = Object.fromEntries( + categoryOrder.map((cat) => { + const n = rows + .filter((r) => r.category === cat) + .reduce((s, r) => s + r.count, 0); + return [cat, total > 0 ? n / total : 0]; + }), + ); + return { name, pct }; + }) + .sort((a, b) => { + for (const cat of categoryOrder) { + const diff = b.pct[cat] - a.pct[cat]; + if (Math.abs(diff) > 1e-10) return diff; + } + return 0; }) - .sort((a, b) => b.top2Pct - a.top2Pct) .map((d) => d.name); const barOpts = { @@ -464,7 +486,15 @@ display( stats .filter((d) => d.count > 0) .map((d) => ({ - Instance: d.display_name, + Instance: { + href: `https://vigilo.city/fr/villes/${(d.display_name + "-" + d.iso) + .normalize("NFD") + .replace(/[̀-ͯ]/g, "") + .toLowerCase() + .replace(/[^a-z0-9]+/g, "-") + .replace(/^-|-$/g, "")}`, + text: d.display_name, + }, Statut: d.is_active ? "Actif" : "Inactif", Signalements: Number(d.count), "Première obs.": fmt(d.first_ts), @@ -473,7 +503,22 @@ display( { sort: "Signalements", reverse: true, + columns: [ + "Instance", + "Statut", + "Signalements", + "Première obs.", + "Dernière obs.", + ], + width: { Instance: 220 }, format: { + Instance: ({ href, text }) => + Object.assign(document.createElement("a"), { + href, + target: "_blank", + textContent: text, + style: "white-space: nowrap", + }), Statut: (v) => { const span = document.createElement("span"); span.textContent = v; diff --git a/tests/test_vigilo.py b/tests/test_vigilo.py index a6bfd09..c1536e3 100644 --- a/tests/test_vigilo.py +++ b/tests/test_vigilo.py @@ -1,17 +1,14 @@ """Unit tests for the fault-tolerant Vigilo downloader.""" import json -import os -import shutil import sys import tempfile import unittest from pathlib import Path -from unittest.mock import Mock, call, patch +from unittest.mock import Mock, patch import requests - sys.path.insert(0, str(Path(__file__).parents[1] / "importer")) import vigilo @@ -22,48 +19,35 @@ class VigiloDownloadTests(unittest.TestCase): def response(self, payload, status_code=200, text=None): """Build a fake requests.Response object for a controlled test case.""" - response = Mock() - response.status_code = status_code - response.text = text if text is not None else json.dumps(payload) - response.json.return_value = payload + r = Mock() + r.status_code = status_code + r.text = text if text is not None else json.dumps(payload) + r.json.return_value = payload if status_code >= 400: from requests.exceptions import HTTPError + r.raise_for_status.side_effect = HTTPError(f"HTTP {status_code}") + return r - response.raise_for_status.side_effect = HTTPError( - f"HTTP {status_code}" - ) - return response - - # Replace the network call so the test does not access Vigilo. - # Replace print because the failure is expected and is asserted below. @patch("vigilo.print") @patch("vigilo.requests.get") def test_failed_city_does_not_stop_other_cities(self, get, print_mock): """Keep the successful city when another city has a network failure.""" city_list = { - "Good city": { - "prod": True, - "scope": "good", - "api_path": "https://good.example", - }, - "Broken city": { - "prod": True, - "scope": "broken", - "api_path": "https://broken.example", - }, + "instances": [ + {"prod": True, "name": "Good city", "scope": "good", "api_path": "https://good.example"}, + {"prod": True, "name": "Broken city", "scope": "broken", "api_path": "https://broken.example"}, + ] } scope = {"cities": [{"id": 1}]} - # The first three calls are: city list, successful scope, observations. - # The fourth call simulates a connection failure for the broken city. - responses = [ + # Calls: check_result.json, good scope, good observations, broken scope (fails) + get.side_effect = [ self.response(city_list), self.response(scope), self.response({"observations": []}), + requests.exceptions.ConnectionError("offline"), ] - get.side_effect = responses + [requests.exceptions.ConnectionError("offline")] with tempfile.TemporaryDirectory() as directory: - # Redirect generated files to a temporary directory. with patch("vigilo.provider", "vigilo"), patch( "vigilo.wdalib.writeContentToFile", side_effect=lambda filename, content: Path(directory, Path(filename).name).write_text(content), @@ -71,9 +55,7 @@ def test_failed_city_does_not_stop_other_cities(self, get, print_mock): vigilo.download_scopes() scopes = json.loads(Path(directory, "scopes.json").read_text()) - failures = json.loads( - Path(directory, "download_failures.json").read_text() - ) + failures = json.loads(Path(directory, "download_failures.json").read_text()) self.assertEqual(len(scopes), 1) self.assertEqual(scopes[0]["name"], "Good city") @@ -81,180 +63,73 @@ def test_failed_city_does_not_stop_other_cities(self, get, print_mock): self.assertEqual(failures[0]["name"], "Broken city") print_mock.assert_called_once() - # Simulate an invalid API response without producing expected error output. @patch("vigilo.print") @patch("vigilo.requests.get") def test_invalid_scope_json_is_recorded(self, get, print_mock): """Record a malformed scope response and continue without crashing.""" city_list = { - "Broken city": { - "prod": True, - "scope": "broken", - "api_path": "https://broken.example", - } + "instances": [ + {"prod": True, "name": "Broken city", "scope": "broken", "api_path": "https://broken.example"}, + ] } invalid_response = self.response(None, text="not json") - # requests raises a JSON decoding error when response.json() is called. invalid_response.json.side_effect = ValueError("invalid json") get.side_effect = [self.response(city_list), invalid_response] with tempfile.TemporaryDirectory() as directory: - # Redirect generated files to a temporary directory. with patch( "vigilo.wdalib.writeContentToFile", side_effect=lambda filename, content: Path(directory, Path(filename).name).write_text(content), ): vigilo.download_scopes() - failures = json.loads( - Path(directory, "download_failures.json").read_text() - ) + failures = json.loads(Path(directory, "download_failures.json").read_text()) self.assertEqual(failures[0]["name"], "Broken city") self.assertEqual(failures[0]["error"], "ValueError") print_mock.assert_called_once() + @patch("vigilo.print") + @patch("vigilo.requests.get") + def test_nonprod_instances_are_skipped(self, get, _print_mock): + """Instances with prod=False are ignored.""" + city_list = { + "instances": [ + {"prod": False, "name": "Dev city", "scope": "dev", "api_path": "https://dev.example"}, + ] + } + get.side_effect = [self.response(city_list)] -class VigiloObservationsBackupTests(unittest.TestCase): - """Test backup and restore of historical observation partitions.""" - - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - self.orig_cwd = os.getcwd() - os.chdir(self.tmpdir) - os.makedirs("dataset/vigilo", exist_ok=True) - os.makedirs("downloaded/vigilo", exist_ok=True) - - def tearDown(self): - os.chdir(self.orig_cwd) - shutil.rmtree(self.tmpdir) - - def _make_partition(self, base, scopeid, content=b"fake"): - p = Path(base) / f"scopeid={scopeid}" - p.mkdir(parents=True, exist_ok=True) - (p / "0.parquet").write_bytes(content) - - def test_save_observations_copies_all_partitions(self): - """All existing scope partitions are copied to the backup location.""" - self._make_partition("dataset/vigilo/observations.parquet", "13_aix") - self._make_partition("dataset/vigilo/observations.parquet", "44_nantes") - - vigilo._save_observations_historical() - - hist = Path("downloaded/vigilo/observations_historical") - self.assertTrue((hist / "scopeid=13_aix").exists()) - self.assertTrue((hist / "scopeid=44_nantes").exists()) - - def test_save_observations_noop_when_no_current_data(self): - """No error when observations.parquet does not exist yet (first run).""" - vigilo._save_observations_historical() - self.assertFalse(Path("downloaded/vigilo/observations_historical").exists()) - - def test_save_observations_replaces_stale_backup(self): - """A stale backup from a previous run is replaced, not merged.""" - stale = Path("downloaded/vigilo/observations_historical/scopeid=old_stale") - stale.mkdir(parents=True) - self._make_partition("dataset/vigilo/observations.parquet", "13_aix") - - vigilo._save_observations_historical() - - hist = Path("downloaded/vigilo/observations_historical") - self.assertFalse((hist / "scopeid=old_stale").exists()) - self.assertTrue((hist / "scopeid=13_aix").exists()) - - def test_restore_copies_inactive_scope_partitions(self): - """Partitions absent from the new SQL output are restored from backup.""" - self._make_partition( - "downloaded/vigilo/observations_historical", "13_aix", b"historical" - ) - self._make_partition( - "downloaded/vigilo/observations_historical", "44_nantes", b"historical" - ) - # SQL only wrote nantes (active scope); aix is inactive - self._make_partition("dataset/vigilo/observations.parquet", "44_nantes", b"new") - - vigilo._restore_historical_observations() - - obs = Path("dataset/vigilo/observations.parquet") - self.assertTrue((obs / "scopeid=13_aix").exists(), "inactive scope should be restored") - self.assertEqual( - (obs / "scopeid=44_nantes" / "0.parquet").read_bytes(), - b"new", - "active scope partition must not be overwritten", - ) - - def test_restore_noop_when_no_backup(self): - """No error when there is no historical backup (first run).""" - Path("dataset/vigilo/observations.parquet").mkdir(parents=True) - vigilo._restore_historical_observations() # must not raise - - -class VigiloScopesBackupTests(unittest.TestCase): - """Test that _save_scopes_historical generates the correct DuckDB SQL.""" - - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - self.orig_cwd = os.getcwd() - os.chdir(self.tmpdir) - os.makedirs("dataset/vigilo", exist_ok=True) - os.makedirs("downloaded/vigilo", exist_ok=True) - os.makedirs("db", exist_ok=True) - Path("db/wda.duckdb").touch() - - def tearDown(self): - os.chdir(self.orig_cwd) - shutil.rmtree(self.tmpdir) - - def _capture_sql_calls(self, returncode_sequence): - """Return (side_effect, captured) where captured accumulates SQL from temp files.""" - captured = [] - codes = iter(returncode_sequence) - - def _side_effect(cmd, **kwargs): - if "< " in cmd: - sql_path = cmd.split("< ")[1] - try: - captured.append(Path(sql_path).read_text()) - except FileNotFoundError: - captured.append("") - return Mock(returncode=next(codes, 0)) - - return _side_effect, captured + with tempfile.TemporaryDirectory() as directory: + with patch( + "vigilo.wdalib.writeContentToFile", + side_effect=lambda filename, content: Path(directory, Path(filename).name).write_text(content), + ): + vigilo.download_scopes() - def test_empty_parquet_created_when_no_current_scopes(self): - """An empty historical parquet with the correct schema is created on first run.""" - side_effect, captured = self._capture_sql_calls([0]) - with patch("vigilo.subprocess.run", side_effect=side_effect): - vigilo._save_scopes_historical() + scopes = json.loads(Path(directory, "scopes.json").read_text()) - self.assertEqual(len(captured), 1) - self.assertIn("CREATE OR REPLACE TABLE _h", captured[0]) - self.assertIn("is_active BOOLEAN", captured[0]) - self.assertIn("first_seen_at DATE", captured[0]) + self.assertEqual(len(scopes), 0) - def test_migration_adds_tracking_columns_with_default_date(self): - """Old-schema scopes.parquet gets is_active=false and the migration date.""" - Path("dataset/vigilo/scopes.parquet").write_bytes(b"old_parquet") - # returncode=1 → schema check fails (old schema); returncode=0 → COPY succeeds - side_effect, captured = self._capture_sql_calls([1, 0]) - with patch("vigilo.subprocess.run", side_effect=side_effect): - vigilo._save_scopes_historical() + @patch("vigilo.print") + @patch("vigilo.requests.get") + def test_citylist_failure_writes_empty_results(self, get, _print_mock): + """A failure fetching check_result.json writes empty scopes and records the failure.""" + get.side_effect = requests.exceptions.ConnectionError("no network") - copy_sql = captured[-1] - self.assertIn("false AS is_active", copy_sql) - self.assertIn(vigilo.MIGRATION_DATE, copy_sql) + with tempfile.TemporaryDirectory() as directory: + with patch( + "vigilo.wdalib.writeContentToFile", + side_effect=lambda filename, content: Path(directory, Path(filename).name).write_text(content), + ): + vigilo.download_scopes() - def test_new_schema_passes_tracking_columns_through(self): - """New-schema scopes.parquet (already has tracking cols) is copied as-is.""" - Path("dataset/vigilo/scopes.parquet").write_bytes(b"new_parquet") - # returncode=0 → schema check passes (new schema); returncode=0 → COPY succeeds - side_effect, captured = self._capture_sql_calls([0, 0]) - with patch("vigilo.subprocess.run", side_effect=side_effect): - vigilo._save_scopes_historical() + scopes = json.loads(Path(directory, "scopes.json").read_text()) + failures = json.loads(Path(directory, "download_failures.json").read_text()) - copy_sql = captured[-1] - self.assertIn("is_active, first_seen_at, last_seen_at", copy_sql) - self.assertNotIn("false AS is_active", copy_sql) + self.assertEqual(scopes, []) + self.assertEqual(len(failures), 1) + self.assertEqual(failures[0]["scope"], "citylist") if __name__ == "__main__":