Merge pull request #1 from openstates/sqlmesh-audit-definition

Audit definitions

Author: alexobaseki

.gitignore

@@ -0,0 +1,6 @@
+logs
+_data
+.cache
+merged_*
+db.db
+__pycache__

README.md

@@ -1 +1,78 @@
-# scraper-audit
+# Scraper Audit
+
+This is a utility for running audits on scraper output for legislative data entities (`bill` or `event`) using [SQLMesh](https://docs.sqlmesh.com/). The script processes JSON data into a DuckDB database and runs a SQLMesh plan, returning any audit errors.
+
+## Features
+
+- Merges entity-level JSON files into one dataset.
+- Initializes a DuckDB database with merged data.
+- Runs `sqlmesh plan` on the staged models (`staged.bill` or `staged.event`).
+- Extracts and prints any audit-related warnings or errors.
+
+## Requirements
+
+- Python 3.9+
+- [`poetry`](https://python-poetry.org/docs/#installation)
+
+## Installation
+
+Clone the repository and install dependencies using Poetry:
+
+```bash
+git clone [email protected]:openstates/scraper-audit.git
+cd scraper-audit
+poetry install
+```
+
+## Usage
+
+Ensure that the main directory contains a data folder with the JSON output files to audit.
+These files are typically generated by the OpenStates Scraper and should follow the naming pattern:
+`*/*/<entity>*.json`
+For example:
+
+```bash
+_data/or/bill_0a3faf9c-1969-11f0-aaa5-4ef1b5972379.json
+_data/or/bill_0a21cf0c-196b-11f0-aaa5-4ef1b5972379.json
+```
+
+Run `poetry run python main.py --entity <entity name>` for example `poetry run python main.py --entity bill`. This should produce output similar to: 
+
+```bash
+INFO:openstates:Initializing data with arguments: entity=bill, jurisdiction=None
+INFO:openstates:Merging JSON files matching pattern: ./*/*/bill*.json
+INFO:openstates:Merged 1179 records into merged_entities.json
+INFO:openstates:Creating DuckDB schema and loading data...
+INFO:openstates:bill: initialized successfully
+INFO:openstates:Running SQLMesh plan via subprocess...
+INFO:openstates:SQLMesh plan output:
+
+`prod` environment will be initialized
+
+Models:
+└── Added:
+    └── staged.bill
+Models needing backfill:
+└── staged.bill: [2025-04-24 - 2025-05-05]
+
+staged.bill  created
+Updating physical layer ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 1/1 • 0:00:00
+
+✔ Physical layer updated
+
+
+[WARNING] staged.bill: 'assert_bills_have_sponsor' audit error: 1179 rows failed. Learn more in logs: ~/scraper-audit/logs/sqlmesh_2025_05_06_20_58_30.log
+
+[1/1] staged.bill  [insert/update rows, audits ❌1] 0.03s   
+Executing model batches ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 1/1 • 0:00:00                                                                                                                                                                                                                          
+                                                                                                                                                                                                                                                                                                                 ✔ Model batches executed
+
+staged.bill  created
+Updating virtual layer  ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 1/1 • 0:00:00
+
+✔ Virtual layer updated
+
+
+Audit failed:
+ [WARNING] staged.bill: 'assert_bills_have_sponsor' audit error: 1179 rows failed. Learn more in logs: ~/scraper-audit/logs/sqlmesh_2025_05_06_20_58_30.log
+```

__init__.py

@@ -0,0 +1,7 @@
+-- Does bill have sponsors?
+AUDIT (
+  name assert_bills_have_sponsor,
+  blocking false
+);
+SELECT * from scraper.bill
+WHERE sponsorships IS NULL;

audits/.gitkeep

@@ -0,0 +1,6 @@
+AUDIT (
+    name assert_events_are_classified,
+    blocking false
+);
+SELECT * from scraper.event
+WHERE classification IS NULL;

audits/bill.sql

@@ -0,0 +1,11 @@
+gateways:
+  duckdb:
+    connection:
+      type: duckdb
+      database: db.db
+
+default_gateway: duckdb
+
+model_defaults:
+  dialect: duckdb
+  start: 2025-04-28

audits/event.sql

@@ -0,0 +1,35 @@
+import argparse
+
+
+from sqlmesh_tasks import sqlmesh_plan
+
+if __name__ == "__main__":
+    default_parser = argparse.ArgumentParser(add_help=False)
+
+    parser = argparse.ArgumentParser(
+        parents=[default_parser],
+        description="Run audits on Scraper output and returns report as string",
+    )
+    parser.add_argument(
+        "--jurisdiction",
+        "-j",
+        type=str,
+        help="Specific jurisdiction to query from",
+    )
+    parser.add_argument(
+        "--entity",
+        "-e",
+        required=True,
+        choices=["bill", "event"],
+        type=str,
+        help="Entity type: bill or event",
+    )
+
+    args = parser.parse_args()
+    entity = args.entity
+    jurisdiction = args.jurisdiction
+    report = sqlmesh_plan(entity, jurisdiction)
+    if report:
+        print("Audit failed:\n", report)
+    else:
+        print("Audit passed.")

config.yaml

@@ -0,0 +1,35 @@
+MODEL (
+    name staged.bill,
+    kind FULL,
+    start '2024-04-24',
+    cron '0 5 * * *',
+    interval_unit 'day',
+    grain (id),
+    audits (
+    assert_bills_have_sponsor,
+    ),
+);
+
+SELECT
+    legislative_session::TEXT AS legislative_session,
+    identifier::TEXT AS identifier,
+    title::TEXT AS title,
+    from_organization::TEXT AS from_organization,
+    classification::JSON AS classification,
+    subject::JSON AS subject,
+    abstracts::JSON AS abstracts,
+    other_titles::JSON AS other_titles,
+    other_identifiers::JSON AS other_identifiers,
+    actions::JSON AS actions,
+    sponsorships::JSON AS sponsorships,
+    related_bills::JSON AS related_bills,
+    versions::JSON AS versions,
+    documents::JSON AS documents,
+    citations::JSON AS citations,
+    sources::JSON AS sources,
+    extras::JSON AS extras,
+    jurisdiction::JSON AS jurisdiction,
+    scraped_at::TIMESTAMP AS scraped_at,
+    _id::TEXT AS _id
+FROM
+    scraper.bill;

macros/.gitkeep

@@ -0,0 +1,32 @@
+MODEL (
+    name staged.event,
+    kind FULL,
+    start '2024-04-24',
+    cron '0 5 * * *',
+    interval_unit 'day',
+    grains (jurisdiction_id, start_date, 'name'),
+    audits (assert_events_are_classified),
+);
+
+SELECT
+    name::TEXT AS name,
+    all_day::BOOLEAN AS all_day,
+    NULLIF(start_date, '')::TIMESTAMP AS start_date,
+    NULLIF(end_date, '')::TIMESTAMP AS end_date,
+    status::TEXT AS status,
+    classification::TEXT AS classification,
+    description::TEXT AS description,
+    upstream_id::TEXT AS upstream_id,
+    location::JSON AS location,
+    media::JSON AS media,
+    documents::JSON AS documents,
+    links::JSON AS links,
+    participants::JSON AS participants,
+    agenda::JSON AS agenda,
+    sources::JSON AS sources,
+    extras::JSON AS extras,
+    jurisdiction::JSON AS jurisdiction,
+    scraped_at::TIMESTAMP AS scraped_at,
+    _id::TEXT AS _id
+FROM
+    scraper.event;