Friction
Engine

-- Enable UUID extension for collision-resistant, decentralised IDs
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

-- ── 1. DEPARTMENTS ─────────────────────────────────────────────────────────
CREATE TABLE departments (
    id               UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    name             VARCHAR(255) NOT NULL,
    short_code       VARCHAR(20)  NOT NULL UNIQUE,
    jurisdiction     VARCHAR(50)  CHECK (jurisdiction IN ('Municipal','State','Central')),
    state            VARCHAR(100),
    city             VARCHAR(100),
    portal_url       TEXT,
    created_at       TIMESTAMPTZ  DEFAULT CURRENT_TIMESTAMP,
    updated_at       TIMESTAMPTZ  DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT unique_dept_name_state UNIQUE (name, state)
);

-- ── 2. OFFICIALS ───────────────────────────────────────────────────────────
CREATE TABLE officials (
    id               UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    department_id    UUID NOT NULL REFERENCES departments(id) ON DELETE CASCADE,
    name             VARCHAR(255) NOT NULL,
    designation      VARCHAR(150),   -- PIO, CPIO, Appellate Authority, Nodal Officer
    contact_email    VARCHAR(255),
    is_active        BOOLEAN DEFAULT TRUE,
    created_at       TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (department_id, name, designation)
);

-- ── 3. INQUIRIES ───────────────────────────────────────────────────────────
CREATE TABLE inquiries (
    id                    UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    tracking_number       VARCHAR(100) NOT NULL UNIQUE,
    department_id         UUID NOT NULL REFERENCES departments(id),
    assigned_official_id  UUID REFERENCES officials(id) ON DELETE SET NULL,
    inquiry_type          VARCHAR(50)  CHECK (inquiry_type IN ('RTI','Tender','Grievance','Other')),
    category              VARCHAR(100),
    date_filed            DATE NOT NULL,
    statutory_deadline    DATE NOT NULL,
    current_status        VARCHAR(50)  DEFAULT 'Pending'
        CHECK (current_status IN ('Pending','Transferred','Rejected','Resolved','Appealed')),
    friction_score        DECIMAL(5,2) DEFAULT 0.00,  -- cached for rapid API reads
    delay_days            INTEGER GENERATED ALWAYS AS (
        CASE WHEN current_status = 'Pending'
             THEN GREATEST(0, CURRENT_DATE - statutory_deadline)
             ELSE 0 END
    ) STORED,
    created_at            TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
    updated_at            TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- ── 4. FRICTION EVENTS ─────────────────────────────────────────────────────
-- Append-only audit log. Never update or delete rows in this table.
CREATE TABLE friction_events (
    id                 UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    inquiry_id         UUID NOT NULL REFERENCES inquiries(id) ON DELETE CASCADE,
    event_date         TIMESTAMPTZ NOT NULL,
    event_category     VARCHAR(50) NOT NULL
        CHECK (event_category IN (
            'Status_Change','Department_Transfer','Deadline_Missed',
            'Document_Requested','Rejected'
        )),
    description        TEXT,
    delay_days_incurred INTEGER DEFAULT 0,  -- atomic unit of the friction score
    created_at         TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

-- ── 5. INDEXES ─────────────────────────────────────────────────────────────
CREATE INDEX idx_inquiries_department   ON inquiries(department_id);
CREATE INDEX idx_inquiries_status       ON inquiries(current_status);
CREATE INDEX idx_inquiries_deadline     ON inquiries(statutory_deadline)
    WHERE current_status = 'Pending';
CREATE INDEX idx_friction_events_inq    ON friction_events(inquiry_id);
CREATE INDEX idx_friction_events_cat    ON friction_events(event_category);
CREATE INDEX idx_friction_events_date   ON friction_events(event_date DESC);

-- ── 6. FRICTION SCORE REFRESH (runs via pg_cron every hour) ───────────────
CREATE OR REPLACE FUNCTION refresh_friction_score(p_inquiry_id UUID)
RETURNS VOID AS $$
DECLARE
    v_score DECIMAL := 0;
BEGIN
    SELECT
        COALESCE(SUM(
            CASE event_category
                WHEN 'Deadline_Missed'      THEN 15
                WHEN 'Department_Transfer'  THEN 20
                WHEN 'Rejected'             THEN 10
                WHEN 'Document_Requested'   THEN 8
                ELSE 0
            END
        ) + SUM(delay_days_incurred) * 0.5, 0)
    INTO v_score
    FROM friction_events
    WHERE inquiry_id = p_inquiry_id;

    UPDATE inquiries
    SET friction_score = LEAST(ROUND(v_score, 2), 100),
        updated_at = CURRENT_TIMESTAMP
    WHERE id = p_inquiry_id;
END;
$$ LANGUAGE plpgsql;

-- ── 7. DEPARTMENT FRICTION VIEW ────────────────────────────────────────────
CREATE OR REPLACE VIEW department_friction AS
SELECT
    d.short_code,
    d.name,
    d.city,
    COUNT(i.id)                                         AS total_inquiries,
    COUNT(i.id) FILTER (WHERE i.current_status = 'Pending'
        AND CURRENT_DATE > i.statutory_deadline)        AS sla_breaches,
    ROUND(AVG(i.friction_score), 1)                    AS avg_friction_score,
    ROUND(AVG(i.delay_days))                           AS avg_delay_days,
    MAX(i.friction_score)                              AS max_friction_score
FROM departments d
LEFT JOIN inquiries i ON i.department_id = d.id
GROUP BY d.id, d.short_code, d.name, d.city
ORDER BY avg_friction_score DESC;

You are a stateless, deterministic data extraction agent operating within a
civic transparency engine. Your sole objective is to ingest noisy, OCR-extracted
text from Indian bureaucratic documents (RTIs, Tenders, Grievances, Official
Responses) and output a strictly formatted JSON object.

You will NOT summarize. You will NOT provide conversational text.
You will output ONLY valid JSON.

══════════════════════════════════════════════════════════════════════════
RULES OF ENGAGEMENT
══════════════════════════════════════════════════════════════════════════

1. DATE STANDARDIZATION
   All dates MUST be converted to ISO 8601 format (YYYY-MM-DD).
   If a date is partial or illegible, output null.

2. EVASION CLASSIFICATION PROTOCOL
   Map evasive bureaucratic language to correct status codes:
   - "does not fall under purview"           → current_status: "Rejected"
   - "data not maintained in this format"    → current_status: "Rejected"
   - "transferred under section 6(3)"        → current_status: "Transferred"
   - "information is third-party in nature"  → current_status: "Rejected"
   - "application under process"             → current_status: "Pending"
   - Any acknowledgment without resolution   → current_status: "Pending"

3. HALLUCINATION ZERO-TOLERANCE
   If a field is not explicitly present or highly inferable from the text,
   output null. Do NOT guess tracking numbers, officer names, or dates.

4. STATUTORY DEADLINE CALCULATION
   For RTI applications: deadline = date_filed + 30 days.
   For appeals to CIC: deadline = date_filed + 45 days.
   If date_filed is null, set statutory_deadline to null.

══════════════════════════════════════════════════════════════════════════
TARGET JSON SCHEMA
══════════════════════════════════════════════════════════════════════════

{
  "tracking_number": "string | null",
  "department": {
    "name": "string — specific department receiving/responding",
    "jurisdiction": "Municipal | State | Central | Unknown"
  },
  "official": {
    "name": "string | null",
    "designation": "PIO | CPIO | Appellate Authority | Nodal Officer | null"
  },
  "inquiry_data": {
    "type": "RTI | Tender | Grievance | Other",
    "category": "string | null — subject matter (Road Construction, Water Supply…)",
    "date_filed": "YYYY-MM-DD | null",
    "statutory_deadline": "YYYY-MM-DD | null",
    "current_status": "Pending | Transferred | Rejected | Resolved | Appealed"
  },
  "friction_events": [
    {
      "event_date": "YYYY-MM-DD | null",
      "event_category": "Status_Change | Department_Transfer | Deadline_Missed | Document_Requested | Rejected",
      "description": "string — 1-sentence summary of the bureaucratic action or excuse",
      "delay_days_incurred": "integer — quantified time wasted by this specific event"
    }
  ]
}

══════════════════════════════════════════════════════════════════════════
STATUS CLASSIFICATION LOGIC
══════════════════════════════════════════════════════════════════════════

Resolved   → Requested information or action was fully provided.
Rejected   → Denied, exempt, dismissed, or "wrong format" refusal.
Transferred → Explicitly forwarded to another department.
Pending    → Acknowledgment only; no final resolution.
Appealed   → Citizen has escalated to First Appellate Authority or CIC.

Process the following input text and return the JSON object only:

[INSERT_OCR_TEXT_HERE]

#!/usr/bin/env python3
"""
Aether OS — RTI Portal Scraper + Extraction Pipeline
Scrapes RTI status from Indian government portals, extracts structured
data via Claude API, and writes to PostgreSQL.
"""

import asyncio
import json
import os
import random
import re
from datetime import datetime
from pathlib import Path

import anthropic
import asyncpg
from playwright.async_api import async_playwright
from pydantic import BaseModel, validator


# ── SCHEMA MODELS ──────────────────────────────────────────────────────────

class FrictionEvent(BaseModel):
    event_date: str | None
    event_category: str
    description: str
    delay_days_incurred: int = 0

    @validator("event_category")
    def validate_category(cls, v):
        allowed = {"Status_Change","Department_Transfer","Deadline_Missed",
                   "Document_Requested","Rejected"}
        if v not in allowed:
            raise ValueError(f"Invalid event_category: {v}")
        return v


class InquiryExtraction(BaseModel):
    tracking_number: str | None
    department: dict
    official: dict
    inquiry_data: dict
    friction_events: list[FrictionEvent]

    @validator("inquiry_data")
    def validate_status(cls, v):
        allowed = {"Pending","Transferred","Rejected","Resolved","Appealed"}
        if v.get("current_status") not in allowed:
            v["current_status"] = "Pending"
        return v


# ── PORTAL CONFIGURATIONS ───────────────────────────────────────────────────

PORTALS = [
    {
        "name": "Central RTI Portal",
        "base_url": "https://rtionline.gov.in",
        "search_path": "/request/view_status.php",
        "selectors": {
            "tracking_input": "#regNo",
            "submit_btn":     "input[type='submit']",
            "status_cell":    "td.status",
            "response_text":  "div.response-content",
        },
    },
    {
        "name": "MCGM RTI Portal",
        "base_url": "https://portal.mcgm.gov.in",
        "search_path": "/rti/status",
        "selectors": {
            "tracking_input": "input[name='rti_no']",
            "submit_btn":     "button[type='submit']",
            "status_cell":    ".rti-status-value",
            "response_text":  ".rti-reply-text",
        },
    },
]


# ── SCRAPER ─────────────────────────────────────────────────────────────────

async def scrape_rti(playwright, portal: dict, tracking_id: str) -> dict | None:
    """Scrape a single RTI status from a portal. Returns raw text or None."""
    browser = await playwright.chromium.launch(
        headless=True,
        args=["--no-sandbox", "--disable-dev-shm-usage"],
    )
    context = await browser.new_context(
        user_agent=(
            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/122.0 Safari/537.36"
        ),
        viewport={"width": 1280, "height": 720},
        java_script_enabled=True,
    )

    try:
        page = await context.new_page()
        url = portal["base_url"] + portal["search_path"]
        await page.goto(url, wait_until="networkidle", timeout=30_000)

        sel = portal["selectors"]
        await page.fill(sel["tracking_input"], tracking_id)

        # Stochastic delay before submit — evades primitive rate limiting
        await asyncio.sleep(random.uniform(1.5, 4.5))
        await page.click(sel["submit_btn"])
        await page.wait_for_load_state("networkidle", timeout=20_000)

        status_el   = await page.query_selector(sel["status_cell"])
        response_el = await page.query_selector(sel["response_text"])

        raw_text = ""
        if status_el:
            raw_text += f"Status: {await status_el.inner_text()}\n"
        if response_el:
            raw_text += f"Response:\n{await response_el.inner_text()}"

        return {
            "tracking_number": tracking_id,
            "portal": portal["name"],
            "raw_text": raw_text.strip(),
            "scraped_at": datetime.utcnow().isoformat(),
        }

    except Exception as exc:
        print(f"  ✗ {tracking_id}: {exc}")
        return None

    finally:
        await browser.close()


# ── LLM EXTRACTION ──────────────────────────────────────────────────────────

SYSTEM_PROMPT = Path("extract-prompt.txt").read_text()
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])


