Commit `137b3438e9`

137b3438e9abacc7fc404054ced30663a5747dec

parent: 63f759d70a

Verified · cmc

cmc <hello@cleberg.net> · 2026-08-23 18:22 UTC

Keep raw feed payloads; sweep in full weekly, pull twice daily

raw_records stores the JSON each feed served for every version of
every record, keyed like incident_amendments. A wrong parse or a
field added later can only be applied to history if the bytes were
kept, and the rolling feeds give no second chance. Costs ~0.7 MB
gzipped a day; the published archive goes from 20 MB to 52 MB.

A 30-day window cannot see an agency amending a record filed months
ago, which Omaha does, so Sundays sweep every feed in full.

Twice-daily pulls halve what the rolling feeds drop before capture.

Layout: unified · split

.github/workflows/daily-pull.yml +22 −5

 on:
   schedule:
-    # 06:00 America/Chicago in summer, 05:00 in winter. Offset from the hour
+    # Twice a day, because the cadence sets how much the rolling feeds drop
-    # because GitHub drops on-the-hour scheduled runs under load.
+    # before it is captured: a five-hour gap cost 13 Sarpy records once.
+    # Offset from the hour, GitHub drops on-the-hour runs under load.
     - cron: "17 11 * * *"
+    - cron: "17 23 * * *"
   workflow_dispatch:
     inputs:
       bootstrap:
 jobs:
   pull:
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v4
             sqlite3 -noheader -separator ' ' "$DB" \
               "SELECT source || '+amend', COUNT(*) FROM incident_amendments
                GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
+            sqlite3 -noheader -separator ' ' "$DB" \
+              "SELECT source || '+raw', COUNT(*) FROM raw_records
+               GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
+            sqlite3 -noheader -separator ' ' "$DB" \
+              "SELECT source || '+raw', COUNT(*) FROM raw_records
+               GROUP BY source ORDER BY source" >> before.txt 2>/dev/null || true
           fi
           cat before.txt
       - name: Pull feeds
         run: |
-          if [ "${{ inputs.full }}" = "true" ] || [ "${{ inputs.bootstrap }}" = "true" ]; then
+          # A 30-day window cannot see an agency amending a record filed months
+          # ago, and OPD does exactly that, so sweep the whole feed on Sundays.
+          if [ "${{ inputs.full }}" = "true" ] || [ "${{ inputs.bootstrap }}" = "true" ] \
+             || [ "$(date -u +%u)" = "7" ]; then
+            echo "full sweep"
             python ingest.py --full
           else
             python ingest.py
           sqlite3 -noheader -separator ' ' "$DB" \
             "SELECT source || '+amend', COUNT(*) FROM incident_amendments
              GROUP BY source ORDER BY source" >> after.txt
+          sqlite3 -noheader -separator ' ' "$DB" \
+            "SELECT source || '+raw', COUNT(*) FROM raw_records
+             GROUP BY source ORDER BY source" >> after.txt
           cat after.txt
           test -s after.txt || { echo "::error::archive is empty"; exit 1; }
           # Keyed on FILENAME, not NR == FNR: before.txt is empty on a bootstrap
             echo "incidents holds each record as first published; every later"
             echo "version the feed served is a row in incident_amendments"
             echo "($(sqlite3 -noheader "$DB" 'SELECT COUNT(*) FROM incident_amendments') so far)."
-            echo "incidents_current is the newest version of each."
+            echo "incidents_current is the newest version of each, and"
+            echo "raw_records keeps the feed's own JSON for every version"
+            echo "so a parse can be redone against what actually arrived."
             echo
             echo '```'
             sqlite3 -header -column "$DB" \

README.nfo +18 −1

       .venv/bin/python app.py
 ARCHIVE
-  .github/workflows/daily-pull.yml runs the pull at 11:17 utc and
+  .github/workflows/daily-pull.yml runs at 11:17 and 23:17 utc and
   keeps the database as metro.db.gz on the "archive" release, so the
   archive does not depend on any one machine. each run restores that
   asset, pulls, refuses to publish if any source came back with fewer
   rows than it started with, then uploads and fails loudly if a feed
   has not moved in seven days.
