bulk_verify.py

"""
Bulk verify emails using the OrbiSearch API.
"""

import time
import requests

API_KEY = "your_api_key_here"
BASE_URL = "https://api.orbisearch.com"


def submit_bulk_job(emails: list[str]) -> dict:
    """Submit a bulk verification job."""
    response = requests.post(
        f"{BASE_URL}/v1/bulk",
        json=emails,
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json",
        },
    )
    response.raise_for_status()
    return response.json()


def get_job_status(job_id: str) -> dict:
    """Get the status of a bulk job."""
    response = requests.get(
        f"{BASE_URL}/v1/bulk/{job_id}",
        headers={"X-API-Key": API_KEY},
    )
    response.raise_for_status()
    return response.json()


def get_job_results(job_id: str) -> dict:
    """Get the results of a completed bulk job."""
    response = requests.get(
        f"{BASE_URL}/v1/bulk/{job_id}/results",
        headers={"X-API-Key": API_KEY},
    )
    response.raise_for_status()
    return response.json()


def verify_bulk(emails: list[str], poll_interval: int = 5) -> list[dict]:
    """
    Verify a list of emails and wait for results.

    Args:
        emails: List of email addresses to verify
        poll_interval: Seconds between status checks

    Returns:
        List of verification results
    """
    # Submit the job
    job = submit_bulk_job(emails)
    job_id = job["job_id"]
    print(f"Job submitted: {job_id}")
    print(f"Total emails: {job['total_emails']}")
    print(f"Estimated cost: {job['estimated_cost']} credits")

    # Poll for completion
    while True:
        status = get_job_status(job_id)
        print(f"Status: {status['status']} ({status['emails_processed']}/{status['total_emails']})")

        if status["status"] == "complete":
            break
        elif status["status"] == "failed":
            raise Exception("Job failed")

        time.sleep(poll_interval)

    # Get results
    results = get_job_results(job_id)
    return results["results"]


if __name__ == "__main__":
    emails = [
        "john@company.com",
        "jane@company.com",
        "info@company.com",
    ]

    results = verify_bulk(emails)

    for result in results:
        print(f"{result['email']}: {result['status']} ({result.get('substatus', 'n/a')})")