archivebox.core.models

Module Contents

Classes

Tag
SnapshotTag
SnapshotQuerySet	Custom QuerySet for Snapshot model with export methods that persist through .filter() etc.
SnapshotManager	Manager for Snapshot model - uses SnapshotQuerySet for chainable methods
Snapshot
SnapshotMachine	State machine for managing Snapshot lifecycle.
ArchiveResult

API

class archivebox.core.models.Tag

Bases: archivebox.base_models.models.ModelWithUUID

id

‘AutoField(…)’

created_by

‘ForeignKey(…)’

created_at

‘DateTimeField(…)’

modified_at

‘DateTimeField(…)’

name

‘CharField(…)’

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

None

class Meta

Bases: archivebox.base_models.models.ModelWithUUID.Meta

app_label

‘core’

verbose_name

‘Tag’

verbose_name_plural

‘Tags’

__str__()

property slug: str

ASCII-safe slugified form of the tag name (derived, not stored).

property api_url: str

to_json() → dict

Convert Tag model instance to a JSON-serializable dict.

static from_json(record: dict[str, Any], overrides: dict[str, Any] | None = None)

Create/update Tag from JSON dict.

Args: record: JSON dict with ‘name’ field overrides: Optional dict with ‘snapshot’ to auto-attach tag

Returns: Tag instance or None

class archivebox.core.models.SnapshotTag

Bases: django.db.models.Model

id

‘AutoField(…)’

snapshot

‘ForeignKey(…)’

tag

‘ForeignKey(…)’

class Meta

app_label

‘core’

db_table

‘core_snapshot_tags’

unique_together

[(‘snapshot’, ‘tag’)]

class archivebox.core.models.SnapshotQuerySet

Bases: django.db.models.QuerySet

Custom QuerySet for Snapshot model with export methods that persist through .filter() etc.

FILTER_TYPES

None

filter_by_patterns(patterns: list[str], filter_type: str = 'exact') → archivebox.core.models.SnapshotQuerySet

Filter snapshots by URL patterns using specified filter type

search(patterns: list[str]) → archivebox.core.models.SnapshotQuerySet

Search snapshots using the configured search backend

to_json(with_headers: bool = False) → str

Generate JSON index from snapshots

to_csv(cols: list[str] | None = None, header: bool = True, separator: str = ',', ljust: int = 0) → str

Generate CSV output from snapshots

to_html(with_headers: bool = True) → str

Generate main index HTML from snapshots

class archivebox.core.models.SnapshotManager

Bases: models.Manager.from_queryset(SnapshotQuerySet)

Manager for Snapshot model - uses SnapshotQuerySet for chainable methods

filter(*args, **kwargs)

get_queryset()

remove(atomic: bool = False) → tuple

Remove snapshots from the database

class archivebox.core.models.Snapshot

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

id

‘UUIDField(…)’

created_at

‘DateTimeField(…)’

modified_at

‘DateTimeField(…)’

url

‘URLField(…)’

timestamp

‘CharField(…)’

bookmarked_at

‘DateTimeField(…)’

crawl: archivebox.crawls.models.Crawl

‘ForeignKey(…)’

parent_snapshot

‘ForeignKey(…)’

title

‘CharField(…)’

downloaded_at

‘DateTimeField(…)’

depth

‘PositiveSmallIntegerField(…)’

fs_version

‘CharField(…)’

current_step

‘PositiveSmallIntegerField(…)’

retry_at

‘RetryAtField(…)’

status

‘StatusField(…)’

config

‘JSONField(…)’

notes

‘TextField(…)’

tags

‘ManyToManyField(…)’

state_machine_name

‘archivebox.core.models.SnapshotMachine’

state_field_name

‘status’

retry_at_field_name

‘retry_at’

StatusChoices

None

active_state

None

crawl_id: uuid.UUID

None

parent_snapshot_id: uuid.UUID | None

None

_prefetched_objects_cache: dict[str, Any]

None

objects

‘SnapshotManager(…)’

archiveresult_set: django.db.models.Manager[ArchiveResult]

None

class Meta

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

app_label

‘core’

verbose_name

‘Snapshot’

verbose_name_plural

‘Snapshots’

constraints

None

__str__()

property created_by

Convenience property to access the user who created this snapshot via its crawl.

property process_set

Get all Process objects related to this snapshot’s ArchiveResults.

property binary_set

Get all Binary objects used by processes related to this snapshot.

save(*args, **kwargs)

static _fs_current_version() → str

Get current ArchiveBox filesystem version (normalized to x.x.0 format)

