archivebox.crawls.models

Module Contents

Classes

CrawlSchedule
Crawl
CrawlMachine

API

class archivebox.crawls.models.CrawlSchedule

Bases: archivebox.base_models.models.ModelWithUUID, archivebox.base_models.models.ModelWithNotes

id

‘UUIDField(…)’

created_at

‘DateTimeField(…)’

created_by

‘ForeignKey(…)’

modified_at

‘DateTimeField(…)’

template: Crawl

‘ForeignKey(…)’

schedule

‘CharField(…)’

is_enabled

‘BooleanField(…)’

label

‘CharField(…)’

notes

‘TextField(…)’

crawl_set: django.db.models.Manager[Crawl]

None

class Meta

Bases: archivebox.base_models.models.ModelWithUUID.Meta, archivebox.base_models.models.ModelWithNotes.Meta

app_label

‘crawls’

verbose_name

‘Scheduled Crawl’

verbose_name_plural

‘Scheduled Crawls’

__str__() → str

property api_url: str

save(*args, **kwargs)

property last_run_at

property next_run_at

is_due(now=None) → bool

enqueue(queued_at=None) → archivebox.crawls.models.Crawl

class archivebox.crawls.models.Crawl

Bases: archivebox.base_models.models.ModelWithOutputDir, archivebox.base_models.models.ModelWithConfig, archivebox.base_models.models.ModelWithHealthStats, archivebox.workers.models.ModelWithStateMachine

id

‘UUIDField(…)’

created_at

‘DateTimeField(…)’

created_by

‘ForeignKey(…)’

modified_at

‘DateTimeField(…)’

urls

‘TextField(…)’

config

‘JSONField(…)’

max_depth

‘PositiveSmallIntegerField(…)’

max_urls

‘IntegerField(…)’

max_size

‘BigIntegerField(…)’

tags_str

‘CharField(…)’

persona_id

‘UUIDField(…)’

label

‘CharField(…)’

notes

‘TextField(…)’

schedule

‘ForeignKey(…)’

status

‘StatusField(…)’

retry_at

‘RetryAtField(…)’

state_machine_name

‘archivebox.crawls.models.CrawlMachine’

retry_at_field_name

‘retry_at’

state_field_name

‘status’

StatusChoices

None

active_state

None

schedule_id: uuid.UUID | None

None

sm: CrawlMachine

None

snapshot_set: django.db.models.Manager[archivebox.core.models.Snapshot]

None

class Meta

Bases: archivebox.base_models.models.ModelWithOutputDir.Meta, archivebox.base_models.models.ModelWithConfig.Meta, archivebox.base_models.models.ModelWithHealthStats.Meta, archivebox.workers.models.ModelWithStateMachine.Meta

app_label

‘crawls’

verbose_name

‘Crawl’

verbose_name_plural

‘Crawls’

__str__()

save(*args, **kwargs)

property api_url: str

to_json() → dict

Convert Crawl model instance to a JSON-serializable dict.

static from_json(record: dict, overrides: dict | None = None)

Create or get a Crawl from a JSON dict.

Args: record: Dict with ‘urls’ (required), optional ‘max_depth’, ‘tags_str’, ‘label’ overrides: Dict of field overrides (e.g., created_by_id)

Returns: Crawl instance or None if invalid

property output_dir: pathlib.Path

Construct output directory: archive/users/{username}/crawls/{YYYYMMDD}/{domain}/{crawl-id} Domain is extracted from the first URL in the crawl.

get_urls_list() → list[str]

Get list of URLs from urls field, filtering out comments and empty lines.

static normalize_domain(value: str) → str

static split_filter_patterns(value) → list[str]

classmethod _pattern_matches_url(url: str, pattern: str) → bool

get_url_allowlist(*, use_effective_config: bool = False, snapshot=None) → list[str]

get_url_denylist(*, use_effective_config: bool = False, snapshot=None) → list[str]

url_passes_filters(url: str, *, snapshot=None, use_effective_config: bool = True) → bool

set_url_filters(allowlist, denylist) → None

apply_crawl_config_filters() → dict[str, int]

