End-to-end example

pip install requests
export SCOUT_API_KEY="your-api-key"
python scout_example.py https://www.example-shop.cz CZ CZK

1. Add the website
  ANALYZING
  TRAINING COLLECTING_PAGES
  ...
  COMPLETED
2. Look at the sample
  1000 sample products, for example: Bio kokosový olej 500 ml
3. Scrape the full catalog
Approve EUR 12.88, at most EUR 19.32? [y/N] y
  RUNNING 38
  ...
  COMPLETED
  Charged EUR 12.57 for 17950 products
4. Download the products
  products.jsonl.gz has 17950 products

"""Add a website, scrape its full catalog and download its products."""

import argparse
import gzip
import json
import os
import shutil
import sys
import time

import requests

BASE = "https://api.deepscout.ai/v1"
END = {"COMPLETED", "FAILED", "CANCELLED"}

api = requests.Session()
api.headers["Authorization"] = f"Bearer {os.environ['SCOUT_API_KEY']}"


class ApiError(Exception):
    def __init__(self, status, body):
        super().__init__(f"{status} {body.get('code')}: {body.get('detail')}")
        self.code = body.get("code")
        self.body = body


def call(method, path, **kwargs):
    response = api.request(method, BASE + path, **kwargs)
    if response.status_code >= 400:
        raise ApiError(response.status_code, response.json())
    return response.json() if response.content else None


def wait(path, seconds, progress):
    """Read the job until it is in an end status or waits for approval."""
    while True:
        job = call("GET", path)
        print(f"  {job['status']} {progress(job)}")
        if job["status"] in END or job["status"] == "AWAITING_APPROVAL":
            return job
        time.sleep(seconds)


def confirm(estimate):
    answer = input(f"Approve EUR {estimate['total']}, at most EUR {estimate['cap']}? [y/N] ")
    if answer.strip().lower() != "y":
        sys.exit("Not approved.")


def approve(job):
    """Approve the quote. Ask again if the price went up."""
    confirm(job["estimate"])
    while True:
        try:
            return call("POST", f"/jobs/scraping/{job['job_id']}/approve")
        except ApiError as error:
            if error.code == "ESTIMATE_CHANGED":
                print("The price went up.")
                confirm(error.body["estimate"])
            elif error.code == "INSUFFICIENT_FUNDS":
                sys.exit(f"Top up at least EUR {error.body['missing_topup']}, then run again.")
            else:
                raise


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("url")
    parser.add_argument("country")
    parser.add_argument("currency")
    args = parser.parse_args()

    project_id = call("GET", "/projects")[0]["project_id"]

    print("1. Add the website")
    setup = call("POST", "/jobs/setup", json={
        "project_id": project_id,
        "url": args.url,
        "country": args.country,
        "currency": args.currency,
    })
    setup = wait(
        f"/jobs/setup/{setup['job_id']}", 60,
        lambda job: job.get("train_progress", {}).get("phase", ""),
    )
    if setup["status"] != "COMPLETED":
        sys.exit(f"The setup ended as {setup['status']}: {setup.get('error')}")
    website_id = setup["website_id"]

    print("2. Look at the sample")
    try:
        sample_url = call("GET", f"/samples/{website_id}", params={"project_id": project_id})["url"]
        sample = [json.loads(line) for line in requests.get(sample_url).text.splitlines() if line]
        print(f"  {len(sample)} sample products, for example: {sample[0]['name']}")
    except ApiError as error:
        if error.code != "SAMPLE_NOT_FOUND":
            raise
        print("  The sample is not ready yet.")

    print("3. Scrape the full catalog")
    job = call("POST", "/jobs/scraping", json={"project_id": project_id, "website_id": website_id})
    if job["status"] == "AWAITING_APPROVAL":
        approve(job)
    job = wait(
        f"/jobs/scraping/{job['job_id']}", 120,
        lambda job: job.get("scrape_progress", {}).get("percent", ""),
    )
    if job["status"] != "COMPLETED":
        sys.exit(f"The scrape ended as {job['status']}: {job.get('error')}")
    print(f"  Charged EUR {job['settlement']['amount']} for {job['settlement']['quantity']} products")

    print("4. Download the products")
    params = {"project_id": project_id, "website_id": website_id}
    with api.get(f"{BASE}/exports/all_products_with_details", params=params, stream=True) as response:
        response.raise_for_status()
        with open("products.jsonl.gz", "wb") as file:
            shutil.copyfileobj(response.raw, file)
    with gzip.open("products.jsonl.gz", "rt", encoding="utf-8") as file:
        count = sum(1 for _ in file)
    print(f"  products.jsonl.gz has {count} products")


if __name__ == "__main__":
    main()