property fs_migration_needed: bool

Check if snapshot needs filesystem migration

_fs_next_version(version: str) → str

Get next version in migration chain (0.7/0.8 had same layout, only 0.8→0.9 migration needed)

static is_legacy_archive_dir(path: pathlib.Path) → bool

Return True for old-style archive/{timestamp} snapshot directories.

migrate_filesystem_to_current_version(source_dir: pathlib.Path | None = None, config: ArchiveBoxBaseConfig | None = None) → None

Copy legacy snapshot output into the current layout and defer old-dir cleanup.

The ordering is intentionally crash-safe:

1. Copy from the legacy directory into the new directory idempotently.

2. Verify the new directory has every old file.

3. Convert metadata in the new directory.

4. Update fs_version in memory for the caller to save.

5. Cleanup is scheduled only after the DB commit succeeds.

_fs_migrate_from_0_7_0_to_0_9_0(source_dir: pathlib.Path | None = None, config: ArchiveBoxBaseConfig | None = None)

_fs_migrate_from_0_8_0_to_0_9_0(source_dir: pathlib.Path | None = None, config: ArchiveBoxBaseConfig | None = None)

_fs_migrate_legacy_to_0_9_0(source_dir: pathlib.Path | None = None, target_dir: pathlib.Path | None = None, config: ArchiveBoxBaseConfig | None = None)

Migrate from flat to nested structure.

0.8.x: archive/{timestamp}/ 0.9.x: archive/users/{user}/snapshots/YYYYMMDD/{domain}/{uuid}/

_cleanup_old_migration_dir(old_dir: pathlib.Path, new_dir: pathlib.Path)

Delete old directory and create symlink after successful migration.

static extract_domain_from_url(url: str) → str

Extract domain from URL for 0.9.x path structure. Uses full hostname with sanitized special chars.

Examples: https://example.com:8080 → example.com_8080 https://sub.example.com → sub.example.com file:///path → localhost data:text/html → data

get_storage_path_for_version(version: str, config: ArchiveBoxBaseConfig | None = None) → pathlib.Path

Calculate storage path for specific filesystem version. Centralizes path logic so it’s reusable.

0.7.x/0.8.x: archive/{timestamp} 0.9.x: archive/users/{username}/snapshots/YYYYMMDD/{domain}/{uuid}/

classmethod load_from_directory(snapshot_dir: pathlib.Path) → Optional[archivebox.core.models.Snapshot]

Load existing Snapshot from DB by reading index.jsonl or index.json.

Reads index file, extracts url+timestamp, queries DB. Returns existing Snapshot or None if not found/invalid. Does NOT create new snapshots.

ONLY used by: archivebox update (for orphan detection)

classmethod create_from_directory(snapshot_dir: pathlib.Path) → Optional[archivebox.core.models.Snapshot]

Create new Snapshot from orphaned directory.

Validates timestamp, ensures uniqueness. Returns new UNSAVED Snapshot or None if invalid.

ONLY used by: archivebox update (for orphan import)

static _select_best_timestamp(index_timestamp: object | None, folder_name: str) → str | None

Select best timestamp from index.json vs folder name.

Validates range (1995-2035). When a valid legacy folder name is available it is the stable filesystem identity, so preserve it over normalized variants like “1508259732.0” found in old index files.

classmethod _ensure_unique_timestamp(url: str, timestamp: str) → str

Ensure timestamp is globally unique. If there is a collision, add a tiny fractional suffix until unique.

static _detect_fs_version_from_index(data: dict) → str

Detect fs_version from index.json structure.

• Has fs_version field: use it

• Has history dict: 0.7.0

• Has archive_results list: 0.8.0

• Default: 0.7.0

reconcile_with_index(output_dir: pathlib.Path | None = None, update_existing_archive_results: bool = True)

Merge index.json/index.jsonl with DB. DB is source of truth.

• Title: longest non-URL

• Tags: union

• ArchiveResults: keep both (by plugin+start_ts)

Converts index.json to index.jsonl if needed, then writes back in JSONL format.

Used by: archivebox update (to sync index with DB)

reconcile_with_index_json(output_dir: pathlib.Path | None = None, update_existing_archive_results: bool = True)

Deprecated: use reconcile_with_index() instead.

_merge_title_from_index(index_data: dict)

Merge title - prefer longest non-URL title.

_merge_tags_from_index(index_data: dict)

Merge tags - union of both sources.

_merge_archive_results_from_index(index_data: dict, update_existing: bool = True)

Merge ArchiveResults - keep both (by plugin+start_ts).

