diff --git a/ArchiveBox.conf b/ArchiveBox.conf
deleted file mode 100644
index a7eaff41..00000000
--- a/ArchiveBox.conf
+++ /dev/null
@@ -1,3 +0,0 @@
-[SERVER_CONFIG]
-SECRET_KEY = y6fw9wcaqls9sx_dze6ahky9ggpkpzoaw5g5v98_u3ro5j0_4f
-
diff --git a/archivebox/api/v1_workers.py b/archivebox/api/v1_workers.py
index 95678ef5..9e138e16 100644
--- a/archivebox/api/v1_workers.py
+++ b/archivebox/api/v1_workers.py
@@ -35,12 +35,8 @@ class WorkerSchema(Schema):
model: str
max_tick_time: int
max_concurrent_tasks: int
- poll_interval: float
- idle_timeout: int
running_count: int
running_workers: List[dict[str, Any]]
- queue_count: int
- queue: List[QueueItemSchema]
@staticmethod
def resolve_model(obj) -> str:
@@ -55,38 +51,21 @@ class WorkerSchema(Schema):
def resolve_max_concurrent_tasks(obj) -> int:
return obj.MAX_CONCURRENT_TASKS
- @staticmethod
- def resolve_poll_interval(obj) -> float:
- return obj.POLL_INTERVAL
-
- @staticmethod
- def resolve_idle_timeout(obj) -> int:
- return obj.IDLE_TIMEOUT
-
@staticmethod
def resolve_running_count(obj) -> int:
- return len(obj.get_running_workers())
+ return obj.get_worker_count()
@staticmethod
def resolve_running_workers(obj) -> List[dict[str, Any]]:
return obj.get_running_workers()
- @staticmethod
- def resolve_queue_count(obj) -> int:
- return obj.get_queue().count()
-
- @staticmethod
- def resolve_queue(obj) -> List[QueueItemSchema]:
- return list(obj.get_queue()[:50]) # Limit to 50 items
-
class OrchestratorSchema(Schema):
"""Schema for the Orchestrator."""
is_running: bool
poll_interval: float
idle_timeout: int
- max_workers_per_type: int
- max_total_workers: int
+ max_crawl_workers: int
total_worker_count: int
workers: List[WorkerSchema]
@@ -95,23 +74,20 @@ class OrchestratorSchema(Schema):
def get_orchestrator(request):
"""Get the orchestrator status and all worker queues."""
from archivebox.workers.orchestrator import Orchestrator
- from archivebox.workers.worker import CrawlWorker, SnapshotWorker, ArchiveResultWorker
+ from archivebox.workers.worker import CrawlWorker
orchestrator = Orchestrator()
# Create temporary worker instances to query their queues
workers = [
CrawlWorker(worker_id=-1),
- SnapshotWorker(worker_id=-1),
- ArchiveResultWorker(worker_id=-1),
]
return {
'is_running': orchestrator.is_running(),
'poll_interval': orchestrator.POLL_INTERVAL,
'idle_timeout': orchestrator.IDLE_TIMEOUT,
- 'max_workers_per_type': orchestrator.MAX_WORKERS_PER_TYPE,
- 'max_total_workers': orchestrator.MAX_TOTAL_WORKERS,
+ 'max_crawl_workers': orchestrator.MAX_CRAWL_WORKERS,
'total_worker_count': orchestrator.get_total_worker_count(),
'workers': workers,
}
@@ -120,41 +96,12 @@ def get_orchestrator(request):
@router.get("/workers", response=List[WorkerSchema], url_name="get_workers")
def get_workers(request):
"""List all worker types and their current status."""
- from archivebox.workers.worker import CrawlWorker, SnapshotWorker, ArchiveResultWorker
+ from archivebox.workers.worker import CrawlWorker
# Create temporary instances to query their queues
return [
CrawlWorker(worker_id=-1),
- SnapshotWorker(worker_id=-1),
- ArchiveResultWorker(worker_id=-1),
]
-@router.get("/worker/{worker_name}", response=WorkerSchema, url_name="get_worker")
-def get_worker(request, worker_name: str):
- """Get status and queue for a specific worker type."""
- from archivebox.workers.worker import WORKER_TYPES
-
- if worker_name not in WORKER_TYPES:
- from ninja.errors import HttpError
- raise HttpError(404, f"Unknown worker type: {worker_name}. Valid types: {list(WORKER_TYPES.keys())}")
-
- WorkerClass = WORKER_TYPES[worker_name]
- return WorkerClass(worker_id=-1)
-
-
-@router.get("/worker/{worker_name}/queue", response=List[QueueItemSchema], url_name="get_worker_queue")
-def get_worker_queue(request, worker_name: str, limit: int = 100):
- """Get the current queue for a specific worker type."""
- from archivebox.workers.worker import WORKER_TYPES
-
- if worker_name not in WORKER_TYPES:
- from ninja.errors import HttpError
- raise HttpError(404, f"Unknown worker type: {worker_name}. Valid types: {list(WORKER_TYPES.keys())}")
-
- WorkerClass = WORKER_TYPES[worker_name]
- worker = WorkerClass(worker_id=-1)
- return list(worker.get_queue()[:limit])
-
-
# Progress endpoint moved to core.views.live_progress_view for simplicity
diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py
index 5043f3ed..25b0815b 100644
--- a/archivebox/cli/archivebox_add.py
+++ b/archivebox/cli/archivebox_add.py
@@ -96,10 +96,9 @@ def add(urls: str | list[str],
first_url = crawl.get_urls_list()[0] if crawl.get_urls_list() else ''
print(f' [dim]First URL: {first_url}[/dim]')
- # 3. The CrawlMachine will create the root Snapshot when started
- # If URLs are from a file: first URL = file:///path/to/sources/...txt
- # Parser extractors will run on it and discover more URLs
- # Those URLs become child Snapshots (depth=1)
+ # 3. The CrawlMachine will create Snapshots from all URLs when started
+ # Parser extractors run on snapshots and discover more URLs
+ # Discovered URLs become child Snapshots (depth+1)
if index_only:
# Just create the crawl but don't start processing
@@ -119,10 +118,9 @@ def add(urls: str | list[str],
# 5. Start the orchestrator to process the queue
# The orchestrator will:
- # - Process Crawl -> create root Snapshot
- # - Process root Snapshot -> run parser extractors -> discover URLs
- # - Create child Snapshots from discovered URLs
- # - Process child Snapshots -> run extractors
+ # - Process Crawl -> create Snapshots from all URLs
+ # - Process Snapshots -> run extractors
+ # - Parser extractors discover new URLs -> create child Snapshots
# - Repeat until max_depth reached
if bg:
diff --git a/archivebox/cli/archivebox_init.py b/archivebox/cli/archivebox_init.py
index 5ef6c9ca..34b10faa 100755
--- a/archivebox/cli/archivebox_init.py
+++ b/archivebox/cli/archivebox_init.py
@@ -160,10 +160,12 @@ def init(force: bool=False, quick: bool=False, install: bool=False) -> None:
CONSTANTS.PERSONAS_DIR.mkdir(parents=True, exist_ok=True)
CONSTANTS.DEFAULT_TMP_DIR.mkdir(parents=True, exist_ok=True)
CONSTANTS.DEFAULT_LIB_DIR.mkdir(parents=True, exist_ok=True)
-
+ (CONSTANTS.DEFAULT_LIB_DIR / 'bin').mkdir(parents=True, exist_ok=True)
+
from archivebox.config.common import STORAGE_CONFIG
STORAGE_CONFIG.TMP_DIR.mkdir(parents=True, exist_ok=True)
STORAGE_CONFIG.LIB_DIR.mkdir(parents=True, exist_ok=True)
+ (STORAGE_CONFIG.LIB_DIR / 'bin').mkdir(parents=True, exist_ok=True)
if install:
from archivebox.cli.archivebox_install import install as install_method
diff --git a/archivebox/cli/archivebox_pluginmap.py b/archivebox/cli/archivebox_pluginmap.py
index b168a480..04a8cba6 100644
--- a/archivebox/cli/archivebox_pluginmap.py
+++ b/archivebox/cli/archivebox_pluginmap.py
@@ -96,33 +96,45 @@ ARCHIVERESULT_MACHINE_DIAGRAM = """
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
-│ │ QUEUED │◄────────────────┐ │
-│ │ (initial) │ │ │
-│ └──────┬──────┘ │ │
-│ │ │ tick() unless can_start() │
-│ │ tick() when │ │
-│ │ can_start() │ │
-│ ▼ │ │
-│ ┌─────────────┐ │ │
-│ │ STARTED │─────────────────┘ │
-│ │ │◄────────────────┐ │
-│ │ enter: │ │ tick() unless is_finished() │
-│ │ result.run()│─────────────────┘ │
-│ │ (execute │ │
-│ │ hook via │ │
-│ │ run_hook())│ │
-│ └──────┬──────┘ │
-│ │ │
-│ │ tick() checks status set by hook output │
-│ ├────────────────┬────────────────┬────────────────┐ │
-│ ▼ ▼ ▼ ▼ │
-│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
-│ │ SUCCEEDED │ │ FAILED │ │ SKIPPED │ │ BACKOFF │ │
-│ │ (final) │ │ (final) │ │ (final) │ │ │ │
-│ └───────────┘ └───────────┘ └───────────┘ └─────┬─────┘ │
-│ │ │
-│ can_start()───┘ │
-│ loops back to STARTED │
+│ │ QUEUED │◄─────────────────┐ │
+│ │ (initial) │ │ │
+│ └──┬───────┬──┘ │ │
+│ │ │ │ tick() unless can_start() │
+│ │ │ exceeded_max_ │ │
+│ │ │ attempts │ │
+│ │ ▼ │ │
+│ │ ┌──────────┐ │ │
+│ │ │ SKIPPED │ │ │
+│ │ │ (final) │ │ │
+│ │ └──────────┘ │ │
+│ │ tick() when │ │
+│ │ can_start() │ │
+│ ▼ │ │
+│ ┌─────────────┐ │ │
+│ │ STARTED │──────────────────┘ │
+│ │ │◄─────────────────────────────────────────────────┐ │
+│ │ enter: │ │ │ │
+│ │ result.run()│ tick() unless │ │ │
+│ │ (execute │ is_finished() │ │ │
+│ │ hook via │──────────────────────┘ │ │
+│ │ run_hook())│ │ │
+│ └──────┬──────┘ │ │
+│ │ │ │
+│ │ tick() checks status set by hook output │ │
+│ ├─────────────┬─────────────┬─────────────┐ │ │
+│ ▼ ▼ ▼ ▼ │ │
+│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │
+│ │ SUCCEEDED │ │ FAILED │ │ SKIPPED │ │ BACKOFF │ │ │
+│ │ (final) │ │ (final) │ │ (final) │ │ │ │ │
+│ └───────────┘ └───────────┘ └───────────┘ └──┬──────┬─┘ │ │
+│ │ │ │ │
+│ exceeded_max_ │ │ can_start()│ │
+│ attempts │ │ loops back │ │
+│ ▼ │ └────────────┘ │
+│ ┌──────────┐ │ │
+│ │ SKIPPED │◄─┘ │
+│ │ (final) │ │
+│ └──────────┘ │
│ │
│ Each ArchiveResult runs ONE specific hook (stored in .hook_name field) │
└─────────────────────────────────────────────────────────────────────────────┘
@@ -137,35 +149,38 @@ BINARY_MACHINE_DIAGRAM = """
│ │ QUEUED │◄────────────────┐ │
│ │ (initial) │ │ │
│ └──────┬──────┘ │ │
-│ │ │ tick() unless can_start() │
+│ │ │ tick() unless can_install() │
+│ │ │ (stays queued if failed) │
│ │ tick() when │ │
-│ │ can_start() │ │
-│ ▼ │ │
-│ ┌─────────────┐ │ │
-│ │ STARTED │─────────────────┘ │
-│ │ │◄────────────────┐ │
-│ │ enter: │ │ │
-│ │ binary.run()│ │ tick() unless is_finished() │
-│ │ (discover │─────────────────┘ │
-│ │ Binary │ │
-│ │ hooks, │ │
-│ │ try each │ │
-│ │ provider) │ │
-│ └──────┬──────┘ │
+│ │ can_install() │ │
+│ │ │ │
+│ │ on_install() runs │ │
+│ │ during transition: │ │
+│ │ • binary.run() │ │
+│ │ (discover Binary │ │
+│ │ hooks, try each │ │
+│ │ provider until │ │
+│ │ one succeeds) │ │
+│ │ • Sets abspath, │ │
+│ │ version, sha256 │ │
+│ │ │ │
+│ │ If install fails: │ │
+│ │ raises exception──────┘ │
+│ │ (retry_at bumped) │
│ │ │
-│ │ tick() checks status set by hook output │
-│ ├────────────────────────────────┐ │
-│ ▼ ▼ │
-│ ┌─────────────┐ ┌─────────────┐ │
-│ │ SUCCEEDED │ │ FAILED │ │
-│ │ (final) │ │ (final) │ │
-│ │ │ │ │ │
-│ │ abspath, │ │ no provider │ │
-│ │ version set │ │ succeeded │ │
-│ └─────────────┘ └─────────────┘ │
+│ ▼ │
+│ ┌─────────────┐ │
+│ │ INSTALLED │ │
+│ │ (final) │ │
+│ │ │ │
+│ │ Binary is │ │
+│ │ ready to │ │
+│ │ use │ │
+│ └─────────────┘ │
│ │
-│ Hooks triggered: on_Binary__* (provider hooks during STARTED.enter) │
+│ Hooks triggered: on_Binary__* (provider hooks during transition) │
│ Providers tried in sequence until one succeeds: apt, brew, pip, npm, etc. │
+│ Installation is synchronous - no intermediate STARTED state │
└─────────────────────────────────────────────────────────────────────────────┘
"""
diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py
index 2fbd05c0..f780a289 100644
--- a/archivebox/cli/archivebox_update.py
+++ b/archivebox/cli/archivebox_update.py
@@ -109,15 +109,18 @@ def drain_old_archive_dirs(resume_from: str = None, batch_size: int = 100) -> di
if not archive_dir.exists():
return stats
- print('[*] Scanning for old directories in archive/...')
+ print('[DEBUG Phase1] Scanning for old directories in archive/...')
# Scan for real directories only (skip symlinks - they're already migrated)
+ all_entries = list(os.scandir(archive_dir))
+ print(f'[DEBUG Phase1] Total entries in archive/: {len(all_entries)}')
entries = [
(e.stat().st_mtime, e.path)
- for e in os.scandir(archive_dir)
+ for e in all_entries
if e.is_dir(follow_symlinks=False) # Skip symlinks
]
entries.sort(reverse=True) # Newest first
+ print(f'[DEBUG Phase1] Real directories (not symlinks): {len(entries)}')
print(f'[*] Found {len(entries)} old directories to drain')
for mtime, entry_path in entries:
@@ -142,14 +145,48 @@ def drain_old_archive_dirs(resume_from: str = None, batch_size: int = 100) -> di
print(f" [{stats['processed']}] Invalid: {entry_path.name}")
continue
+ # Ensure snapshot has a valid crawl (migration 0024 may have failed)
+ from archivebox.crawls.models import Crawl
+ has_valid_crawl = False
+ if snapshot.crawl_id:
+ # Check if the crawl actually exists
+ has_valid_crawl = Crawl.objects.filter(id=snapshot.crawl_id).exists()
+
+ if not has_valid_crawl:
+ # Create a new crawl (created_by will default to system user)
+ crawl = Crawl.objects.create(urls=snapshot.url)
+ # Use queryset update to avoid triggering save() hooks
+ from archivebox.core.models import Snapshot as SnapshotModel
+ SnapshotModel.objects.filter(pk=snapshot.pk).update(crawl=crawl)
+ # Refresh the instance
+ snapshot.crawl = crawl
+ snapshot.crawl_id = crawl.id
+ print(f"[DEBUG Phase1] Created missing crawl for snapshot {str(snapshot.id)[:8]}")
+
# Check if needs migration (0.8.x → 0.9.x)
+ print(f"[DEBUG Phase1] Snapshot {str(snapshot.id)[:8]}: fs_version={snapshot.fs_version}, needs_migration={snapshot.fs_migration_needed}")
if snapshot.fs_migration_needed:
try:
- # Manually trigger filesystem migration without full save()
- # This avoids UNIQUE constraint issues while still migrating files
- cleanup_info = None
- if hasattr(snapshot, '_fs_migrate_from_0_8_0_to_0_9_0'):
- cleanup_info = snapshot._fs_migrate_from_0_8_0_to_0_9_0()
+ # Calculate paths using actual directory (entry_path), not snapshot.timestamp
+ # because snapshot.timestamp might be truncated
+ old_dir = entry_path
+ new_dir = snapshot.get_storage_path_for_version('0.9.0')
+ print(f"[DEBUG Phase1] Migrating {old_dir.name} → {new_dir}")
+
+ # Manually migrate files
+ if not new_dir.exists() and old_dir.exists():
+ new_dir.mkdir(parents=True, exist_ok=True)
+ import shutil
+ file_count = 0
+ for old_file in old_dir.rglob('*'):
+ if old_file.is_file():
+ rel_path = old_file.relative_to(old_dir)
+ new_file = new_dir / rel_path
+ if not new_file.exists():
+ new_file.parent.mkdir(parents=True, exist_ok=True)
+ shutil.copy2(old_file, new_file)
+ file_count += 1
+ print(f"[DEBUG Phase1] Copied {file_count} files")
# Update only fs_version field using queryset update (bypasses validation)
from archivebox.core.models import Snapshot as SnapshotModel
@@ -158,9 +195,8 @@ def drain_old_archive_dirs(resume_from: str = None, batch_size: int = 100) -> di
# Commit the transaction
transaction.commit()
- # Manually call cleanup since we bypassed normal save() flow
- if cleanup_info:
- old_dir, new_dir = cleanup_info
+ # Cleanup: delete old dir and create symlink
+ if old_dir.exists() and old_dir != new_dir:
snapshot._cleanup_old_migration_dir(old_dir, new_dir)
stats['migrated'] += 1
@@ -207,19 +243,39 @@ def process_all_db_snapshots(batch_size: int = 100) -> dict:
continue
try:
- # Reconcile index.json with DB
- snapshot.reconcile_with_index_json()
+ print(f"[DEBUG Phase2] Snapshot {str(snapshot.id)[:8]}: fs_version={snapshot.fs_version}, needs_migration={snapshot.fs_migration_needed}")
+
+ # Check if snapshot has a directory on disk
+ from pathlib import Path
+ output_dir = Path(snapshot.output_dir)
+ has_directory = output_dir.exists() and output_dir.is_dir()
+
+ # Only reconcile if directory exists (don't create empty directories for orphans)
+ if has_directory:
+ snapshot.reconcile_with_index_json()
# Clean up invalid field values from old migrations
if not isinstance(snapshot.current_step, int):
snapshot.current_step = 0
+ # If still needs migration, it's an orphan (no directory on disk)
+ # Mark it as migrated to prevent save() from triggering filesystem migration
+ if snapshot.fs_migration_needed:
+ if has_directory:
+ print(f"[DEBUG Phase2] WARNING: Snapshot {str(snapshot.id)[:8]} has directory but still needs migration")
+ else:
+ print(f"[DEBUG Phase2] Orphan snapshot {str(snapshot.id)[:8]} - marking as migrated without filesystem operation")
+ # Use queryset update to set fs_version without triggering save() hooks
+ from archivebox.core.models import Snapshot as SnapshotModel
+ SnapshotModel.objects.filter(pk=snapshot.pk).update(fs_version='0.9.0')
+ snapshot.fs_version = '0.9.0'
+
# Queue for archiving (state machine will handle it)
snapshot.status = Snapshot.StatusChoices.QUEUED
snapshot.retry_at = timezone.now()
snapshot.save()
- stats['reconciled'] += 1
+ stats['reconciled'] += 1 if has_directory else 0
stats['queued'] += 1
except Exception as e:
# Skip snapshots that can't be processed (e.g., missing crawl)
diff --git a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py
index 5b0666c5..8ac9d889 100644
--- a/archivebox/core/migrations/0027_copy_archiveresult_to_process.py
+++ b/archivebox/core/migrations/0027_copy_archiveresult_to_process.py
@@ -4,6 +4,7 @@
from django.db import migrations, connection
import json
from pathlib import Path
+from archivebox.uuid_compat import uuid7
def parse_cmd_field(cmd_raw):
@@ -39,7 +40,6 @@ def parse_cmd_field(cmd_raw):
def get_or_create_current_machine(cursor):
"""Get or create Machine.current() using raw SQL."""
- import uuid
import socket
from datetime import datetime
@@ -55,7 +55,8 @@ def get_or_create_current_machine(cursor):
return row[0]
# Create new machine
- machine_id = str(uuid.uuid4())
+ # Django UUIDField stores UUIDs as 32-char hex (no dashes) in SQLite
+ machine_id = uuid7().hex
now = datetime.now().isoformat()
# Check which columns exist (schema differs between 0.8.x and 0.9.x)
@@ -103,7 +104,6 @@ def get_or_create_binary(cursor, machine_id, name, abspath, version):
Returns:
binary_id (str)
"""
- import uuid
from datetime import datetime
# If abspath is just a name without slashes, it's not a full path
@@ -123,7 +123,8 @@ def get_or_create_binary(cursor, machine_id, name, abspath, version):
return row[0]
# Create new binary
- binary_id = str(uuid.uuid4())
+ # Django UUIDField stores UUIDs as 32-char hex (no dashes) in SQLite
+ binary_id = uuid7().hex
now = datetime.now().isoformat()
# Check which columns exist (schema differs between 0.8.x and 0.9.x)
@@ -186,10 +187,10 @@ def create_process(cursor, machine_id, pwd, cmd, status, exit_code, started_at,
Returns:
process_id (str)
"""
- import uuid
from datetime import datetime
- process_id = str(uuid.uuid4())
+ # Django UUIDField stores UUIDs as 32-char hex (no dashes) in SQLite
+ process_id = uuid7().hex
now = datetime.now().isoformat()
# Convert cmd array to JSON
diff --git a/archivebox/core/migrations/0028_alter_snapshot_fs_version.py b/archivebox/core/migrations/0028_alter_snapshot_fs_version.py
new file mode 100644
index 00000000..eb86883d
--- /dev/null
+++ b/archivebox/core/migrations/0028_alter_snapshot_fs_version.py
@@ -0,0 +1,18 @@
+# Generated by Django 6.0 on 2026-01-02 08:43
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('core', '0027_copy_archiveresult_to_process'),
+ ]
+
+ operations = [
+ migrations.AlterField(
+ model_name='snapshot',
+ name='fs_version',
+ field=models.CharField(default='0.9.0', help_text='Filesystem version of this snapshot (e.g., "0.7.0", "0.8.0", "0.9.0"). Used to trigger lazy migration on save().', max_length=10),
+ ),
+ ]
diff --git a/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py b/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py
new file mode 100644
index 00000000..36b9f14c
--- /dev/null
+++ b/archivebox/core/migrations/0029_migrate_archiveresult_to_uuid_pk.py
@@ -0,0 +1,181 @@
+# Generated by hand on 2026-01-02
+# Migrate ArchiveResult from integer PK to UUID PK (matching Snapshot)
+
+from django.db import migrations, models, connection
+from uuid import UUID
+from archivebox.uuid_compat import uuid7
+
+
+def migrate_archiveresult_id_to_uuid(apps, schema_editor):
+ """
+ Migrate ArchiveResult from integer PK to UUID PK.
+
+ Strategy:
+ 1. Add old_id field to store current integer IDs
+ 2. Generate UUIDs for any records missing them
+ 3. Swap id and uuid fields (uuid becomes PK, old integer id becomes old_id)
+ """
+ cursor = connection.cursor()
+
+ # Check if table exists and has data
+ cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='core_archiveresult'")
+ if not cursor.fetchone():
+ print('ArchiveResult table does not exist, skipping migration')
+ return
+
+ cursor.execute("SELECT COUNT(*) FROM core_archiveresult")
+ row_count = cursor.fetchone()[0]
+
+ if row_count == 0:
+ print('No ArchiveResult records to migrate')
+ return
+
+ print(f'Migrating {row_count} ArchiveResult records from integer PK to UUID PK...')
+
+ # Step 0: Check if machine_process table exists, if not NULL out process_id values
+ cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='machine_process'")
+ machine_process_exists = cursor.fetchone() is not None
+
+ if not machine_process_exists:
+ print('machine_process table does not exist yet, setting process_id to NULL')
+ cursor.execute("UPDATE core_archiveresult SET process_id = NULL WHERE process_id IS NOT NULL")
+
+ # Step 1: Create new table with UUID as primary key
+ cursor.execute("""
+ CREATE TABLE core_archiveresult_new (
+ id TEXT PRIMARY KEY NOT NULL,
+ old_id INTEGER,
+ uuid TEXT UNIQUE,
+ created_at DATETIME NOT NULL,
+ modified_at DATETIME NOT NULL,
+
+ snapshot_id TEXT NOT NULL,
+ plugin VARCHAR(32) NOT NULL,
+ hook_name VARCHAR(255) NOT NULL DEFAULT '',
+
+ status VARCHAR(15) NOT NULL DEFAULT 'queued',
+ retry_at DATETIME,
+
+ start_ts DATETIME,
+ end_ts DATETIME,
+
+ output_str TEXT NOT NULL DEFAULT '',
+ output_json TEXT,
+ output_files TEXT NOT NULL DEFAULT '{}',
+ output_size BIGINT NOT NULL DEFAULT 0,
+ output_mimetypes VARCHAR(512) NOT NULL DEFAULT '',
+
+ config TEXT NOT NULL DEFAULT '{}',
+ notes TEXT NOT NULL DEFAULT '',
+ num_uses_succeeded INTEGER NOT NULL DEFAULT 0,
+ num_uses_failed INTEGER NOT NULL DEFAULT 0,
+
+ process_id TEXT,
+
+ FOREIGN KEY (snapshot_id) REFERENCES core_snapshot(id) ON DELETE CASCADE,
+ FOREIGN KEY (process_id) REFERENCES machine_process(id) ON DELETE SET NULL
+ );
+ """)
+
+ # Step 2: Generate UUIDs for records that don't have them
+ cursor.execute("SELECT id, uuid FROM core_archiveresult")
+ records = cursor.fetchall()
+
+ id_to_uuid = {}
+ for old_id, existing_uuid in records:
+ if existing_uuid:
+ # Normalize existing UUID to 32-char hex format (Django SQLite UUIDField format)
+ # (existing UUIDs might be stored with or without dashes in old schema)
+ id_to_uuid[old_id] = UUID(existing_uuid).hex
+ else:
+ # Generate new UUIDv7 (time-ordered) as 32-char hex
+ id_to_uuid[old_id] = uuid7().hex
+
+ # Step 3: Copy data with UUIDs as new primary key
+ cursor.execute("SELECT * FROM core_archiveresult")
+ old_records = cursor.fetchall()
+
+ # Get column names
+ cursor.execute("PRAGMA table_info(core_archiveresult)")
+ columns = cursor.fetchall()
+ col_names = [col[1] for col in columns]
+
+ for i, record in enumerate(old_records):
+ old_id = record[col_names.index('id')]
+ new_uuid = id_to_uuid[old_id]
+
+ # Build insert with new structure
+ values = {col_names[i]: record[i] for i in range(len(col_names))}
+
+ # Check which fields exist in new table
+ fields_to_copy = [
+ 'created_at', 'modified_at', 'snapshot_id', 'plugin', 'hook_name',
+ 'status', 'retry_at', 'start_ts', 'end_ts',
+ 'output_str', 'output_json', 'output_files', 'output_size', 'output_mimetypes',
+ 'config', 'notes', 'num_uses_succeeded', 'num_uses_failed', 'process_id'
+ ]
+
+ # Build INSERT statement
+ existing_fields = [f for f in fields_to_copy if f in values]
+ placeholders = ', '.join(['?'] * (len(existing_fields) + 3)) # +3 for id, old_id, uuid
+ field_list = 'id, old_id, uuid, ' + ', '.join(existing_fields)
+
+ insert_values = [new_uuid, old_id, new_uuid] + [values.get(f) for f in existing_fields]
+
+ cursor.execute(
+ f"INSERT INTO core_archiveresult_new ({field_list}) VALUES ({placeholders})",
+ insert_values
+ )
+
+ # Step 4: Replace old table with new table
+ cursor.execute("DROP TABLE core_archiveresult")
+ cursor.execute("ALTER TABLE core_archiveresult_new RENAME TO core_archiveresult")
+
+ # Step 5: Create indexes
+ cursor.execute("CREATE INDEX core_archiveresult_snapshot_id_idx ON core_archiveresult(snapshot_id)")
+ cursor.execute("CREATE INDEX core_archiveresult_plugin_idx ON core_archiveresult(plugin)")
+ cursor.execute("CREATE INDEX core_archiveresult_status_idx ON core_archiveresult(status)")
+ cursor.execute("CREATE INDEX core_archiveresult_retry_at_idx ON core_archiveresult(retry_at)")
+ cursor.execute("CREATE INDEX core_archiveresult_created_at_idx ON core_archiveresult(created_at)")
+ cursor.execute("CREATE INDEX core_archiveresult_hook_name_idx ON core_archiveresult(hook_name)")
+ cursor.execute("CREATE INDEX core_archiveresult_process_id_idx ON core_archiveresult(process_id)")
+ cursor.execute("CREATE INDEX core_archiveresult_old_id_idx ON core_archiveresult(old_id)")
+
+ print(f'✓ Migrated {row_count} ArchiveResult records to UUID primary key')
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('core', '0028_alter_snapshot_fs_version'),
+ ]
+
+ operations = [
+ migrations.SeparateDatabaseAndState(
+ database_operations=[
+ migrations.RunPython(
+ migrate_archiveresult_id_to_uuid,
+ reverse_code=migrations.RunPython.noop,
+ ),
+ ],
+ state_operations=[
+ # Remove old uuid field
+ migrations.RemoveField(
+ model_name='archiveresult',
+ name='uuid',
+ ),
+ # Change id from AutoField to UUIDField
+ migrations.AlterField(
+ model_name='archiveresult',
+ name='id',
+ field=models.UUIDField(primary_key=True, default=uuid7, editable=False, unique=True),
+ ),
+ # Add old_id field to preserve legacy integer IDs
+ migrations.AddField(
+ model_name='archiveresult',
+ name='old_id',
+ field=models.IntegerField(null=True, blank=True, db_index=True, help_text='Legacy integer ID from pre-0.9.0 versions'),
+ ),
+ ],
+ ),
+ ]
diff --git a/archivebox/core/migrations/0030_alter_archiveresult_id.py b/archivebox/core/migrations/0030_alter_archiveresult_id.py
new file mode 100644
index 00000000..0c5e54b0
--- /dev/null
+++ b/archivebox/core/migrations/0030_alter_archiveresult_id.py
@@ -0,0 +1,19 @@
+# Generated by Django 6.0 on 2026-01-02 10:02
+
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('core', '0029_migrate_archiveresult_to_uuid_pk'),
+ ]
+
+ operations = [
+ migrations.AlterField(
+ model_name='archiveresult',
+ name='id',
+ field=models.UUIDField(default=uuid.uuid7, editable=False, primary_key=True, serialize=False, unique=True),
+ ),
+ ]
diff --git a/archivebox/core/models.py b/archivebox/core/models.py
index 3a21041a..3de5b4f8 100755
--- a/archivebox/core/models.py
+++ b/archivebox/core/models.py
@@ -362,6 +362,7 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
# Migrate filesystem if needed (happens automatically on save)
if self.pk and self.fs_migration_needed:
+ print(f"[DEBUG save()] Triggering filesystem migration for {str(self.id)[:8]}: {self.fs_version} → {self._fs_current_version()}")
# Walk through migration chain automatically
current = self.fs_version
target = self._fs_current_version()
@@ -372,6 +373,7 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
# Only run if method exists (most are no-ops)
if hasattr(self, method):
+ print(f"[DEBUG save()] Running {method}()")
getattr(self, method)()
current = next_ver
@@ -449,10 +451,18 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
old_dir = self.get_storage_path_for_version('0.8.0')
new_dir = self.get_storage_path_for_version('0.9.0')
- if not old_dir.exists() or old_dir == new_dir or new_dir.exists():
- # Even if no directory migration needed, still convert index format
- self.convert_index_json_to_jsonl()
- return
+ print(f"[DEBUG _fs_migrate] {self.timestamp}: old_exists={old_dir.exists()}, same={old_dir == new_dir}, new_exists={new_dir.exists()}")
+
+ if not old_dir.exists() or old_dir == new_dir:
+ # No migration needed
+ print(f"[DEBUG _fs_migrate] Returning None (early return)")
+ return None
+
+ if new_dir.exists():
+ # New directory already exists (files already copied), but we still need cleanup
+ # Return cleanup info so old directory can be cleaned up
+ print(f"[DEBUG _fs_migrate] Returning cleanup info (new_dir exists)")
+ return (old_dir, new_dir)
new_dir.mkdir(parents=True, exist_ok=True)
@@ -495,47 +505,32 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
def _cleanup_old_migration_dir(self, old_dir: Path, new_dir: Path):
"""
Delete old directory and create symlink after successful migration.
- Called via transaction.on_commit() after DB commit succeeds.
"""
import shutil
import logging
- print(f"[DEBUG] _cleanup_old_migration_dir called: old_dir={old_dir}, new_dir={new_dir}")
-
# Delete old directory
if old_dir.exists() and not old_dir.is_symlink():
- print(f"[DEBUG] Attempting to delete old directory: {old_dir}")
try:
shutil.rmtree(old_dir)
- print(f"[DEBUG] Successfully deleted old directory: {old_dir}")
except Exception as e:
- # Log but don't raise - migration succeeded, this is just cleanup
- print(f"[DEBUG] Failed to delete old directory {old_dir}: {e}")
logging.getLogger('archivebox.migration').warning(
f"Could not remove old migration directory {old_dir}: {e}"
)
return # Don't create symlink if cleanup failed
- else:
- print(f"[DEBUG] Old directory doesn't exist or is already a symlink: {old_dir}")
# Create backwards-compat symlink (after old dir is deleted)
symlink_path = old_dir # Same path as old_dir
if symlink_path.is_symlink():
- print(f"[DEBUG] Unlinking existing symlink: {symlink_path}")
symlink_path.unlink()
if not symlink_path.exists():
- print(f"[DEBUG] Creating symlink: {symlink_path} -> {new_dir}")
try:
symlink_path.symlink_to(new_dir, target_is_directory=True)
- print(f"[DEBUG] Successfully created symlink")
except Exception as e:
- print(f"[DEBUG] Failed to create symlink: {e}")
logging.getLogger('archivebox.migration').warning(
f"Could not create symlink from {symlink_path} to {new_dir}: {e}"
)
- else:
- print(f"[DEBUG] Symlink path already exists: {symlink_path}")
# =========================================================================
# Path Calculation and Migration Helpers
@@ -660,13 +655,28 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
if not timestamp:
return None
- # Look up existing
+ # Look up existing (try exact match first, then fuzzy match for truncated timestamps)
try:
- return cls.objects.get(url=url, timestamp=timestamp)
+ snapshot = cls.objects.get(url=url, timestamp=timestamp)
+ print(f"[DEBUG load_from_directory] Found existing snapshot for {url} @ {timestamp}: {str(snapshot.id)[:8]}")
+ return snapshot
except cls.DoesNotExist:
+ print(f"[DEBUG load_from_directory] NOT FOUND (exact): {url} @ {timestamp}")
+ # Try fuzzy match - index.json may have truncated timestamp
+ # e.g., index has "1767000340" but DB has "1767000340.624737"
+ candidates = cls.objects.filter(url=url, timestamp__startswith=timestamp)
+ if candidates.count() == 1:
+ snapshot = candidates.first()
+ print(f"[DEBUG load_from_directory] Found via fuzzy match: {snapshot.timestamp}")
+ return snapshot
+ elif candidates.count() > 1:
+ print(f"[DEBUG load_from_directory] Multiple fuzzy matches, using first")
+ return candidates.first()
+ print(f"[DEBUG load_from_directory] NOT FOUND (fuzzy): {url} @ {timestamp}")
return None
except cls.MultipleObjectsReturned:
# Should not happen with unique constraint
+ print(f"[DEBUG load_from_directory] Multiple snapshots found for {url} @ {timestamp}")
return cls.objects.filter(url=url, timestamp=timestamp).first()
@classmethod
@@ -1668,83 +1678,20 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea
return archiveresults
- def advance_step_if_ready(self) -> bool:
- """
- Advance current_step if all foreground hooks in current step are finished.
-
- Called by the state machine to check if step can advance.
- Background hooks (.bg) don't block step advancement.
-
- Step advancement rules:
- - All foreground ARs in current step must be finished (SUCCEEDED/FAILED/SKIPPED)
- - Background ARs (hook_name contains '.bg.') are ignored for advancement
- - When ready, increments current_step by 1 (up to 9)
-
- Returns:
- True if step was advanced, False if not ready or already at step 9.
- """
- from archivebox.hooks import extract_step, is_background_hook
-
- if self.current_step >= 9:
- return False # Already at final step
-
- # Get all ARs for current step that are foreground
- current_step_ars = self.archiveresult_set.filter(
- hook_name__isnull=False
- ).exclude(hook_name='')
-
- # Check each AR in current step
- for ar in current_step_ars:
- ar_step = extract_step(ar.hook_name)
- if ar_step != self.current_step:
- continue # Not in current step
-
- if is_background_hook(ar.hook_name):
- continue # Background hooks don't block
-
- # Foreground hook in current step - check if finished
- if ar.status not in ArchiveResult.FINAL_OR_ACTIVE_STATES:
- # Still pending/queued - can't advance
- return False
-
- if ar.status == ArchiveResult.StatusChoices.STARTED:
- # Still running - can't advance
- return False
-
- # All foreground hooks in current step are finished - advance!
- self.current_step += 1
- self.save(update_fields=['current_step', 'modified_at'])
- return True
def is_finished_processing(self) -> bool:
"""
- Check if this snapshot has finished processing.
+ Check if all ArchiveResults are finished.
- Used by SnapshotMachine.is_finished() to determine if snapshot is complete.
-
- Returns:
- True if all archiveresults are finished (or no work to do), False otherwise.
+ Note: This is only called for observability/progress tracking.
+ SnapshotWorker owns the execution and doesn't poll this.
"""
- # if no archiveresults exist yet, it's not finished
- if not self.archiveresult_set.exists():
- return False
+ # Check if any ARs are still pending/started
+ pending = self.archiveresult_set.exclude(
+ status__in=ArchiveResult.FINAL_OR_ACTIVE_STATES
+ ).exists()
- # Try to advance step if ready (handles step-based hook execution)
- # This will increment current_step when all foreground hooks in current step are done
- while self.advance_step_if_ready():
- pass # Keep advancing until we can't anymore
-
- # if archiveresults exist but are still pending, it's not finished
- if self.pending_archiveresults().exists():
- return False
-
- # Don't wait for background hooks - they'll be cleaned up on entering sealed state
- # Background hooks in STARTED state are excluded by pending_archiveresults()
- # (STARTED is in FINAL_OR_ACTIVE_STATES) so once all results are FINAL or ACTIVE,
- # we can transition to sealed and cleanup() will kill the background hooks
-
- # otherwise archiveresults exist and are all finished, so it's finished
- return True
+ return not pending
def get_progress_stats(self) -> dict:
"""
@@ -2242,7 +2189,6 @@ class SnapshotMachine(BaseStateMachine, strict_states=True):
tick = (
queued.to.itself(unless='can_start') |
queued.to(started, cond='can_start') |
- started.to.itself(unless='is_finished') |
started.to(sealed, cond='is_finished')
)
@@ -2253,6 +2199,10 @@ class SnapshotMachine(BaseStateMachine, strict_states=True):
can_start = bool(self.snapshot.url)
return can_start
+ def is_finished(self) -> bool:
+ """Check if all ArchiveResults for this snapshot are finished."""
+ return self.snapshot.is_finished_processing()
+
@queued.enter
def enter_queued(self):
self.snapshot.update_and_requeue(
@@ -2262,29 +2212,10 @@ class SnapshotMachine(BaseStateMachine, strict_states=True):
@started.enter
def enter_started(self):
- import sys
-
- print(f'[cyan] 🔄 SnapshotMachine.enter_started() - creating archiveresults for {self.snapshot.url}[/cyan]', file=sys.stderr)
-
- # Run the snapshot - creates pending archiveresults for all enabled plugins
- self.snapshot.run()
-
- # Check if any archiveresults were created
- ar_count = self.snapshot.archiveresult_set.count()
- print(f'[cyan] 🔄 ArchiveResult count: {ar_count}[/cyan]', file=sys.stderr)
-
- if ar_count == 0:
- # No archiveresults created, seal immediately
- print(f'[cyan] 🔄 No archiveresults created, sealing snapshot immediately[/cyan]', file=sys.stderr)
- self.seal()
- else:
- # Set status = started with retry_at far future (so workers don't claim us - we're waiting for ARs)
- # Last AR will manually call self.seal() when done
- self.snapshot.update_and_requeue(
- retry_at=timezone.now() + timedelta(days=365),
- status=Snapshot.StatusChoices.STARTED,
- )
- print(f'[cyan] 🔄 {ar_count} archiveresults created, waiting for them to finish[/cyan]', file=sys.stderr)
+ """Just mark as started - SnapshotWorker will create ARs and run hooks."""
+ self.snapshot.status = Snapshot.StatusChoices.STARTED
+ self.snapshot.retry_at = None # No more polling
+ self.snapshot.save(update_fields=['status', 'retry_at', 'modified_at'])
@sealed.enter
def enter_sealed(self):
@@ -2329,12 +2260,11 @@ class ArchiveResult(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWi
plugins = [get_plugin_name(e) for e in get_plugins()]
return tuple((e, e) for e in plugins)
- # Keep AutoField for backward compatibility with 0.7.x databases
- # UUID field is added separately by migration for new records
- id = models.AutoField(primary_key=True, editable=False)
- # Note: unique constraint is added by migration 0027 - don't set unique=True here
- # or SQLite table recreation in earlier migrations will fail
- uuid = models.UUIDField(default=uuid7, null=True, blank=True, db_index=True)
+ # UUID primary key (migrated from integer in 0029)
+ id = models.UUIDField(primary_key=True, default=uuid7, editable=False, unique=True)
+ # old_id preserves the legacy integer ID for backward compatibility
+ old_id = models.IntegerField(null=True, blank=True, db_index=True, help_text='Legacy integer ID from pre-0.9.0 versions')
+ # Note: uuid field was removed in migration 0029 when id became UUID
created_at = models.DateTimeField(default=timezone.now, db_index=True)
modified_at = models.DateTimeField(auto_now=True)
@@ -2684,13 +2614,11 @@ class ArchiveResult(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWi
plugin_dir = Path(self.snapshot.output_dir) / self.plugin
start_ts = timezone.now()
- is_bg_hook = False
+ process = None
for hook in hooks:
- # Check if this is a background hook
- is_bg_hook = is_background_hook(hook.name)
-
- result = run_hook(
+ # Run hook using Process.launch() - returns Process model
+ process = run_hook(
hook,
output_dir=plugin_dir,
config=config,
@@ -2700,27 +2628,25 @@ class ArchiveResult(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWi
depth=self.snapshot.depth,
)
- # Background hooks return None
- if result is None:
- is_bg_hook = True
-
- # Update status based on hook execution
- if is_bg_hook:
- # BACKGROUND HOOK - still running, return immediately
- # Status stays STARTED, will be finalized by Snapshot.cleanup()
- self.status = self.StatusChoices.STARTED
+ # Link ArchiveResult to Process
+ self.process = process
self.start_ts = start_ts
- if self.process_id:
- self.process.pwd = str(plugin_dir)
- self.process.save()
+ self.save(update_fields=['process_id', 'start_ts', 'modified_at'])
+
+ if not process:
+ # No hooks ran
+ self.status = self.StatusChoices.FAILED
+ self.output_str = 'No hooks executed'
self.save()
return
+ # Update status based on hook execution
+ if process.status == process.StatusChoices.RUNNING:
+ # BACKGROUND HOOK - still running, return immediately
+ # Status is already STARTED from enter_started(), will be finalized by Snapshot.cleanup()
+ return
+
# FOREGROUND HOOK - completed, update from filesystem
- self.start_ts = start_ts
- if self.process_id:
- self.process.pwd = str(plugin_dir)
- self.process.save()
self.update_from_output()
# Clean up empty output directory if no files were created
@@ -3037,26 +2963,30 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
skipped = State(value=ArchiveResult.StatusChoices.SKIPPED, final=True)
# Tick Event - transitions based on conditions
+ # Flow: queued → started → (succeeded|failed|skipped)
+ # queued → skipped (if exceeded max attempts)
+ # started → backoff → started (retry)
tick = (
+ queued.to(skipped, cond='is_exceeded_max_attempts') | # Check skip first
queued.to.itself(unless='can_start') |
queued.to(started, cond='can_start') |
- started.to.itself(unless='is_finished') |
started.to(succeeded, cond='is_succeeded') |
started.to(failed, cond='is_failed') |
started.to(skipped, cond='is_skipped') |
started.to(backoff, cond='is_backoff') |
+ backoff.to(skipped, cond='is_exceeded_max_attempts') | # Check skip from backoff too
backoff.to.itself(unless='can_start') |
- backoff.to(started, cond='can_start') |
- backoff.to(succeeded, cond='is_succeeded') |
- backoff.to(failed, cond='is_failed') |
- backoff.to(skipped, cond='is_skipped')
+ backoff.to(started, cond='can_start')
+ # Removed redundant transitions: backoff.to(succeeded/failed/skipped)
+ # Reason: backoff should always retry→started, then started→final states
)
def can_start(self) -> bool:
- if not self.archiveresult.snapshot.url:
- return False
+ """Pure function - check if AR can start (has valid URL)."""
+ return bool(self.archiveresult.snapshot.url)
- # Check if snapshot has exceeded MAX_URL_ATTEMPTS failed results
+ def is_exceeded_max_attempts(self) -> bool:
+ """Check if snapshot has exceeded MAX_URL_ATTEMPTS failed results."""
from archivebox.config.configset import get_config
config = get_config(
@@ -3070,15 +3000,7 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
status=ArchiveResult.StatusChoices.FAILED
).count()
- if failed_count >= max_attempts:
- # Mark this result as skipped since we've hit the limit
- self.archiveresult.status = ArchiveResult.StatusChoices.SKIPPED
- self.archiveresult.output_str = f'Skipped: snapshot exceeded MAX_URL_ATTEMPTS ({max_attempts} failures)'
- self.archiveresult.retry_at = None
- self.archiveresult.save()
- return False
-
- return True
+ return failed_count >= max_attempts
def is_succeeded(self) -> bool:
"""Check if extractor plugin succeeded (status was set by run())."""
@@ -3101,12 +3023,35 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
)
def is_finished(self) -> bool:
- """Check if extraction has completed (success, failure, or skipped)."""
- return self.archiveresult.status in (
+ """
+ Check if extraction has completed (success, failure, or skipped).
+
+ For background hooks in STARTED state, checks if their Process has finished and reaps them.
+ """
+ # If already in final state, return True
+ if self.archiveresult.status in (
ArchiveResult.StatusChoices.SUCCEEDED,
ArchiveResult.StatusChoices.FAILED,
ArchiveResult.StatusChoices.SKIPPED,
- )
+ ):
+ return True
+
+ # If in STARTED state with a Process, check if Process has finished running
+ if self.archiveresult.status == ArchiveResult.StatusChoices.STARTED:
+ if self.archiveresult.process_id:
+ process = self.archiveresult.process
+
+ # If process is NOT running anymore, reap the background hook
+ if not process.is_running():
+ self.archiveresult.update_from_output()
+ # Check if now in final state after reaping
+ return self.archiveresult.status in (
+ ArchiveResult.StatusChoices.SUCCEEDED,
+ ArchiveResult.StatusChoices.FAILED,
+ ArchiveResult.StatusChoices.SKIPPED,
+ )
+
+ return False
@queued.enter
def enter_queued(self):
@@ -3148,7 +3093,12 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
)
def _check_and_seal_parent_snapshot(self):
- """Check if this is the last ArchiveResult to finish - if so, seal the parent Snapshot."""
+ """
+ Check if this is the last ArchiveResult to finish - if so, seal the parent Snapshot.
+
+ Note: In the new architecture, SnapshotWorker handles step advancement and sealing.
+ This method is kept for backwards compatibility with manual CLI commands.
+ """
import sys
snapshot = self.archiveresult.snapshot
@@ -3189,6 +3139,8 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
def enter_failed(self):
import sys
+ print(f'[red] ❌ ArchiveResult.enter_failed() called for {self.archiveresult.plugin}[/red]', file=sys.stderr)
+
self.archiveresult.update_and_requeue(
retry_at=None,
status=ArchiveResult.StatusChoices.FAILED,
@@ -3207,6 +3159,16 @@ class ArchiveResultMachine(BaseStateMachine, strict_states=True):
def enter_skipped(self):
import sys
+ # Set output_str if not already set (e.g., when skipped due to max attempts)
+ if not self.archiveresult.output_str and self.is_exceeded_max_attempts():
+ from archivebox.config.configset import get_config
+ config = get_config(
+ crawl=self.archiveresult.snapshot.crawl,
+ snapshot=self.archiveresult.snapshot,
+ )
+ max_attempts = config.get('MAX_URL_ATTEMPTS', 50)
+ self.archiveresult.output_str = f'Skipped: snapshot exceeded MAX_URL_ATTEMPTS ({max_attempts} failures)'
+
self.archiveresult.update_and_requeue(
retry_at=None,
status=ArchiveResult.StatusChoices.SKIPPED,
diff --git a/archivebox/crawls/admin.py b/archivebox/crawls/admin.py
index da08b0ac..110fe941 100644
--- a/archivebox/crawls/admin.py
+++ b/archivebox/crawls/admin.py
@@ -281,25 +281,11 @@ class CrawlAdmin(ConfigEditorMixin, BaseModelAdmin):
"""Editor for crawl URLs."""
widget_id = f'crawl_urls_{obj.pk}'
- # Check if it's a local file we can edit
- source_file = obj.get_file_path()
- is_file = source_file is not None
- file_contents = ""
- error = None
-
- if is_file and source_file:
- try:
- file_contents = source_file.read_text().strip()
- except Exception as e:
- error = f'Error reading {source_file}: {e}'
-
# Escape for safe HTML embedding
escaped_urls = (obj.urls or '').replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"')
- escaped_file_contents = file_contents.replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"')
# Count lines for auto-expand logic
line_count = len((obj.urls or '').split('\n'))
- file_line_count = len(file_contents.split('\n')) if file_contents else 0
uri_rows = min(max(3, line_count), 10)
html = f'''
@@ -318,21 +304,6 @@ class CrawlAdmin(ConfigEditorMixin, BaseModelAdmin):
{line_count} URL{'s' if line_count != 1 else ''} · Note: URLs displayed here for reference only
-
- {"" if not is_file else f'''
-
-
-
- {"
" + error + "
" if error else ""}
-
-