krz/omaha-metro-blotter

  1# Sarpy and Council Bluffs serve a rolling 12-month window; records that age out
  2# of those feeds exist nowhere else. This job is the only thing keeping them, so
  3# it refuses to publish an archive smaller than the one it started with.
  4name: daily pull
  5
  6on:
  7  schedule:
  8    # Twice a day, because the cadence sets how much the rolling feeds drop
  9    # before it is captured: a five-hour gap cost 13 Sarpy records once.
 10    # Offset from the hour, GitHub drops on-the-hour runs under load.
 11    - cron: "17 11 * * *"
 12    - cron: "17 23 * * *"
 13  workflow_dispatch:
 14    inputs:
 15      bootstrap:
 16        description: "Start a new archive instead of restoring the published one"
 17        type: boolean
 18        default: false
 19      full:
 20        description: "Pull each feed in full rather than the last 30 days"
 21        type: boolean
 22        default: false
 23
 24permissions:
 25  contents: write
 26
 27concurrency:
 28  group: archive
 29  cancel-in-progress: false
 30
 31env:
 32  TAG: archive
 33  DB: raw_data/metro.db
 34  GH_TOKEN: ${{ github.token }}
 35
 36jobs:
 37  pull:
 38    runs-on: ubuntu-latest
 39    timeout-minutes: 45
 40
 41    steps:
 42      - uses: actions/checkout@v4
 43
 44      - uses: actions/setup-python@v5
 45        with:
 46          python-version: "3.13"
 47          cache: pip
 48          cache-dependency-path: requirements-ingest.txt
 49
 50      - run: pip install -r requirements-ingest.txt
 51
 52      - name: Restore archive
 53        run: |
 54          if gh release download "$TAG" --pattern metro.db.gz --dir .; then
 55            gunzip -c metro.db.gz > "$DB"
 56            rm metro.db.gz
 57            echo "restored $(du -h "$DB" | cut -f1)"
 58          elif [ "${{ inputs.bootstrap }}" = "true" ]; then
 59            echo "no published archive; starting a new one"
 60          else
 61            echo "::error::No metro.db.gz on release '$TAG'. Anything that has" \
 62                 "already aged out of the Sarpy and Council Bluffs feeds cannot" \
 63                 "be recovered. Re-run with bootstrap only if that is intended."
 64            exit 1
 65          fi
 66
 67      - name: Count rows before
 68        run: |
 69          : > before.txt
 70          if [ -f "$DB" ]; then
 71            sqlite3 -noheader -separator ' ' "$DB" \
 72              "SELECT source, COUNT(*) FROM incidents GROUP BY source ORDER BY source" \
 73              > before.txt
 74            # absent until the amendment migration has run against this archive
 75            sqlite3 -noheader -separator ' ' "$DB" \
 76              "SELECT source || '+amend', COUNT(*) FROM incident_amendments
 77               GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
 78            sqlite3 -noheader -separator ' ' "$DB" \
 79              "SELECT source || '+raw', COUNT(*) FROM raw_records
 80               GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
 81            sqlite3 -noheader -separator ' ' "$DB" \
 82              "SELECT agency || '+search', COUNT(*) FROM alpr_searches
 83               GROUP BY agency ORDER BY agency" >> before.txt 2>/dev/null || true
 84            sqlite3 -noheader -separator ' ' "$DB" \
 85              "SELECT source || '+raw', COUNT(*) FROM raw_records
 86               GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
 87            sqlite3 -noheader -separator ' ' "$DB" \
 88              "SELECT agency || '+search', COUNT(*) FROM alpr_searches
 89               GROUP BY agency ORDER BY agency" >> before.txt 2>/dev/null || true
 90          fi
 91          cat before.txt
 92
 93      - name: Pull feeds
 94        run: |
 95          # A 30-day window cannot see an agency amending a record filed months
 96          # ago, and OPD does exactly that, so sweep the whole feed on Sundays.
 97          if [ "${{ inputs.full }}" = "true" ] || [ "${{ inputs.bootstrap }}" = "true" ] \
 98             || [ "$(date -u +%u)" = "7" ]; then
 99            echo "full sweep"
