import os
import requests
from src.config import Config
from src.toolbox.common_tools import CommonTools
import time
class INaturalistAPI:
def init(self, app_id, secret, username, password, site=“https://www.inaturalist.org”):
self.app_id = app_id
self.secret = secret
self.username = username
self.password = password
self.site = site.rstrip(“/”)
self.api_site = “https://api.inaturalist.org/v1” # Modern v1 Node API
# 1. FIX: Custom User-Agent keeps you from getting flagged by WAF firewalls
self.user_agent = "PlanetScottSyncApp/1.0 (Contact: admin@planetscott.com)"
self.oauth_token = None
self.jwt_token = None
self.authenticate()
def authenticate(self):
"""Authenticates using Password Grant, then upgrades the token to JWT."""
# Step A: Get Standard OAuth Access Token from rails backend
payload = {
"client_id": self.app_id,
"client_secret": self.secret,
"grant_type": "password",
"username": self.username,
"password": self.password
}
url = f"{self.site}/oauth/token"
headers = {"User-Agent": self.user_agent}
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
self.oauth_token = response.json().get("access_token")
time.sleep(1.2) # Defensive pacing
# Step B: Exchange OAuth Token for a modern Node JWT token
jwt_url = f"{self.site}/users/api_token"
jwt_headers = {
"User-Agent": self.user_agent,
"Authorization": f"Bearer {self.oauth_token}"
}
jwt_response = requests.get(jwt_url, headers=jwt_headers)
jwt_response.raise_for_status()
self.jwt_token = jwt_response.json().get("api_token")
time.sleep(1.2)
return True
def _get_api_headers(self, is_multipart=False):
"""Helper to safely build distinct endpoint headers."""
headers = {
"User-Agent": self.user_agent,
"Authorization": f"Bearer {self.jwt_token}" # Node routes require JWT
}
if not is_multipart:
headers["Content-Type"] = "application/json"
return headers
def build_description(self, species, images):
"""Builds Markdown friendly descriptions (iNat prefers Markdown over HTML)."""
description = f"More information on my website: {CommonTools().build_external_url(CommonTools().build_species_url(species['id'], species['common_name'], species['genus_latin_name'], species['latin_name']))}"
if not images:
description += "\n\nNOTE: This observation does not have photos available. It is a casual observation where a photo was not possible but the data potentially useful.\nThe species was identified by field marks and/or unrecorded vocalizations in the field."
return description
def post_observation(self, observation):
"""Creates an observation on the modern v1 API engine."""
url = f"{self.api_site}/observations"
wrapped_payload = {"observation": observation}
headers = self._get_api_headers(is_multipart=False)
response = requests.post(url, json=wrapped_payload, headers=headers)
response.raise_for_status()
time.sleep(1.5)
# Note: api.inaturalist.org/v1/observations returns a direct dictionary containing the ID
return response.json()["id"]
def post_observation_image(self, observation_id, image_path):
"""Uploads images using correct Multipart boundaries without header clash."""
url = f"{self.api_site}/observation_photos"
if not os.path.exists(image_path):
print(f"File missing, skipping: {image_path}")
return False
# FIX: Explicitly exclude Content-Type so requests can inject the multi-part boundary
headers = self._get_api_headers(is_multipart=True)
with open(image_path, "rb") as img_file:
files = {"file": img_file}
# Match the legacy naming parameters used by Node wrappers
data = {"observation_photo[observation_id]": observation_id}
response = requests.post(url, headers=headers, data=data, files=files)
response.raise_for_status()
time.sleep(1.5) # Mandatory cooling gap to avoid secondary 403 blocks
return True
def observation_images_insert(self, observation_id, images):
"""Loops through dynamic disk paths to process image items."""
for i in images:
image_path = os.path.join(
Config.WEB_FILES_STATIC_DIRECTORY,
"images",
"upload",
"large",
f"{i['id']}.jpg"
)
self.post_observation_image(observation_id, image_path)
return True
def observation_insert(self, species, site, visit, images):
"""Builds payload structure and fires the sequencers."""
species_name = species["latin_name"].split("_")[0]
description = self.build_description(species, images)
observation = {
"species_guess": f"{species['genus_latin_name']} {species_name}",
"observed_on_string": visit["visit_date"].strftime("%Y-%m-%d"),
"description": description,
"place_guess": site["name"],
"latitude": site["latitude"],
"longitude": site["longitude"],
"positional_accuracy": site["accuracy_meters"],
"location_is_exact": False,
"license": "CC BY-NC 4.0"
}
observation_id = self.post_observation(observation)
print(f"Successfully posted observation {species['common_name']} -> ID: {observation_id}")
self.observation_images_insert(observation_id, images)
return observation_id
def observation_update(self, observation_id, species, images):
"""Updates fields on an existing observation."""
url = f"{self.api_site}/observations/{observation_id}"
new_description = self.build_description(species, images)
updated_fields = {
"description": new_description,
}
headers = self._get_api_headers(is_multipart=False)
response = requests.put(url, json={"observation": updated_fields}, headers=headers)
response.raise_for_status()
time.sleep(1.5)
self.observation_images_insert(observation_id, images)
return response.json()