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

‘CompactUUIDField(…)’

created_at

‘DateTimeField(…)’

created_by

‘ForeignKey(…)’

modified_at

‘DateTimeField(…)’

template: Crawl

‘ForeignKey(…)’

schedule

‘CharField(…)’

is_enabled

‘BooleanField(…)’

config

‘JSONField(…)’

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.ModelWithDeleteAfter, archivebox.base_models.models.ModelWithOutputDir, archivebox.base_models.models.ModelWithConfig, archivebox.base_models.models.ModelWithHealthStats, archivebox.workers.models.ModelWithStateMachine

id

‘CompactUUIDField(…)’

created_at

‘DateTimeField(…)’

created_by

‘ForeignKey(…)’

modified_at

‘DateTimeField(…)’

urls

‘TextField(…)’

config

‘JSONField(…)’

permissions

‘GeneratedField(…)’

max_depth

‘PositiveSmallIntegerField(…)’

tags_str

‘CharField(…)’

persona

‘ForeignKey(…)’

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

delete_after_final_statuses

()

RUNNABLE_STATES

()

INACTIVE_STATES

()

schedule_id: uuid.UUID | None

None

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

None

class Meta

Bases: archivebox.base_models.models.ModelWithDeleteAfter.Meta, 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’

indexes

None

__str__()

get_delete_after_config_value()

pause(*, save: bool = True) → bool

resume(*, when=None, save: bool = True) → bool

cancel() → None

schedule_child_snapshots_for_sealing() → int

schedule_child_snapshots_for_pause() → int

classmethod missing_delete_at_candidates()

save(*args, **kwargs)

update_child_snapshot_permissions(old_permissions: str | None, new_permissions: str | None) → int

property api_url: str

static parse_tag_names(tags: collections.abc.Iterable[str] | str, *, pattern: str = ',') → list[str]

current_tag_names() → list[str]

apply_snapshot_tag_diff(*, added_tag_names: collections.abc.Iterable[str], removed_tag_names: collections.abc.Iterable[str]) → None

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

get_urls_list() → list[str]

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

has_internal_input_root() → bool

Return True when Crawl.urls is preserved source text, not the work queue.

The runner creates a synthetic root snapshot for raw import text so parser hooks use the same Snapshot lifecycle as every other extractor. In that mode the submitted text must remain in Crawl.urls forever; parsed URLs live as child Snapshot rows and should not be appended back.

static normalize_domain(value: str) → str

static split_filter_patterns(value) → list[str]

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

get_current_config(*, refresh: bool = False) → dict[str, Any]

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

url_passes_compiled_filters(url: str, *, allowlist: list[str], denylist: list[str]) → 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()

static _config_value(config: collections.abc.Mapping[str, Any] | Any, key: str, default: Any = None) → Any

classmethod create_scheduler_row(**kwargs) → archivebox.crawls.models.Crawl

limit_stop_reason(*, config: collections.abc.Mapping[str, Any] | Any | None = None, output_dir: pathlib.Path | None = None, num_snapshots: int | None = None) → str

lifecycle_stop_reason(*, num_snapshots: int | None = None, num_sealed_snapshots: int | None = None) → str

stop_reason(*, config: collections.abc.Mapping[str, Any] | Any | None = None, output_dir: pathlib.Path | None = None, num_snapshots: int | None = None, num_sealed_snapshots: int | None = None) → str

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.

create_discovered_snapshots(parent_snapshot, records: collections.abc.Iterable[collections.abc.Mapping[str, Any]], *, depth: int, created_by_id: int | None = None) → list[archivebox.core.models.Snapshot]

Create child snapshots from discovered URL records after filtering and deduping once.

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

Install crawl-declared binaries through their unified state machine.

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(…)’

paused

‘State(…)’

sealed

‘State(…)’

tick

None

seal

None

pause_requested

None

resume_requested

‘to(…)’

can_start() → bool

is_finished() → bool

Check if all Snapshots for this crawl are finished.

has_finished_snapshots() → bool

A queued crawl with only final Snapshot rows was interrupted before sealing.

enter_queued()

enter_started()

enter_paused()

enter_sealed()