+  sundays it sweeps every feed in full instead of the last 30 days,
+  because a 30-day window cannot see an agency amending a record it
+  filed months ago, and omaha does that.
   first run: trigger it manually with bootstrap enabled, which pulls
   every feed in full and creates the release. after that the restore
   step is mandatory -- a bootstrap over a live archive throws away
       0 6 * * *  cd /path/to/omaha-incidents && .venv/bin/python ingest.py
+RAW
+  raw_records keeps the feed's own json for every version of every
+  record, keyed the same way amendments are. a parse that turns out
+  wrong, or a field a feed adds later, can only be applied to history
+  if the bytes were kept, and the rolling feeds mean there is no
+  second chance to fetch them. the payloads already carry fields
+  ingest does not map: council bluffs response times and priority,
+  sarpy case status.
+  it costs about 0.7 mb gzipped a day and roughly triples the
+  database: 20 mb published without it, 52 mb with. 319 records
+  predate it and their raw is gone; the feeds no longer serve them.
 AMENDMENTS
   agencies edit records after publishing them: a disposition changes,
   a case reopens, a record is withdrawn. nothing in the incidents

ingest.py +39 −24

 def upsert(conn, rows):
     """Insert records not seen before; file a changed record as an amendment.
-    Nothing in incidents is ever updated. A record whose payload differs from
+    Takes (values, raw) pairs, where raw is the feature exactly as the feed
-    the one on file is appended to incident_amendments, so the version the
+    served it. Nothing in incidents is ever updated: a record whose payload
-    agency published first stays readable next to what it published later."""
+    differs from the one on file is appended to incident_amendments, so the
+    version the agency published first stays readable next to what it published
+    later. Every version's raw payload is kept too, so a parse can be redone
+    against what actually arrived."""
     now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
-    marks = ",".join("?" * (len(COLUMNS) + 1))
+    marks = ",".join("?" * (len(COLUMNS) + 2))
-    staged = [r + (digest(r[2:]),) for r in rows]
+    staged = [r + (digest(r[2:]), raw) for r, raw in rows]
     conn.execute("DROP TABLE IF EXISTS temp.incoming")
-    conn.execute(f"CREATE TEMP TABLE incoming ({','.join(COLUMNS)}, digest)")
+    conn.execute(f"CREATE TEMP TABLE incoming ({','.join(COLUMNS)}, digest, raw)")
     conn.executemany(f"INSERT INTO temp.incoming VALUES ({marks})", staged)
     conn.execute("CREATE INDEX temp.incoming_key ON incoming (source, source_key)")
               ON o.source = i.source AND o.source_key = i.source_key
             WHERE o.digest <> i.digest""", (now,)).rowcount
+    # OR IGNORE keyed on the version, so a run that re-serves a known record
+    # stores nothing and the first full run backfills whatever is still served.
+    conn.execute(
+        """INSERT OR IGNORE INTO raw_records (source, source_key, digest,
+                                              fetched_at, payload)
+           SELECT source, source_key, digest, ?, raw FROM temp.incoming
+            WHERE raw IS NOT NULL""", (now,))
     conn.execute("DROP TABLE temp.incoming")
     return len(rows), amended
         occurred = local_iso(a["dteMidpoint"])
         if occurred is None:
             continue
-        rows.append(("opd", str(a["PK"]), "Omaha PD", a.get("RB"), occurred,
+        rows.append((("opd", str(a["PK"]), "Omaha PD", a.get("RB"), occurred,
-                     a.get("NIBRSCategory"), None, None, None, 0,
+                      a.get("NIBRSCategory"), None, None, None, 0,
-                     a.get("AddressBlock"), a.get("LatBlock"), a.get("LonBlock")))
+                      a.get("AddressBlock"), a.get("LatBlock"), a.get("LonBlock")),
+                     json.dumps(f, sort_keys=True)))
     return upsert(conn, rows)
             unmapped.add(prefix)
             agency = f"Unmapped {prefix}"
         g = f.get("geometry") or {}