100            python ingest.py --full
101          else
102            python ingest.py
103          fi
104
105      - name: Check nothing was lost
106        run: |
107          sqlite3 -noheader -separator ' ' "$DB" \
108            "SELECT source, COUNT(*) FROM incidents GROUP BY source ORDER BY source" \
109            > after.txt
110          sqlite3 -noheader -separator ' ' "$DB" \
111            "SELECT source || '+amend', COUNT(*) FROM incident_amendments
112             GROUP BY source ORDER BY source" >> after.txt
113          sqlite3 -noheader -separator ' ' "$DB" \
114            "SELECT source || '+raw', COUNT(*) FROM raw_records
115             GROUP BY source ORDER BY source" >> after.txt
116          sqlite3 -noheader -separator ' ' "$DB" \
117            "SELECT agency || '+search', COUNT(*) FROM alpr_searches
118             GROUP BY agency ORDER BY agency" >> after.txt
119          cat after.txt
120          test -s after.txt || { echo "::error::archive is empty"; exit 1; }
121          # Keyed on FILENAME, not NR == FNR: before.txt is empty on a bootstrap
122          # run, and awk never resets FNR for a zero-length file.
123          awk -v first=before.txt '
124               FILENAME == first { was[$1] = $2; next }
125               { now[$1] = $2 }
126               END {
127                 for (s in was)
128                   if (now[s] + 0 < was[s] + 0) {
129                     printf "::error::%s lost rows: %d -> %d\n", s, was[s], now[s]
130                     bad = 1
131                   }
132                 exit bad
133               }' before.txt after.txt
134
135          # An amendment identical to its original means the digest drifted
136          # against what SQLite stores, and every run would file the same
137          # phantom again. Cheap to check, silent and cumulative if it happens.
138          phantom=$(sqlite3 -noheader "$DB" "
139            SELECT COUNT(*) FROM incident_amendments a
140            JOIN incidents o ON o.source = a.source AND o.source_key = a.source_key
141            WHERE a.agency IS o.agency AND a.case_id IS o.case_id
142              AND a.occurred_at IS o.occurred_at AND a.category IS o.category
143              AND a.call_type IS o.call_type AND a.disposition IS o.disposition
144              AND a.offense_desc IS o.offense_desc AND a.is_stop IS o.is_stop
145              AND a.address IS o.address AND a.lat IS o.lat AND a.lon IS o.lon")
146          if [ "$phantom" -ne 0 ]; then
147            echo "::error::$phantom amendments are identical to their original"
148            exit 1
149          fi
150
151      - name: Publish archive
152        run: |
153          sqlite3 "$DB" "VACUUM;"
154          gzip -c "$DB" > metro.db.gz
155          gh release view "$TAG" >/dev/null 2>&1 \
156            || gh release create "$TAG" --title "Incident archive" --notes "building"
157          gh release upload "$TAG" metro.db.gz --clobber
158          {
159            echo "SQLite archive of Omaha metro police incident feeds, rebuilt daily."
160            echo "Sarpy County and Council Bluffs publish a rolling 12-month window,"
161            echo "so this holds records their own feeds no longer serve."
162            echo
163            echo "Updated $(date -u '+%Y-%m-%d %H:%M UTC'). Schema: schema.sql."
164            echo
165            echo "incidents holds each record as first published; every later"
166            echo "version the feed served is a row in incident_amendments"
167            echo "($(sqlite3 -noheader "$DB" 'SELECT COUNT(*) FROM incident_amendments') so far)."
168            echo "incidents_current is the newest version of each, and"
169            echo "raw_records keeps the feed's own JSON for every version"
170            echo "so a parse can be redone against what actually arrived."
171            echo
172            echo '```'
173            sqlite3 -header -column "$DB" \
174              "SELECT agency, COUNT(*) AS rows, SUM(is_stop) AS stops,
175                      MIN(occurred_at) AS earliest, MAX(occurred_at) AS latest
176               FROM incidents GROUP BY agency ORDER BY rows DESC"
177            echo '```'
178          } > notes.md
179          gh release edit "$TAG" --notes-file notes.md
180
181      # Runs after the upload on purpose: a feed that stopped updating should
182      # raise the alarm without also blocking the archive from being published.
183      - name: Check the feeds are still moving
184        run: |
185          sqlite3 -noheader "$DB" \
186            "SELECT source || ' ' || MAX(occurred_at) FROM incidents
187             WHERE source <> 'opd_csv' GROUP BY source
188             HAVING MAX(occurred_at) < datetime('now', '-7 days')" > stale.txt
189          if [ -s stale.txt ]; then
190            while read -r line; do echo "::error::feed is stale: $line"; done < stale.txt
191            exit 1
192          fi
193          echo "all feeds current"
194
195      - name: Summary
196        if: always()
197        run: |
198          {
199            echo "| source | before | after |"
200            echo "|---|---|---|"
201            awk -v first=before.txt '
202                 FILENAME == first { was[$1] = $2; next }
203                 { printf "| %s | %s | %s |\n", $1, ($1 in was ? was[$1] : 0), $2 }' \
204                before.txt after.txt
205          } >> "$GITHUB_STEP_SUMMARY"