archivebox.core.models

Module Contents

Classes

UngroupedSubquery	Scalar subquery that should not be copied into the outer GROUP BY.
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
ArchiveResult

API

exception archivebox.core.models.SnapshotMigrationError

Bases: RuntimeError

Raised when a snapshot filesystem migration fails validation.

Initialization

Initialize self. See help(type(self)) for accurate signature.

class archivebox.core.models.UngroupedSubquery

Bases: django.db.models.Subquery

Scalar subquery that should not be copied into the outer GROUP BY.

get_group_by_cols()

class archivebox.core.models.Tag

Bases: archivebox.base_models.models.ModelWithUUID

id

‘AutoField(…)’

created_by

‘ForeignKey(…)’

created_at

‘DateTimeField(…)’

modified_at

‘DateTimeField(…)’

name

‘CharField(…)’

classmethod get_or_create_by_name(name: str, *, defaults: collections.abc.Mapping[str, Any] | None = None) → tuple[archivebox.core.models.Tag, bool]

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__()

save(*args, **kwargs)

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: ClassVar[list[tuple[str, str]]]

[(‘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.

bulk_create(objs, *args, **kwargs)

paged_iterator(chunk_size: int = 500)

Iterate snapshots using bounded keyset pages instead of one streaming cursor.

Django’s iterator(chunk_size=…) still keeps a single SQLite SELECT cursor open until the full queryset is exhausted. That is fine for read-only exports, but update/migration code does filesystem work and writes while iterating; a long-lived read cursor there can stretch lock waits across thousands of rows. This respects the queryset’s existing filters, order_by(), select_related(), and prefetch_related() state; if no ordering is defined, it falls back to primary-key order.

FILTER_TYPES: ClassVar[dict[str, Any]]

None

FILTER_TYPE_CHOICES

‘tuple(…)’

FILTER_ARG_KEYS

(‘after’, ‘before’, ‘filter_type’, ‘filter_patterns’, ‘status’, ‘url__icontains’, ‘url__istartswith’…

SPECIAL_FILTER_ARG_KEYS

‘frozenset(…)’

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

Filter snapshots by URL patterns using specified filter type

search(**kwargs) → archivebox.core.models.SnapshotQuerySet

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

BROWSER_EXTENSION_UPLOAD_HOOK_NAME

‘on_Snapshot__archivebox_browser_extension_upload’

id

‘CompactUUIDField(…)’

created_at

‘DateTimeField(…)’

modified_at

‘DateTimeField(…)’

url

‘TextField(…)’

timestamp

‘CharField(…)’

bookmarked_at

‘DateTimeField(…)’

crawl: archivebox.crawls.models.Crawl

‘ForeignKey(…)’

parent_snapshot

‘ForeignKey(…)’

title

‘CharField(…)’

downloaded_at

‘DateTimeField(…)’

depth

‘PositiveSmallIntegerField(…)’

fs_version

‘CharField(…)’

retry_at

‘RetryAtField(…)’

status

‘StatusField(…)’

config

‘JSONField(…)’

permissions

‘GeneratedField(…)’

output_size

‘BigIntegerField(…)’

notes

‘TextField(…)’

tags

‘ManyToManyField(…)’

state_field_name

‘status’

retry_at_field_name

‘retry_at’

StatusChoices

None

INITIAL_STATE

None

ACTIVE_STATE

None

FINAL_STATES

()

FINAL_OR_ACTIVE_STATES

()

active_state

None

delete_after_final_statuses

()

RUNNABLE_STATES

()

OPEN_STATES

()

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

add_tag_ids(tag_ids: collections.abc.Iterable[int | str]) → None

remove_tag_ids(tag_ids: collections.abc.Iterable[int | str]) → int

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.ModelWithNotes.Meta, archivebox.base_models.models.ModelWithHealthStats.Meta, archivebox.workers.models.ModelWithQueue.Meta

app_label

‘core’

verbose_name

‘Snapshot’

verbose_name_plural

‘Snapshots’

indexes: ClassVar[list[django.db.models.Index]]

None

constraints: ClassVar[list[django.db.models.BaseConstraint]]

None

__str__()

classmethod crawl_count_subquery(*, status: str | None = None, outer_ref: str = 'pk') → django.db.models.QuerySet

Return a scalar subquery counting Snapshots for one outer Crawl.

classmethod crawl_count_expr(*, status: str | None = None, outer_ref: str = 'pk')

classmethod crawl_total_and_status_counts(crawl_ids: collections.abc.Iterable[Any], *, status: str) → dict[str, dict[str, int]]

Return total and status-filtered Snapshot counts keyed by Crawl ID.

update_and_requeue(**kwargs) → bool

Update this Snapshot through the shared retry_at ownership path.

Any non-final Snapshot work means the parent Crawl must also be visible to the runner. Keep that invariant here so CLI/admin callers do not hand-edit the parent Crawl state every time they retry a hook.

queue_for_extraction(*, when=None) → bool

Queue this Snapshot for the runner using the normal state path.

schedule_plugin_run(plugins: collections.abc.Iterable[str], *, when=None) → bool

Persist one snapshot-scoped plugin request until abx-dl completes it.

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

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

restore_paused_scheduler_marker() → None

Restore the indefinite scheduler marker owned by the PAUSED lifecycle.

reconcile_parent_lifecycle(*, lock_seconds: int = 60) → bool | None

Follow parent Crawl pause/seal state before any Snapshot work runs.

Crawl.pause()/cancel() only wake child rows. The runner claims each due Snapshot and lets this method perform the actual child transition, so cancellation stays fast and Snapshot cleanup still runs from the normal lifecycle owner.

finalize_completed_upload_results() → int

start_processing() → bool

Atomically move a claimed queued Snapshot into its active lease.

seal() → bool

Atomically finalize this Snapshot and reconcile its output metadata.

advance_lifecycle() → bool

Advance one explicit lifecycle step after the runner claims this row.

cancel() → None

get_delete_after_config_value()

classmethod missing_delete_at_candidates()

classmethod is_archivebox_internal_url(url: str, *, config: collections.abc.Mapping[str, Any] | Any | None = None) → bool

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.

ensure_permissions_config(crawl_permissions: str | None = None) → bool

validate_url_for_archiving(*, config: collections.abc.Mapping[str, Any] | Any | None = None) → None

save(*args, **kwargs)

static _fs_current_version() → str

Get current ArchiveBox filesystem layout version.

_FS_VERSION_MIGRATION_PATHS: ClassVar[dict[str, str]]

None

property fs_migration_needed: bool

Check if snapshot needs filesystem migration

_fs_next_version(version: str) → str

Get the next declared version in the filesystem migration chain.

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 safely remove the old tree.

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. Remove the verified legacy source.

5. Persist fs_version last, so an interruption remains selected by the indexed stale-version query and resumes naturally.

Re-running this method also reconciles a legacy timestamp directory left behind when a machine stopped after the database commit but before the on-commit cleanup callback ran.

_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_from_0_9_0_to_0_9_4(source_dir: pathlib.Path | None = None, config: ArchiveBoxBaseConfig | None = None)

hydrate_archiveresult_output_metadata(snapshot_dir: pathlib.Path | None = None) → int

Populate missing ArchiveResult file metadata from existing outputs.

_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}/

static _migration_trees_match(old_dir: pathlib.Path, new_dir: pathlib.Path) → bool

Verify every legacy entry exists unchanged at the destination.

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

Delete the old directory after its contents are verified at the new path.

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) → 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 one row per hook; retries update the existing row.

_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 and creates index.jsonl while preserving the original JSON byte-for-byte for unknown legacy metadata. 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() → str | None

icons(path: str | None = None, prefix: str = '/', quote_paths: bool = False) → 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_crawl_symlink(*, crawl_dir: pathlib.Path | None = None, snapshot_dir: pathlib.Path | None = None) → None

Ensure snapshot is symlinked under its crawl output directory.

remove_legacy_archive_symlink() → None

Remove a stale archive/ compatibility projection.

reconcile_filesystem_links() → None

Repair filesystem projections after a save or migration.

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] = (), *, created_by: Any = None) → None