def extract_structured(raw_text: str) -> InquiryExtraction | None:
    """Send raw OCR text to Claude API and parse structured output."""
    prompt = SYSTEM_PROMPT.replace("[INSERT_OCR_TEXT_HERE]", raw_text)

    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        temperature=0,                # deterministic extraction
        system="Output only valid JSON. No prose.",
        messages=[{"role": "user", "content": prompt}],
    )

    try:
        data = json.loads(message.content[0].text)
        return InquiryExtraction(**data)
    except Exception as exc:
        # Route to dead-letter queue for manual review
        dead_letter_path = Path("dead_letter") / f"{datetime.utcnow().isoformat()}.json"
        dead_letter_path.parent.mkdir(exist_ok=True)
        dead_letter_path.write_text(json.dumps({
            "raw_text": raw_text,
            "error": str(exc),
            "raw_response": message.content[0].text,
        }, indent=2))
        print(f"  ✗ Extraction failed → dead_letter/{dead_letter_path.name}")
        return None


# ── DATABASE WRITE ───────────────────────────────────────────────────────────

async def upsert_inquiry(conn, extraction: InquiryExtraction, dept_id: str):
    """Upsert inquiry and append new friction events."""
    rti = extraction.inquiry_data

    inquiry_id = await conn.fetchval("""
        INSERT INTO inquiries
            (tracking_number, department_id, inquiry_type, category,
             date_filed, statutory_deadline, current_status)
        VALUES ($1,$2,$3,$4,$5,$6,$7)
        ON CONFLICT (tracking_number) DO UPDATE
            SET current_status = EXCLUDED.current_status,
                updated_at     = CURRENT_TIMESTAMP
        RETURNING id
    """, extraction.tracking_number, dept_id,
        rti.get("type"), rti.get("category"),
        rti.get("date_filed"), rti.get("statutory_deadline"),
        rti.get("current_status", "Pending"))

    for ev in extraction.friction_events:
        await conn.execute("""
            INSERT INTO friction_events
                (inquiry_id, event_date, event_category, description, delay_days_incurred)
            VALUES ($1,$2,$3,$4,$5)
            ON CONFLICT DO NOTHING
        """, inquiry_id, ev.event_date, ev.event_category,
            ev.description, ev.delay_days_incurred)

    # Refresh cached friction score
    await conn.execute("SELECT refresh_friction_score($1)", inquiry_id)


