Colin Powell
c3673e2679
- New nature app with SpeciesObservation model (scrobblable) - SpeciesObservationLogData for observation-specific fields - Atom feed parser + iNat REST API enrichment - Photo download and local storage - Celery periodic import (every 4 hours) - Management command: import_from_inaturalist - Species fields: title, scientific_name, taxon_id, iconic_taxon_name, rank, wikipedia_url - Log fields: observation_id, photo_urls, quality_grade, place_guess, lat/lon, uri, identifications_count - Updates existing scrobbles when observation data changes - iNaturalist user mapping on UserProfile - Templates, admin, API, URLs - Default 5 minute run time for species observations
313 lines
11 KiB
Python
313 lines
11 KiB
Python
import logging
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import timedelta
|
|
|
|
import requests
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.files.base import ContentFile
|
|
from django.utils.dateparse import parse_datetime
|
|
from nature.models import SpeciesObservation, SpeciesObservationLogData
|
|
from scrobbles.models import Scrobble
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
User = get_user_model()
|
|
|
|
INATURALIST_ATOM_URL = "https://www.inaturalist.org/observations.atom"
|
|
INATURALIST_API_URL = "https://api.inaturalist.org/v1/observations"
|
|
ATOM_NS = "http://www.w3.org/2005/Atom"
|
|
GEORSS_NS = "http://www.georss.org/georss"
|
|
|
|
|
|
def fetch_atom_feed(username: str) -> str:
|
|
"""Fetch the Atom feed for a given iNaturalist username."""
|
|
url = (
|
|
f"{INATURALIST_ATOM_URL}?verifiable=any&page=1&spam=&place_id=any"
|
|
f"&user_id={username}&project_id="
|
|
)
|
|
response = requests.get(url, timeout=30)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
|
|
def parse_atom_entries(xml_text: str) -> list[dict]:
|
|
"""Parse Atom XML and extract observation entries."""
|
|
root = ET.fromstring(xml_text)
|
|
entries = []
|
|
|
|
for entry in root.findall(f"{{{ATOM_NS}}}entry"):
|
|
entry_id = entry.findtext(f"{{{ATOM_NS}}}id", "")
|
|
observation_id = entry_id.split("/")[-1] if "/" in entry_id else entry_id
|
|
|
|
title = entry.findtext(f"{{{ATOM_NS}}}title", "")
|
|
published = entry.findtext(f"{{{ATOM_NS}}}published", "")
|
|
updated = entry.findtext(f"{{{ATOM_NS}}}updated", "")
|
|
|
|
content_el = entry.find(f"{{{ATOM_NS}}}content")
|
|
photo_urls = []
|
|
if content_el is not None and content_el.text:
|
|
photo_urls = re.findall(r'src="([^"]+)"', content_el.text)
|
|
|
|
georss_point = entry.findtext(f"{{{GEORSS_NS}}}point", "")
|
|
georss_feature = entry.findtext(f"{{{GEORSS_NS}}}featureName", "")
|
|
|
|
lat, lon = None, None
|
|
if georss_point:
|
|
parts = georss_point.strip().split()
|
|
if len(parts) == 2:
|
|
lat, lon = float(parts[0]), float(parts[1])
|
|
|
|
entries.append(
|
|
{
|
|
"observation_id": int(observation_id),
|
|
"title": title,
|
|
"published": published,
|
|
"updated": updated,
|
|
"photo_urls": photo_urls,
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"place_guess": georss_feature,
|
|
}
|
|
)
|
|
|
|
return entries
|
|
|
|
|
|
def fetch_observation_detail(observation_id: int) -> dict:
|
|
"""Fetch detailed observation data from the iNaturalist REST API."""
|
|
url = f"{INATURALIST_API_URL}/{observation_id}"
|
|
response = requests.get(url, timeout=30)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if data.get("results"):
|
|
return data["results"][0]
|
|
return {}
|
|
|
|
|
|
def download_photo(url: str) -> tuple[str, bytes] | None:
|
|
"""Download a photo and return (filename, content)."""
|
|
try:
|
|
response = requests.get(url, timeout=30)
|
|
response.raise_for_status()
|
|
filename = url.split("/")[-1].split("?")[0]
|
|
if not filename.endswith((".jpg", ".jpeg", ".png", ".webp")):
|
|
filename = "observation.jpg"
|
|
return filename, response.content
|
|
except Exception:
|
|
logger.exception(f"Failed to download photo: {url}")
|
|
return None
|
|
|
|
|
|
def _update_species_photo(species: SpeciesObservation, taxon: dict) -> None:
|
|
"""Download and save the default taxon photo for a species, if needed."""
|
|
if species.photo is not None:
|
|
return
|
|
default_photo = taxon.get("default_photo", {})
|
|
photo_url = default_photo.get("url")
|
|
if not photo_url:
|
|
return
|
|
photo_url = photo_url.replace("square", "medium")
|
|
downloaded = download_photo(photo_url)
|
|
if downloaded:
|
|
filename, content = downloaded
|
|
species.photo.save(filename, ContentFile(content), save=True)
|
|
|
|
|
|
def import_observations_for_user(user) -> list[int]:
|
|
"""Import iNaturalist observations for a single user.
|
|
|
|
Returns a list of created Scrobble IDs.
|
|
"""
|
|
profile = getattr(user, "profile", None)
|
|
if not profile or not profile.inaturalist_username:
|
|
logger.info(f"No iNaturalist username for user {user}, skipping")
|
|
return []
|
|
|
|
username = profile.inaturalist_username
|
|
logger.info(f"Importing iNaturalist observations for {username}")
|
|
|
|
created_scrobble_ids = []
|
|
|
|
try:
|
|
xml_text = fetch_atom_feed(username)
|
|
entries = parse_atom_entries(xml_text)
|
|
except Exception:
|
|
logger.exception(f"Failed to fetch/parse Atom feed for {username}")
|
|
return []
|
|
|
|
for entry_data in entries:
|
|
try:
|
|
scrobble_id = _process_entry(user, entry_data)
|
|
if scrobble_id:
|
|
created_scrobble_ids.append(scrobble_id)
|
|
except Exception:
|
|
logger.exception(
|
|
f"Failed to process observation {entry_data['observation_id']}"
|
|
)
|
|
|
|
logger.info(f"Imported {len(created_scrobble_ids)} observations for {username}")
|
|
return created_scrobble_ids
|
|
|
|
|
|
def _process_entry(user, entry_data: dict) -> int | None:
|
|
"""Process a single Atom entry into a SpeciesObservation and Scrobble.
|
|
|
|
Each iNaturalist observation becomes a new Scrobble. If a Scrobble
|
|
already exists for this observation, check whether the title or
|
|
species data has changed and update in place.
|
|
|
|
Returns the Scrobble ID.
|
|
"""
|
|
observation_id = entry_data["observation_id"]
|
|
entry_title = entry_data["title"]
|
|
|
|
existing_scrobble = Scrobble.objects.filter(
|
|
species_observation__isnull=False,
|
|
user=user,
|
|
source_id=str(observation_id),
|
|
).first()
|
|
|
|
detail = fetch_observation_detail(observation_id)
|
|
|
|
species = None
|
|
taxon = detail.get("taxon", {})
|
|
if taxon:
|
|
common_name = taxon.get("preferred_common_name", entry_title)
|
|
scientific_name = taxon.get("name", "")
|
|
taxon_id = taxon.get("id")
|
|
iconic_taxon_name = taxon.get("iconic_taxon_name", "")
|
|
taxon_rank = taxon.get("rank", "")
|
|
wikipedia_url = taxon.get("wikipedia_url", "") or ""
|
|
if taxon_id and common_name:
|
|
species = SpeciesObservation.find_or_create(
|
|
common_name,
|
|
scientific_name,
|
|
taxon_id,
|
|
iconic_taxon_name=iconic_taxon_name,
|
|
rank=taxon_rank,
|
|
wikipedia_url=wikipedia_url,
|
|
)
|
|
_update_species_photo(species, taxon)
|
|
changed_fields = []
|
|
if species.title != common_name:
|
|
species.title = common_name
|
|
changed_fields.append("title")
|
|
if species.iconic_taxon_name != iconic_taxon_name:
|
|
species.iconic_taxon_name = iconic_taxon_name
|
|
changed_fields.append("iconic_taxon_name")
|
|
if species.rank != taxon_rank:
|
|
species.rank = taxon_rank
|
|
changed_fields.append("rank")
|
|
if species.wikipedia_url != wikipedia_url:
|
|
species.wikipedia_url = wikipedia_url
|
|
changed_fields.append("wikipedia_url")
|
|
if changed_fields:
|
|
species.save(update_fields=changed_fields)
|
|
|
|
quality_grade = detail.get("quality_grade", "")
|
|
place_guess = detail.get("place_guess", entry_data.get("place_guess", ""))
|
|
description = detail.get("description", "")
|
|
|
|
coords = detail.get("geojson", {}).get("coordinates")
|
|
if coords and len(coords) == 2:
|
|
lon, lat = coords[0], coords[1]
|
|
else:
|
|
lat = entry_data.get("lat")
|
|
lon = entry_data.get("lon")
|
|
|
|
observed_at = detail.get("time_observed_at") or detail.get("observed_on")
|
|
|
|
photo_url = entry_data["photo_urls"][0] if entry_data["photo_urls"] else None
|
|
|
|
all_photo_urls = entry_data.get("photo_urls", [])
|
|
if detail.get("photos"):
|
|
all_photo_urls = [
|
|
p.get("url", "").replace("square", "medium") for p in detail["photos"]
|
|
]
|
|
|
|
uri = detail.get("uri", "")
|
|
identifications_count = detail.get("identifications_count", 0)
|
|
|
|
log_data = SpeciesObservationLogData(
|
|
observation_id=observation_id,
|
|
photo_urls=all_photo_urls,
|
|
quality_grade=quality_grade,
|
|
place_guess=place_guess,
|
|
lat=lat,
|
|
lon=lon,
|
|
description=description or "",
|
|
uri=uri or "",
|
|
identifications_count=identifications_count,
|
|
)
|
|
|
|
if existing_scrobble:
|
|
changed = False
|
|
|
|
if species and existing_scrobble.species_observation_id != species.id:
|
|
existing_scrobble.species_observation = species
|
|
changed = True
|
|
|
|
if existing_scrobble.log.get("place_guess") != place_guess:
|
|
changed = True
|
|
|
|
if existing_scrobble.log.get("quality_grade") != quality_grade:
|
|
changed = True
|
|
|
|
if existing_scrobble.log.get("photo_urls") != all_photo_urls:
|
|
changed = True
|
|
|
|
if existing_scrobble.log.get("uri") != uri:
|
|
changed = True
|
|
|
|
if existing_scrobble.log.get("identifications_count") != identifications_count:
|
|
changed = True
|
|
|
|
if not changed:
|
|
logger.debug(f"Observation {observation_id} unchanged, skipping")
|
|
return existing_scrobble.id
|
|
|
|
logger.info(f"Observation {observation_id} updated, refreshing scrobble")
|
|
existing_scrobble.log = log_data.asdict
|
|
existing_scrobble.save(update_fields=["log", "species_observation"])
|
|
return existing_scrobble.id
|
|
|
|
timestamp = parse_datetime(observed_at) if observed_at else None
|
|
stop_timestamp = timestamp + timedelta(minutes=5) if timestamp else None
|
|
|
|
scrobble = Scrobble.objects.create(
|
|
user=user,
|
|
species_observation=species,
|
|
media_type=Scrobble.MediaType.SPECIES_OBSERVATION,
|
|
source="iNaturalist",
|
|
source_id=str(observation_id),
|
|
timestamp=timestamp,
|
|
stop_timestamp=stop_timestamp,
|
|
played_to_completion=True,
|
|
in_progress=False,
|
|
log=log_data.asdict,
|
|
)
|
|
|
|
if photo_url:
|
|
downloaded = download_photo(photo_url)
|
|
if downloaded:
|
|
filename, content = downloaded
|
|
scrobble.screenshot.save(filename, ContentFile(content), save=True)
|
|
|
|
return scrobble.id
|
|
|
|
|
|
def import_observations_for_all_users() -> list[int]:
|
|
"""Import iNaturalist observations for all users with auto-import enabled."""
|
|
users = User.objects.filter(
|
|
profile__inaturalist_auto_import=True,
|
|
profile__inaturalist_username__isnull=False,
|
|
).exclude(profile__inaturalist_username="")
|
|
|
|
all_scrobble_ids = []
|
|
for user in users:
|
|
scrobble_ids = import_observations_for_user(user)
|
|
all_scrobble_ids.extend(scrobble_ids)
|
|
|
|
return all_scrobble_ids
|