91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""Conflict / military-movement overlay, backed by ACLED.
|
|
|
|
There is no single authoritative, free, real-time public feed of "military
|
|
movements" — ACLED (Armed Conflict Location & Event Data, acleddata.com) is
|
|
the closest widely-used open dataset of georeferenced, sourced conflict and
|
|
political-violence events, and offers free access for registered users. If
|
|
ACLED_API_KEY / ACLED_EMAIL are not configured, this module is a no-op and
|
|
the overlay simply stays empty in the UI rather than erroring.
|
|
"""
|
|
|
|
import datetime as dt
|
|
import logging
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .models import ConflictEvent
|
|
|
|
log = logging.getLogger("newsatlas.conflict")
|
|
|
|
ACLED_URL = "https://api.acleddata.com/acled/read"
|
|
|
|
|
|
def enabled() -> bool:
|
|
return bool(settings.acled_api_key and settings.acled_email)
|
|
|
|
|
|
def poll_conflict_events(session: Session) -> int:
|
|
if not enabled():
|
|
log.info("ACLED credentials not configured; skipping conflict-event poll")
|
|
return 0
|
|
|
|
since = (dt.datetime.utcnow() - dt.timedelta(days=7)).strftime("%Y-%m-%d")
|
|
params = {
|
|
"key": settings.acled_api_key,
|
|
"email": settings.acled_email,
|
|
"event_date": since,
|
|
"event_date_where": ">=",
|
|
"limit": 500,
|
|
}
|
|
|
|
try:
|
|
resp = httpx.get(ACLED_URL, params=params, timeout=30)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
except Exception:
|
|
log.exception("Failed to fetch ACLED data")
|
|
return 0
|
|
|
|
added = 0
|
|
for row in payload.get("data", []):
|
|
external_id = str(row.get("event_id_cnty") or row.get("data_id"))
|
|
exists = session.execute(
|
|
select(ConflictEvent.id).where(
|
|
ConflictEvent.source == "acled",
|
|
ConflictEvent.external_id == external_id,
|
|
)
|
|
).first()
|
|
if exists:
|
|
continue
|
|
|
|
try:
|
|
lat = float(row["latitude"])
|
|
lon = float(row["longitude"])
|
|
event_date = dt.datetime.strptime(row["event_date"], "%Y-%m-%d")
|
|
except (KeyError, ValueError, TypeError):
|
|
continue
|
|
|
|
session.add(
|
|
ConflictEvent(
|
|
source="acled",
|
|
external_id=external_id,
|
|
event_type=row.get("event_type", ""),
|
|
actor1=row.get("actor1", ""),
|
|
actor2=row.get("actor2", ""),
|
|
fatalities=int(row["fatalities"]) if str(row.get("fatalities", "")).isdigit() else None,
|
|
notes=row.get("notes", ""),
|
|
location_name=row.get("location", ""),
|
|
country=row.get("country", ""),
|
|
lat=lat,
|
|
lon=lon,
|
|
event_date=event_date,
|
|
)
|
|
)
|
|
added += 1
|
|
|
|
session.commit()
|
|
return added
|