__package__ = "archivebox.core"
import json
import os
import uuid
from collections.abc import Iterable, Mapping, Sequence
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, Optional
from urllib.parse import urlparse
from django.conf import settings
from django.contrib import admin
from django.core.exceptions import FieldDoesNotExist, ObjectDoesNotExist, ValidationError
from django.db import IntegrityError, models, transaction
from django.db.models import F, Q, QuerySet, Sum, Value
from django.db.models.fields.json import KT
from django.db.models.functions import Coalesce
from django.urls import reverse_lazy
from django.utils import timezone
from django.utils.functional import cached_property
from django.utils.safestring import mark_safe
from django.utils.text import slugify
from archivebox.base_models.models import (
ModelWithConfig,
ModelWithDeleteAfter,
ModelWithHealthStats,
ModelWithNotes,
ModelWithOutputDir,
ModelWithUUID,
get_or_create_system_user_pk,
)
from archivebox.config import CONSTANTS
from archivebox.config.common import get_config, rprint
from archivebox.crawls.models import Crawl
from archivebox.machine.models import Binary
from archivebox.misc.system import atomic_write
from archivebox.misc.util import (
domain as url_domain,
)
from archivebox.misc.util import (
htmldecode,
parse_date,
sanitize_html_text,
to_json,
ts_to_date_str,
validate_url,
)
from archivebox.plugins.discovery import (
get_plugin_icon,
get_plugin_name,
get_plugins,
)
from archivebox.uuid_compat import CompactUUIDField, uuid7
from archivebox.workers.models import ACTIVE_STATE_LEASE_SECONDS, RETRY_AT_MAX, ModelWithQueue
if TYPE_CHECKING:
from archivebox.config.common import ArchiveBoxBaseConfig
class SnapshotMigrationError(RuntimeError):
"""Raised when a snapshot filesystem migration fails validation."""
class UngroupedSubquery(models.Subquery):
"""Scalar subquery that should not be copied into the outer GROUP BY."""
def get_group_by_cols(self):
return []
class Tag(ModelWithUUID):
# Keep AutoField for compatibility with main branch migrations
# Don't use UUIDField here - requires complex FK transformation
id = models.AutoField(primary_key=True, serialize=False, verbose_name="ID")
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
default=get_or_create_system_user_pk,
null=True,
related_name="tag_set",
)
created_at = models.DateTimeField(default=timezone.now, db_index=True, null=True)
modified_at = models.DateTimeField(auto_now=True)
name = models.CharField(unique=True, blank=False, max_length=100)
@classmethod
def get_or_create_by_name(cls, name: str, *, defaults: Mapping[str, Any] | None = None) -> tuple["Tag", bool]:
tag = cls.objects.filter(name__iexact=name).first()
if tag:
return tag, False
try:
return cls.objects.create(name=name, **(defaults or {})), True
except IntegrityError:
tag = cls.objects.filter(name__iexact=name).first()
if tag is None:
raise
return tag, False
snapshot_set: models.Manager["Snapshot"]
class Meta(ModelWithUUID.Meta):
app_label = "core"
verbose_name = "Tag"
verbose_name_plural = "Tags"
def __str__(self):
return self.name
def save(self, *args, **kwargs):
update_fields = kwargs.get("update_fields")
if update_fields is None or "name" in update_fields:
self.name = sanitize_html_text(self.name).strip()
super().save(*args, **kwargs)
@property
def slug(self) -> str:
"""ASCII-safe slugified form of the tag name (derived, not stored)."""
return slugify(self.name or "") or "tag"
@property
def api_url(self) -> str:
return str(reverse_lazy("api-1:get_tag", args=[self.id]))
def to_json(self) -> dict:
"""
Convert Tag model instance to a JSON-serializable dict.
"""
from archivebox.config import VERSION
return {
"type": "Tag",
"schema_version": VERSION,
"id": str(self.id),
"name": self.name,
}
@staticmethod
def 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
"""
name = record.get("name")
if not name:
return None
tag, _ = Tag.get_or_create_by_name(name)
# Auto-attach to snapshot if in overrides
if overrides and "snapshot" in overrides and tag:
overrides["snapshot"].add_tag_ids([tag.pk])
return tag
class SnapshotTag(models.Model):
id = models.AutoField(primary_key=True)
snapshot = models.ForeignKey("Snapshot", db_column="snapshot_id", on_delete=models.CASCADE, to_field="id")
tag = models.ForeignKey(Tag, db_column="tag_id", on_delete=models.CASCADE, to_field="id")
class Meta:
app_label = "core"
db_table = "core_snapshot_tags"
unique_together: ClassVar[list[tuple[str, str]]] = [("snapshot", "tag")]
class SnapshotQuerySet(models.QuerySet):
"""Custom QuerySet for Snapshot model with export methods that persist through .filter() etc."""
def bulk_create(self, objs, *args, **kwargs):
objs = list(objs)
missing_crawl_ids = set()
from archivebox.core.permissions import PERMISSIONS_VALUES
for obj in objs:
if isinstance(obj, self.model):
config = dict(obj.config or {})
permission = str(config.get("PERMISSIONS") or "").strip().lower()
if permission not in PERMISSIONS_VALUES and obj.crawl_id:
crawl = getattr(obj, "crawl", None)
if not getattr(crawl, "permissions", None):
missing_crawl_ids.add(str(obj.crawl_id))
crawl_permissions_by_id = {}
if missing_crawl_ids:
crawl_permissions_by_id = {
str(crawl_id): permissions
for crawl_id, permissions in Crawl.objects.filter(pk__in=missing_crawl_ids).values_list("pk", "permissions")
}
from archivebox.misc.db import truncate_overlong_charfields
for obj in objs:
if isinstance(obj, self.model):
obj.ensure_permissions_config(crawl_permissions=crawl_permissions_by_id.get(str(obj.crawl_id)))
# bulk_create bypasses pre_save, so clamp CharFields here too
# (e.g. page titles) to stay within postgres VARCHAR limits.
truncate_overlong_charfields(obj)
return super().bulk_create(objs, *args, **kwargs)
def paged_iterator(self, 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.
"""
pk_field = self.model._meta.pk.name
raw_ordering = tuple(self.query.order_by or self.model._meta.ordering or (pk_field,))
if any(not isinstance(term, str) or term == "?" for term in raw_ordering):
offset = 0
while True:
batch = list(self[offset : offset + chunk_size])
if not batch:
break
yield from batch
offset += chunk_size
return
ordering = []
for term in raw_ordering:
descending = term.startswith("-")
field_name = term[1:] if descending else term
if field_name == "pk":
field_name = pk_field
ordering.append(f"-{field_name}" if descending else field_name)
ordered_field_names = [term.removeprefix("-") for term in ordering]
try:
if any(self.model._meta.get_field(field_name).null for field_name in ordered_field_names):
offset = 0
while True:
batch = list(self[offset : offset + chunk_size])
if not batch:
break
yield from batch
offset += chunk_size
return
except (AttributeError, FieldDoesNotExist):
offset = 0
while True:
batch = list(self[offset : offset + chunk_size])
if not batch:
break
yield from batch
offset += chunk_size
return
unique_field_names = {pk_field, *(field.name for field in self.model._meta.fields if field.unique)}
if not any(field_name in unique_field_names for field_name in ordered_field_names):
offset = 0
while True:
batch = list(self[offset : offset + chunk_size])
if not batch:
break
yield from batch
offset += chunk_size
return
last_values = None
value_field_names = tuple(dict.fromkeys([*ordered_field_names, pk_field]))
while True:
batch_qs = self.order_by(*ordering)
if last_values is not None:
page_filter = models.Q()
for idx, term in enumerate(ordering):
descending = term.startswith("-")
field_name = term[1:] if descending else term
prefix = {ordered_field_names[i]: last_values[i] for i in range(idx)}
comparison = "lt" if descending else "gt"
page_filter |= models.Q(**prefix, **{f"{field_name}__{comparison}": last_values[idx]})
batch_qs = batch_qs.filter(page_filter)
batch_rows = list(batch_qs.values_list(*value_field_names)[:chunk_size])
if not batch_rows:
break
pk_idx = value_field_names.index(pk_field)
snapshot_ids = [row[pk_idx] for row in batch_rows]
snapshots_by_id = {snapshot.pk: snapshot for snapshot in self.filter(pk__in=snapshot_ids).order_by()}
for row in batch_rows:
snapshot_id = row[pk_idx]
snapshot = snapshots_by_id.get(snapshot_id)
if snapshot is not None:
yield snapshot
last_values = batch_rows[-1][: len(ordered_field_names)]
# =========================================================================
# Filtering Methods
# =========================================================================
FILTER_TYPES: ClassVar[dict[str, Any]] = {
"exact": lambda pattern: models.Q(url=pattern),
"substring": lambda pattern: models.Q(url__icontains=pattern),
"regex": lambda pattern: models.Q(url__iregex=pattern),
"domain": lambda pattern: (
models.Q(url__istartswith=f"http://{pattern}")
| models.Q(url__istartswith=f"https://{pattern}")
| models.Q(url__istartswith=f"ftp://{pattern}")
),
"tag": lambda pattern: models.Q(tags__name=pattern),
"timestamp": lambda pattern: models.Q(timestamp=pattern),
}
FILTER_TYPE_CHOICES = tuple(FILTER_TYPES)
FILTER_ARG_KEYS = (
"after",
"before",
"filter_type",
"filter_patterns",
"status",
"url__icontains",
"url__istartswith",
"tag",
"crawl_id",
"limit",
"sort",
"search",
)
SPECIAL_FILTER_ARG_KEYS = frozenset({"filter_patterns", "filter_type", "query", "search", "tag", "before", "after", "limit", "sort"})
def filter_by_patterns(self, patterns: list[str], filter_type: str = "exact") -> "SnapshotQuerySet":
"""Filter snapshots by URL patterns using specified filter type"""
from archivebox.misc.logging import stderr
q_filter = models.Q()
for pattern in patterns:
try:
q_filter = q_filter | self.FILTER_TYPES[filter_type](pattern)
except KeyError:
stderr()
stderr(f"[X] Got invalid pattern for --filter-type={filter_type}:", color="red")
stderr(f" {pattern}")
raise SystemExit(2)
return self.filter(q_filter)
def search(self, **kwargs) -> "SnapshotQuerySet":
from archivebox.core.snapshot_status import filter_snapshots_by_status
from archivebox.search.query import apply_snapshot_search
queryset = self
filter_patterns = tuple(str(pattern) for pattern in kwargs.get("filter_patterns") or ())
filter_type = kwargs.get("filter_type") or "substring"
query = kwargs.get("query")
if isinstance(query, (list, tuple)):
query = " ".join(str(part) for part in query)
query = (query or (" ".join(filter_patterns) if kwargs.get("search") else "")).strip()
field_names = {field.name for field in self.model._meta.get_fields()}
field_names.update(field.attname for field in self.model._meta.fields)
field_filters = {
key: value
for key, value in kwargs.items()
if value is not None and key not in self.SPECIAL_FILTER_ARG_KEYS and key.split("__", 1)[0] in field_names
}
status = field_filters.pop("status", None)
queryset = filter_snapshots_by_status(queryset, status)
if field_filters:
queryset = queryset.filter(**field_filters)
if kwargs.get("tag"):
queryset = queryset.filter(tags__name__iexact=kwargs["tag"])
if kwargs.get("before") is not None:
queryset = queryset.filter(bookmarked_at__lt=datetime.fromtimestamp(float(kwargs["before"]), tz=UTC))
if kwargs.get("after") is not None:
queryset = queryset.filter(bookmarked_at__gt=datetime.fromtimestamp(float(kwargs["after"]), tz=UTC))
if query:
queryset = apply_snapshot_search(
queryset,
query,
search_mode=kwargs.get("search"),
ordering=("-created_at",) if not kwargs.get("sort") else None,
max_results=kwargs.get("limit"),
skip_backend_when_metadata_satisfies_limit=True,
include_metadata_for_forced_backend=True,
)
elif filter_patterns:
queryset = queryset.filter_by_patterns(list(filter_patterns), filter_type)
if kwargs.get("sort"):
queryset = queryset.order_by(kwargs["sort"])
elif not queryset.query.order_by:
queryset = queryset.order_by("-created_at")
limit = kwargs.get("limit")
if limit is not None and limit > 0:
queryset = queryset[:limit]
return queryset
# =========================================================================
# Export Methods
# =========================================================================
def to_json(self, with_headers: bool = False) -> str:
"""Generate JSON index from snapshots"""
import sys
from datetime import datetime
from archivebox.config import VERSION
config = get_config()
MAIN_INDEX_HEADER = (
{
"info": "This is an index of site data archived by ArchiveBox: The self-hosted web archive.",
"schema": "archivebox.index.json",
"copyright_info": config.FOOTER_INFO,
"meta": {
"project": "ArchiveBox",
"version": VERSION,
"git_sha": VERSION,
"website": "https://ArchiveBox.io",
"docs": "https://github.com/ArchiveBox/ArchiveBox/wiki",
"source": "https://github.com/ArchiveBox/ArchiveBox",
"issues": "https://github.com/ArchiveBox/ArchiveBox/issues",
"dependencies": {},
},
}
if with_headers
else {}
)
snapshot_dicts = [s.to_dict(extended=True, static_export=True) for s in self.iterator(chunk_size=500)]
if with_headers:
output = {
**MAIN_INDEX_HEADER,
"num_links": len(snapshot_dicts),
"updated": datetime.now(UTC),
"last_run_cmd": sys.argv,
"links": snapshot_dicts,
}
else:
output = snapshot_dicts
return to_json(output, indent=4, sort_keys=True)
def to_csv(self, cols: list[str] | None = None, header: bool = True, separator: str = ",", ljust: int = 0) -> str:
"""Generate CSV output from snapshots"""
cols = cols or ["timestamp", "is_archived", "url"]
header_str = separator.join(col.ljust(ljust) for col in cols) if header else ""
row_strs = (s.to_csv(cols=cols, ljust=ljust, separator=separator) for s in self.iterator(chunk_size=500))
return "\n".join((header_str, *row_strs))
def to_html(self, with_headers: bool = True) -> str:
"""Generate main index HTML from snapshots"""
from datetime import datetime
from django.template.loader import render_to_string
from archivebox.config import VERSION
from archivebox.config.version import get_COMMIT_HASH
config = get_config()
template = "static_index.html" if with_headers else "minimal_index.html"
snapshot_list = list(self.iterator(chunk_size=500))
manifest_records = []
for snapshot in snapshot_list:
outputs = snapshot.discover_outputs(include_filesystem_fallback=True)
output_paths = [str(output.get("path") or "") for output in outputs]
snapshot._public_preview_paths = [
path for preferred in ("screenshot/screenshot.png", "screenshot.png") for path in output_paths if path == preferred
]
snapshot._public_favicon_paths = [path for path in output_paths if path in ("favicon/favicon.ico", "favicon.ico")]
snapshot.write_html_details()
if with_headers:
# Use the same portable schema as the JSON export. Rendering
# above has already populated result-count caches, archive_size
# reuses the sealed output_size field, and tags are prefetched.
manifest_records.append(snapshot.to_dict(extended=True, static_export=True))
if with_headers:
manifest = "".join(f"{to_json(record, indent=None, sort_keys=True)}\n" for record in manifest_records)
atomic_write(str(CONSTANTS.DATA_DIR / CONSTANTS.JSONL_INDEX_FILENAME), manifest)
return render_to_string(
template,
{
"version": VERSION,
"git_sha": get_COMMIT_HASH() or VERSION,
"num_links": str(len(snapshot_list)),
"date_updated": datetime.now(UTC).strftime("%Y-%m-%d"),
"time_updated": datetime.now(UTC).strftime("%Y-%m-%d %H:%M"),
"links": snapshot_list,
"FOOTER_INFO": config.FOOTER_INFO,
"STATIC_EXPORT": True,
"STATIC_EXPORT_DIR": CONSTANTS.DATA_DIR,
},
)
class SnapshotManager(models.Manager.from_queryset(SnapshotQuerySet)): # ty: ignore[unsupported-base]
"""Manager for Snapshot model - uses SnapshotQuerySet for chainable methods"""
def filter(self, *args, **kwargs):
domain = kwargs.pop("domain", None)
qs = super().filter(*args, **kwargs)
if domain:
qs = qs.filter(url__icontains=f"://{domain}")
return qs
def get_queryset(self):
# Don't prefetch by default - it causes "too many open files" during bulk operations
# Views/templates can add .prefetch_related('tags', 'archiveresult_set') where needed
return super().get_queryset()
# =========================================================================
# Import Methods
# =========================================================================
def remove(self, atomic: bool = False) -> tuple:
"""Remove snapshots from the database"""
from django.db import transaction
if atomic:
with transaction.atomic():
return self.get_queryset().delete()
return self.get_queryset().delete()
class Snapshot(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHealthStats, ModelWithQueue):
BROWSER_EXTENSION_UPLOAD_HOOK_NAME = "on_Snapshot__archivebox_browser_extension_upload"
id = CompactUUIDField(primary_key=True, default=uuid7, editable=False, unique=True)
created_at = models.DateTimeField(default=timezone.now, db_index=True)
modified_at = models.DateTimeField(auto_now=True)
# Stored as a variable-length TextField so short URLs don't reserve space and very long
# URLs (up to MAX_URL_LENGTH chars, enforced in save()) are supported, while keeping a
# normal index so exact, prefix, and substring lookups all stay fast.
url = models.TextField(db_index=True) # URLs can appear in multiple crawls
timestamp = models.CharField(max_length=32, unique=True, db_index=True, editable=False)
bookmarked_at = models.DateTimeField(default=timezone.now, db_index=True)
crawl: Crawl = models.ForeignKey(Crawl, on_delete=models.CASCADE, null=False, related_name="snapshot_set", db_index=True) # type: ignore[assignment]
parent_snapshot = models.ForeignKey(
"self",
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name="child_snapshots",
db_index=True,
help_text="Parent snapshot that discovered this URL (for recursive crawling)",
)
title = models.CharField(max_length=512, null=True, blank=True, db_index=True)
downloaded_at = models.DateTimeField(default=None, null=True, editable=False, db_index=True, blank=True)
depth = models.PositiveSmallIntegerField(default=0, db_index=True) # 0 for root snapshot, 1+ for discovered URLs
fs_version = models.CharField(
max_length=10,
default="0.9.0",
db_index=True,
help_text='Filesystem version of this snapshot (e.g., "0.7.0", "0.8.0", "0.9.4").',
)
retry_at = ModelWithQueue.RetryAtField(default=timezone.now)
status = ModelWithQueue.StatusField(
choices=ModelWithQueue.StatusChoices,
default=ModelWithQueue.StatusChoices.QUEUED,
)
config = models.JSONField(default=dict, null=False, blank=False, editable=True)
permissions = models.GeneratedField(
expression=Coalesce(KT("config__PERMISSIONS"), Value("private"), output_field=models.CharField(max_length=16)),
output_field=models.CharField(max_length=16, null=False),
db_persist=True,
db_index=True,
editable=False,
)
output_size = models.BigIntegerField(
default=0,
db_index=True,
editable=False,
help_text="Total bytes of all ArchiveResult output files",
)
notes = models.TextField(blank=True, null=False, default="")
# output_dir is computed via @cached_property from fs_version and get_storage_path_for_version()
tags = models.ManyToManyField(Tag, blank=True, through=SnapshotTag, related_name="snapshot_set", through_fields=("snapshot", "tag"))
state_field_name = "status"
retry_at_field_name = "retry_at"
StatusChoices = ModelWithQueue.StatusChoices
INITIAL_STATE = StatusChoices.QUEUED
ACTIVE_STATE = StatusChoices.STARTED
FINAL_STATES = (StatusChoices.SEALED,)
FINAL_OR_ACTIVE_STATES = (*FINAL_STATES, ACTIVE_STATE)
active_state = StatusChoices.STARTED
delete_after_final_statuses = (StatusChoices.SEALED,)
RUNNABLE_STATES = (StatusChoices.QUEUED, StatusChoices.STARTED)
OPEN_STATES = (*RUNNABLE_STATES, StatusChoices.PAUSED)
crawl_id: uuid.UUID
parent_snapshot_id: uuid.UUID | None
_prefetched_objects_cache: dict[str, Any]
objects = SnapshotManager()
archiveresult_set: models.Manager["ArchiveResult"]
def add_tag_ids(self, tag_ids: Iterable[int | str]) -> None:
tag_ids = [tag_id for tag_id in dict.fromkeys(tag_ids) if tag_id]
for tag_id in tag_ids:
try:
SnapshotTag(snapshot_id=self.pk, tag_id=tag_id).save(force_insert=True)
except IntegrityError:
# Only the unique (snapshot, tag) conflict is idempotent. Do
# not hide foreign-key or other integrity failures.
if SnapshotTag.objects.filter(snapshot_id=self.pk, tag_id=tag_id).exists():
continue
raise
def remove_tag_ids(self, tag_ids: Iterable[int | str]) -> int:
tag_ids = [tag_id for tag_id in dict.fromkeys(tag_ids) if tag_id]
if not tag_ids:
return 0
# QuerySet.delete() wraps even a fast through-table DELETE in atomic().
# SnapshotTag has no delete hooks or child rows, so issue the same
# idempotent DELETE as one autocommit statement.
return SnapshotTag.objects.filter(snapshot_id=self.pk, tag_id__in=tag_ids)._raw_delete(SnapshotTag.objects.db)
class Meta(
ModelWithDeleteAfter.Meta,
ModelWithOutputDir.Meta,
ModelWithConfig.Meta,
ModelWithNotes.Meta,
ModelWithHealthStats.Meta,
ModelWithQueue.Meta,
):
app_label = "core"
verbose_name = "Snapshot"
verbose_name_plural = "Snapshots"
indexes: ClassVar[list[models.Index]] = [
models.Index(fields=["-bookmarked_at", "-created_at"], name="snapshot_public_order_idx"),
models.Index(fields=["crawl", "status", "modified_at"], name="snapshot_progress_idx"),
]
constraints: ClassVar[list[models.BaseConstraint]] = [
# Allow same URL in different crawls, but not duplicates within same crawl
models.UniqueConstraint(fields=["url", "crawl"], name="unique_url_per_crawl"),
# Global timestamp uniqueness for 1:1 symlink mapping
models.UniqueConstraint(fields=["timestamp"], name="unique_timestamp"),
]
def __str__(self):
return f"[{self.id}] {self.url[:64]}"
@classmethod
def crawl_count_subquery(cls, *, status: str | None = None, outer_ref: str = "pk") -> QuerySet:
"""Return a scalar subquery counting Snapshots for one outer Crawl."""
qs = cls.objects.filter(crawl_id=models.OuterRef(outer_ref))
if status is not None:
qs = qs.filter(status=status)
return qs.order_by().values("crawl_id").annotate(count=models.Count("pk")).values("count")
@classmethod
def crawl_count_expr(cls, *, status: str | None = None, outer_ref: str = "pk"):
# Use scalar subqueries for sortable Crawl admin counters: SQLite can
# probe the (crawl_id, status, modified_at) index per Crawl row instead
# of joining/grouping all visible Snapshot rows.
return Coalesce(
models.Subquery(cls.crawl_count_subquery(status=status, outer_ref=outer_ref), output_field=models.IntegerField()),
models.Value(0),
)
@classmethod
def crawl_total_and_status_counts(cls, crawl_ids: Iterable[Any], *, status: str) -> dict[str, dict[str, int]]:
"""Return total and status-filtered Snapshot counts keyed by Crawl ID."""
crawl_ids = list(crawl_ids)
if not crawl_ids:
return {}
return {
str(row["crawl_id"]): {
"total": row["total"],
"status": row["status_count"],
}
for row in cls.objects.filter(crawl_id__in=crawl_ids)
.values("crawl_id")
.annotate(
total=models.Count("pk"),
status_count=models.Count("pk", filter=Q(status=status)),
)
}
def update_and_requeue(self, **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.
"""
updated = super().update_and_requeue(**kwargs)
if not updated:
return False
next_status = kwargs.get("status", self.status)
if next_status not in (self.StatusChoices.QUEUED, self.StatusChoices.STARTED) or not self.crawl_id:
return True
crawl = self.crawl
crawl_status = crawl.StatusChoices.STARTED if crawl.status == crawl.StatusChoices.STARTED else crawl.StatusChoices.QUEUED
crawl.update_and_requeue(
status=crawl_status,
retry_at=kwargs.get("retry_at") or timezone.now(),
)
return True
def queue_for_extraction(self, *, when=None) -> bool:
"""Queue this Snapshot for the runner using the normal state path."""
return self.update_and_requeue(
status=self.StatusChoices.QUEUED,
retry_at=when or timezone.now(),
)
def schedule_plugin_run(self, plugins: Iterable[str], *, when=None) -> bool:
"""Persist one snapshot-scoped plugin request until abx-dl completes it."""
plugin_names = sorted({name.strip() for name in plugins if name.strip()})
if not plugin_names:
return False
retry_at = when or timezone.now()
for _attempt in range(8):
current = type(self).objects.select_related("crawl").get(pk=self.pk)
pending_plugins = {str(name).strip() for name in (current.config or {}).get("RETRY_PLUGINS", []) if str(name).strip()}
config = {**(current.config or {}), "RETRY_PLUGINS": sorted(pending_plugins | set(plugin_names))}
status = current.status if current.status == self.StatusChoices.SEALED else self.StatusChoices.QUEUED
updated = (
type(self)
.objects.filter(
pk=self.pk,
config=current.config,
status=current.status,
retry_at=current.retry_at,
)
.update(
config=config,
status=status,
retry_at=retry_at,
modified_at=timezone.now(),
)
)
if updated:
crawl = current.crawl
break
else:
raise RuntimeError(f"Snapshot {self.pk} changed repeatedly while scheduling plugins")
self.refresh_from_db()
if status in self.RUNNABLE_STATES and self.crawl_id:
crawl_status = crawl.StatusChoices.STARTED if crawl.status == crawl.StatusChoices.STARTED else crawl.StatusChoices.QUEUED
crawl.update_and_requeue(status=crawl_status, retry_at=retry_at)
return True
def pause(self, *, save: bool = True) -> bool:
return super().pause(save=save)
def resume(self, *, when: datetime | None = None, save: bool = True) -> bool:
return super().resume(when=when, save=save)
def restore_paused_scheduler_marker(self) -> None:
"""
Restore the indefinite scheduler marker owned by the PAUSED lifecycle.
"""
type(self).objects.filter(
pk=self.pk,
status=self.StatusChoices.PAUSED,
).update(
retry_at=RETRY_AT_MAX,
modified_at=timezone.now(),
)
self.status = self.StatusChoices.PAUSED
self.retry_at = RETRY_AT_MAX
def reconcile_parent_lifecycle(self, *, 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.
"""
parent_status = Crawl.objects.filter(id=self.crawl_id).values_list("status", flat=True).first()
if parent_status == Crawl.StatusChoices.SEALED and self.status != self.StatusChoices.SEALED:
if not self.claim_processing_lock(lock_seconds=lock_seconds):
return False
self.refresh_from_db()
parent_status = Crawl.objects.filter(id=self.crawl_id).values_list("status", flat=True).first()
if parent_status == Crawl.StatusChoices.SEALED and self.status != self.StatusChoices.SEALED:
self.seal()
return True
if parent_status == Crawl.StatusChoices.PAUSED and self.status not in (self.StatusChoices.PAUSED, self.StatusChoices.SEALED):
if not self.claim_processing_lock(lock_seconds=lock_seconds):
return False
self.refresh_from_db()
parent_status = Crawl.objects.filter(id=self.crawl_id).values_list("status", flat=True).first()
if parent_status == Crawl.StatusChoices.PAUSED and self.status not in (
self.StatusChoices.PAUSED,
self.StatusChoices.SEALED,
):
self.pause()
return True
return None
def finalize_completed_upload_results(self) -> int:
now = timezone.now()
result_ids = []
upload_results = (
self.archiveresult_set.filter(
status=ArchiveResult.StatusChoices.QUEUED,
hook_name=self.BROWSER_EXTENSION_UPLOAD_HOOK_NAME,
output_size__gt=0,
)
.exclude(output_files={})
.only("id", "output_files")
)
for result in upload_results:
if ArchiveResult.output_files_upload_complete(result.output_files or {}):
result_ids.append(result.id)
if not result_ids:
return 0
# Browser-extension uploads are already-finished external writes. If the
# PATCH request saved files but omitted status, finalize only this
# Snapshot's complete uploads without scanning ArchiveResult globally.
return ArchiveResult.objects.filter(id__in=result_ids, status=ArchiveResult.StatusChoices.QUEUED).update(
status=ArchiveResult.StatusChoices.SUCCEEDED,
modified_at=now,
)
def start_processing(self) -> bool:
"""Atomically move a claimed queued Snapshot into its active lease."""
owned_retry_at = self.retry_at
now = timezone.now()
lease_until = now + timedelta(seconds=ACTIVE_STATE_LEASE_SECONDS)
updated = (
type(self)
.objects.filter(
pk=self.pk,
retry_at=owned_retry_at,
status=self.StatusChoices.QUEUED,
)
.update(
status=self.StatusChoices.STARTED,
retry_at=lease_until,
modified_at=now,
)
)
self.refresh_from_db()
return updated == 1
def seal(self) -> bool:
"""Atomically finalize this Snapshot and reconcile its output metadata."""
if self.status == self.StatusChoices.SEALED:
return True
now = timezone.now()
updated = (
type(self)
.objects.filter(
pk=self.pk,
retry_at=self.retry_at,
status__in=self.OPEN_STATES,
)
.update(
status=self.StatusChoices.SEALED,
retry_at=None,
modified_at=now,
)
)
self.refresh_from_db()
if updated == 1:
self.finalize_output_metadata()
return updated == 1
def advance_lifecycle(self) -> bool:
"""Advance one explicit lifecycle step after the runner claims this row."""
if self.status == self.StatusChoices.PAUSED:
return False
if self.status == self.StatusChoices.QUEUED:
return bool(self.url) and self.start_processing()
# abx-dl emits SnapshotCompletedEvent after the complete hook sequence;
# ArchiveResult projection state never drives Snapshot completion.
return False
def cancel(self) -> None:
if self.status != self.StatusChoices.SEALED:
self.seal()
def get_delete_after_config_value(self):
from archivebox.config.common import resolve_delete_after_config_value
return resolve_delete_after_config_value(self.config, self.crawl.config)
@classmethod
def missing_delete_at_candidates(cls):
return cls.objects.filter(delete_at__isnull=True).filter(
Q(config__has_key="DELETE_AFTER") | Q(crawl__config__has_key="DELETE_AFTER"),
)
@classmethod
def is_archivebox_internal_url(cls, url: str, *, config: Mapping[str, Any] | Any | None = None) -> bool:
parsed = urlparse((url or "").strip())
if parsed.scheme not in ("http", "https") or not parsed.hostname:
return False
from archivebox.core.routes_util import (
get_admin_host,
get_api_host,
get_base_host,
get_listen_host,
get_web_host,
split_host_port,
)
if config is None:
config = get_config()
elif isinstance(config, Mapping):
route_config = config
class RouteConfig:
BIND_ADDR = str(route_config.get("BIND_ADDR") or "")
BASE_URL = str(route_config.get("BASE_URL") or "")
CSRF_TRUSTED_ORIGINS = str(route_config.get("CSRF_TRUSTED_ORIGINS") or "")
SERVER_SECURITY_MODE = str(route_config.get("SERVER_SECURITY_MODE") or "")
@property
def USES_SUBDOMAIN_ROUTING(self) -> bool:
return self.SERVER_SECURITY_MODE == "safe-subdomains-fullreplay"
config = RouteConfig()
host = parsed.hostname.lower().strip(".")
port = str(parsed.port) if parsed.port else None
protected_subdomains = {"admin", "web", "api"}
protected_hosts: set[tuple[str, str | None]] = set()
protected_roots: set[tuple[str, str | None]] = set()
protected_local_roots: set[tuple[str, str | None]] = set()
for host_value in (
get_listen_host(config=config),
get_base_host(config=config),
get_admin_host(config=config),
get_web_host(config=config),
get_api_host(config=config),
):
if not host_value:
continue
protected_host, protected_port = split_host_port(host_value)
protected_host = protected_host.strip(".")
if not protected_host:
continue
protected_hosts.add((protected_host, protected_port))
if protected_host in {"", "0.0.0.0", "::", "127.0.0.1", "::1", "localhost"}:
for local_alias in ("127.0.0.1", "localhost"):
protected_hosts.add((local_alias, protected_port))
protected_local_roots.add(("archivebox.localhost", protected_port))
elif protected_host == "archivebox.localhost":
protected_local_roots.add((protected_host, protected_port))
parts = protected_host.split(".", 1)
if len(parts) == 2 and (parts[0] in protected_subdomains or parts[0].startswith("snap-")):
protected_roots.add((parts[1], protected_port))
else:
protected_roots.add((protected_host, protected_port))
for protected_host, protected_port in protected_hosts:
if host == protected_host and (protected_port is None or port == protected_port):
return True
role_roots = protected_roots if config.USES_SUBDOMAIN_ROUTING else protected_local_roots
for protected_root, protected_port in role_roots:
if protected_port is not None and port != protected_port:
continue
if not protected_root or not host.endswith(f".{protected_root}"):
continue
subdomain = host[: -(len(protected_root) + 1)]
if subdomain in protected_subdomains or subdomain.startswith("snap-"):
return True
return False
@property
def created_by(self):
"""Convenience property to access the user who created this snapshot via its crawl."""
return self.crawl.created_by
@property
def process_set(self):
"""Get all Process objects related to this snapshot's ArchiveResults."""
from archivebox.machine.models import Process
return Process.objects.filter(archiveresult__snapshot_id=self.id)
@property
def binary_set(self):
"""Get all Binary objects used by processes related to this snapshot."""
return Binary.objects.filter(process_set__archiveresult__snapshot_id=self.id).distinct()
def ensure_permissions_config(self, crawl_permissions: str | None = None) -> bool:
config = dict(self.config or {})
permission = str(config.get("PERMISSIONS") or "").strip().lower()
from archivebox.core.permissions import PERMISSIONS_PUBLIC, PERMISSIONS_VALUES, normalize_permissions
if permission not in PERMISSIONS_VALUES:
if self.crawl_id and not crawl_permissions:
crawl_permissions = Crawl.objects.filter(pk=self.crawl_id).values_list("permissions", flat=True).first()
config["PERMISSIONS"] = normalize_permissions(
crawl_permissions,
default=PERMISSIONS_PUBLIC,
)
self.config = config
return True
elif config.get("PERMISSIONS") != permission:
config["PERMISSIONS"] = permission
self.config = config
return True
return False
def validate_url_for_archiving(self, *, config: Mapping[str, Any] | Any | None = None) -> None:
try:
validate_url(self.url or "")
except ValueError as err:
raise ValidationError({"url": str(err)}) from err
if self.is_archivebox_internal_url(self.url, config=config):
raise ValidationError({"url": "ArchiveBox cannot archive its own admin, web, api, or snapshot URLs."})
def save(self, *args, **kwargs):
update_fields = kwargs.get("update_fields")
validate_url_field = self._state.adding or update_fields is None or "url" in update_fields
crawl_config_for_save = None
crawl_permissions_for_save = None
if self.crawl_id and validate_url_field:
crawl_row = Crawl.objects.filter(pk=self.crawl_id).values("config", "permissions").first()
if crawl_row:
crawl_config_for_save = crawl_row.get("config") or {}
crawl_permissions_for_save = crawl_row.get("permissions")
if self.ensure_permissions_config(crawl_permissions=crawl_permissions_for_save) and update_fields is not None:
kwargs["update_fields"] = tuple(dict.fromkeys([*update_fields, "config"]))
if validate_url_field:
self.validate_url_for_archiving(config=crawl_config_for_save if self.crawl_id else None)
if not self.bookmarked_at:
self.bookmarked_at = self.created_at or timezone.now()
if not self.timestamp:
self.timestamp = str(self.bookmarked_at.timestamp())
if self._state.adding or update_fields is None or "title" in update_fields:
self.title = self._normalize_title_candidate(self.title, snapshot_url=self.url or "") or None
if self._state.adding or update_fields is None or "notes" in update_fields:
self.notes = sanitize_html_text(self.notes)
super().save(*args, **kwargs)
from django.db import transaction
def finish_snapshot_save():
self.reconcile_filesystem_links()
crawl = Crawl.objects.filter(pk=self.crawl_id).first()
if crawl is None:
return
crawl_tag_names = crawl.current_tag_names()
if crawl_tag_names:
# Snapshots can be created by parser hook side-effect records,
# direct ORM creates, or legacy crawl URL expansion. Crawl tags
# are user-facing metadata on the whole import, so attach them
# at the Snapshot.save() boundary instead of relying on every
# caller to remember to duplicate this fanout logic.
tags_by_name = {tag.name: tag for tag in Tag.objects.filter(name__in=crawl_tag_names)}
missing_tags = [Tag(name=name) for name in crawl_tag_names if name not in tags_by_name]
if missing_tags:
Tag.objects.bulk_create(missing_tags, ignore_conflicts=True)
tags_by_name = {tag.name: tag for tag in Tag.objects.filter(name__in=crawl_tag_names)}
self.add_tag_ids([tag.pk for name in crawl_tag_names if (tag := tags_by_name.get(name))])
# Crawl.urls remains the original submitted source. Snapshot rows
# are the normalized work queue and discovery projection.
# get_or_create/update_or_create wrap save() in atomic(); defer filesystem
# work and crawl maintenance so SQLite commits before touching the disk.
transaction.on_commit(finish_snapshot_save)
# =========================================================================
# Filesystem Migration Methods
# =========================================================================
@staticmethod
def _fs_current_version() -> str:
"""Get current ArchiveBox filesystem layout version."""
return "0.9.4"
_FS_VERSION_MIGRATION_PATHS: ClassVar[dict[str, str]] = {
"0.7.0": "0.9.0",
"0.8.0": "0.9.0",
"0.8.5": "0.9.0",
"0.9.0": "0.9.4",
"0.9.1": "0.9.4",
"0.9.2": "0.9.4",
"0.9.3": "0.9.4",
}
@property
def fs_migration_needed(self) -> bool:
"""Check if snapshot needs filesystem migration"""
return self.fs_version != self._fs_current_version()
def _fs_next_version(self, version: str) -> str:
"""Get the next declared version in the filesystem migration chain."""
return self._FS_VERSION_MIGRATION_PATHS.get(version, self._fs_current_version())
@staticmethod
def is_legacy_archive_dir(path: Path) -> bool:
"""Return True for old-style archive/{timestamp} snapshot directories."""
if path.name in CONSTANTS.RESERVED_ARCHIVE_DIR_NAMES or path.name.startswith("."):
return False
try:
ts_int = int(float(path.name))
except (TypeError, ValueError, OverflowError):
return False
return 788918400 <= ts_int <= 2082758400
def migrate_filesystem_to_current_version(self, source_dir: 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.
"""
current = self.fs_version
target = self._fs_current_version()
cleanup: tuple[Path, Path] | None = None
runtime_config = config or get_config()
if current == target:
current_dir = self.get_storage_path_for_version(target)
legacy_dir = Path(source_dir) if source_dir else CONSTANTS.ARCHIVE_DIR / self.timestamp
cleanup = self._fs_migrate_legacy_to_0_9_0(source_dir=legacy_dir, target_dir=current_dir)
crawl_dir = self.crawl.output_dir
old_crawl_dir = crawl_dir.with_name(str(uuid.UUID(hex=self.crawl.id.hex)))
if old_crawl_dir.exists() and not crawl_dir.exists() and not old_crawl_dir.is_symlink():
crawl_dir.parent.mkdir(parents=True, exist_ok=True)
old_crawl_dir.rename(crawl_dir)
if current_dir.exists():
self.hydrate_archiveresult_output_metadata(snapshot_dir=current_dir)
if cleanup:
old_dir, new_dir = cleanup
if not self._cleanup_old_migration_dir(old_dir, new_dir):
raise SnapshotMigrationError(f"Could not clean up verified migration directory: {old_dir}")
self.reconcile_filesystem_links()
return
while current != target:
next_ver = self._fs_next_version(current)
migrations = {
("0.7.0", "0.9.0"): self._fs_migrate_from_0_7_0_to_0_9_0,
("0.8.0", "0.9.0"): self._fs_migrate_from_0_8_0_to_0_9_0,
("0.8.5", "0.9.0"): self._fs_migrate_from_0_8_0_to_0_9_0,
("0.9.0", "0.9.4"): self._fs_migrate_from_0_9_0_to_0_9_4,
("0.9.1", "0.9.4"): self._fs_migrate_from_0_9_0_to_0_9_4,
("0.9.2", "0.9.4"): self._fs_migrate_from_0_9_0_to_0_9_4,
("0.9.3", "0.9.4"): self._fs_migrate_from_0_9_0_to_0_9_4,
}
migration = migrations.get((current, next_ver))
if migration is None:
raise ValueError(f"No filesystem migration path from {current} to {next_ver}")
cleanup = migration(source_dir=source_dir, config=runtime_config) or cleanup
current = next_ver
self.fs_version = current
source_dir = None
target_dir = self.get_storage_path_for_version(target)
if target_dir.exists():
self.hydrate_archiveresult_output_metadata(snapshot_dir=target_dir)
if cleanup:
old_dir, new_dir = cleanup
if not self._cleanup_old_migration_dir(old_dir, new_dir):
raise SnapshotMigrationError(f"Could not clean up verified migration directory: {old_dir}")
if self.pk:
now = timezone.now()
type(self).objects.filter(pk=self.pk).update(fs_version=target, modified_at=now)
self.modified_at = now
self.reconcile_filesystem_links()
def _fs_migrate_from_0_7_0_to_0_9_0(self, source_dir: Path | None = None, config: "ArchiveBoxBaseConfig | None" = None):
return self._fs_migrate_legacy_to_0_9_0(source_dir=source_dir, config=config)
def _fs_migrate_from_0_8_0_to_0_9_0(self, source_dir: Path | None = None, config: "ArchiveBoxBaseConfig | None" = None):
return self._fs_migrate_legacy_to_0_9_0(source_dir=source_dir, config=config)
def _fs_migrate_from_0_9_0_to_0_9_4(self, source_dir: Path | None = None, config: "ArchiveBoxBaseConfig | None" = None):
runtime_config = config or get_config()
target_dir = self.get_storage_path_for_version("0.9.4")
cleanup = self._fs_migrate_legacy_to_0_9_0(source_dir=source_dir or self.output_dir, target_dir=target_dir, config=runtime_config)
crawl_dir = self.crawl.output_dir
old_crawl_dir = crawl_dir.with_name(str(uuid.UUID(hex=self.crawl.id.hex)))
if old_crawl_dir.exists() and not crawl_dir.exists() and not old_crawl_dir.is_symlink():
crawl_dir.parent.mkdir(parents=True, exist_ok=True)
old_crawl_dir.rename(crawl_dir)
return cleanup
def hydrate_archiveresult_output_metadata(self, snapshot_dir: Path | None = None) -> int:
"""Populate missing ArchiveResult file metadata from existing outputs."""
hydrated = 0
for result in self.archiveresult_set.filter(output_files={}).iterator():
hydrated += int(result.update_output_metadata_from_filesystem(snapshot_dir=snapshot_dir))
return hydrated
def _fs_migrate_legacy_to_0_9_0(
self,
source_dir: Path | None = None,
target_dir: 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}/
"""
import filecmp
import shutil
old_dir = Path(source_dir) if source_dir else self.get_storage_path_for_version("0.8.0")
new_dir = Path(target_dir) if target_dir else self.get_storage_path_for_version("0.9.0")
if old_dir == new_dir:
self.convert_index_json_to_jsonl(output_dir=new_dir)
return None
if old_dir.is_symlink():
try:
points_to_target = new_dir.exists() and old_dir.resolve() == new_dir.resolve()
except OSError:
points_to_target = False
if not points_to_target:
raise SnapshotMigrationError(f"Legacy output symlink does not point to the expected target: {old_dir}")
self.convert_index_json_to_jsonl(output_dir=new_dir)
return (old_dir, new_dir)
if not old_dir.exists():
if new_dir.exists():
self.convert_index_json_to_jsonl(output_dir=new_dir)
return None
return None
if not new_dir.exists():
new_dir.parent.mkdir(parents=True, exist_ok=True)
try:
old_dir.rename(new_dir)
except OSError:
pass
else:
self.convert_index_json_to_jsonl(output_dir=new_dir)
return (old_dir, new_dir)
def copy_file_without_overwriting(source: str, destination: str):
destination_path = Path(destination)
if os.path.lexists(destination_path):
if not destination_path.is_symlink() and destination_path.is_file() and filecmp.cmp(source, destination, shallow=False):
return destination
raise SnapshotMigrationError(f"Migration would overwrite a different output: {destination_path}")
return shutil.copy2(source, destination)
# copytree preserves unknown directories and symlinks. Remove only
# already-copied identical symlinks so interrupted migrations can retry.
for source in old_dir.rglob("*"):
if not source.is_symlink():
continue
destination = new_dir / source.relative_to(old_dir)
if destination.is_symlink() and destination.readlink() == source.readlink():
destination.unlink()
elif os.path.lexists(destination):
raise SnapshotMigrationError(f"Migration would overwrite a different output: {destination}")
shutil.copytree(
old_dir,
new_dir,
copy_function=copy_file_without_overwriting,
dirs_exist_ok=True,
symlinks=True,
)
# Verify every source entry before the old tree is eligible for cleanup.
for source in old_dir.rglob("*"):
destination = new_dir / source.relative_to(old_dir)
if source.is_symlink():
copied = destination.is_symlink() and destination.readlink() == source.readlink()
elif source.is_dir():
copied = destination.is_dir() and not destination.is_symlink()
elif source.is_file():
copied = destination.is_file() and not destination.is_symlink() and filecmp.cmp(source, destination, shallow=False)
else:
raise SnapshotMigrationError(f"Migration cannot safely copy special output: {source}")
if not copied:
raise SnapshotMigrationError(f"Migration incomplete: {source.relative_to(old_dir)}")
# Convert index.json to index.jsonl in the new directory.
self.convert_index_json_to_jsonl(output_dir=new_dir)
return (old_dir, new_dir)
@staticmethod
def _migration_trees_match(old_dir: Path, new_dir: Path) -> bool:
"""Verify every legacy entry exists unchanged at the destination."""
import filecmp
if old_dir.is_symlink():
try:
return new_dir.exists() and old_dir.resolve() == new_dir.resolve()
except OSError:
return False
if not old_dir.exists():
return True
if not new_dir.is_dir() or new_dir.is_symlink():
return False
for source in old_dir.rglob("*"):
destination = new_dir / source.relative_to(old_dir)
if source.is_symlink():
copied = destination.is_symlink() and destination.readlink() == source.readlink()
elif source.is_dir():
copied = destination.is_dir() and not destination.is_symlink()
elif source.is_file():
copied = destination.is_file() and not destination.is_symlink() and filecmp.cmp(source, destination, shallow=False)
else:
copied = False
if not copied:
return False
return True
def _cleanup_old_migration_dir(self, old_dir: Path, new_dir: Path) -> bool:
"""Delete the old directory after its contents are verified at the new path."""
import logging
import shutil
from archivebox.config.permissions import SudoPermission
if not self._migration_trees_match(old_dir, new_dir):
logging.getLogger("archivebox.migration").warning(
f"Refusing to remove unverified migration directory {old_dir}",
)
return False
# Delete old directory
if old_dir.exists() and not old_dir.is_symlink():
try:
# Root-launched commands retain a saved root EUID after dropping
# privileges, which is needed for legacy trees owned by old UIDs.
with SudoPermission(uid=0, fallback=True):
shutil.rmtree(old_dir)
except OSError as e:
logging.getLogger("archivebox.migration").warning(
f"Could not remove old migration directory {old_dir}: {e}",
)
return False
# Older migration runs may already have left a timestamp projection.
if old_dir.is_symlink():
old_dir.unlink(missing_ok=True)
return True
# =========================================================================
# Path Calculation and Migration Helpers
# =========================================================================
@staticmethod
def 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
"""
from urllib.parse import urlparse
try:
parsed = urlparse(url)
if parsed.scheme in ("http", "https"):
if parsed.port:
return f"{parsed.hostname}_{parsed.port}".replace(":", "_")
return parsed.hostname or "unknown"
elif parsed.scheme == "file":
return "localhost"
elif parsed.scheme:
return parsed.scheme
else:
return "unknown"
except (TypeError, ValueError):
return "unknown"
def get_storage_path_for_version(self, version: str) -> 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}/
"""
if version in ("0.7.0", "0.8.0", "0.8.5"):
return CONSTANTS.ARCHIVE_DIR / self.timestamp
elif version in ("0.9.0", "0.9.1", "0.9.2", "0.9.3", "0.9.4", "1.0.0"):
username = self.created_by.username
date_base = self.bookmarked_at or self.created_at
date_str = date_base.strftime("%Y%m%d") if date_base else "unknown"
domain = self.extract_domain_from_url(self.url)
return CONSTANTS.USERS_DIR / username / CONSTANTS.SNAPSHOTS_DIR_NAME / date_str / domain / str(self.id)
else:
# Unknown version - use current
return self.get_storage_path_for_version(self._fs_current_version())
# =========================================================================
# Loading and Creation from Filesystem (Used by archivebox update ONLY)
# =========================================================================
@classmethod
def load_from_directory(cls, snapshot_dir: Path) -> Optional["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)
"""
from archivebox.machine.models import Process
# Try index.jsonl first (new format), then index.json (legacy)
jsonl_path = snapshot_dir / CONSTANTS.JSONL_INDEX_FILENAME
json_path = snapshot_dir / CONSTANTS.JSON_INDEX_FILENAME
data = None
if jsonl_path.exists():
try:
records = Process.parse_records_from_text(jsonl_path.read_text())
for record in records:
if record.get("type") == "Snapshot":
data = record
break
except OSError:
pass
if data is None and json_path.exists():
try:
with open(json_path) as f:
data = json.load(f)
except (json.JSONDecodeError, OSError):
pass
if not data:
timestamp = cls._select_best_timestamp(
index_timestamp=None,
folder_name=snapshot_dir.name,
)
if not timestamp:
return None
try:
return cls.objects.select_related("crawl__created_by").get(timestamp=timestamp)
except cls.DoesNotExist:
return None
except cls.MultipleObjectsReturned:
return cls.objects.select_related("crawl__created_by").filter(timestamp=timestamp).first()
url = data.get("url")
if not url:
timestamp = cls._select_best_timestamp(
index_timestamp=data.get("timestamp"),
folder_name=snapshot_dir.name,
)
if not timestamp:
return None
try:
return cls.objects.select_related("crawl__created_by").get(timestamp=timestamp)
except cls.DoesNotExist:
return None
except cls.MultipleObjectsReturned:
return cls.objects.select_related("crawl__created_by").filter(timestamp=timestamp).first()
# Get timestamp - prefer index file, fallback to folder name
timestamp = cls._select_best_timestamp(
index_timestamp=data.get("timestamp"),
folder_name=snapshot_dir.name,
)
folder_timestamp = cls._select_best_timestamp(
index_timestamp=None,
folder_name=snapshot_dir.name,
)
if not timestamp:
return None
# Look up existing (try exact match first, then fuzzy match for truncated timestamps)
try:
snapshot = cls.objects.select_related("crawl__created_by").get(url=url, timestamp=timestamp)
return snapshot
except cls.DoesNotExist:
# Try fuzzy match - index.json may have truncated timestamp
# e.g., index has "1767000340" but DB has "1767000340.624737"
# Do not fuzzy-match when the legacy folder name itself is a valid
# timestamp; distinct dirs like 1508259732 and 1508259732.0 must
# remain distinct snapshots.
if not folder_timestamp or timestamp != folder_timestamp:
candidates = cls.objects.select_related("crawl__created_by").filter(url=url, timestamp__startswith=timestamp)
if candidates.count() == 1:
snapshot = candidates.first()
if snapshot is None:
return None
return snapshot
elif candidates.count() > 1:
return candidates.first()
return None
except cls.MultipleObjectsReturned:
# Should not happen with unique constraint
return cls.objects.select_related("crawl__created_by").filter(url=url, timestamp=timestamp).first()
@classmethod
def create_from_directory(cls, snapshot_dir: Path) -> Optional["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)
"""
from archivebox.machine.models import Process
# Try index.jsonl first (new format), then index.json (legacy)
jsonl_path = snapshot_dir / CONSTANTS.JSONL_INDEX_FILENAME
json_path = snapshot_dir / CONSTANTS.JSON_INDEX_FILENAME
data = None
if jsonl_path.exists():
try:
records = Process.parse_records_from_text(jsonl_path.read_text())
for record in records:
if record.get("type") == "Snapshot":
data = record
break
except OSError:
pass
if data is None and json_path.exists():
try:
with open(json_path) as f:
data = json.load(f)
except (json.JSONDecodeError, OSError):
pass
if not data or not data.get("url"):
archive_org_path = snapshot_dir / "archive.org.txt"
try:
archived_url = archive_org_path.read_text(encoding="utf-8", errors="replace").strip().splitlines()[0].strip()
except (IndexError, OSError):
archived_url = ""
if archived_url.startswith(("http://", "https://")):
if "://web.archive.org/web/" in archived_url and "/web/" in archived_url:
archive_target = archived_url.split("/web/", 1)[1].split("/", 1)
if len(archive_target) == 2:
candidate = archive_target[1]
if not candidate.startswith(("http://", "https://")) and "/" in candidate:
candidate = candidate.split("/", 1)[1]
if candidate.startswith(("http://", "https://")):
archived_url = candidate
data = {
"url": archived_url,
"timestamp": snapshot_dir.name,
"title": "",
}
if not data:
return None
url = data.get("url")
if not url:
return None
# Get and validate timestamp
timestamp = cls._select_best_timestamp(
index_timestamp=data.get("timestamp"),
folder_name=snapshot_dir.name,
)
if not timestamp:
return None
# Ensure uniqueness (reuses existing logic from create_or_update_from_dict)
timestamp = cls._ensure_unique_timestamp(url, timestamp)
# Detect version
fs_version = cls._detect_fs_version_from_index(data)
system_user_id = get_or_create_system_user_pk()
catchall_crawl, _ = Crawl.objects.get_or_create(
label="[migration] orphaned snapshots",
defaults={
"urls": f"# Orphaned snapshot: {url}",
"max_depth": 0,
"created_by_id": system_user_id,
},
)
if cls.objects.filter(crawl=catchall_crawl, url=url).exists():
catchall_crawl = Crawl.objects.create(
label=f"[migration] orphaned snapshot {timestamp}",
urls=url,
max_depth=0,
created_by_id=system_user_id,
)
snapshot_kwargs = {
"url": url,
"timestamp": timestamp,
"title": data.get("title", ""),
"fs_version": fs_version,
"crawl": catchall_crawl,
}
try:
bookmarked_at = parse_date(data.get("bookmarked_at") or timestamp)
except (TypeError, ValueError, OSError):
bookmarked_at = None
try:
created_at = parse_date(data.get("created_at"))
except (TypeError, ValueError, OSError):
created_at = None
if bookmarked_at:
snapshot_kwargs["bookmarked_at"] = bookmarked_at
if created_at:
snapshot_kwargs["created_at"] = created_at
return cls(
**snapshot_kwargs,
)
@staticmethod
def _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.
"""
def is_valid_timestamp(ts: object | None) -> bool:
if not isinstance(ts, (str, int, float)):
return False
try:
ts_int = int(float(ts))
# 1995-01-01 to 2035-12-31
return 788918400 <= ts_int <= 2082758400
except (TypeError, ValueError, OverflowError):
return False
index_valid = is_valid_timestamp(index_timestamp) if index_timestamp else False
folder_valid = is_valid_timestamp(folder_name)
if folder_valid:
return str(folder_name).strip()
if index_valid and index_timestamp is not None:
return str(index_timestamp).strip()
return None
@classmethod
def _ensure_unique_timestamp(cls, url: str, timestamp: str) -> str:
"""
Ensure timestamp is globally unique.
If there is a collision, add a tiny fractional suffix until unique.
"""
candidate = str(timestamp)
base = float(timestamp)
suffix = 0
while cls.objects.filter(timestamp=candidate).exists():
suffix += 1
candidate = f"{base + (suffix / 1_000_000):.6f}".rstrip("0").rstrip(".")
return candidate
@staticmethod
def _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
"""
if "fs_version" in data:
return data["fs_version"]
if "history" in data and "archive_results" not in data:
return "0.7.0"
if "archive_results" in data:
return "0.8.0"
return "0.7.0"
# =========================================================================
# Index.json Reconciliation
# =========================================================================
def reconcile_with_index(self, output_dir: 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)
"""
import json
# Try to convert index.json to index.jsonl first
output_dir = Path(output_dir) if output_dir is not None else Path(self.output_dir)
self.convert_index_json_to_jsonl(output_dir=output_dir)
# Check for index.jsonl (preferred) or index.json (legacy)
jsonl_path = output_dir / CONSTANTS.JSONL_INDEX_FILENAME
json_path = output_dir / CONSTANTS.JSON_INDEX_FILENAME
index_data = {}
if jsonl_path.exists():
# Read from JSONL format
jsonl_data = self.read_index_jsonl(output_dir=output_dir)
if jsonl_data["snapshot"]:
index_data = jsonl_data["snapshot"]
# Convert archive_results list to expected format
index_data["archive_results"] = jsonl_data["archive_results"]
elif json_path.exists():
# Fallback to legacy JSON format
try:
with open(json_path) as f:
index_data = json.load(f)
except (OSError, TypeError, ValueError, json.JSONDecodeError):
pass
# Merge title
self._merge_title_from_index(index_data)
# Merge tags
self._merge_tags_from_index(index_data)
# Merge ArchiveResults
self._merge_archive_results_from_index(index_data, update_existing=update_existing_archive_results)
if not self._normalize_title_candidate(self.title, snapshot_url=self.url):
title_results = (
self.archiveresult_set.filter(
plugin="title",
status=ArchiveResult.StatusChoices.SUCCEEDED,
)
.exclude(output_str="")
.order_by("-start_ts", "-end_ts", "-created_at")
)
for title_result in title_results.only("output_str"):
result_title = self._normalize_title_candidate(title_result.output_str, snapshot_url=self.url)
if result_title:
self.title = result_title
break
# Write back in JSONL format
self.write_index_jsonl(output_dir=output_dir)
def reconcile_with_index_json(self, output_dir: Path | None = None, update_existing_archive_results: bool = True):
"""Deprecated: use reconcile_with_index() instead."""
return self.reconcile_with_index(output_dir=output_dir, update_existing_archive_results=update_existing_archive_results)
def _merge_title_from_index(self, index_data: dict):
"""Merge title - prefer longest non-URL title."""
index_title = self._normalize_title_candidate(index_data.get("title"), snapshot_url=self.url)
db_title = self._normalize_title_candidate(self.title, snapshot_url=self.url)
candidates = [t for t in [index_title, db_title] if t]
if candidates:
best_title = max(candidates, key=len)
if self.title != best_title:
self.title = best_title
elif self.title:
self.title = None
def _merge_tags_from_index(self, index_data: dict):
"""Merge tags - union of both sources."""
index_tags = set(index_data.get("tags", "").split(",")) if index_data.get("tags") else set()
index_tags = {t.strip() for t in index_tags if t.strip()}
db_tags = set(self.tags.values_list("name", flat=True))
new_tags = index_tags - db_tags
if new_tags:
for tag_name in new_tags:
tag, _ = Tag.get_or_create_by_name(tag_name)
self.add_tag_ids([tag.pk])
def _merge_archive_results_from_index(self, index_data: dict, update_existing: bool = True):
"""Merge ArchiveResults one row per hook; retries update the existing row."""
existing = {(ar.plugin, ar.hook_name): ar for ar in ArchiveResult.objects.filter(snapshot=self)}
if update_existing:
for archiveresult in existing.values():
normalized_status = ArchiveResult.normalize_status(archiveresult.status)
if archiveresult.status != normalized_status:
archiveresult.status = normalized_status
archiveresult.save(update_fields=["status", "modified_at"])
# Handle 0.8.x format (archive_results list)
for result_data in index_data.get("archive_results", []):
self._create_archive_result_if_missing(result_data, existing, update_existing=update_existing)
# Handle 0.7.x format (history dict)
if "history" in index_data and isinstance(index_data["history"], dict):
for plugin, result_list in index_data["history"].items():
if isinstance(result_list, list):
for result_data in result_list:
# Support both old 'extractor' and new 'plugin' keys for backwards compat
result_data["plugin"] = result_data.get("plugin") or result_data.get("extractor") or plugin
self._create_archive_result_if_missing(result_data, existing, update_existing=update_existing)
def _create_archive_result_if_missing(self, result_data: dict, existing: dict, update_existing: bool = True):
"""Create ArchiveResult if not already in DB."""
from dateutil import parser
from django.db import transaction
from archivebox.machine.models import Machine, Process
# Support both old 'extractor' and new 'plugin' keys for backwards compat
plugin = (result_data.get("plugin") or result_data.get("extractor", ""))[:32]
if not plugin:
return
start_ts = None
if result_data.get("start_ts"):
try:
start_ts = parser.parse(result_data["start_ts"])
if start_ts and timezone.is_naive(start_ts):
start_ts = timezone.make_aware(start_ts, timezone.get_current_timezone())
except (TypeError, ValueError, OverflowError):
pass
end_ts = None
if result_data.get("end_ts"):
try:
end_ts = parser.parse(result_data["end_ts"])
if end_ts and timezone.is_naive(end_ts):
end_ts = timezone.make_aware(end_ts, timezone.get_current_timezone())
except (TypeError, ValueError, OverflowError):
pass
# Support both 'output' (legacy) and 'output_str' (new JSONL) field names
output_str = result_data.get("output_str") or result_data.get("output", "")
status = ArchiveResult.normalize_status(result_data.get("status") or ArchiveResult.StatusChoices.FAILED)
process = None
cmd = result_data.get("cmd") or []
pwd = result_data.get("pwd") or ""
output_files = ArchiveResult._normalize_output_files(result_data.get("output_files"))
output_size = ArchiveResult._coerce_output_file_size(result_data.get("output_size"))
output_json = result_data.get("output_json")
output_mimetypes = result_data.get("output_mimetypes", "")
hook_name = result_data.get("hook_name", "")
existing_result = existing.get((plugin, hook_name))
if existing_result:
if not update_existing:
return
update_fields = []
if existing_result.status != status:
existing_result.status = status
update_fields.append("status")
if output_str and existing_result.output_str != output_str:
existing_result.output_str = output_str
update_fields.append("output_str")
if output_json and existing_result.output_json != output_json:
existing_result.output_json = output_json
update_fields.append("output_json")
if output_files and existing_result.output_files != output_files:
existing_result.output_files = output_files
update_fields.append("output_files")
if "output_size" in result_data and existing_result.output_size != output_size:
existing_result.output_size = output_size
update_fields.append("output_size")
if output_mimetypes and existing_result.output_mimetypes != output_mimetypes:
existing_result.output_mimetypes = output_mimetypes
update_fields.append("output_mimetypes")
if start_ts and existing_result.start_ts != start_ts:
existing_result.start_ts = start_ts
update_fields.append("start_ts")
if end_ts and existing_result.end_ts != end_ts:
existing_result.end_ts = end_ts
update_fields.append("end_ts")
if update_fields:
existing_result.save(update_fields=[*update_fields, "modified_at"])
return
# Machine.current() can probe the host and sanitize config. Do that before
# atomic() so the transaction below only covers the two related row writes.
machine = Machine.current() if cmd or pwd else None
with transaction.atomic():
if machine is not None:
process = Process.objects.create(
machine=machine,
process_type=Process.TypeChoices.HOOK,
worker_type="archiveresult",
cmd=cmd,
pwd=pwd,
status=Process.StatusChoices.EXITED,
exit_code=0 if status in ("succeeded", "skipped", "noresults") else 1,
started_at=start_ts,
ended_at=end_ts,
)
archiveresult = ArchiveResult.objects.create(
snapshot=self,
plugin=plugin,
hook_name=hook_name,
status=status,
output_str=output_str,
output_json=output_json,
output_files=output_files,
output_size=output_size,
output_mimetypes=output_mimetypes,
start_ts=start_ts,
end_ts=end_ts,
process=process,
)
existing[(plugin, hook_name)] = archiveresult
def write_index_json(self):
"""Write index.json in 0.9.x format (deprecated, use write_index_jsonl)."""
import json
index_path = Path(self.output_dir) / "index.json"
data = {
"url": self.url,
"timestamp": self.timestamp,
"title": self.title or "",
"tags": ",".join(sorted(self.tags.values_list("name", flat=True))),
"fs_version": self.fs_version,
"bookmarked_at": self.bookmarked_at.isoformat() if self.bookmarked_at else None,
"created_at": self.created_at.isoformat() if self.created_at else None,
"archive_results": [
{
"plugin": ar.plugin,
"status": ar.status,
"start_ts": ar.start_ts.isoformat() if ar.start_ts else None,
"end_ts": ar.end_ts.isoformat() if ar.end_ts else None,
"output": ar.output_str or "",
"cmd": ar.cmd if isinstance(ar.cmd, list) else [],
"pwd": ar.pwd,
}
for ar in ArchiveResult.objects.filter(snapshot=self).order_by("start_ts")
],
}
index_path.parent.mkdir(parents=True, exist_ok=True)
with open(index_path, "w") as f:
json.dump(data, f, indent=2, sort_keys=True)
def write_index_jsonl(self, output_dir: 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.)
"""
import json
output_dir = Path(output_dir) if output_dir is not None else Path(self.output_dir)
index_path = output_dir / CONSTANTS.JSONL_INDEX_FILENAME
index_path.parent.mkdir(parents=True, exist_ok=True)
archive_results = list(self.archiveresult_set.select_related("process__binary").order_by("start_ts"))
# Build canonical records before replacing the file so legacy records
# without a corresponding DB row can be retained byte-for-byte.
binaries_seen = set()
processes_seen = set()
records = [self.to_json()]
for ar in archive_results:
process = ar.process_record
if process and process.binary and process.binary_id not in binaries_seen:
binaries_seen.add(process.binary_id)
records.append(process.binary.to_json())
if process and process.id not in processes_seen:
processes_seen.add(process.id)
records.append(process.to_json())
records.append(ar.to_json(snapshot_output_dir=output_dir))
canonical_keys = {
(record.get("type"), str(record.get("id"))) for record in records if record.get("type") and record.get("id") is not None
}
preserved_lines = []
if index_path.exists():
for line in index_path.read_text(encoding="utf-8").splitlines(keepends=True):
try:
existing_record = json.loads(line)
except json.JSONDecodeError:
preserved_lines.append(line)
continue
if existing_record.get("type") == "Snapshot":
records[0] = {**existing_record, **records[0]}
continue
key = (existing_record.get("type"), str(existing_record.get("id")))
if existing_record.get("id") is None or key not in canonical_keys:
preserved_lines.append(line)
tmp_index_path = index_path.with_name(f".{index_path.name}.tmp")
with open(tmp_index_path, "w", encoding="utf-8") as f:
f.write(json.dumps(records[0]) + "\n")
for line in preserved_lines:
f.write(line)
if not line.endswith("\n"):
f.write("\n")
for record in records[1:]:
f.write(json.dumps(record) + "\n")
os.replace(tmp_index_path, index_path)
def read_index_jsonl(self, output_dir: Path | None = None) -> dict:
"""
Read index.jsonl and return parsed records grouped by type.
Returns dict with keys: 'snapshot', 'archive_results', 'binaries', 'processes'
"""
from archivebox.machine.models import Process
from archivebox.misc.jsonl import (
TYPE_ARCHIVERESULT,
TYPE_BINARY,
TYPE_BINARYREQUEST,
TYPE_PROCESS,
TYPE_SNAPSHOT,
)
output_dir = Path(output_dir) if output_dir is not None else Path(self.output_dir)
index_path = output_dir / CONSTANTS.JSONL_INDEX_FILENAME
result: dict[str, Any] = {
"snapshot": None,
"archive_results": [],
"binaries": [],
"processes": [],
}
if not index_path.exists():
return result
records = Process.parse_records_from_text(index_path.read_text())
for record in records:
record_type = record.get("type")
if record_type == TYPE_SNAPSHOT:
result["snapshot"] = record
elif record_type == TYPE_ARCHIVERESULT:
result["archive_results"].append(record)
elif record_type in {TYPE_BINARYREQUEST, TYPE_BINARY}:
result["binaries"].append(record)
elif record_type == TYPE_PROCESS:
result["processes"].append(record)
return result
def convert_index_json_to_jsonl(self, output_dir: 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.
"""
import json
output_dir = Path(output_dir) if output_dir is not None else Path(self.output_dir)
json_path = output_dir / CONSTANTS.JSON_INDEX_FILENAME
jsonl_path = output_dir / CONSTANTS.JSONL_INDEX_FILENAME
# Skip if already converted or no json file exists. Keep a divergent
# legacy JSON index intact instead of silently discarding its metadata.
if jsonl_path.exists():
return False
if not json_path.exists():
return False
try:
with open(json_path) as f:
data = json.load(f)
except (json.JSONDecodeError, OSError):
return False
# Detect format version and extract records
fs_version = data.get("fs_version", "0.7.0")
records = []
snapshot_record = {
"type": "Snapshot",
"id": str(self.id),
"crawl_id": str(self.crawl_id) if self.crawl_id else None,
"url": data.get("url", self.url),
"timestamp": data.get("timestamp", self.timestamp),
"title": data.get("title", self.title or ""),
"tags": data.get("tags", ""),
"fs_version": fs_version,
"bookmarked_at": data.get("bookmarked_at"),
"created_at": data.get("created_at"),
}
records.append(snapshot_record)
# Handle 0.8.x/0.9.x format (archive_results list)
for result_data in data.get("archive_results", []):
ar_record = {
"type": "ArchiveResult",
"snapshot_id": str(self.id),
"plugin": result_data.get("plugin", ""),
"hook_name": result_data.get("hook_name", ""),
"status": result_data.get("status") or ArchiveResult.StatusChoices.FAILED,
"output_str": result_data.get("output_str") or result_data.get("output", ""),
"output_json": result_data.get("output_json"),
"output_files": result_data.get("output_files"),
"output_size": result_data.get("output_size"),
"output_mimetypes": result_data.get("output_mimetypes", ""),
"start_ts": result_data.get("start_ts"),
"end_ts": result_data.get("end_ts"),
}
if result_data.get("cmd"):
ar_record["cmd"] = result_data["cmd"]
if result_data.get("pwd"):
ar_record["pwd"] = result_data["pwd"]
records.append(ar_record)
# Handle 0.7.x format (history dict)
if "history" in data and isinstance(data["history"], dict):
for plugin, result_list in data["history"].items():
if not isinstance(result_list, list):
continue
for result_data in result_list:
ar_record = {
"type": "ArchiveResult",
"snapshot_id": str(self.id),
"plugin": result_data.get("plugin") or result_data.get("extractor") or plugin,
"hook_name": result_data.get("hook_name", ""),
"status": result_data.get("status") or ArchiveResult.StatusChoices.FAILED,
"output_str": result_data.get("output_str") or result_data.get("output", ""),
"output_json": result_data.get("output_json"),
"output_files": result_data.get("output_files"),
"output_size": result_data.get("output_size"),
"output_mimetypes": result_data.get("output_mimetypes", ""),
"start_ts": result_data.get("start_ts"),
"end_ts": result_data.get("end_ts"),
}
if result_data.get("cmd"):
ar_record["cmd"] = result_data["cmd"]
if result_data.get("pwd"):
ar_record["pwd"] = result_data["pwd"]
records.append(ar_record)
jsonl_path.parent.mkdir(parents=True, exist_ok=True)
tmp_jsonl_path = jsonl_path.with_name(f".{jsonl_path.name}.tmp")
with open(tmp_jsonl_path, "w", encoding="utf-8") as f:
f.write("".join(json.dumps(record) + "\n" for record in records))
os.replace(tmp_jsonl_path, jsonl_path)
return True
# =========================================================================
# Snapshot Utilities
# =========================================================================
@staticmethod
def move_directory_to_invalid(snapshot_dir: Path):
"""
Move invalid directory to data/invalid/YYYYMMDD/.
Used by: archivebox update (when encountering invalid directories)
"""
import shutil
invalid_dir = CONSTANTS.DATA_DIR / "invalid" / datetime.now(UTC).strftime("%Y%m%d")
invalid_dir.mkdir(parents=True, exist_ok=True)
dest = invalid_dir / snapshot_dir.name
counter = 1
while dest.exists():
dest = invalid_dir / f"{snapshot_dir.name}_{counter}"
counter += 1
try:
shutil.move(str(snapshot_dir), str(dest))
except OSError:
return
@classmethod
def find_and_merge_duplicates(cls) -> int:
"""
Find and merge snapshots with same url:timestamp.
Returns count of duplicate sets merged.
Used by: archivebox update (Phase 3: deduplication)
"""
from django.db.models import Count
duplicates = cls.objects.values("url", "timestamp").annotate(count=Count("id")).filter(count__gt=1)
merged = 0
for dup in duplicates.iterator(chunk_size=500):
snapshots = list(
cls.objects.filter(url=dup["url"], timestamp=dup["timestamp"]).order_by("created_at"), # Keep oldest
)
if len(snapshots) > 1:
try:
cls._merge_snapshots(snapshots)
merged += 1
except OSError:
continue
return merged
@classmethod
def _merge_snapshots(cls, snapshots: Sequence["Snapshot"]):
"""
Merge exact duplicates.
Keep oldest, union files + ArchiveResults.
"""
import shutil
keeper = snapshots[0]
duplicates = snapshots[1:]
keeper_dir = Path(keeper.output_dir)
for dup in duplicates:
dup_dir = Path(dup.output_dir)
# Merge files
if dup_dir.exists() and dup_dir != keeper_dir:
for dup_file in dup_dir.rglob("*"):
if not dup_file.is_file():
continue
rel = dup_file.relative_to(dup_dir)
keeper_file = keeper_dir / rel
if not keeper_file.exists():
keeper_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(dup_file, keeper_file)
try:
shutil.rmtree(dup_dir)
except OSError:
continue
# Merge tags
for tag in dup.tags.all():
keeper.add_tag_ids([tag.pk])
# Move each hook result, merging only an exact identity collision.
for result in ArchiveResult.objects.filter(snapshot=dup):
existing = ArchiveResult.objects.filter(
snapshot=keeper,
plugin=result.plugin,
hook_name=result.hook_name,
).first()
if existing is None:
result.snapshot = keeper
result.save(update_fields=["snapshot", "modified_at"])
continue
output_files = {**(existing.output_files or {}), **(result.output_files or {})}
existing.output_files = output_files
existing.output_size = max(
sum(
ArchiveResult._coerce_output_file_size(metadata.get("size"))
for metadata in output_files.values()
if isinstance(metadata, dict)
),
existing.output_size,
result.output_size,
)
if result.modified_at >= existing.modified_at:
existing.status = result.status
existing.output_str = result.output_str
existing.output_json = result.output_json
existing.start_ts = result.start_ts
existing.end_ts = result.end_ts
existing.output_mimetypes = ",".join(
sorted(
{
mimetype.strip()
for value in (existing.output_mimetypes, result.output_mimetypes)
for mimetype in value.split(",")
if mimetype.strip()
},
),
)
existing.save()
result.delete()
# Delete
dup.delete()
# =========================================================================
# Output Directory Properties
# =========================================================================
@property
def output_dir_parent(self) -> str:
return "archive"
@property
def output_dir_name(self) -> str:
return str(self.timestamp)
def archive(self, overwrite=False, methods=None):
updates = {
"status": self.StatusChoices.QUEUED,
"retry_at": timezone.now(),
}
if overwrite:
updates["downloaded_at"] = None
return int(self.update_and_requeue(**updates))
@admin.display(description="Tags")
def tags_str(self) -> str | None:
if "_tags_str_cached" in self.__dict__:
return self.__dict__["_tags_str_cached"]
calc_tags_str = lambda: ",".join(sorted(tag.name for tag in self.tags.all()))
prefetched_cache = self.__dict__.get("_prefetched_objects_cache", {})
if "tags" in prefetched_cache:
return calc_tags_str()
return calc_tags_str()
def icons(self, path: str | None = None, prefix: str = "/", quote_paths: bool = False) -> str:
"""Generate HTML icons showing which extractor plugins have succeeded for this snapshot"""
from urllib.parse import quote
from django.utils.html import format_html
compact_icons = self.__dict__.get("_icons_compact", False)
def calc_icons():
if compact_icons and self.status == self.StatusChoices.STARTED:
progress_stats = self.__dict__.get("_icons_progress_stats") or self.get_progress_stats()
total = int(progress_stats.get("total") or 0)
succeeded = int(progress_stats.get("succeeded") or 0)
failed = int(progress_stats.get("failed") or 0)
skipped = int(progress_stats.get("skipped") or 0)
noresults = int(progress_stats.get("noresults") or 0)
running = int(progress_stats.get("running") or 0)
completed = succeeded + failed + skipped + noresults
percent = int((completed / total * 100) if total > 0 else 0)
successful_plugins = sorted(self.__dict__.get("_icons_archive_results") or ())
visible_plugins = successful_plugins[:8]
plugin_icon_spans = []
for plugin in visible_plugins:
icon = get_plugin_icon(plugin)
if str(icon).strip():
plugin_icon_spans.append(str(format_html('{}', plugin, mark_safe(icon))))
successful_icons = format_html(
'{}
',
mark_safe(
"".join(plugin_icon_spans)
+ (
str(
format_html(
'+{}',
len(successful_plugins) - 8,
len(successful_plugins) - 8,
),
)
if len(successful_plugins) > 8
else ""
),
),
)
return format_html(
''
'
'
''
'{}/{} hooks'
"
"
'
"
'
'
"✓{} ✗{} ⏳{}"
"
"
"{}"
"