finalize_output_metadata() → None

Clean up background ArchiveResult hooks and empty results.

Called after entering the sealed state. Reconcile late background outputs and hydrate result metadata.

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

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

Queue the parent Snapshot to rerun plugins with failed facts.

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, archive_results: list[archivebox.core.models.Snapshot.discover_outputs.ArchiveResult] | None = None) → list[dict]

Discover output files from ArchiveResults and filesystem.

property static_archive_path: str

Snapshot output path relative to the data root, for portable exports.

to_dict(extended: bool = False, static_export: 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

get_html_details_context(request=None, *, static_export_dir: pathlib.Path | None = None) → dict[str, Any]

Build the one context used by both served and on-disk snapshot pages.

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

Write the unified snapshot detail page with portable filesystem URLs.

get_detail_page_auxiliary_items(outputs: list[dict] | None = None, hidden_card_plugins: set[str] | None = None, archive_results: list[archivebox.core.models.Snapshot.get_detail_page_auxiliary_items.ArchiveResult] | 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.ArchiveResult

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

class StatusChoices

Bases: django.db.models.TextChoices

QUEUED

(‘queued’, ‘Queued’)

STARTED

(‘started’, ‘Started’)

PAUSED

(‘paused’, ‘Paused’)

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

()

delete_after_final_statuses

None

classmethod normalize_status(status: str | None) → str

classmethod get_or_create_by_hook(snapshot: archivebox.core.models.Snapshot, plugin: str, hook_name: str, *, defaults: collections.abc.Mapping[str, Any] | None = None) → tuple[archivebox.core.models.ArchiveResult, bool]

static output_files_upload_complete(output_files: dict[str, dict[str, Any]]) → bool

classmethod get_plugin_choices()

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

classmethod snapshot_count_subquery(*, status: str | None = None, outer_ref: str = 'pk') → django.db.models.QuerySet

Return a scalar subquery counting ArchiveResults for one outer Snapshot.

Use this instead of filtered join aggregates for per-row Snapshot counts: the scalar form lets SQLite probe the covering (snapshot_id, status) or (status, snapshot_id) indexes once per visible Snapshot row, instead of joining and grouping the whole candidate Snapshot queryset.

classmethod snapshot_half_count_subquery(*, outer_ref: str = 'snapshot_id') → django.db.models.QuerySet

classmethod snapshot_count_expr(*, status: str | None = None, outer_ref: str = 'pk')

classmethod status_counts(queryset: django.db.models.QuerySet | None = None, statuses: collections.abc.Iterable[str] | None = None) → dict[str, int]

Count requested statuses with separate indexed COUNT probes.

classmethod snapshot_ids_with_majority_status(status: str | collections.abc.Iterable[str]) → django.db.models.QuerySet

Return Snapshot IDs where more than half of ArchiveResults have status.

Start from ArchiveResult.status for every majority-status filter. The (status, snapshot_id) index keeps the plan predictable even when a user’s collection has an unusual status distribution.

id

‘CompactUUIDField(…)’

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

retry_at

‘DateTimeField(…)’

notes

‘TextField(…)’

snapshot_id: uuid.UUID

None

process_id: uuid.UUID | None

None

class Meta

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

app_label

‘core’

verbose_name

‘Archive Result’

verbose_name_plural

‘Archive Results’

indexes: ClassVar[list[django.db.models.Index]]

None

constraints: ClassVar[list[django.db.models.BaseConstraint]]

None

__str__()

static _format_output_line_for_display(line: str) → str

output_str_for_display() → str

get_delete_after_config_value()

classmethod missing_delete_at_candidates()

property created_by

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

to_json(*, snapshot_output_dir: pathlib.Path | None = None) → 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)

safe_update(update_fields: collections.abc.Mapping[str, Any], *, refresh: bool = True) → bool

Compare-and-swap one loaded ArchiveResult without opening a transaction.

schedule_delete_cleanup(*, using: str | None = None) → None

Remove shared plugin output and refresh persisted Snapshot metadata after commit.

static refresh_snapshot_output_sizes(snapshot_ids)

snapshot_dir()

url()

property api_url: str

get_absolute_url()

property is_paused: bool

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]

update_output_metadata_from_filesystem(snapshot_dir: pathlib.Path | None = None, save: bool = True) → 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(output_file_map: dict[str, dict[str, Any]] | None = None) → 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 process_record

property pwd: str

Working directory, derived from the snapshot/plugin path if the Process row is gone.

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).

_url_passes_filters(url: str) → bool

Check if URL passes URL_ALLOWLIST and URL_DENYLIST config filters.

Uses the centralized config resolver so frozen crawl/snapshot values and live Machine/Persona execution values apply in their scoped order.

property output_dir: pathlib.Path

Get the output directory for this plugin’s results.