_create_archive_result_if_missing(result_data: dict, existing: dict, update_existing: bool = True)

Create ArchiveResult if not already in DB.

write_index_json()

Write index.json in 0.9.x format (deprecated, use write_index_jsonl).

write_index_jsonl(output_dir: pathlib.Path | None = None)

Write index.jsonl in flat JSONL format.

Each line is a JSON record with a ‘type’ field:

• Snapshot: snapshot metadata (crawl_id, url, tags, etc.)

• ArchiveResult: extractor results (plugin, status, output, etc.)

• Binary: binary info used for the extraction

• Process: process execution details (cmd, exit_code, timing, etc.)

read_index_jsonl(output_dir: pathlib.Path | None = None) → dict

Read index.jsonl and return parsed records grouped by type.

Returns dict with keys: ‘snapshot’, ‘archive_results’, ‘binaries’, ‘processes’

convert_index_json_to_jsonl(output_dir: pathlib.Path | None = None) → bool

Convert index.json to index.jsonl format.

Reads existing index.json, creates index.jsonl, and removes index.json. Returns True if conversion was performed, False if no conversion needed.

static move_directory_to_invalid(snapshot_dir: pathlib.Path)

Move invalid directory to data/invalid/YYYYMMDD/.

Used by: archivebox update (when encountering invalid directories)

classmethod find_and_merge_duplicates() → int

Find and merge snapshots with same url:timestamp. Returns count of duplicate sets merged.

Used by: archivebox update (Phase 3: deduplication)

classmethod _merge_snapshots(snapshots: collections.abc.Sequence[archivebox.core.models.Snapshot])

Merge exact duplicates. Keep oldest, union files + ArchiveResults.

property output_dir_parent: str

property output_dir_name: str

archive(overwrite=False, methods=None)

tags_str(nocache=True) → str | None

icons(path: str | None = None) → str

Generate HTML icons showing which extractor plugins have succeeded for this snapshot

property api_url: str

get_absolute_url()

domain() → str

property title_stripped: str

static _normalize_title_candidate(candidate: str | None, *, snapshot_url: str) → str

property resolved_title: str

hashes_index() → dict[str, dict[str, Any]]

property output_dir: pathlib.Path

The filesystem path to the snapshot’s output directory.

ensure_legacy_archive_symlink() → None

Ensure the legacy archive/ path resolves to this snapshot.

ensure_crawl_symlink(*, crawl_dir: pathlib.Path | None = None, snapshot_dir: pathlib.Path | None = None) → None

Ensure snapshot is symlinked under its crawl output directory.

legacy_archive_path() → str

archive_path_from_db() → str

Best-effort public URL path derived from DB fields only.

url_path() → str

URL path matching the current snapshot output_dir layout.

archive_path()

archive_size()

save_tags(tags: collections.abc.Iterable[str] = ()) → None

pending_archiveresults() → django.db.models.QuerySet[archivebox.core.models.ArchiveResult]

run() → list[archivebox.core.models.ArchiveResult]

Execute snapshot by creating pending ArchiveResults for all enabled hooks.

Returns: list[ArchiveResult]: Newly created pending results

cleanup()

Clean up background ArchiveResult hooks and empty results.

Called by the state machine when entering the ‘sealed’ state. Deletes empty ArchiveResults after the abx-dl cleanup phase has finished.

to_json() → dict

Convert Snapshot model instance to a JSON-serializable dict. Includes all fields needed to fully reconstruct/identify this snapshot.

static from_json(record: dict[str, Any], overrides: dict[str, Any] | None = None, queue_for_extraction: bool = True)

Create/update Snapshot from JSON dict.

Unified method that handles:

• ID-based patching: {“id”: “…”, “title”: “new title”}

• URL-based create/update: {“url”: “…”, “title”: “…”, “tags”: “…”}

• Auto-creates Crawl if not provided

• Optionally queues for extraction

Args: record: Dict with ‘url’ (for create) or ‘id’ (for patch), plus other fields overrides: Dict with ‘crawl’, ‘snapshot’ (parent), ‘created_by_id’ queue_for_extraction: If True, sets status=QUEUED and retry_at (default: True)

Returns: Snapshot instance or None

create_pending_archiveresults() → list[archivebox.core.models.ArchiveResult]

Create ArchiveResult records for all enabled hooks.

Uses the hooks system to discover available hooks from:

• abx_plugins/plugins//on_Snapshot__.{py,sh,js}

• data/custom_plugins//on_Snapshot__.{py,sh,js}