_iter_url_lines() → list[tuple[str, str]]

count_urls_for_limit() → int

Count unique URLs already queued or snapshotted for this crawl.

max_urls is a crawl-wide cap on snapshots, so direct URL entries and recursively discovered snapshots both have to consume the same budget.

remaining_url_capacity() → int | None

has_remaining_url_capacity() → bool

remaining_snapshot_capacity() → int | None

has_remaining_snapshot_capacity() → bool

prune_urls(predicate) → list[str]

prune_url(url: str) → int

exclude_domain(domain: str) → dict[str, int | str | bool]

get_system_task() → str | None

resolve_persona()

add_url(entry: dict) → bool

Add a URL to the crawl queue if not already present.

Args: entry: dict with ‘url’, optional ‘depth’, ‘title’, ‘timestamp’, ‘tags’, ‘via_snapshot’, ‘plugin’

Returns: True if URL was added, False if skipped (duplicate or depth exceeded)

create_snapshots_from_urls() → list[archivebox.core.models.Snapshot]

Create Snapshot objects for each URL in self.urls that doesn’t already exist.

Returns: List of newly created Snapshot objects

create_discovered_snapshot(parent_snapshot, *, url: str, depth: int, title: str = '', tags: str = '', created_by_id: int | None = None)

Create one child snapshot if it passes crawl filters and limits.

install_declared_binaries(binary_names: set[str], machine=None) → None

Install crawl-declared Binary rows without violating the retry_at lock lifecycle.

Correct calling pattern:

1. Crawl hooks declare Binary records and queue them with retry_at <= now

2. Exactly one actor claims each Binary by moving retry_at into the future

3. Only that owner executes .sm.tick() and performs install side effects

4. Everyone else waits for the claimed owner to finish instead of launching a second install against shared state such as the pip or npm trees

This helper follows that contract by claiming each Binary before ticking it, and by waiting when another worker already owns the row. That keeps synchronous crawl execution compatible with the shared background runner and avoids duplicate installs of the same dependency.

run() → Snapshot | None

Execute this Crawl: run hooks, process JSONL, create snapshots.

Called by the state machine when entering the ‘started’ state.

Returns: The root Snapshot for this crawl, or None for system crawls that don’t create snapshots

is_finished() → bool

Check if crawl is finished (all snapshots sealed or no snapshots exist).

cleanup()

Clean up background hooks and run on_CrawlEnd hooks.

class archivebox.crawls.models.CrawlMachine(obj, *args, **kwargs)

Bases: archivebox.workers.models.BaseStateMachine

crawl: archivebox.crawls.models.Crawl

None

State machine for managing Crawl lifecycle.

Hook Lifecycle: ┌─────────────────────────────────────────────────────────────┐ │ QUEUED State │ │ • Waiting for crawl to be ready (has URLs) │ └─────────────────────────────────────────────────────────────┘ ↓ tick() when can_start() ┌─────────────────────────────────────────────────────────────┐ │ STARTED State → enter_started() │ │ 1. crawl.run() │ │ • discover_hooks(‘Crawl’) → finds all crawl hooks │ │ • For each hook: │ │ - run_hook(script, output_dir, …) │ │ - Parse JSONL from hook output │ │ - process_hook_records() → creates Snapshots │ │ • create_snapshots_from_urls() → from self.urls field │ │ │ │ 2. Snapshots process independently with their own │ │ state machines (see SnapshotMachine) │ └─────────────────────────────────────────────────────────────┘ ↓ tick() when is_finished() ┌─────────────────────────────────────────────────────────────┐ │ SEALED State → enter_sealed() │ │ • cleanup() → runs on_CrawlEnd hooks, kills background │ │ • Set retry_at=None (no more processing) │ └─────────────────────────────────────────────────────────────┘

model_attr_name

‘crawl’

queued

‘State(…)’

started

‘State(…)’

sealed

‘State(…)’

tick

None

seal

‘to(…)’

can_start() → bool

is_finished() → bool

Check if all Snapshots for this crawl are finished.

enter_started()

enter_sealed()