Detecting Bulk Deletions in an OSM Diff Stream Jump to heading#

#!/usr/bin/env python3
"""Detect bulk deletions across a sliding window of OSM diffs."""
from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

import osmium

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
logger = logging.getLogger(__name__)

ABSOLUTE_THRESHOLD = 1_000     # deletions in the window
SHARE_THRESHOLD = 0.40         # deletions as a fraction of the changeset's operations
MIN_OPERATIONS = 50            # below this, a share is noise
WINDOW = timedelta(hours=1)    # how long a changeset stays open in the accumulator


@dataclass
class Window:
    """Per-changeset accumulator, flushed once the changeset goes quiet."""
    changeset: int
    user: str = ""
    deleted_nodes: int = 0
    deleted_ways: int = 0
    deleted_relations: int = 0
    other_ops: int = 0
    deleted_ids: list[tuple[str, int]] = field(default_factory=list)
    last_seen: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @property
    def deletions(self) -> int:
        return self.deleted_nodes + self.deleted_ways + self.deleted_relations

    @property
    def operations(self) -> int:
        return self.deletions + self.other_ops

    @property
    def share(self) -> float:
        return self.deletions / self.operations if self.operations else 0.0


class DeletionWatcher(osmium.SimpleHandler):
    """Accumulate deletions per changeset across successive diffs."""

    #: keep this many ids per changeset for the report — enough to investigate,
    #: bounded so a 400 000-object redaction does not become a 400 000-entry list
    SAMPLE_IDS = 20

    def __init__(self) -> None:
        super().__init__()
        self.windows: dict[int, Window] = {}
        self.reported: set[int] = set()

    def _record(self, obj, kind: str) -> None:
        w = self.windows.get(obj.changeset)
        if w is None:
            w = self.windows[obj.changeset] = Window(changeset=obj.changeset)
        w.user = obj.user or w.user
        w.last_seen = datetime.now(timezone.utc)
        if obj.visible:
            w.other_ops += 1
            return
        setattr(w, f"deleted_{kind}", getattr(w, f"deleted_{kind}") + 1)
        if len(w.deleted_ids) < self.SAMPLE_IDS:
            w.deleted_ids.append((kind[0], obj.id))

    def node(self, n) -> None:
        self._record(n, "nodes")

    def way(self, w) -> None:
        self._record(w, "ways")

    def relation(self, r) -> None:
        self._record(r, "relations")

    def detections(self) -> list[Window]:
        """Changesets that have crossed the threshold and not yet been reported."""
        out: list[Window] = []
        for cs_id, w in self.windows.items():
            if cs_id in self.reported:
                continue
            big = w.deletions >= ABSOLUTE_THRESHOLD
            skewed = w.operations >= MIN_OPERATIONS and w.share >= SHARE_THRESHOLD
            if big and skewed:
                out.append(w)
                self.reported.add(cs_id)
        return out

    def evict(self, now: datetime | None = None) -> int:
        """Drop changesets that have gone quiet, so the accumulator stays bounded."""
        now = now or datetime.now(timezone.utc)
        stale = [cid for cid, w in self.windows.items() if now - w.last_seen > WINDOW]
        for cid in stale:
            del self.windows[cid]
        self.reported -= set(stale)
        return len(stale)


def scan(diff_paths: list[str]) -> list[Window]:
    watcher = DeletionWatcher()
    found: list[Window] = []
    for path in diff_paths:
        watcher.apply_file(path)
        for w in watcher.detections():
            logger.warning(
                "bulk deletion: changeset %d by %s — %d deletions (%.0f%% of %d ops); sample %s",
                w.changeset, w.user, w.deletions, w.share * 100, w.operations,
                w.deleted_ids[:5],
            )
            found.append(w)
        evicted = watcher.evict()
        if evicted:
            logger.debug("evicted %d quiet changeset(s)", evicted)
    return found

# Grab an hour of minutely diffs and count deletions per changeset by hand.
for seq in $(seq 6123400 6123459); do
  p=$(printf "%09d" $seq | sed 's|\(...\)\(...\)\(...\)|\1/\2/\3|')
  curl -sO "https://planet.osm.org/replication/minute/${p}.osc.gz"
done
zcat *.osc.gz | grep -oP '(?<=<delete>)' | wc -l

def test_large_import_is_not_flagged():
    w = Window(changeset=1, deleted_nodes=1_200, other_ops=95_000)
    assert w.deletions >= ABSOLUTE_THRESHOLD          # absolute passes
    assert w.share < SHARE_THRESHOLD                  # share does not
    # therefore not a detection

def test_small_deletion_burst_is_not_flagged():
    w = Window(changeset=2, deleted_ways=40, other_ops=2)
    assert w.share > SHARE_THRESHOLD
    assert w.deletions < ABSOLUTE_THRESHOLD

def test_mass_removal_is_flagged():
    w = Window(changeset=3, deleted_ways=8_000, other_ops=120)
    assert w.deletions >= ABSOLUTE_THRESHOLD and w.share >= SHARE_THRESHOLD