Creates one ArchiveResult per hook (not per plugin), with hook_name set. This enables step-based execution where all hooks in a step can run in parallel.

is_finished_processing() → bool

Check if all ArchiveResults are finished.

Note: This is only called for observability/progress tracking. The shared runner owns execution and does not poll this.

get_progress_stats() → dict

Get progress statistics for this snapshot’s archiving process.

Returns dict with: - total: Total number of archive results - succeeded: Number of succeeded results - failed: Number of failed results - running: Number of currently running results - pending: Number of pending/queued results - percent: Completion percentage (0-100) - output_size: Total output size in bytes - is_sealed: Whether the snapshot is in a final state

retry_failed_archiveresults() → int

Reset failed/skipped ArchiveResults to queued for retry.

Returns count of ArchiveResults reset.

url_hash() → str

scheme() → str

path() → str

basename() → str

extension() → str

base_url() → str

is_static() → bool

is_archived() → bool

bookmarked_date() → str | None

downloaded_datestr() → str | None

archive_dates() → list[datetime.datetime]

oldest_archive_date() → datetime.datetime | None

newest_archive_date() → datetime.datetime | None

num_outputs() → int

num_failures() → int

latest_outputs(status: str | None = None) → dict[str, Any]

Get the latest output that each plugin produced

discover_outputs(include_filesystem_fallback: bool = True) → list[dict]

Discover output files from ArchiveResults and filesystem.

to_dict(extended: bool = False) → dict[str, Any]

Convert Snapshot to a dictionary (replacement for Link._asdict())

to_json_str(indent: int = 4) → str

Convert to JSON string (legacy method, use to_json() for dict)

to_csv(cols: list[str] | None = None, separator: str = ',', ljust: int = 0) → str

Convert to CSV string

write_json_details(out_dir: pathlib.Path | str | None = None) → None

Write JSON index file for this snapshot to its output directory

write_html_details(out_dir: pathlib.Path | str | None = None) → None

Write HTML detail page for this snapshot to its output directory

get_detail_page_auxiliary_items(outputs: list[dict] | None = None, hidden_card_plugins: set[str] | None = None) → tuple[list[dict[str, object]], list[dict[str, object]]]

static _ts_to_date_str(dt: datetime.datetime | None) → str | None

class archivebox.core.models.SnapshotMachine(obj, *args, **kwargs)

Bases: archivebox.workers.models.BaseStateMachine

State machine for managing Snapshot lifecycle.

Hook Lifecycle: ┌─────────────────────────────────────────────────────────────┐ │ QUEUED State │ │ • Waiting for snapshot to be ready │ └─────────────────────────────────────────────────────────────┘ ↓ tick() when can_start() ┌─────────────────────────────────────────────────────────────┐ │ STARTED State → enter_started() │ │ 1. snapshot.run() │ │ • discover_hooks(‘Snapshot’) → finds all plugin hooks │ │ • create_pending_archiveresults() → creates ONE │ │ ArchiveResult per hook (NO execution yet) │ │ 2. The shared abx-dl runner executes hooks and the │ │ projector updates ArchiveResult rows from events │ │ 3. Advance through steps 0-9 as foreground hooks complete │ └─────────────────────────────────────────────────────────────┘ ↓ tick() when is_finished() ┌─────────────────────────────────────────────────────────────┐ │ SEALED State → enter_sealed() │ │ • cleanup() → kills any background hooks still running │ │ • Set retry_at=None (no more processing) │ └─────────────────────────────────────────────────────────────┘

https://github.com/ArchiveBox/ArchiveBox/wiki/ArchiveBox-Architecture-Diagrams

Initialization

model_attr_name

‘snapshot’

queued

‘State(…)’

started

‘State(…)’

sealed

‘State(…)’

tick

None

seal

‘to(…)’

snapshot: archivebox.core.models.Snapshot

None

can_start() → bool

is_finished() → bool

Check if all ArchiveResults for this snapshot are finished.

enter_queued()

enter_started()

Just mark as started. The shared runner creates ArchiveResults and runs hooks.

enter_sealed()

class archivebox.core.models.ArchiveResult

Bases: archivebox.base_models.models.ModelWithOutputDir, archivebox.base_models.models.ModelWithConfig, archivebox.base_models.models.ModelWithNotes

class StatusChoices

Bases: django.db.models.TextChoices

QUEUED

(‘queued’, ‘Queued’)

STARTED

(‘started’, ‘Started’)

BACKOFF

(‘backoff’, ‘Waiting to retry’)

SUCCEEDED

(‘succeeded’, ‘Succeeded’)

FAILED

