SwhaleAI/marine-mammals-api

1"""2Récupère les observations depuis OBIS (primaire) et GBIF (complément).3 4Deux tables alimentées :5- observations       : points géo pour la carte, cappés à MAX_GEO_PER_SPECIES6- observation_counts : agrégats annuels réels pour les graphiques de tendance7 8Stratégie source :9- OBIS pour toutes les espèces marines.10- GBIF en fallback si OBIS retourne 0 résultats (dauphins de rivière).11 12Usage :13    cd scripts && python fetch_observations.py14    cd scripts && python fetch_observations.py --since 2024-01-01   # incrémental15"""16 17import asyncio18import argparse19from collections import defaultdict20from datetime import date, datetime, timezone21from pathlib import Path22import sys23 24import asyncpg25import httpx26from dotenv import load_dotenv27from tqdm import tqdm28 29sys.path.insert(0, str(Path(__file__).parent))30from utils.db_connector import get_connection, log_sync31 32DEFAULT_START_DATE = "2010-01-01"33OBIS_GEO_PAGE_SIZE = 50034OBIS_COUNT_PAGE_SIZE = 5000   # payloads minimaux (fields=date_year)35GBIF_PAGE_SIZE = 30036OBIS_DELAY = 1.037GBIF_DELAY = 0.538REQUEST_TIMEOUT = 6039INSERT_CHUNK = 50040MAX_GEO_PER_SPECIES = 1500041 42 43def _parse_date(s: str | None) -> date | None:44    if not s:45        return None46    try:47        return date.fromisoformat(s[:10])48    except ValueError:49        return None50 51 52def _parse_int(v) -> int | None:53    try:54        return int(v)55    except (TypeError, ValueError):56        return None57 58 59async def fetch_obis_geo(60    http: httpx.AsyncClient, scientific_name: str, start_date: str61) -> list[dict]:62    """Points géo OBIS pour la carte, cappés à MAX_GEO_PER_SPECIES."""63    records = []64    offset = 065    while True:66        resp = await asyncio.wait_for(67            http.get(68                "https://api.obis.org/v3/occurrence",69                params={70                    "scientificname": scientific_name,71                    "startdate": start_date,72                    "size": OBIS_GEO_PAGE_SIZE,73                    "offset": offset,74                },75            ),76            timeout=REQUEST_TIMEOUT,77        )78        await asyncio.sleep(OBIS_DELAY)79        resp.raise_for_status()80        data = resp.json()81        batch = data.get("results", [])82        total = data.get("total", 0)83 84        for r in batch:85            if r.get("dropped") or r.get("absence"):86                continue87            lat = r.get("decimalLatitude")88            lon = r.get("decimalLongitude")89            record_id = str(r.get("occurrenceID", ""))[:100]90            if lat is None or lon is None or not record_id:91                continue92            records.append({93                "latitude": lat,94                "longitude": lon,95                "observed_at": _parse_date(r.get("eventDate")),96                "individual_count": _parse_int(r.get("individualCount")),97                "depth_m": None,98                "source_record_id": record_id,99            })100 101        offset += len(batch)102        if not batch or offset >= total or len(records) >= MAX_GEO_PER_SPECIES:103            break104    return records[:MAX_GEO_PER_SPECIES]105 106 107async def fetch_obis_counts(108    http: httpx.AsyncClient, scientific_name: str, start_date: str109) -> dict[int, int]:110    """Agrégat annuel OBIS complet (fields=date_year, sans cap)."""111    counts: dict[int, int] = defaultdict(int)112    offset = 0113    while True:114        resp = await asyncio.wait_for(115            http.get(116                "https://api.obis.org/v3/occurrence",117                params={118                    "scientificname": scientific_name,119                    "startdate": start_date,120                    "size": OBIS_COUNT_PAGE_SIZE,121                    "offset": offset,122                    "fields": "date_year",123                },124            ),125            timeout=REQUEST_TIMEOUT,126        )127        await asyncio.sleep(OBIS_DELAY)128        resp.raise_for_status()129        data = resp.json()130        batch = data.get("results", [])131        total = data.get("total", 0)132 133        for r in batch:134            year = r.get("date_year")135            if year:136                counts[int(year)] += 1137 138        offset += len(batch)139        if not batch or offset >= total:140            break141    return dict(counts)142 143 144async def fetch_gbif_geo_and_counts(145    http: httpx.AsyncClient, scientific_name: str, start_year: int146) -> tuple[list[dict], dict[int, int]]:147    """Fetch GBIF en une seule passe : retourne points géo ET agrégat annuel."""148    records = []149    counts: dict[int, int] = defaultdict(int)150    offset = 0151    end_year = date.today().year152    while True:153        resp = await asyncio.wait_for(154            http.get(155                "https://api.gbif.org/v1/occurrence/search",156                params={157                    "scientificName": scientific_name,158                    "year": f"{start_year},{end_year}",159                    "limit": GBIF_PAGE_SIZE,160                    "offset": offset,161                    "hasCoordinate": "true",162                    "hasGeospatialIssue": "false",163                },164            ),165            timeout=REQUEST_TIMEOUT,166        )167        await asyncio.sleep(GBIF_DELAY)168        resp.raise_for_status()169        data = resp.json()170        batch = data.get("results", [])171 172        for r in batch:173            lat = r.get("decimalLatitude")174            lon = r.get("decimalLongitude")175            record_id = str(r.get("key", ""))[:100]176            if lat is None or lon is None or not record_id:177                continue178            records.append({179                "latitude": lat,180                "longitude": lon,181                "observed_at": _parse_date(r.get("eventDate")),182                "individual_count": _parse_int(r.get("individualCount")),183                "depth_m": r.get("depth"),184                "source_record_id": record_id,185            })186            year = r.get("year")187            if year:188                counts[int(year)] += 1189 190        offset += len(batch)191        if data.get("endOfRecords", True) or not batch:192            break193    return records[:MAX_GEO_PER_SPECIES], dict(counts)194 195 196async def insert_geo(197    conn: asyncpg.Connection, species_id: int, source: str, records: list[dict]198) -> int:199    if not records:200        return 0201    202    # Préparation des données203    rows = [204        (205            species_id,206            r["latitude"],207            r["longitude"],208            r["observed_at"],209            source,210            r["individual_count"],211            r["depth_m"],212            r["source_record_id"],213        )214        for r in records215    ]216 217    for i in range(0, len(rows), INSERT_CHUNK):218        await conn.executemany(219            """220            INSERT INTO observations221                (species_id, latitude, longitude, observed_at, source,222                 individual_count, depth_m, source_record_id)223            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)224            ON CONFLICT (source, source_record_id) DO NOTHING225            """,226            rows[i : i + INSERT_CHUNK],227        )228    return len(rows)229 230 231async def insert_counts(232    conn: asyncpg.Connection, species_id: int, source: str, counts: dict[int, int]233) -> int:234    if not counts:235        return 0236    rows = [(species_id, year, count, source) for year, count in counts.items()]237    await conn.executemany(238        """239        INSERT INTO observation_counts (species_id, year, count, source)240        VALUES ($1, $2, $3, $4)241        ON CONFLICT (species_id, year, source) DO UPDATE SET count = EXCLUDED.count242        """,243        rows,244    )245    return len(rows)246 247 248async def main() -> None:249    parser = argparse.ArgumentParser()250    parser.add_argument(251        "--since",252        default=DEFAULT_START_DATE,253        help="Date de départ ISO (défaut: 2010-01-01). Ex: --since 2024-01-01",254    )255    args = parser.parse_args()256 257    load_dotenv(Path(__file__).parent.parent / ".env")258    started_at = datetime.now(timezone.utc)259    total_geo = 0260 261    conn = await get_connection()262    try:263        species_rows = await conn.fetch(264            "SELECT id, scientific_name FROM species ORDER BY scientific_name"265        )266        species_map = {r["scientific_name"]: r["id"] for r in species_rows}267        print(f"{len(species_map)} espèces. Fetch depuis {args.since}…", flush=True)268 269        async with httpx.AsyncClient(270            timeout=30,271            headers={"User-Agent": "CetaScope/1.0 (portfolio; contact: lucienlaumont36@gmail.com)"},272        ) as http:273            progress = tqdm(species_map.items(), total=len(species_map), unit="espèce")274            for name, sp_id in progress:275                progress.set_description(f"{name[:35]:<35}")276                try:277                    geo = await fetch_obis_geo(http, name, args.since)278                    counts = await fetch_obis_counts(http, name, args.since)279                    source = "OBIS"280 281                    if not geo and not counts:282                        start_year = date.fromisoformat(args.since).year283                        geo, counts = await fetch_gbif_geo_and_counts(284                            http, name, start_year285                        )286                        source = "GBIF"287 288                    n_geo = await insert_geo(conn, sp_id, source, geo)289                    n_years = await insert_counts(conn, sp_id, source, counts)290                    total_geo += n_geo291                    progress.write(f"  {name}: {n_geo} geo, {n_years} années ({source})")292 293                except Exception as e:294                    progress.write(f"  ERREUR {name}: {e}")295 296        print(f"Terminé : {total_geo} points géo.", flush=True)297        await log_sync(298            conn,299            source="OBIS+GBIF",300            sync_type="observations",301            added=total_geo,302            updated=0,303            status="success",304            started_at=started_at,305        )306 307    except Exception as exc:308        await log_sync(309            conn,310            source="OBIS+GBIF",311            sync_type="observations",312            added=total_geo,313            updated=0,314            status="error",315            started_at=started_at,316            error=str(exc),317        )318        raise319 320    finally:321        await conn.close()322 323 324if __name__ == "__main__":325    asyncio.run(main())326