# ── MAIN PIPELINE ────────────────────────────────────────────────────────────

async def main():
    tracking_ids = Path("tracking_ids.txt").read_text().strip().splitlines()
    conn = await asyncpg.connect(os.environ["DATABASE_URL"])

    async with async_playwright() as pw:
        for portal in PORTALS:
            print(f"\n▶ Scraping {portal['name']} ({len(tracking_ids)} RTIs)…")
            for tid in tracking_ids:
                # Stochastic inter-request delay
                await asyncio.sleep(random.uniform(3.0, 9.0))

                raw = await scrape_rti(pw, portal, tid)
                if not raw or not raw["raw_text"]:
                    continue

                print(f"  → Extracting {tid}…")
                extraction = extract_structured(raw["raw_text"])
                if not extraction:
                    continue

                # Resolve department UUID from short code
                dept_id = await conn.fetchval(
                    "SELECT id FROM departments WHERE short_code = $1",
                    portal["name"].split()[0].upper()
                )
                if dept_id:
                    await upsert_inquiry(conn, extraction, dept_id)
                    print(f"  ✓ {tid} written to DB")

    await conn.close()
    print("\n✓ Pipeline complete.")


if __name__ == "__main__":
    asyncio.run(main())

// app/api/friction/departments/route.ts
// GET /api/friction/departments
// Returns ranked department friction scores for the public leaderboard

