How to Bulk Harvest¶

Use HarvesterAPI to create web harvest tasks (screenshots, asset downloads, video capture) and retrieve the results.

Use async for bulk operations

For harvesting multiple URLs, use the async methods shown below to run tasks concurrently. This is significantly faster than processing URLs one at a time.

Task Types¶

`task_type`	Description	Params key
`"visual"`	Full-page screenshot, PDF, or MHTML capture	`vis_params`
`"asset"`	Recursive file/asset download via wget	`wget_params`
`"video"`	Video download and transcription via yt-dlp	`vid_params`
`"single"`	Single-URL non-recursive fetch; same `wget_params` as `asset`; available from all egress locations	`wget_params`

See the Web Harvesting API reference for all supported task_params keys per type.

Single Task (Synchronous)¶

from silo_sdk import HarvesterAPI, load_config

config = load_config("config/default.json")
harvester = HarvesterAPI(config)

# Visual capture — full-page image
task_id = harvester.create_harvest_task(
    task_type="visual",
    urls=["https://example.com"],
    egress_info={"name": "New York, NY"},
    dest_path="/screenshots/",
    task_params={"output_image": True},
)
print(f"Task created: {task_id}")

# Poll until done
result = harvester.wait_for_completion(task_id, max_wait_time=300, check_interval=5)
if result and result.get("status") == "done":
    print("Capture complete")
elif result and result.get("status") == "error":
    print(f"Failed: {result.get('last_error')}")

Single Task (Async)¶

import asyncio
from silo_sdk import HarvesterAPI, load_config

config = load_config("config/default.json")
harvester = HarvesterAPI(config)

async def harvest_one(url: str) -> dict:
    task_id = harvester.create_harvest_task(
        task_type="visual",
        urls=[url],
        egress_info={"name": "New York, NY"},
        dest_path="/screenshots/",
        task_params={"output_pdf": True, "output_image": True},
    )
    return await harvester.wait_for_completion_async(
        task_id, check_interval=5, max_wait_time=300
    )

result = asyncio.run(harvest_one("https://example.com"))
print(result.get("status"))

Bulk Tasks (Async Concurrent)¶

Use async for bulk harvesting — tasks run concurrently:

import asyncio
from silo_sdk import HarvesterAPI, load_config

config = load_config("config/default.json")
harvester = HarvesterAPI(config)

urls = [
    "https://site1.example.com",
    "https://site2.example.com",
    "https://site3.example.com",
]

async def harvest_all(urls: list[str]) -> None:
    # Create all tasks first
    task_ids = []
    for url in urls:
        task_id = harvester.create_harvest_task(
            task_type="visual",
            urls=[url],
            egress_info={"name": "New York, NY"},
            dest_path="/screenshots/",
            task_params={"output_image": True},
        )
        task_ids.append(task_id)
        print(f"Created: {task_id} → {url}")

    # Poll all tasks concurrently
    results = await asyncio.gather(*[
        harvester.wait_for_completion_async(task_id, check_interval=5, max_wait_time=300)
        for task_id in task_ids
    ])

    for task_id, result in zip(task_ids, results):
        status = result.get("status") if result else "timeout"
        print(f"Task {task_id}: {status}")

asyncio.run(harvest_all(urls))

Scheduled Daily Captures (`run_after`)¶

Use run_after to queue tasks at a future time. Accepts a Unix epoch integer or ISO 8601 UTC string:

import time

task_id = harvester.create_harvest_task(
    task_type="visual",
    urls=["https://example.com"],
    egress_info={"name": "London"},
    dest_path="/daily/",
    task_params={"output_pdf": True},
    run_after=int(time.time()) + 3600,  # 1 hour from now
)

For automated daily scheduling with jitter, use the scheduled_visual_harvest.py script:

# 7 days starting now, London egress, screenshot + image
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress London --output-image

# PDF capture, A4 landscape, specific start time
python scripts/scheduled_visual_harvest.py https://target.com \
    --egress "New York, NY" --start 2026-03-01T09:00:00Z \
    --days 7 --dest-path /daily-captures/ \
    --output-pdf --paper A4 --landscape

# Preview the schedule without submitting
python scripts/scheduled_visual_harvest.py https://example.com \
    --egress London --output-image --dry-run

Tasks enter wait status until their scheduled time, then activate automatically.

Task Status Values¶

Status	Meaning
`wait`	Queued but not yet started (includes `run_after` deferred tasks)
`claimed`	Assigned to a Silo node
`running`	Actively executing
`done`	Completed successfully
`error`	Failed — check `last_error` field

Asset Download Example¶

task_id = harvester.create_harvest_task(
    task_type="asset",
    urls=["https://example.com"],
    egress_info={"name": "New York City", "categories": {"connectivity": "datacenter"}},
    dest_path="/assets/",
    task_params={
        "recursive": "true",
        "level": "2",
        "page-requisites": "true",
        "quota": "100m",
    },
)

Egress Locations¶

egress_info={"name": "New York, NY"}
egress_info={"name": "London"}
egress_info={"name": "Dallas, TX"}

# Silo for Research (datacenter egress required for asset tasks)
egress_info={
    "name": "New York, NY",
    "categories": {"connectivity": "datacenter", "availability": "public"},
}

Use BrowsingAPI.get_egress_locations() to list available location names.

For enriched metadata including connectivity type, availability, and supported protocols per city, call BrowsingAPI.get_egress_locations(include_details=True):

from silo_sdk.browsing import BrowsingAPI

locations = BrowsingAPI.get_egress_locations(include_details=True)
# Returns: {"North America": [{"name": "New York, NY", "connectivity": ["datacenter", "isp", "wireless-carrier"],
#            "availability": ["public"], "protocol": ["direct"]}, ...], ...}

TOR Routing¶

TOR egress is available for visual, video, and single task types only (not asset). Two cities support TOR: Sydney (direct + TOR) and Moncks Corner, SC (TOR only).

# TOR routing — random exit node (recommended)
egress_info={"name": "World", "categories": {"protocol": "tor"}}

# TOR routing — constrained to a specific TOR-capable city
egress_info={"name": "Sydney", "categories": {"protocol": "tor"}}
egress_info={"name": "Moncks Corner, SC", "categories": {"protocol": "tor"}}

# Geographic hierarchy also works (resolves to nearest TOR-capable city)
egress_info={"name": "Australia", "categories": {"protocol": "tor"}}   # → Sydney
egress_info={"name": "United States", "categories": {"protocol": "tor"}}  # → Moncks Corner

Download Results¶

Harvest results are stored in the bucket specified by BUCKET_ID in your config (or override per-task with dest_bucket_id). Download with FileAPI:

from silo_sdk import FileAPI

files = FileAPI(config)
results = files.find_files(path="/screenshots/")
for f in results:
    files.download_file(
        file_id=f["file_id"],
        output_path=f"./downloads/{f['name']}",
    )