WildDex/dev_scripts/INaturalist_requester.py

import requests
from pathlib import Path
import csv
import time
 
# Portland, OR place ID on iNaturalist
PLACE_ID = 2143
OUTPUT_FILE = Path(__file__).parent.parent / "wildlife" / "portland_wildlife.csv"
 
TAXON_GROUPS = ["Aves", "Mammalia", "Reptilia", "Amphibia", "Insecta"]
 
FIELDNAMES = [
    "taxon_id",
    "scientific_name",
    "common_name",
    "kingdom",
    "phylum",
    "class",
    "order",
    "family",
    "genus",
    "iconic_taxon",
    "observation_count",
    "photo_url_thumb",
    "photo_url_med",
    "wikipedia_url"
]

def fetch_species_for_group(iconic_taxon: str) -> list[dict]:
    """Fetch unique species list for one iconic taxon group using species_counts."""
    results = []
    page = 1
 
    print(f"Fetching {iconic_taxon}...")
 
    while True:
        response = requests.get(
            "https://api.inaturalist.org/v1/observations/species_counts",
            params={
                "iconic_taxa": iconic_taxon,
                "quality_grade": "research",
                "per_page": 500,
                "page": page,
                "place_id" : PLACE_ID,
            },
        )
        response.raise_for_status()
        data = response.json()
 
        results_page = data.get("results", [])
        if not results_page:
            break
 
        for entry in results_page:
            taxon = entry.get("taxon", {})
 
            # Skip anything that isn't a species (genus-level matches etc.)
            if taxon.get("rank") != "species":
                continue
 
            ancestors = {
                a["rank"]: a["name"]
                for a in taxon.get("ancestors", [])
            }
 
            photo = taxon.get("default_photo") or {}
 
            results.append({
                "taxon_id": taxon.get("id"),
                "scientific_name": taxon.get("name"),
                "common_name": taxon.get("preferred_common_name", ""),
                "kingdom": ancestors.get("kingdom", ""),
                "phylum": ancestors.get("phylum", ""),
                "class": ancestors.get("class", ""),
                "order": ancestors.get("order", ""),
                "family": ancestors.get("family", ""),
                "genus": ancestors.get("genus", ""),
                "iconic_taxon": taxon.get("iconic_taxon_name", ""),
                "observation_count": entry.get("count", 0),
                "photo_url_thumb": photo.get("square_url", ""),
                "photo_url_med": photo.get("medium_url", ""),
                "wikipedia_url": taxon.get("wikipedia_url", "")
            })
 
        total = data.get("total_results", 0)
        fetched_so_far = (page - 1) * 500 + len(results_page)
        print(f"  Page {page} — {fetched_so_far}/{total} species")
 
        if fetched_so_far >= total:
            break
 
        page += 1
        time.sleep(1)  # stay within API rate limits (~1 req/sec)
 
    print(f"  Done — {len(results)} species found for {iconic_taxon}\n")
    return results

def enrich_with_ancestors(species_list: list[dict]) -> list[dict]:
    """
    Batch-fetch full taxon details from /v1/taxa to fill in ancestor fields.
    The species_counts endpoint omits ancestors, so we fetch them separately.
    iNaturalist supports up to 30 comma-separated IDs per request.
    """
    BATCH_SIZE = 30
    ids = [str(s["taxon_id"]) for s in species_list]
    taxon_details = {}
 
    print(f"Enriching {len(ids)} species with ancestor data...")
 
    for i in range(0, len(ids), BATCH_SIZE):
        batch = ids[i : i + BATCH_SIZE]
        response = requests.get(
            f"https://api.inaturalist.org/v1/taxa/{','.join(batch)}",
        )
        response.raise_for_status()
        data = response.json()
 
        for taxon in data.get("results", []):
            ancestors = {
                a["rank"]: a["name"]
                for a in taxon.get("ancestors", [])
            }
            taxon_details[taxon["id"]] = {
                "ancestors": ancestors,
                "wikipedia_url": taxon.get("wikipedia_url", ""),
            }
 
        print(f"  Fetched batch {i // BATCH_SIZE + 1}/{-(-len(ids) // BATCH_SIZE)}")
        time.sleep(1)
 
    # Merge ancestor data back into species list
    for s in species_list:
        detail = taxon_details.get(s["taxon_id"], {})
        ancestors = detail.get("ancestors", {})
        s["kingdom"]       = ancestors.get("kingdom", "")
        s["phylum"]        = ancestors.get("phylum", "")
        s["class"]         = ancestors.get("class", "")
        s["order"]         = ancestors.get("order", "")
        s["family"]        = ancestors.get("family", "")
        s["genus"]         = ancestors.get("genus", "")
        s["wikipedia_url"] = detail.get("wikipedia_url", "") or s.get("wikipedia_url", "")
 
    print("Enrichment complete.\n")
    return species_list

if __name__ == "__main__":
    all_species = []
 
    for group in TAXON_GROUPS:
        species = fetch_species_for_group(group)
        all_species.extend(species)
 
    # Deduplicate by taxon_id just in case
    seen = set()
    unique_species = []
    for s in all_species:
        if s["taxon_id"] not in seen:
            seen.add(s["taxon_id"])
            unique_species.append(s)
 
    # Sort by observation count descending — most-seen species first
    # This makes it easy to cap at top ~200 for the POC
    unique_species.sort(key=lambda x: x["observation_count"], reverse=True)
 
    # Second pass: fill in ancestor taxonomy from /v1/taxa
    unique_species = enrich_with_ancestors(unique_species)
 
    print(f"Total unique species: {len(unique_species)}")
 
    with open(OUTPUT_FILE, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
        writer.writeheader()
        writer.writerows(unique_species)
 
    print(f"Written to {OUTPUT_FILE}")