(‘failed’, ‘Failed’)

SKIPPED

(‘skipped’, ‘Skipped’)

NORESULTS

(‘noresults’, ‘No Results’)

INITIAL_STATE

None

ACTIVE_STATE

None

FINAL_STATES

()

FINAL_OR_ACTIVE_STATES

()

classmethod normalize_status(status: str | None) → str

classmethod get_plugin_choices()

Get plugin choices from discovered hooks (for forms/admin).

id

‘UUIDField(…)’

created_at

‘DateTimeField(…)’

modified_at

‘DateTimeField(…)’

snapshot: archivebox.core.models.Snapshot

‘ForeignKey(…)’

plugin

‘CharField(…)’

hook_name

‘CharField(…)’

process

‘OneToOneField(…)’

output_str

‘TextField(…)’

output_json

‘JSONField(…)’

output_files

‘JSONField(…)’

output_size

‘BigIntegerField(…)’

output_mimetypes

‘CharField(…)’

start_ts

‘DateTimeField(…)’

end_ts

‘DateTimeField(…)’

status

‘CharField(…)’

notes

‘TextField(…)’

snapshot_id: uuid.UUID

None

process_id: uuid.UUID | None

None

class Meta

Bases: archivebox.base_models.models.ModelWithOutputDir.Meta, archivebox.base_models.models.ModelWithConfig.Meta, archivebox.base_models.models.ModelWithNotes.Meta

app_label

‘core’

verbose_name

‘Archive Result’

verbose_name_plural

‘Archive Results Log’

indexes

None

__str__()

property created_by

Convenience property to access the user who created this archive result via its snapshot’s crawl.

to_json() → dict

Convert ArchiveResult model instance to a JSON-serializable dict.

static from_json(record: dict[str, Any], overrides: dict[str, Any] | None = None)

Create/update ArchiveResult from JSON dict.

Args: record: JSON dict with ‘snapshot_id’, ‘plugin’, etc. overrides: Optional dict of field overrides

Returns: ArchiveResult instance or None

save(*args, **kwargs)

snapshot_dir()

url()

property api_url: str

get_absolute_url()

reset_for_retry(*, save: bool = True) → None

property plugin_module: Any | None

static _normalize_output_files(raw_output_files: Any) → dict[str, dict[str, Any]]

static _coerce_output_file_size(value: Any) → int

output_file_map() → dict[str, dict[str, Any]]

output_file_paths() → list[str]

output_file_count() → int

output_size_from_files() → int

update_output_metadata_from_filesystem(snapshot_dir: pathlib.Path | None = None, save: bool = True) → bool

output_exists() → bool

static _looks_like_output_path(raw_output: str | None, plugin_name: str | None = None) → bool

_existing_output_path(raw_output: str | None) → str | None

static _fallback_output_file_path(output_file_paths: collections.abc.Sequence[str], plugin_name: str | None = None, output_file_map: dict[str, dict[str, Any]] | None = None) → str | None

static _find_best_output_file(dir_path: pathlib.Path, plugin_name: str | None = None) → pathlib.Path | None

embed_path_db() → str | None

embed_path() → str | None

Get the relative path to the embeddable output file for this result.

This is intentionally DB-backed only so snapshot/admin rendering stays fast and predictable without filesystem probes.

property output_dir_name: str

property output_dir_parent: str

property pwd: str

Working directory (from Process).

property cmd: list

Command array (from Process).

property cmd_version: str

Command version (from Process.binary).

property binary

Binary FK (from Process).

property iface

Network interface FK (from Process).

property machine

Machine FK (from Process).

property timeout: int

Timeout in seconds (from Process).

save_search_index()

update_from_output()

Update this ArchiveResult from filesystem logs and output files.

Used for Snapshot cleanup / orphan recovery when a hook’s output exists on disk but the projector did not finalize the row in the database.

Updates:

• status, output_str, output_json from ArchiveResult JSONL record

• output_files, output_size, output_mimetypes by walking filesystem

• end_ts, cmd, cmd_version, binary FK

• Processes side-effect records (Snapshot, Tag, etc.) via process_hook_records()

_set_binary_from_cmd(cmd: list) → None

Find Binary for command and set binary FK.

Tries matching by absolute path first, then by binary name. Only matches binaries on the current machine.

_url_passes_filters(url: str) → bool

Check if URL passes URL_ALLOWLIST and URL_DENYLIST config filters.

Uses proper config hierarchy: defaults -> file -> env -> machine -> user -> crawl -> snapshot

property output_dir: pathlib.Path

Get the output directory for this plugin’s results.