-- Run this in the BigQuery console
CREATE TABLE your_dataset.browse_ai_leads (
    task_id STRING,
    robot_id STRING,
    first_name STRING,
    last_name STRING,
    email STRING,
    company_name STRING,
    phone STRING,
    website STRING,
    job_title STRING,
    origin_url STRING,
    scraped_at TIMESTAMP
);

from google.cloud import bigquery
from flask import Flask, request, jsonify
from datetime import datetimeapp = Flask(__name__)# Set GOOGLE_APPLICATION_CREDENTIALS env var to your service account JSON path
client = bigquery.Client()
TABLE_ID = "your-project.your_dataset.browse_ai_leads"@app.route("/browse-ai-webhook", methods=["POST"])
def handle_webhook():
    payload = request.get_json()    if payload.get("event") != "taskFinishedSuccessfully":
        return jsonify({"status": "ignored"}), 200    task = payload["task"]
    captured = task.get("capturedTexts", {})    row = {
        "task_id": task.get("id", ""),
        "robot_id": task.get("robotId", ""),
        "first_name": captured.get("first_name", ""),
        "last_name": captured.get("last_name", ""),
        "email": captured.get("email", ""),
        "company_name": captured.get("company_name", ""),
        "phone": captured.get("phone", ""),
        "website": captured.get("website", ""),
        "job_title": captured.get("job_title", ""),
        "origin_url": task.get("inputParameters", {}).get("originUrl", ""),
        "scraped_at": datetime.utcnow().isoformat()
    }    errors = client.insert_rows_json(TABLE_ID, [row])
    if errors:
        return jsonify({"status": "error", "errors": errors}), 500    return jsonify({"status": "inserted"}), 200if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

curl -X POST "https://api.browse.ai/v2/robots/YOUR_ROBOT_ID/webhooks" \
  -H "Authorization: Bearer YOUR_BROWSE_AI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://yourdomain.com/browse-ai-webhook",
    "events": ["taskFinishedSuccessfully"]
  }'

from google.cloud import bigquery
import requests
from datetime import datetime, timedeltaBROWSE_AI_API_KEY = "your_browse_ai_api_key"
ROBOT_ID = "your_robot_id"
TABLE_ID = "your-project.your_dataset.browse_ai_leads"client = bigquery.Client()def get_recent_tasks(since_hours=1):
    resp = requests.get(
        f"https://api.browse.ai/v2/robots/{ROBOT_ID}/tasks",
        headers={"Authorization": f"Bearer {BROWSE_AI_API_KEY}"},
        params={"pageSize": 100}
    )
    tasks = resp.json().get("result", {}).get("robotTasks", {}).get("items", [])
    cutoff = datetime.utcnow() - timedelta(hours=since_hours)
    return [t for t in tasks if t.get("status") == "successful"
            and datetime.fromisoformat(t["finishedAt"].replace("Z","")) > cutoff]def batch_load_to_bigquery(tasks):
    rows = []
    for task in tasks:
        captured = task.get("capturedTexts", {})
        rows.append({
            "task_id": task.get("id", ""),
            "robot_id": task.get("robotId", ""),
            "email": captured.get("email", ""),
            "company_name": captured.get("company_name", ""),
            "origin_url": task.get("inputParameters", {}).get("originUrl", ""),
            "scraped_at": task.get("finishedAt", "")
        })    if rows:
        errors = client.insert_rows_json(TABLE_ID, rows)
        if errors:
            print(f"Errors: {errors}")
        else:
            print(f"Inserted {len(rows)} rows")# Run on a schedule (e.g. cron job every hour)

-- Create a deduplicated view
CREATE OR REPLACE VIEW your_dataset.browse_ai_leads_deduped AS
SELECT * EXCEPT(row_num)
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY task_id ORDER BY scraped_at DESC) AS row_num
    FROM your_dataset.browse_ai_leads
)
WHERE row_num = 1;

CREATE TABLE your_dataset.browse_ai_leads (
    ...
    scraped_at TIMESTAMP
)
PARTITION BY DATE(scraped_at);