-        rows.append(("sarpy", iid, agency, iid, occurred, a.get("Category"),
+        rows.append((("sarpy", iid, agency, iid, occurred, a.get("Category"),
-                     a.get("CadTypeDesc"), a.get("CadDisposition"),
+                      a.get("CadTypeDesc"), a.get("CadDisposition"),
-                     a.get("StatuteDesc"),
+                      a.get("StatuteDesc"),
-                     int(a.get("Category") == SARPY_STOP_CATEGORY),
+                      int(a.get("Category") == SARPY_STOP_CATEGORY),
-                     a.get("BlkAddress"), g.get("y"), g.get("x")))
+                      a.get("BlkAddress"), g.get("y"), g.get("x")),
+                     json.dumps(f, sort_keys=True)))
     result = upsert(conn, rows)
     if unmapped:
         print(f"  unmapped IncidentId prefixes: {sorted(unmapped)}")
         g = f.get("geometry") or {}
         code = a.get("incident_code")
         # Council Bluffs withholds the street address; the point is still exact.
-        rows.append(("cbpd", a["cfs_number"], "Council Bluffs PD",
+        rows.append((("cbpd", a["cfs_number"], "Council Bluffs PD",
-                     a.get("case_number") or a.get("cfs_number"), occurred,
+                      a.get("case_number") or a.get("cfs_number"), occurred,
-                     a.get("incident_category"), code, a.get("disp_code"), None,
+                      a.get("incident_category"), code, a.get("disp_code"), None,
-                     int(code == CBPD_STOP_CODE),
+                      int(code == CBPD_STOP_CODE),
-                     None, g.get("y"), g.get("x")))
+                      None, g.get("y"), g.get("x")),
+                     json.dumps(f, sort_keys=True)))
     return upsert(conn, rows)
                     when = datetime.strptime(f"{date} {tm}", "%m/%d/%Y %H:%M")
                 except ValueError:
                     continue
-                rows.append(("opd_csv", f"{path.stem}:{i}", "Omaha PD", rb,
+                rows.append((("opd_csv", f"{path.stem}:{i}", "Omaha PD", rb,
-                             when.strftime("%Y-%m-%dT%H:%M:%S"), None, None, None,
+                              when.strftime("%Y-%m-%dT%H:%M:%S"), None, None,
-                             desc, 0, loc,
+                              None, desc, 0, loc,
-                             float(lat) if lat else None,
+                              float(lat) if lat else None,
-                             float(lon) if lon else None))
+                              float(lon) if lon else None),
+                             json.dumps(r)))
     return upsert(conn, rows)

schema.sql +13

	@@ -76,6 +76,19 @@ FROM (
76	)	76	)
77	WHERE rn = 1;	77	WHERE rn = 1;
78		78
		79	-- What the feed actually served, one row per version, keyed the same way
		80	-- incident_amendments is. A parse that turns out wrong or a field a feed adds
		81	-- later can only be applied to history if the bytes were kept, and the feeds
		82	-- age out, so there is no second chance to fetch them.
		83	CREATE TABLE IF NOT EXISTS raw_records (
		84	source TEXT NOT NULL,
		85	source_key TEXT NOT NULL,
		86	digest TEXT NOT NULL, -- the version of the record this payload produced
		87	fetched_at TEXT NOT NULL,
		88	payload TEXT NOT NULL, -- the feature object as served, JSON
		89	PRIMARY KEY (source, source_key, digest)
		90	);
		91
79	-- ALPR cameras from OpenStreetMap (ODbL). first_seen/last_seen track when a node	92	-- ALPR cameras from OpenStreetMap (ODbL). first_seen/last_seen track when a node
80	-- entered and was last present in the Overpass result, so cameras that appear or	93	-- entered and was last present in the Overpass result, so cameras that appear or
81	-- are removed are visible over time.	94	-- are removed are visible over time.

Commit 137b3438e9

Commit `137b3438e9`