import { NextResponse } from "next/server";
import { sql } from "@vercel/postgres";

export const revalidate = 3600; // ISR: regenerate every hour

export async function GET(req: Request) {
  const { searchParams } = new URL(req.url);
  const jurisdiction = searchParams.get("jurisdiction"); // Municipal | State | Central

  const result = await sql`
    SELECT
      d.short_code,
      d.name,
      d.city,
      d.state,
      df.total_inquiries,
      df.sla_breaches,
      ROUND(df.sla_breaches::numeric / NULLIF(df.total_inquiries,0) * 100, 1) AS breach_rate_pct,
      df.avg_friction_score,
      df.avg_delay_days,
      df.max_friction_score
    FROM department_friction df
    JOIN departments d ON d.short_code = df.short_code
    WHERE ($1::text IS NULL OR d.jurisdiction = $1)
    ORDER BY df.avg_friction_score DESC
    LIMIT 50
  `;

  return NextResponse.json({
    data: result.rows,
    generated_at: new Date().toISOString(),
    schema_version: "1.0",
  });
}


// app/api/friction/inquiries/route.ts
// GET /api/friction/inquiries?dept=MCGM&status=Pending&page=1
// Paginated RTI inquiry list with friction scores

export async function GET_INQUIRIES(req: Request) {
  const { searchParams } = new URL(req.url);
  const dept   = searchParams.get("dept");
  const status = searchParams.get("status");
  const page   = parseInt(searchParams.get("page") ?? "1");
  const limit  = 25;
  const offset = (page - 1) * limit;

  const result = await sql`
    SELECT
      i.tracking_number,
      d.short_code         AS department,
      i.inquiry_type,
      i.category,
      i.date_filed,
      i.statutory_deadline,
      i.current_status,
      i.friction_score,
      i.delay_days,
      o.name               AS assigned_officer,
      o.designation        AS officer_designation
    FROM inquiries i
    JOIN departments d ON d.id = i.department_id
    LEFT JOIN officials o ON o.id = i.assigned_official_id
    WHERE ($1::text IS NULL OR d.short_code = $1)
      AND ($2::text IS NULL OR i.current_status = $2)
    ORDER BY i.friction_score DESC, i.delay_days DESC
    LIMIT ${limit} OFFSET ${offset}
  `;

  return NextResponse.json({
    data: result.rows,
    page,
    per_page: limit,
    generated_at: new Date().toISOString(),
  });
}


// app/api/friction/webhook/route.ts
// POST — internal endpoint called by pg_cron when threshold exceeded
// Broadcasts to configured X bot and Substack newsletter

export async function POST_WEBHOOK(req: Request) {
  const body = await req.json();
  const { department_short_code, friction_score, threshold } = body;

  const secret = req.headers.get("x-aether-secret");
  if (secret !== process.env.WEBHOOK_SECRET) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  // Post to X via API v2
  const tweet = `🚨 FRICTION ALERT: ${department_short_code} has crossed ` +
    `friction threshold ${threshold}. Current score: ${friction_score}/100. ` +
    `View breakdown: aetheros.in/friction/${department_short_code.toLowerCase()} ` +
    `#RTI #Accountability`;

  await fetch("https://api.twitter.com/2/tweets", {
    method: "POST",
    headers: {
      "Authorization": `Bearer ${process.env.X_BEARER_TOKEN}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ text: tweet }),
  });

  return NextResponse.json({ ok: true, tweet_length: tweet.length });
}