The Intelligence Pipeline

{
  "data": {
    "active": [
      {
        "wellNumber": "W41156",
        "operator":   "Slawson Petroleum Company",
        "wellName":   "BEAR DEN FEDERAL 1-19H",
        "county":     "Mountrail",
        "formation":  "Bakken",
        "permitDate": "2025-01-15"
      }
      // ... 2,171 total entries across 8 sections
    ],
    "conf":                  [...], // confidential
    "approved":              [...], // approved, not yet spudded
    "approved_confidential": [...],
    "conf_plugged_or_prod":  [...],
    "released_confidential": [...],
    "plugged":               [...],
    "other":                 [...]
  }
}

def identify_new_permits(self):
    cutoff = datetime.now() - timedelta(days=CONFIG['lookback_days'])
    processed = self.db.get_processed_wells()  # set from SQLite
    new_wells = []

    for section in nd_daily['data'].values():
        for well in section:
            permit_dt = datetime.strptime(well['permitDate'], '%Y-%m-%d')
            if permit_dt >= cutoff and well['wellNumber'] not in processed:
                new_wells.append(well['wellNumber'])

    return new_wells[:CONFIG['max_per_run']]  # cap at 100


# Cron schedule — macOS launchd / crontab
# 0 2 * * * /usr/bin/python3 /path/to/orchestrate_wellfile_pipeline.py >> /var/log/pipeline.log 2>&1

NDIC_PDF_URL = "https://www.dmr.nd.gov/oilgas/Wellfiles/{well_num}.pdf"

for well_num in new_wells:
    url = NDIC_PDF_URL.format(well_num=well_num)

    for attempt in range(3):
        try:
            resp = requests.get(url, timeout=30)
            if resp.status_code == 200:
                path = PDFS_DIR / f"{well_num}.pdf"
                path.write_bytes(resp.content)
                db.mark_downloaded(well_num)
                break
        except requests.RequestException as e:
            logging.warning(f"Attempt {attempt+1} failed: {e}")

    time.sleep(2.0)  # 0.5 req/sec to avoid NDIC rate limits

from google.cloud import vision

def ocr_page(image_bytes: bytes) -> str:
    client = vision.ImageAnnotatorClient()
    image  = vision.Image(content=image_bytes)
    resp   = client.document_text_detection(image=image)
    return resp.full_text_annotation.text


# Local fallback — requires: brew install ocrmypdf poppler
def ocr_local_fallback(pdf_path: Path) -> str:
    ocr_path = pdf_path.with_suffix('.ocr.pdf')
    subprocess.run(
        ['ocrmypdf', '--force-ocr', str(pdf_path), str(ocr_path)],
        check=True
    )
    result = subprocess.run(
        ['pdftotext', str(ocr_path), '-'],
        capture_output=True, text=True
    )
    return result.stdout

NORTH DAKOTA INDUSTRIAL COMMISSION
OIL AND GAS DIVISION

PERMIT TO DRILL

WELL NUMBER:   41156
OPERATOR:      Slawson Petroleum Company, Inc.
WELL NAME:     BEAR DEN FEDERAL 1-19H
LOCATION:      SE/4SE/4 Sec. 19, T153N, R90W
COUNTY:        Mountrail
FORMATION:     Bakken
SURFACE ELEV:  2,212 ft.
TOTAL DEPTH:   21,500 ft.
H2S PRESENT:   Yes — safety equipment required
SETBACK:       Variance granted per Docket OG-22-0471
...

{
  "wellNumber":       "W41156",
  "operator":         "Slawson Petroleum Company, Inc.",
  "wellName":         "BEAR DEN FEDERAL 1-19H",
  "county":           "Mountrail",
  "formation":        "Bakken",
  "surfaceElevation": 2212,
  "totalDepth":       21500,
  "spudDate":         "2025-01-20",
  "permitDate":       "2025-01-15",
  "signals": {
    "h2s":               true,
    "flaring":           false,
    "setbackVariance":   true,
    "multipleFormations":false
  },
  "signalCount":  2,
  "tier":         "T1",   // T1 = score 50+, high-value opportunity
  "model":        "grok-2",
  "extractedAt":  "2025-01-22T02:14:33Z"
}

{
  "wellNumber":  "W41156",
  "model":       "claude-haiku-3-5",
  "operator":    "Slawson Petroleum Company",
  "formation":   "Bakken",
  "intelligenceSummary": "High-value Bakken horizontal with H2S
    present and setback variance — likely targeting Three Forks
    secondary bench. Setback variance suggests proximity to
    residential or public land boundary. Prioritize H2S-rated
    equipment proposal.",
  "hazardFlags":     ["H2S", "SETBACK_VARIANCE"],
  "opportunityScore": 72,
  "tier":            "T1",
  "actionWindow":    "24-48 hours",
  "comparesWithGrok": {
    "operatorMatch":  true,
    "formationMatch": true,
    "signalMatch":    true
  },
  "processedAt": "2025-01-22T02:15:01Z"
}

-- snowflake/02_analytics/E_canonical_signals.sql
-- Normalize inconsistent NDIC operator name casing at the view level.
-- Source data is never modified — normalization lives here only.

CREATE OR REPLACE VIEW CANONICAL_WELL_SIGNALS AS
SELECT
  well_number,
  permit_date,
  CASE
    WHEN UPPER(operator) LIKE '%KODA RESOURCES%'
      THEN 'Koda Resources Operating, LLC'
    WHEN UPPER(operator) LIKE '%OASIS PETROLEUM%'
      THEN 'Oasis Petroleum North America, LLC'
    WHEN UPPER(operator) LIKE '%DEVON ENERGY%'
      THEN 'Devon Energy Williston, LLC'
    WHEN UPPER(operator) LIKE '%ENERPLUS%'
      THEN 'Enerplus Resources USA Corporation'
    WHEN UPPER(operator) LIKE '%GRAYSON MILL%'
      THEN 'Grayson Mill Operating, LLC'
    ELSE operator
  END AS operator_normalized,
  signal_count,
  tier,
  h2s_flag,
  flaring_flag,
  setback_variance_flag
FROM RAW_WELLFILE_TEXT;

// POST request to Lambda endpoint
{
  "query": "H2S hazard Mountrail setback variance",
  "limit": 10
}

// Response from Snowflake Cortex Search via Lambda
{
  "results": [
    {
      "wellNumber":     "W41156",
      "documentType":  "wellfile",
      "relevanceScore": 0.87,
      "snippet": "...H2S present at 15 ppm. Setback variance
                  granted per Docket OG-22-0471. Safety equipment
                  required within 500ft buffer...",
      "reportUrl": "/html_reports/W41156.html"
    }
  ],
  "totalDocs":  338,
  "searchedAt": "2026-03-04T02:14:22Z"
}

import snowflake.connector

def lambda_handler(event, context):
    query = event['body']['query']
    limit = event['body'].get('limit', 10)

    # PAT auth — no username/password stored in Lambda env
    conn = snowflake.connector.connect(
        account=os.environ['SNOWFLAKE_ACCOUNT'],
        token=os.environ['SNOWFLAKE_PAT'],
        authenticator='oauth'
    )

    results = conn.cursor().execute("""
        SELECT * FROM TABLE(
          CORTEX_SEARCH_RESULT(
            service => 'WELLFILE_SEARCH',
            query   => %s,
            limit   => %s
          )
        )
    """, (query, limit))

    return {"results": results.fetchall()}

https://oilgasorgrass.com/
  /                           → index.html (landing + well search)
  /signal_coverage.html       → live intelligence dashboard (6 tabs)
  /wellfile_updates.html      → per-well AI report index
  /daily_permits.html         → daily NDIC permit feed
  /pipeline.html              → this page

Per-well reports:
  /html_reports/W{num}.html   → Grok extraction report
  /html_reports_i/W{num}_i.html → Haiku intelligence report

Data files (S3, CloudFront-cached):
  /nd_daily.json              → 2,171 wells, updated daily
  /html_files_manifest.json   → index of all reports