diff --git a/archivebox/cli/archivebox_run.py b/archivebox/cli/archivebox_run.py index 75b8ebd2..e17f5709 100644 --- a/archivebox/cli/archivebox_run.py +++ b/archivebox/cli/archivebox_run.py @@ -244,7 +244,7 @@ def process_stdin_records() -> int: return 0 -def run_runner(daemon: bool = False) -> int: +def run_runner(daemon: bool = False, crawl_id: str | None = None) -> int: """ Run the background runner loop. @@ -269,7 +269,7 @@ def run_runner(daemon: bool = False) -> int: current.save(update_fields=["process_type", "modified_at"]) try: - run_pending_crawls(daemon=daemon) + run_pending_crawls(daemon=daemon, crawl_id=crawl_id) return 0 except KeyboardInterrupt: return 0 @@ -319,19 +319,7 @@ def main(daemon: bool, crawl_id: str, snapshot_id: str, binary_id: str): sys.exit(1) if crawl_id: - try: - from archivebox.services.runner import run_crawl - - run_crawl(crawl_id) - sys.exit(0) - except KeyboardInterrupt: - sys.exit(0) - except Exception as e: - rprint(f"[red]Runner error: {type(e).__name__}: {e}[/red]", file=sys.stderr) - import traceback - - traceback.print_exc() - sys.exit(1) + sys.exit(run_runner(daemon=False, crawl_id=crawl_id)) if daemon: sys.exit(run_runner(daemon=True)) diff --git a/archivebox/cli/archivebox_snapshot.py b/archivebox/cli/archivebox_snapshot.py index b2804d4e..541319e4 100644 --- a/archivebox/cli/archivebox_snapshot.py +++ b/archivebox/cli/archivebox_snapshot.py @@ -35,7 +35,7 @@ from collections.abc import Iterable import rich_click as click from rich import print as rprint -from django.db.models import Q, Sum +from django.db.models import Case, IntegerField, Q, Sum, When from django.db.models.functions import Coalesce from archivebox.cli.cli_utils import apply_filters @@ -213,7 +213,7 @@ def list_snapshots( is_tty = sys.stdout.isatty() and not csv - queryset = Snapshot.objects.annotate(output_size_sum=Coalesce(Sum("archiveresult__output_size"), 0)).order_by("-created_at") + queryset = Snapshot.objects.order_by("-created_at") # Apply filters filter_kwargs = { @@ -264,7 +264,17 @@ def list_snapshots( if sort: queryset = queryset.order_by(sort) - if limit: + + if not is_tty: + if limit: + limited_ids = list(queryset.values_list("id", flat=True)[:limit]) + preserved_order = Case( + *(When(id=snapshot_id, then=position) for position, snapshot_id in enumerate(limited_ids)), + output_field=IntegerField(), + ) + queryset = Snapshot.objects.filter(id__in=limited_ids).order_by(preserved_order) + queryset = queryset.annotate(output_size_sum=Coalesce(Sum("archiveresult__output_size"), 0)).prefetch_related("tags") + elif limit: queryset = queryset[:limit] count = 0 @@ -276,8 +286,36 @@ def list_snapshots( rows: list[str] = [] if with_headers: rows.append(",".join(cols)) + simple_cols = { + "id", + "crawl_id", + "url", + "title", + "timestamp", + "depth", + "status", + "fs_version", + "bookmarked_at", + "created_at", + "modified_at", + "retry_at", + "downloaded_at", + } + from archivebox.misc.util import to_json + for snapshot in queryset.iterator(chunk_size=500): - rows.append(snapshot.to_csv(cols=cols, separator=",")) + if set(cols).issubset(simple_cols): + rows.append( + ",".join( + to_json( + value.isoformat() if hasattr((value := getattr(snapshot, col, "")), "isoformat") else value, + indent=None, + ) + for col in cols + ), + ) + else: + rows.append(snapshot.to_csv(cols=cols, separator=",")) count += 1 output = "\n".join(rows) if output: diff --git a/archivebox/config/common.py b/archivebox/config/common.py index e9727daf..f0a82c9e 100644 --- a/archivebox/config/common.py +++ b/archivebox/config/common.py @@ -1,6 +1,7 @@ __package__ = "archivebox.config" import json +import os import re import secrets import sys @@ -76,6 +77,7 @@ class StorageConfig(BaseConfigSet): # ARCHIVE_DIR / USERS_DIR are resolved dynamically via get_config(). ARCHIVE_DIR: Path = Field(default=CONSTANTS.ARCHIVE_DIR) USERS_DIR: Path = Field(default=CONSTANTS.USERS_DIR) + PERSONAS_DIR: Path = Field(default=CONSTANTS.PERSONAS_DIR) # TMP_DIR must be a local, fast, readable/writable dir by archivebox user, # must be a short path due to unix path length restrictions for socket files (<100 chars) @@ -461,6 +463,7 @@ def get_config( snapshot: Any = None, archiveresult: Any = None, machine: Any = None, + include_machine: bool = True, resolve_plugins: bool = True, ) -> ArchiveBoxBaseConfig: """ @@ -484,7 +487,7 @@ def get_config( if crawl is None and snapshot is not None: crawl = snapshot.crawl - if machine is None: + if include_machine and machine is None: try: from django.apps import apps @@ -520,10 +523,10 @@ def get_config( scope_overrides: ConfigPayload = {} - if machine is not None and machine.config: + if include_machine and machine is not None and machine.config: from archivebox.machine.models import _sanitize_machine_config - scope_overrides.update(_sanitize_machine_config(machine.config)) + scope_overrides.update(_sanitize_machine_config(machine.config, lib_dir=config_data.get("LIB_DIR"))) if persona is not None: scope_overrides.update(persona.get_derived_config()) @@ -564,6 +567,7 @@ def get_config( config_data["ABX_RUNTIME"] = "archivebox" config = ArchiveBoxConfig.model_validate(config_data) + os.environ["ABXPKG_LIB_DIR"] = str(config.LIB_DIR) archiving_warning_key = (config.TIMEOUT, config.USE_COLOR) if archiving_warning_key not in _WARNED_ARCHIVING_CONFIGS: config.warn_if_invalid() diff --git a/archivebox/config/paths.py b/archivebox/config/paths.py index 15e339bb..74d51a28 100644 --- a/archivebox/config/paths.py +++ b/archivebox/config/paths.py @@ -13,7 +13,7 @@ from typing import TYPE_CHECKING from benedict import benedict -from .permissions import SudoPermission, IS_ROOT, ARCHIVEBOX_USER +from .permissions import SudoPermission, IS_ROOT, ARCHIVEBOX_USER, ARCHIVEBOX_GROUP if TYPE_CHECKING: from archivebox.config.common import ArchiveBoxConfig @@ -158,11 +158,23 @@ def assert_dir_can_contain_unix_sockets(dir_path: Path) -> bool: def create_and_chown_dir(dir_path: Path) -> None: + """Create a required runtime dir and fix only that dir's ownership when needed.""" + dir_existed = dir_path.exists() + dir_path.mkdir(parents=True, exist_ok=True) + + try: + stat = dir_path.stat() + except OSError: + return + + if dir_existed and stat.st_uid == ARCHIVEBOX_USER and stat.st_gid == ARCHIVEBOX_GROUP: + return + with SudoPermission(uid=0, fallback=True): - dir_path.mkdir(parents=True, exist_ok=True) - subprocess.run(["chown", str(ARCHIVEBOX_USER), str(dir_path)], stderr=subprocess.DEVNULL) - for child in dir_path.iterdir(): - subprocess.run(["chown", str(ARCHIVEBOX_USER), str(child)], stderr=subprocess.DEVNULL) + try: + os.chown(dir_path, ARCHIVEBOX_USER, ARCHIVEBOX_GROUP) + except (OSError, PermissionError): + pass def tmp_dir_socket_path_is_short_enough(dir_path: Path) -> bool: diff --git a/archivebox/core/admin_site.py b/archivebox/core/admin_site.py index f3a3b177..ec34d251 100644 --- a/archivebox/core/admin_site.py +++ b/archivebox/core/admin_site.py @@ -10,6 +10,9 @@ from admin_data_views.admin import ( get_app_list as adv_get_app_list, ) +from archivebox.config import VERSION +from archivebox.config.version import get_COMMIT_HASH + if TYPE_CHECKING: from django.http import HttpRequest from django.template.response import TemplateResponse @@ -24,6 +27,12 @@ class ArchiveBoxAdmin(admin.AdminSite): site_title = "Admin" namespace = "admin" + def each_context(self, request: "HttpRequest") -> dict[str, Any]: + context = super().each_context(request) + context["VERSION"] = VERSION + context["STATIC_CACHE_KEY"] = (get_COMMIT_HASH() or VERSION or "dev").strip() + return context + @staticmethod def _format_object_count(count: int) -> tuple[int, str, str]: if count >= 1_000_000_000: diff --git a/archivebox/core/models.py b/archivebox/core/models.py index 821bfc77..e7de917a 100755 --- a/archivebox/core/models.py +++ b/archivebox/core/models.py @@ -2537,7 +2537,7 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea lower = (path or "").lower() return lower.endswith(text_exts) - hashes_index = self.hashes_index + hashes_index = self.hashes_index if include_filesystem_fallback else {} for result in self.archiveresult_set.all().order_by("start_ts"): output_file_map = result.output_file_map() embed_path = result.embed_path_db(output_file_map=output_file_map) @@ -2616,6 +2616,8 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea if not include_filesystem_fallback or hashes_index: return outputs + if not snap_dir.is_dir(): + return outputs embeddable_exts = { "html", @@ -2829,7 +2831,8 @@ class Snapshot(ModelWithOutputDir, ModelWithConfig, ModelWithNotes, ModelWithHea outputs: list[dict] | None = None, hidden_card_plugins: set[str] | None = None, ) -> tuple[list[dict[str, object]], list[dict[str, object]]]: - outputs = outputs or self.discover_outputs(include_filesystem_fallback=True) + if outputs is None: + outputs = self.discover_outputs(include_filesystem_fallback=True) hidden_card_plugins = hidden_card_plugins or set() accounted_entries: set[str] = set() for output in outputs: diff --git a/archivebox/core/urls.py b/archivebox/core/urls.py index b249074c..819789f0 100644 --- a/archivebox/core/urls.py +++ b/archivebox/core/urls.py @@ -69,7 +69,7 @@ urlpatterns = [ path("accounts/login/", RedirectView.as_view(url="/admin/login/")), path("accounts/logout/", RedirectView.as_view(url="/admin/logout/")), path("accounts/", include("django.contrib.auth.urls")), - path("admin/live-progress/", live_progress_view, name="live_progress"), + path("admin/live-progress/", archivebox_admin.admin_view(live_progress_view), name="live_progress"), path("admin/", archivebox_admin.urls), path("api/", include("archivebox.api.urls"), name="api"), path("health/", HealthCheckView.as_view(), name="healthcheck"), diff --git a/archivebox/core/views.py b/archivebox/core/views.py index 5e20f62b..4c61ab9e 100644 --- a/archivebox/core/views.py +++ b/archivebox/core/views.py @@ -18,8 +18,7 @@ from django.utils.safestring import mark_safe from django.views import View from django.views.generic.list import ListView from django.views.generic import FormView -from django.db import connection -from django.db.models import Q, Prefetch +from django.db.models import Count, Q, Prefetch from django.contrib import messages from django.contrib.auth.mixins import UserPassesTestMixin from django.views.decorators.csrf import csrf_exempt @@ -174,7 +173,7 @@ class SnapshotView(View): hidden_card_plugins = {"archivedotorg", "favicon", "title"} outputs = [ out - for out in snapshot.discover_outputs(include_filesystem_fallback=True) + for out in snapshot.discover_outputs(include_filesystem_fallback=False) if (out.get("size") or 0) > 0 and out.get("name") not in hidden_card_plugins ] archiveresults = {out["name"]: out for out in outputs} @@ -1319,6 +1318,19 @@ def live_progress_view(request): from archivebox.crawls.models import Crawl from archivebox.core.models import Snapshot, ArchiveResult from archivebox.machine.models import Process, Machine + from archivebox.machine.detect import get_host_guid + + if not request.user.is_authenticated or not request.user.is_active or not request.user.is_staff: + return JsonResponse({"error": "Permission denied"}, status=403) + + now = timezone.now() + crawl_scope = Crawl.objects.all() + snapshot_scope = Snapshot.objects.all() + archiveresult_scope = ArchiveResult.objects.all() + if not request.user.is_superuser: + crawl_scope = crawl_scope.filter(created_by=request.user) + snapshot_scope = snapshot_scope.filter(crawl__created_by=request.user) + archiveresult_scope = archiveresult_scope.filter(snapshot__crawl__created_by=request.user) def is_current_run_timestamp(event_ts, run_started_at) -> bool: if run_started_at is None: @@ -1376,9 +1388,18 @@ def live_progress_view(request): return hook_details(Path(hook_path).name, plugin=Path(hook_path).parent.name or "setup") - machine = Machine.current() - Process.cleanup_stale_running(machine=machine) - Process.cleanup_orphaned_workers() + def snapshot_output_url(snapshot, output_path: str) -> str: + return build_snapshot_url(str(snapshot.id), output_path, request=request, config=getattr(request, "archivebox_config", None)) + + def snapshot_view_url(snapshot, output_path: str = "") -> str: + anchor = f"#{output_path}" if output_path else "" + return build_web_url( + f"/{snapshot.archive_path_from_db}/index.html{anchor}", + request=request, + config=getattr(request, "archivebox_config", None), + ) + + machine = Machine.objects.filter(guid=get_host_guid()).first() orchestrator_proc = ( Process.objects.filter( machine=machine, @@ -1387,6 +1408,8 @@ def live_progress_view(request): ) .order_by("-started_at") .first() + if machine is not None + else None ) runner_worker = None try: @@ -1402,85 +1425,110 @@ def live_progress_view(request): orchestrator_running = orchestrator_proc is not None or runner_worker_running orchestrator_pid = orchestrator_proc.pid if orchestrator_proc else runner_worker_pid - def sqlite_approx_count(model) -> int | None: - if connection.vendor != "sqlite": - return None - with connection.cursor() as cursor: - try: - cursor.execute("SELECT stat FROM sqlite_stat1 WHERE tbl = %s", [model._meta.db_table]) - stats = [int(str(row[0]).split()[0]) for row in cursor.fetchall() if row and row[0]] - except Exception: - stats = [] - return max(stats) if stats else None + def count_statuses(queryset, statuses) -> dict[str, int]: + return {status: queryset.filter(status=status).count() for status in statuses} # Get model counts by status - crawls_pending = Crawl.objects.filter(status=Crawl.StatusChoices.QUEUED).count() - crawls_started = Crawl.objects.filter(status=Crawl.StatusChoices.STARTED).count() + crawl_status_counts = count_statuses(crawl_scope, (Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED)) + crawls_pending = crawl_status_counts.get(Crawl.StatusChoices.QUEUED, 0) + crawls_started = crawl_status_counts.get(Crawl.StatusChoices.STARTED, 0) # Get recent crawls (last 24 hours) from datetime import timedelta - one_day_ago = timezone.now() - timedelta(days=1) - crawls_recent = Crawl.objects.filter(created_at__gte=one_day_ago).count() + one_day_ago = now - timedelta(days=1) + recently_cancelled_after = now - timedelta(minutes=10) + crawls_recent = crawl_scope.filter(created_at__gte=one_day_ago).count() - snapshots_pending = Snapshot.objects.filter(status=Snapshot.StatusChoices.QUEUED).count() - snapshots_started = Snapshot.objects.filter(status=Snapshot.StatusChoices.STARTED).count() + snapshot_status_counts = count_statuses(snapshot_scope, (Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED)) + snapshots_pending = snapshot_status_counts.get(Snapshot.StatusChoices.QUEUED, 0) + snapshots_started = snapshot_status_counts.get(Snapshot.StatusChoices.STARTED, 0) - archiveresults_pending = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.QUEUED).count() - archiveresults_started = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.STARTED).count() - archiveresults_failed = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.FAILED).count() - archiveresults_backoff = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.BACKOFF).count() - archiveresults_skipped = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.SKIPPED).count() - archiveresults_noresults = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.NORESULTS).count() - archiveresults_total = sqlite_approx_count(ArchiveResult) - if archiveresults_total is not None: - archiveresults_succeeded = max( - archiveresults_total - - archiveresults_pending - - archiveresults_started - - archiveresults_failed - - archiveresults_backoff - - archiveresults_skipped - - archiveresults_noresults, - 0, - ) - else: - archiveresults_succeeded = ArchiveResult.objects.filter(status=ArchiveResult.StatusChoices.SUCCEEDED).count() + archiveresult_status_counts = count_statuses( + archiveresult_scope, + ( + ArchiveResult.StatusChoices.QUEUED, + ArchiveResult.StatusChoices.STARTED, + ArchiveResult.StatusChoices.SUCCEEDED, + ArchiveResult.StatusChoices.FAILED, + ), + ) + archiveresults_pending = archiveresult_status_counts.get(ArchiveResult.StatusChoices.QUEUED, 0) + archiveresults_started = archiveresult_status_counts.get(ArchiveResult.StatusChoices.STARTED, 0) + archiveresults_succeeded = archiveresult_status_counts.get(ArchiveResult.StatusChoices.SUCCEEDED, 0) + archiveresults_failed = archiveresult_status_counts.get(ArchiveResult.StatusChoices.FAILED, 0) # Build hierarchical active crawls with nested snapshots and archive results active_crawls_qs = ( - Crawl.objects.filter(status__in=[Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED]) - .prefetch_related("snapshot_set") + crawl_scope.filter( + Q(status__in=[Crawl.StatusChoices.QUEUED, Crawl.StatusChoices.STARTED]) + | Q(status=Crawl.StatusChoices.SEALED, modified_at__gte=recently_cancelled_after), + ) + .select_related("created_by") .distinct() .order_by("-modified_at")[:10] ) + active_crawls_list = list(active_crawls_qs) + active_crawl_ids = [crawl.id for crawl in active_crawls_list] + snapshot_counts_by_crawl: dict[str, dict[str, int]] = {str(crawl_id): {} for crawl_id in active_crawl_ids} + cancelled_snapshot_counts_by_crawl: dict[str, int] = {str(crawl_id): 0 for crawl_id in active_crawl_ids} + if active_crawl_ids: + for row in snapshot_scope.filter(crawl_id__in=active_crawl_ids).values("crawl_id", "status").annotate(count=Count("id")): + snapshot_counts_by_crawl.setdefault(str(row["crawl_id"]), {})[row["status"]] = row["count"] + for row in ( + snapshot_scope.filter( + crawl_id__in=active_crawl_ids, + status=Snapshot.StatusChoices.SEALED, + downloaded_at__isnull=True, + modified_at__gte=recently_cancelled_after, + ) + .values("crawl_id") + .annotate(count=Count("id")) + ): + cancelled_snapshot_counts_by_crawl[str(row["crawl_id"])] = row["count"] - running_processes = Process.objects.filter( - machine=machine, - status=Process.StatusChoices.RUNNING, - process_type__in=[ - Process.TypeChoices.HOOK, - Process.TypeChoices.BINARY, - ], - ) - recent_processes = Process.objects.filter( - machine=machine, - process_type__in=[ - Process.TypeChoices.HOOK, - Process.TypeChoices.BINARY, - ], - modified_at__gte=timezone.now() - timedelta(minutes=10), - ).order_by("-modified_at") + if machine is not None: + running_processes = Process.objects.filter( + machine=machine, + status=Process.StatusChoices.RUNNING, + process_type__in=[ + Process.TypeChoices.HOOK, + Process.TypeChoices.BINARY, + ], + ).only("id", "machine_id", "process_type", "status", "pwd", "cmd", "pid", "exit_code", "started_at", "modified_at") + recent_processes = ( + Process.objects.filter( + machine=machine, + process_type__in=[ + Process.TypeChoices.HOOK, + Process.TypeChoices.BINARY, + ], + modified_at__gte=now - timedelta(minutes=10), + ) + .only("id", "machine_id", "process_type", "status", "pwd", "cmd", "pid", "exit_code", "started_at", "modified_at") + .order_by("-modified_at") + ) + else: + running_processes = Process.objects.none() + recent_processes = Process.objects.none() crawl_process_pids: dict[str, int] = {} snapshot_process_pids: dict[str, int] = {} process_records_by_crawl: dict[str, list[tuple[dict[str, object], object | None]]] = {} process_records_by_snapshot: dict[str, list[tuple[dict[str, object], object | None]]] = {} seen_process_records: set[str] = set() active_snapshot_statuses = {Snapshot.StatusChoices.QUEUED, Snapshot.StatusChoices.STARTED} - snapshots = [ - snapshot for crawl in active_crawls_qs for snapshot in crawl.snapshot_set.all() if snapshot.status in active_snapshot_statuses - ] + recently_cancelled_snapshots_q = Q( + status=Snapshot.StatusChoices.SEALED, + downloaded_at__isnull=True, + modified_at__gte=recently_cancelled_after, + ) + crawls_by_id = {str(crawl.id): crawl for crawl in active_crawls_list} + snapshots = list( + snapshot_scope.filter(Q(status__in=active_snapshot_statuses) | recently_cancelled_snapshots_q, crawl_id__in=active_crawl_ids) + .select_related("crawl") + .order_by("crawl_id", "status", "modified_at")[:100], + ) snapshots_by_id = {str(snapshot.id): snapshot for snapshot in snapshots} def find_snapshot_for_process(proc_pwd: Path) -> Snapshot | None: @@ -1490,14 +1538,29 @@ def live_progress_view(request): return snapshot return None + def find_crawl_for_process(proc_pwd: Path) -> Crawl | None: + for path_part in reversed(proc_pwd.parts): + crawl = crawls_by_id.get(path_part) + if crawl: + return crawl + return None + + running_worker_ids: set[str] = set() for proc in running_processes: if not proc.pwd: continue - matched_snapshot = find_snapshot_for_process(Path(proc.pwd)) + proc_pwd = Path(proc.pwd) + matched_snapshot = find_snapshot_for_process(proc_pwd) + matched_crawl = matched_snapshot.crawl if matched_snapshot is not None else find_crawl_for_process(proc_pwd) if matched_snapshot is None: - continue - crawl_id = str(matched_snapshot.crawl_id) - snapshot_id = str(matched_snapshot.id) + if matched_crawl is None: + continue + crawl_id = str(matched_crawl.id) + snapshot_id = "" + else: + crawl_id = str(matched_snapshot.crawl_id) + snapshot_id = str(matched_snapshot.id) + running_worker_ids.add(str(proc.id)) _plugin, _label, phase, _hook_name = process_label(proc.cmd) if crawl_id and proc.pid: crawl_process_pids.setdefault(crawl_id, proc.pid) @@ -1507,11 +1570,13 @@ def live_progress_view(request): for proc in recent_processes: if not proc.pwd: continue - matched_snapshot = find_snapshot_for_process(Path(proc.pwd)) - if matched_snapshot is None: + proc_pwd = Path(proc.pwd) + matched_snapshot = find_snapshot_for_process(proc_pwd) + matched_crawl = matched_snapshot.crawl if matched_snapshot is not None else find_crawl_for_process(proc_pwd) + if matched_snapshot is None and matched_crawl is None: continue - crawl_id = str(matched_snapshot.crawl_id) - snapshot_id = str(matched_snapshot.id) + crawl_id = str(matched_snapshot.crawl_id if matched_snapshot is not None else matched_crawl.id) + snapshot_id = str(matched_snapshot.id) if matched_snapshot is not None else "" plugin, label, phase, hook_name = process_label(proc.cmd) @@ -1545,19 +1610,21 @@ def live_progress_view(request): process_records_by_crawl.setdefault(crawl_id, []).append((payload, proc_started_at)) active_crawls = [] - total_workers = 0 - for crawl in active_crawls_qs: - # Get ALL snapshots for this crawl to count status (already prefetched) - all_crawl_snapshots = list(crawl.snapshot_set.all()) - - # Count snapshots by status from ALL snapshots - total_snapshots = len(all_crawl_snapshots) - completed_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.SEALED) - started_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.STARTED) - pending_snapshots = sum(1 for s in all_crawl_snapshots if s.status == Snapshot.StatusChoices.QUEUED) + total_workers = len(running_worker_ids) + for crawl in active_crawls_list: + crawl_snapshot_counts = snapshot_counts_by_crawl.get(str(crawl.id), {}) + total_snapshots = sum(crawl_snapshot_counts.values()) + completed_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.SEALED, 0) + started_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.STARTED, 0) + pending_snapshots = crawl_snapshot_counts.get(Snapshot.StatusChoices.QUEUED, 0) + cancelled_snapshots = cancelled_snapshot_counts_by_crawl.get(str(crawl.id), 0) # Get only ACTIVE snapshots to display (limit to 5 most recent) - active_crawl_snapshots = [s for s in all_crawl_snapshots if s.status in active_snapshot_statuses][:5] + active_crawl_snapshots = list( + snapshot_scope.filter(Q(status__in=active_snapshot_statuses) | recently_cancelled_snapshots_q, crawl=crawl) + .select_related("crawl") + .order_by("status", "modified_at")[:5], + ) # Count URLs in the crawl (for when snapshots haven't been created yet) urls_count = 0 @@ -1572,7 +1639,6 @@ def live_progress_view(request): for payload, proc_started_at in process_records_by_crawl.get(str(crawl.id), []) if is_current_run_timestamp(proc_started_at, crawl_run_started_at) ] - total_workers += sum(1 for item in crawl_setup_plugins if item.get("source") == "process" and item.get("status") == "started") crawl_setup_total = len(crawl_setup_plugins) crawl_setup_completed = sum(1 for item in crawl_setup_plugins if item.get("status") == "succeeded") crawl_setup_failed = sum(1 for item in crawl_setup_plugins if item.get("status") == "failed") @@ -1591,10 +1657,31 @@ def live_progress_view(request): if archiveresult_matches_current_run(ar, snapshot_run_started_at) ] - now = timezone.now() plugin_progress_values: list[int] = [] all_plugins: list[dict[str, object]] = [] seen_plugin_keys: set[str] = set() + snapshot_title = snapshot._normalize_title_candidate(snapshot.title, snapshot_url=snapshot.url) + snapshot_favicon_url = "" + snapshot_preview_url = "" + snapshot_preview_link = snapshot_view_url(snapshot) + snapshot_fallback_urls: list[str] = [] + result_by_plugin = {result.plugin: result for result in snapshot_results} + title_result = result_by_plugin.get("title") + if not snapshot_title and title_result is not None: + snapshot_title = snapshot._normalize_title_candidate(title_result.output_str, snapshot_url=snapshot.url) + favicon_result = result_by_plugin.get("favicon") + if favicon_result is not None and favicon_result.status == ArchiveResult.StatusChoices.SUCCEEDED: + favicon_path = favicon_result.embed_path_db() or "favicon/favicon.ico" + snapshot_favicon_url = snapshot_output_url(snapshot, favicon_path) + screenshot_result = result_by_plugin.get("screenshot") + if screenshot_result is not None and screenshot_result.status == ArchiveResult.StatusChoices.SUCCEEDED: + screenshot_path = screenshot_result.embed_path_db() or "screenshot/screenshot.png" + snapshot_preview_url = snapshot_output_url(snapshot, screenshot_path) + snapshot_preview_link = snapshot_view_url(snapshot, screenshot_path) + if snapshot_favicon_url: + snapshot_fallback_urls.append(snapshot_favicon_url) + elif snapshot_favicon_url: + snapshot_preview_url = snapshot_favicon_url def plugin_sort_key(ar): status_order = { @@ -1618,7 +1705,7 @@ def live_progress_view(request): progress_value = 100 elif status == ArchiveResult.StatusChoices.STARTED: started_at = ar.start_ts or (ar.process.started_at if ar.process_id and ar.process else None) - timeout = ar.timeout or 120 + timeout = ar.process.timeout if ar.process_id and ar.process else 120 if started_at and timeout: elapsed = max(0.0, (now - started_at).total_seconds()) progress_value = int(min(99, max(1, (elapsed / float(timeout)) * 100))) @@ -1638,12 +1725,17 @@ def live_progress_view(request): "phase": phase, "status": status, "process_id": str(ar.process_id) if ar.process_id else None, + "admin_url": f"/admin/core/archiveresult/{ar.id}/change/", } + output_path = ar.embed_path_db() + if output_path: + plugin_payload["output_path"] = output_path + plugin_payload["output_url"] = snapshot_view_url(snapshot, output_path) if status == ArchiveResult.StatusChoices.STARTED and ar.process_id and ar.process: plugin_payload["pid"] = ar.process.pid if status == ArchiveResult.StatusChoices.STARTED: plugin_payload["progress"] = progress_value - plugin_payload["timeout"] = ar.timeout or 120 + plugin_payload["timeout"] = ar.process.timeout if ar.process_id and ar.process else 120 plugin_payload["source"] = "archiveresult" all_plugins.append(plugin_payload) seen_plugin_keys.add(str(ar.process_id) if ar.process_id else f"{ar.plugin}:{hook_name}") @@ -1662,7 +1754,6 @@ def live_progress_view(request): plugin_progress_values.append(100) elif proc_status == "started": plugin_progress_values.append(1) - total_workers += 1 else: plugin_progress_values.append(0) @@ -1672,11 +1763,30 @@ def live_progress_view(request): pending_plugins = sum(1 for item in all_plugins if item.get("status") == "queued") snapshot_progress = int(sum(plugin_progress_values) / len(plugin_progress_values)) if plugin_progress_values else 0 + worker_state = "running" if snapshot_process_pids.get(str(snapshot.id)) else "waiting" + if snapshot.status == Snapshot.StatusChoices.SEALED and not snapshot.downloaded_at: + worker_state = "cancelled" + if ( + snapshot.status == Snapshot.StatusChoices.STARTED + and worker_state == "waiting" + and not all_plugins + and snapshot.modified_at + and (now - snapshot.modified_at).total_seconds() > 30 + ): + worker_state = "stalled" if orchestrator_running else "crashed" active_snapshots_for_crawl.append( { "id": str(snapshot.id), "url": snapshot.url[:80], + "full_url": snapshot.url, + "title": snapshot_title, + "admin_url": f"/admin/core/snapshot/{snapshot.id}/change/", + "view_url": snapshot_view_url(snapshot), + "favicon_url": snapshot_favicon_url, + "preview_url": snapshot_preview_url, + "preview_link": snapshot_preview_link, + "preview_fallbacks": snapshot_fallback_urls, "status": snapshot.status, "started": (snapshot.downloaded_at or snapshot.created_at).isoformat() if (snapshot.downloaded_at or snapshot.created_at) @@ -1688,6 +1798,7 @@ def live_progress_view(request): "pending_plugins": pending_plugins, "all_plugins": all_plugins, "worker_pid": snapshot_process_pids.get(str(snapshot.id)), + "worker_state": worker_state, }, ) @@ -1696,8 +1807,19 @@ def live_progress_view(request): urls_preview = crawl.urls[:60] if crawl.urls else None # Check if retry_at is in the future (would prevent worker from claiming) - retry_at_future = crawl.retry_at > timezone.now() if crawl.retry_at else False - seconds_until_retry = int((crawl.retry_at - timezone.now()).total_seconds()) if crawl.retry_at and retry_at_future else 0 + retry_at_future = crawl.retry_at > now if crawl.retry_at else False + seconds_until_retry = int((crawl.retry_at - now).total_seconds()) if crawl.retry_at and retry_at_future else 0 + crawl_worker_state = ( + "running" + if crawl_process_pids.get(str(crawl.id)) or any(snapshot.get("worker_pid") for snapshot in active_snapshots_for_crawl) + else "waiting" + ) + if crawl.status == Crawl.StatusChoices.SEALED and cancelled_snapshots: + crawl_worker_state = "cancelled" + elif ( + crawl.status == Crawl.StatusChoices.STARTED and crawl_worker_state == "waiting" and (started_snapshots or pending_snapshots) + ): + crawl_worker_state = "stalled" if orchestrator_running else "crashed" active_crawls.append( { @@ -1713,6 +1835,7 @@ def live_progress_view(request): "started_snapshots": started_snapshots, "failed_snapshots": 0, "pending_snapshots": pending_snapshots, + "cancelled_snapshots": cancelled_snapshots, "setup_plugins": crawl_setup_plugins, "setup_total_plugins": crawl_setup_total, "setup_completed_plugins": crawl_setup_completed, @@ -1724,6 +1847,7 @@ def live_progress_view(request): "retry_at_future": retry_at_future, "seconds_until_retry": seconds_until_retry, "worker_pid": crawl_process_pids.get(str(crawl.id)), + "worker_state": crawl_worker_state, }, ) diff --git a/archivebox/crawls/models.py b/archivebox/crawls/models.py index 8f7d413e..04ee0727 100755 --- a/archivebox/crawls/models.py +++ b/archivebox/crawls/models.py @@ -1,6 +1,7 @@ __package__ = "archivebox.crawls" -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any +from collections.abc import Iterable, Mapping from io import StringIO import uuid import json @@ -410,7 +411,9 @@ class Crawl(ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWith def url_passes_filters(self, url: str, *, snapshot=None, use_effective_config: bool = True) -> bool: denylist = self.get_url_denylist(use_effective_config=use_effective_config, snapshot=snapshot) allowlist = self.get_url_allowlist(use_effective_config=use_effective_config, snapshot=snapshot) + return self.url_passes_compiled_filters(url, allowlist=allowlist, denylist=denylist) + def url_passes_compiled_filters(self, url: str, *, allowlist: list[str], denylist: list[str]) -> bool: for pattern in denylist: if self._pattern_matches_url(url, pattern): return False @@ -748,56 +751,158 @@ class Crawl(ModelWithOutputDir, ModelWithConfig, ModelWithHealthStats, ModelWith created_by_id: int | None = None, ): """Create one child snapshot if it passes crawl filters and limits.""" - from archivebox.core.models import Snapshot + snapshots = self.create_discovered_snapshots( + parent_snapshot, + [{"url": url, "title": title, "tags": tags}], + depth=depth, + created_by_id=created_by_id, + ) + return snapshots[0] if snapshots else None + + def create_discovered_snapshots( + self, + parent_snapshot, + records: Iterable[Mapping[str, Any]], + *, + depth: int, + created_by_id: int | None = None, + ) -> list["Snapshot"]: + """Create child snapshots from discovered URL records after filtering and deduping once.""" + from archivebox.core.models import Snapshot, SnapshotTag, Tag + from archivebox.config.common import get_config from archivebox.misc.util import fix_url_from_markdown, sanitize_extracted_url + from archivebox.core.host_utils import get_admin_host, get_api_host, get_listen_host, get_public_host, get_web_host, split_host_port if self.status == self.StatusChoices.SEALED: - return None + return [] - url = sanitize_extracted_url(fix_url_from_markdown(str(url or "").strip())) - if not url: - return None - try: - validate_url_length(url) - except ValueError as err: - print(f"[yellow][!] Skipping over-long discovered snapshot URL: {url[:120]}... ({err})[/yellow]") - return None if depth > self.max_depth: - return None - if not self.url_passes_filters(url, snapshot=parent_snapshot): - return None - if self.snapshot_set.filter(url=url).exists(): - return None - if not self.has_remaining_snapshot_capacity(): - return None + return [] + + config = get_config(crawl=self, snapshot=parent_snapshot) + allowlist = self.split_filter_patterns(config.get("URL_ALLOWLIST", "")) + denylist = self.split_filter_patterns(config.get("URL_DENYLIST", "")) + protected_subdomains = {"admin", "web", "api", "public"} + protected_hosts = set() + protected_roots = set() + for host_value in ( + get_listen_host(config=config), + get_admin_host(config=config), + get_web_host(config=config), + get_api_host(config=config), + get_public_host(config=config), + ): + if not host_value: + continue + protected_host = split_host_port(host_value)[0].strip(".") + if not protected_host: + continue + protected_hosts.add(protected_host) + host_parts = protected_host.split(".", 1) + if len(host_parts) == 2 and (host_parts[0] in protected_subdomains or host_parts[0].startswith("snap-")): + protected_roots.add(host_parts[1]) + else: + protected_roots.add(protected_host) + uses_subdomain_routing = bool(config.get("USES_SUBDOMAIN_ROUTING", False)) + + deduped_records: dict[str, Mapping[str, Any]] = {} + for record in records: + url = sanitize_extracted_url(fix_url_from_markdown(str(record.get("url") or "").strip())) + if not url or url in deduped_records: + continue + try: + validate_url_length(url) + except ValueError as err: + print(f"[yellow][!] Skipping over-long discovered snapshot URL: {url[:120]}... ({err})[/yellow]") + continue + parsed = urlparse(url) + host = (parsed.hostname or "").lower().strip(".") + is_internal_url = False + if parsed.scheme in ("http", "https") and host: + if host in protected_hosts: + is_internal_url = True + elif uses_subdomain_routing: + for protected_root in protected_roots: + 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-"): + is_internal_url = True + break + if is_internal_url: + print(f"[yellow][!] Skipping internal ArchiveBox discovered snapshot URL: {url}[/yellow]") + continue + if self.url_passes_compiled_filters(url, allowlist=allowlist, denylist=denylist): + deduped_records[url] = record + + if not deduped_records: + return [] + + existing_urls = set(self.snapshot_set.filter(url__in=deduped_records.keys()).values_list("url", flat=True)) + urls = [url for url in deduped_records.keys() if url not in existing_urls] + remaining = self.remaining_snapshot_capacity() + if remaining is not None: + urls = urls[:remaining] + if not urls: + return [] + + now = timezone.now() + snapshots = [ + Snapshot( + url=url, + timestamp=str((now + timedelta(microseconds=index)).timestamp()), + title=str(deduped_records[url].get("title") or "").strip()[:512] or None, + crawl=self, + parent_snapshot=parent_snapshot, + depth=depth, + status=Snapshot.StatusChoices.QUEUED, + retry_at=now, + bookmarked_at=now, + created_at=now, + ) + for index, url in enumerate(urls) + ] try: - snapshot = Snapshot.from_json( - { - "url": url, - "depth": depth, - "title": title, - "tags": tags, - "parent_snapshot_id": str(parent_snapshot.id), - "crawl_id": str(self.id), - }, - overrides={ - "crawl": self, - "snapshot": parent_snapshot, - "created_by_id": created_by_id or self.created_by_id, - }, - queue_for_extraction=False, - ) + created_snapshots = list(Snapshot.objects.bulk_create(snapshots)) except ValidationError as err: - print(f"[yellow][!] Skipping blocked discovered snapshot URL: {url} ({err})[/yellow]") - return None - if snapshot is None or snapshot.status == Snapshot.StatusChoices.SEALED: - return None + print(f"[yellow][!] Skipping blocked discovered snapshots: {err}[/yellow]") + return [] - snapshot.status = Snapshot.StatusChoices.QUEUED - snapshot.retry_at = timezone.now() - snapshot.save(update_fields=["status", "retry_at", "modified_at"]) - return snapshot + crawl_urls = {url for _raw_line, url in self._iter_url_lines() if url} + new_url_lines = [snapshot.url for snapshot in created_snapshots if snapshot.url not in crawl_urls] + if new_url_lines: + self.urls = (self.urls.rstrip() + "\n" + "\n".join(new_url_lines)).lstrip("\n") + self.save(update_fields=["urls", "modified_at"]) + + tag_names_by_url: dict[str, set[str]] = {} + for snapshot in created_snapshots: + tags = str(deduped_records[snapshot.url].get("tags") or "").strip() + if tags: + tag_names_by_url[snapshot.url] = {tag.strip() for tag in re.split(config.TAG_SEPARATOR_PATTERN, tags) if tag.strip()} + try: + snapshot.ensure_crawl_symlink() + except Exception: + pass + + tag_names = {tag for tags in tag_names_by_url.values() for tag in tags} + if tag_names: + tags_by_name = {tag.name: tag for tag in Tag.objects.filter(name__in=tag_names)} + missing_tags = [Tag(name=name) for name in sorted(tag_names - tags_by_name.keys())] + 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=tag_names)} + SnapshotTag.objects.bulk_create( + [ + SnapshotTag(snapshot=snapshot, tag=tags_by_name[tag_name]) + for snapshot in created_snapshots + for tag_name in tag_names_by_url.get(snapshot.url, set()) + if tag_name in tags_by_name + ], + ignore_conflicts=True, + ) + + return created_snapshots def install_declared_binaries(self, binary_names: set[str], machine=None) -> None: """ diff --git a/archivebox/hooks.py b/archivebox/hooks.py index 14870389..603b2c16 100644 --- a/archivebox/hooks.py +++ b/archivebox/hooks.py @@ -442,6 +442,7 @@ def run_hook( lib_bin_dir = resolved_config.LIB_BIN_DIR if lib_dir: env["LIB_DIR"] = str(lib_dir) + env["ABXPKG_LIB_DIR"] = str(lib_dir) if lib_bin_dir: env["LIB_BIN_DIR"] = str(lib_bin_dir) @@ -468,6 +469,7 @@ def run_hook( SKIP_KEYS = { "PATH", "LIB_DIR", + "ABXPKG_LIB_DIR", "LIB_BIN_DIR", "NODE_PATH", "NODE_MODULES_DIR", diff --git a/archivebox/machine/migrations/0014_process_machine_pro_pid_6eec8b_idx_and_more.py b/archivebox/machine/migrations/0014_process_machine_pro_pid_6eec8b_idx_and_more.py new file mode 100644 index 00000000..1d3b1066 --- /dev/null +++ b/archivebox/machine/migrations/0014_process_machine_pro_pid_6eec8b_idx_and_more.py @@ -0,0 +1,20 @@ +# Generated by Django 6.0.5 on 2026-05-26 18:19 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("machine", "0013_alter_machine_config"), + ] + + operations = [ + migrations.AddIndex( + model_name="process", + index=models.Index(fields=["pid", "started_at"], name="machine_pro_pid_6eec8b_idx"), + ), + migrations.AddIndex( + model_name="process", + index=models.Index(fields=["process_type", "worker_type", "pwd", "started_at"], name="machine_pro_process_b0411b_idx"), + ), + ] diff --git a/archivebox/machine/models.py b/archivebox/machine/models.py index 3a4f1a1b..cf8b6f49 100755 --- a/archivebox/machine/models.py +++ b/archivebox/machine/models.py @@ -52,8 +52,6 @@ PROCESS_RECHECK_INTERVAL = 60 # Re-validate every 60 seconds PID_REUSE_WINDOW = timedelta(hours=24) # Max age for considering a PID match valid PROCESS_TIMEOUT_GRACE = timedelta(seconds=30) # Extra margin before force-cleaning timed-out RUNNING rows START_TIME_TOLERANCE = 5.0 # Seconds tolerance for start time matching -LEGACY_MACHINE_CONFIG_KEYS = frozenset({"CHROMIUM_VERSION"}) -MACHINE_CONFIG_ALWAYS_ALLOWED_KEYS = frozenset({"ABX_INSTALL_CACHE"}) def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Binary | None: @@ -76,6 +74,13 @@ def _find_existing_binary_for_reference(machine: Machine, reference: str) -> Bin return qs.filter(name=reference).order_by("-modified_at").first() +def _canonical_binary_name(name: Any) -> str: + name = str(name or "").strip() + if "/" in name or "\\" in name or name.startswith("~"): + return Path(name).expanduser().name + return name + + def _get_process_binary_env_keys(plugin_name: str, hook_path: str, env: dict[str, Any] | None) -> list[str]: env = env or {} plugin_name = str(plugin_name or "").strip() @@ -122,13 +127,12 @@ def _get_process_binary_env_keys(plugin_name: str, hook_path: str, env: dict[str return keys -def _sanitize_machine_config(config: dict[str, Any] | None) -> dict[str, Any]: +def _sanitize_machine_config(config: dict[str, Any] | None, *, lib_dir: str | Path | None = None) -> dict[str, Any]: if not isinstance(config, dict): return {} - sanitized = {key: value for key, value in config.items() if key in MACHINE_CONFIG_ALWAYS_ALLOWED_KEYS or str(key).endswith("_BINARY")} - for key in LEGACY_MACHINE_CONFIG_KEYS: - sanitized.pop(key, None) + sanitized = {key: value for key, value in config.items() if str(key).endswith("_BINARY")} + active_lib_dir = Path(lib_dir).expanduser().resolve(strict=False) if lib_dir else None for key, value in list(sanitized.items()): if not str(key).endswith("_BINARY"): continue @@ -140,8 +144,16 @@ def _sanitize_machine_config(config: dict[str, Any] | None) -> dict[str, Any]: continue if "/" in value or value.startswith("~"): try: - if not Path(value).expanduser().exists(): + path = Path(value).expanduser() + if not path.exists(): sanitized.pop(key, None) + continue + if active_lib_dir is not None: + resolved_path = path.resolve(strict=False) + try: + resolved_path.relative_to(active_lib_dir) + except ValueError: + sanitized.pop(key, None) except OSError: sanitized.pop(key, None) return sanitized @@ -237,7 +249,9 @@ class Machine(ModelWithHealthStats): @classmethod def _sanitize_config(cls, machine: Machine) -> Machine: - sanitized = _sanitize_machine_config(machine.config) + from archivebox.config.constants import CONSTANTS + + sanitized = _sanitize_machine_config(machine.config, lib_dir=os.environ.get("LIB_DIR") or CONSTANTS.DEFAULT_LIB_DIR) current = machine.config or {} if sanitized != current: machine.config = sanitized @@ -521,7 +535,7 @@ class Binary(ModelWithHealthStats, ModelWithStateMachine): Returns: Binary instance or None """ - name = record.get("name") + name = _canonical_binary_name(record.get("name")) if not name: return None @@ -1094,6 +1108,8 @@ class Process(models.Model): indexes = [ models.Index(fields=["machine", "status", "retry_at"]), models.Index(fields=["binary", "exit_code"]), + models.Index(fields=["pid", "started_at"]), + models.Index(fields=["process_type", "worker_type", "pwd", "started_at"]), ] def __str__(self) -> str: diff --git a/archivebox/services/binary_service.py b/archivebox/services/binary_service.py index 16a44cd7..e9b3080f 100644 --- a/archivebox/services/binary_service.py +++ b/archivebox/services/binary_service.py @@ -1,5 +1,7 @@ from __future__ import annotations +from pathlib import Path + from asgiref.sync import sync_to_async from abx_dl.events import BinaryRequestEvent, BinaryEvent @@ -20,6 +22,7 @@ class BinaryService(BaseService): machine = await sync_to_async(Machine.current, thread_sensitive=True)() existing = await Binary.objects.filter(machine=machine, name=event.name).afirst() + cache_invalidated = False if existing and existing.status == Binary.StatusChoices.INSTALLED: changed = False if event.binproviders and existing.binproviders != event.binproviders: @@ -29,7 +32,10 @@ class BinaryService(BaseService): existing.overrides = event.overrides changed = True if changed: - await existing.asave(update_fields=["binproviders", "overrides", "modified_at"]) + existing.status = Binary.StatusChoices.QUEUED + existing.retry_at = None + cache_invalidated = True + await existing.asave(update_fields=["binproviders", "overrides", "status", "retry_at", "modified_at"]) elif existing is None: await Binary.objects.acreate( machine=machine, @@ -39,19 +45,50 @@ class BinaryService(BaseService): status=Binary.StatusChoices.QUEUED, ) - installed = ( - await Binary.objects.filter(machine=machine, name=event.name, status=Binary.StatusChoices.INSTALLED) - .exclude(abspath="") - .exclude(abspath__isnull=True) - .order_by("-modified_at") - .afirst() - ) + installed = None + if not cache_invalidated: + installed = ( + await Binary.objects.filter(machine=machine, name=event.name, status=Binary.StatusChoices.INSTALLED) + .exclude(abspath="") + .exclude(abspath__isnull=True) + .order_by("-modified_at") + .afirst() + ) + if installed is not None and not await sync_to_async(Path(installed.abspath).expanduser().exists, thread_sensitive=True)(): + installed.status = Binary.StatusChoices.QUEUED + installed.retry_at = None + await installed.asave(update_fields=["status", "retry_at", "modified_at"]) + installed = None + if installed is not None and event.overrides and installed.overrides != event.overrides: + installed.status = Binary.StatusChoices.QUEUED + installed.retry_at = None + await installed.asave(update_fields=["status", "retry_at", "modified_at"]) + installed = None cached = None if installed is not None: + from archivebox.config.common import get_config from abxpkg import BinProvider, PROVIDER_CLASS_BY_NAME binary_env: dict[str, str] = {} + installed_path = Path(installed.abspath).expanduser().resolve(strict=False) + active_lib_dir = ( + Path(str((await sync_to_async(get_config, thread_sensitive=True)()).get("LIB_DIR", ""))) + .expanduser() + .resolve( + strict=False, + ) + ) provider_name = (installed.binprovider or installed.binproviders.split(",", 1)[0]).strip() + if active_lib_dir and provider_name in {"npm", "pip", "puppeteer", "uv", "deno", "gem", "cargo", "goget", "nix", "bash"}: + try: + installed_path.relative_to(active_lib_dir) + except ValueError: + installed.status = Binary.StatusChoices.QUEUED + installed.retry_at = None + await installed.asave(update_fields=["status", "retry_at", "modified_at"]) + installed = None + if installed is None: + return None provider_class = PROVIDER_CLASS_BY_NAME.get(provider_name) if provider_class is not None: provider = provider_class() diff --git a/archivebox/services/machine_service.py b/archivebox/services/machine_service.py index 1d0ba1c4..554e7803 100644 --- a/archivebox/services/machine_service.py +++ b/archivebox/services/machine_service.py @@ -16,15 +16,17 @@ class MachineService(BaseService): async def on_MachineEvent__save_to_db(self, event: MachineEvent) -> None: from archivebox.machine.models import Machine, _sanitize_machine_config + from archivebox.config.common import get_config if event.config_type != "derived": return machine = await sync_to_async(Machine.current, thread_sensitive=True)() + lib_dir = await sync_to_async(lambda: get_config(include_machine=False).LIB_DIR, thread_sensitive=True)() config = dict(machine.config or {}) if event.config is not None: - config.update(_sanitize_machine_config(event.config)) + config.update(_sanitize_machine_config(event.config, lib_dir=lib_dir)) elif event.method == "update": key = event.key.replace("config/", "", 1).strip() if key: @@ -36,5 +38,5 @@ class MachineService(BaseService): else: return - machine.config = _sanitize_machine_config(config) + machine.config = _sanitize_machine_config(config, lib_dir=lib_dir) await machine.asave(update_fields=["config", "modified_at"]) diff --git a/archivebox/services/process_service.py b/archivebox/services/process_service.py index 229fc296..0419b354 100644 --- a/archivebox/services/process_service.py +++ b/archivebox/services/process_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from datetime import datetime from typing import ClassVar @@ -7,7 +8,7 @@ from asgiref.sync import sync_to_async from django.utils import timezone from abxbus import BaseEvent -from abx_dl.events import ProcessCompletedEvent, ProcessStartedEvent +from abx_dl.events import CrawlCleanupEvent, CrawlCompletedEvent, ProcessCompletedEvent, ProcessStartedEvent from abx_dl.services.base import BaseService @@ -26,23 +27,38 @@ def parse_event_datetime(value: str | None): def current_network_interface_with_machine(): from archivebox.machine.models import NetworkInterface - current_iface = NetworkInterface.current(refresh=True) + current_iface = NetworkInterface.current() return NetworkInterface.objects.select_related("machine").get(id=current_iface.id) class ProcessService(BaseService): - LISTENS_TO: ClassVar[list[type[BaseEvent]]] = [ProcessStartedEvent, ProcessCompletedEvent] + LISTENS_TO: ClassVar[list[type[BaseEvent]]] = [ + ProcessStartedEvent, + ProcessCompletedEvent, + CrawlCleanupEvent, + CrawlCompletedEvent, + ] EMITS: ClassVar[list[type[BaseEvent]]] = [] def __init__(self, bus): + self._iface = None + self._completed_queue: asyncio.Queue[ProcessCompletedEvent | None] = asyncio.Queue() + self._completed_worker: asyncio.Task | None = None super().__init__(bus) self.bus.on(ProcessStartedEvent, self.on_ProcessStartedEvent__save_to_db) self.bus.on(ProcessCompletedEvent, self.on_ProcessCompletedEvent__save_to_db) + self.bus.on(CrawlCleanupEvent, self.on_CrawlCleanupEvent__flush_completed) + self.bus.on(CrawlCompletedEvent, self.on_CrawlCompletedEvent__flush_completed) + + async def current_iface(self): + if self._iface is None: + self._iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)() + return self._iface async def on_ProcessStartedEvent__save_to_db(self, event: ProcessStartedEvent) -> None: from archivebox.machine.models import Process - iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)() + iface = await self.current_iface() process_type = event.process_type or ( Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK ) @@ -50,15 +66,15 @@ class ProcessService(BaseService): started_at = parse_event_datetime(event.start_ts) if started_at is None: raise ValueError("ProcessStartedEvent.start_ts is required") - process_query = Process.objects.filter( - process_type=process_type, - worker_type=worker_type, - pwd=event.output_dir, - cmd=[event.hook_path, *event.hook_args], - started_at=started_at, - ) if event.pid: - process_query = process_query.filter(pid=event.pid) + process_query = Process.objects.filter(pid=event.pid, started_at=started_at) + else: + process_query = Process.objects.filter( + process_type=process_type, + worker_type=worker_type, + pwd=event.output_dir, + started_at=started_at, + ) process = await process_query.order_by("-modified_at").afirst() if process is None: process = await Process.objects.acreate( @@ -96,12 +112,50 @@ class ProcessService(BaseService): plugin_name=event.plugin_name, hook_path=event.hook_path, ) - await process.asave() + await Process.objects.filter(id=process.id).aupdate( + pwd=process.pwd, + cmd=process.cmd, + env=process.env, + timeout=process.timeout, + pid=process.pid, + url=process.url, + process_type=process.process_type, + worker_type=process.worker_type, + started_at=process.started_at, + status=process.status, + retry_at=process.retry_at, + binary_id=process.binary_id, + modified_at=timezone.now(), + ) + + async def _completed_worker_loop(self) -> None: + while True: + event = await self._completed_queue.get() + try: + if event is None: + return + await self._save_completed_process_to_db(event) + finally: + self._completed_queue.task_done() + + def _ensure_completed_worker(self) -> None: + if self._completed_worker is None or self._completed_worker.done(): + self._completed_worker = asyncio.create_task(self._completed_worker_loop()) async def on_ProcessCompletedEvent__save_to_db(self, event: ProcessCompletedEvent) -> None: + self._ensure_completed_worker() + await self._completed_queue.put(event) + + async def on_CrawlCleanupEvent__flush_completed(self, event: CrawlCleanupEvent) -> None: + await self._completed_queue.join() + + async def on_CrawlCompletedEvent__flush_completed(self, event: CrawlCompletedEvent) -> None: + await self._completed_queue.join() + + async def _save_completed_process_to_db(self, event: ProcessCompletedEvent) -> None: from archivebox.machine.models import Process - iface = await sync_to_async(current_network_interface_with_machine, thread_sensitive=True)() + iface = await self.current_iface() process_type = event.process_type or ( Process.TypeChoices.BINARY if event.hook_name.startswith("on_BinaryRequest") else Process.TypeChoices.HOOK ) @@ -109,18 +163,18 @@ class ProcessService(BaseService): started_at = parse_event_datetime(event.start_ts) if started_at is None: raise ValueError("ProcessCompletedEvent.start_ts is required") - process_query = Process.objects.filter( - process_type=process_type, - worker_type=worker_type, - pwd=event.output_dir, - cmd=[event.hook_path, *event.hook_args], - started_at=started_at, - ) if event.pid: - process_query = process_query.filter(pid=event.pid) + process_query = Process.objects.filter(pid=event.pid, started_at=started_at) + else: + process_query = Process.objects.filter( + process_type=process_type, + worker_type=worker_type, + pwd=event.output_dir, + started_at=started_at, + ) process = await process_query.order_by("-modified_at").afirst() if process is None: - process = await Process.objects.acreate( + await Process.objects.acreate( machine=iface.machine, iface=iface, process_type=process_type, @@ -135,28 +189,28 @@ class ProcessService(BaseService): status=Process.StatusChoices.RUNNING, retry_at=None, ) - elif process.iface_id != iface.id or process.machine_id != iface.machine_id: - process.iface = iface - process.machine = iface.machine - await process.asave(update_fields=["iface", "machine", "modified_at"]) + process = await process_query.order_by("-modified_at").afirst() + if process is None: + return - process.pwd = event.output_dir - if not process.cmd: - process.cmd = [event.hook_path, *event.hook_args] - process.env = event.env - process.pid = event.pid or process.pid - process.url = event.url or process.url - process.process_type = process_type or process.process_type - process.worker_type = worker_type or process.worker_type - process.started_at = started_at - process.ended_at = parse_event_datetime(event.end_ts) or timezone.now() - process.stdout = event.stdout - process.stderr = event.stderr - process.exit_code = event.exit_code - process.status = process.StatusChoices.EXITED - process.retry_at = None - await sync_to_async(process.hydrate_binary_from_context, thread_sensitive=True)( - plugin_name=event.plugin_name, - hook_path=event.hook_path, - ) - await process.asave() + missing_cmd = not process.cmd + updates = { + "machine_id": iface.machine_id, + "iface_id": iface.id, + "pwd": event.output_dir, + "pid": event.pid or process.pid, + "url": event.url or process.url, + "process_type": process_type or process.process_type, + "worker_type": worker_type or process.worker_type, + "started_at": started_at, + "ended_at": parse_event_datetime(event.end_ts) or timezone.now(), + "stdout": event.stdout, + "stderr": event.stderr, + "exit_code": event.exit_code, + "status": Process.StatusChoices.EXITED, + "retry_at": None, + "modified_at": timezone.now(), + } + if missing_cmd: + updates["cmd"] = [event.hook_path, *event.hook_args] + await Process.objects.filter(id=process.id).aupdate(**updates) diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index ef7ba6d9..42339240 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -4,6 +4,7 @@ import asyncio import contextvars import json import os +import signal import shutil import subprocess import sys @@ -206,6 +207,7 @@ class CrawlRunner: self.initial_snapshot_ids = snapshot_ids self.snapshot_tasks: dict[str, asyncio.Task[None]] = {} self.snapshot_semaphore = asyncio.Semaphore(1) + self.max_concurrent_snapshots = 1 self.persona = None self.base_config: dict[str, Any] = {} self.derived_config: dict[str, Any] = {} @@ -214,11 +216,60 @@ class CrawlRunner: self._live_stream = None self.root_crawl_event_id: str | None = None self.root_crawl_start_event_id: str | None = None + self._run_task: asyncio.Task[None] | None = None self._skip_wait_until_idle = False + self._signal_abort_requested = False + + def _install_signal_handlers(self) -> list[tuple[signal.Signals, Any, bool]]: + loop = asyncio.get_running_loop() + installed: list[tuple[signal.Signals, Any, bool]] = [] + for sig in (signal.SIGINT, signal.SIGTERM): + previous = signal.getsignal(sig) + + def request_abort(sig=sig) -> None: + self._request_abort_from_signal(sig) + + try: + loop.add_signal_handler(sig, request_abort) + installed.append((sig, previous, True)) + except (NotImplementedError, RuntimeError): + signal.signal(sig, lambda _signum, _frame, sig=sig: self._request_abort_from_signal(sig)) + installed.append((sig, previous, False)) + return installed + + def _restore_signal_handlers(self, installed: list[tuple[signal.Signals, Any, bool]]) -> None: + loop = asyncio.get_running_loop() + for sig, previous, installed_on_loop in reversed(installed): + if installed_on_loop: + loop.remove_signal_handler(sig) + signal.signal(sig, previous) + + def _request_abort_from_signal(self, sig: signal.Signals) -> None: + if self._signal_abort_requested: + if self._run_task is not None and not self._run_task.done(): + self._run_task.cancel() + return + self._signal_abort_requested = True + self._skip_wait_until_idle = True + asyncio.create_task(self.abort_from_signal(sig.name)) + + async def abort_from_signal(self, signal_name: str) -> None: + from archivebox.crawls.models import Crawl + + await sync_to_async( + Crawl.objects.filter(id=self.crawl.id).exclude(status=Crawl.StatusChoices.SEALED).update, + thread_sensitive=False, + )( + status=Crawl.StatusChoices.STARTED, + retry_at=timezone.now(), + modified_at=timezone.now(), + ) async def crawl_is_cancelled(self) -> bool: from archivebox.crawls.models import Crawl + if self._signal_abort_requested: + return True return await Crawl.objects.filter(id=self.crawl.id, status=Crawl.StatusChoices.SEALED).aexists() async def watch_for_cancelled_crawl(self, parent_event: BaseEvent, *, poll_interval: float = 1.0) -> None: @@ -239,10 +290,13 @@ class CrawlRunner: runtime="archivebox", crawl_id=str(self.crawl.id), ) + installed_signal_handlers = self._install_signal_handlers() root_snapshot_id: str | None = None try: + self._run_task = asyncio.current_task() snapshot_ids = await sync_to_async(self.load_run_state, thread_sensitive=True)() - max_concurrent_snapshots = max(1, int(self.base_config.CRAWL_MAX_CONCURRENT_SNAPSHOTS)) + max_concurrent_snapshots = max(1, int(self.base_config.get("CRAWL_MAX_CONCURRENT_SNAPSHOTS", 1))) + self.max_concurrent_snapshots = max_concurrent_snapshots self.snapshot_semaphore = asyncio.Semaphore(max_concurrent_snapshots) live_ui = self._create_live_ui() with live_ui if live_ui is not None else nullcontext(): @@ -259,6 +313,8 @@ class CrawlRunner: root_snapshot_id = snapshot_ids[0] await self.run_crawl(root_snapshot_id, snapshot_ids) finally: + self._run_task = None + self._restore_signal_handlers(installed_signal_handlers) await heartbeat.stop() if not self._skip_wait_until_idle: await self.bus.wait_until_idle() @@ -271,6 +327,8 @@ class CrawlRunner: await sync_to_async(self.finalize_run_state, thread_sensitive=True)() async def enqueue_snapshot(self, snapshot_id: str) -> None: + if await self.crawl_is_cancelled(): + return task = self.snapshot_tasks.get(snapshot_id) if task is not None and not task.done(): return @@ -285,6 +343,7 @@ class CrawlRunner: async def wait_for_snapshot_tasks(self) -> None: task_errors: list[Exception] = [] + stop_scheduling = False while True: pending_tasks: list[asyncio.Task[None]] = [] for snapshot_id, task in list(self.snapshot_tasks.items()): @@ -300,15 +359,18 @@ class CrawlRunner: await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)() except Exception as err: task_errors.append(err) + stop_scheduling = True continue pending_tasks.append(task) if not pending_tasks: + if task_errors: + if len(task_errors) == 1: + raise task_errors[0] + raise ExceptionGroup("One or more snapshot tasks failed", task_errors) + if stop_scheduling: + return await self.enqueue_pending_snapshots_from_projection() if not self.snapshot_tasks: - if task_errors: - if len(task_errors) == 1: - raise task_errors[0] - raise ExceptionGroup("One or more snapshot tasks failed", task_errors) return continue done, _pending = await asyncio.wait(pending_tasks, return_when=asyncio.FIRST_COMPLETED) @@ -326,22 +388,55 @@ class CrawlRunner: await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)() except Exception as err: task_errors.append(err) - await self.enqueue_pending_snapshots_from_projection() + stop_scheduling = True + if self.snapshot_tasks and await self.crawl_is_cancelled(): + stop_scheduling = True + if not stop_scheduling: + await self.enqueue_pending_snapshots_from_projection() + + async def drain_snapshot_tasks(self) -> None: + task_errors: list[Exception] = [] + while self.snapshot_tasks: + done, _pending = await asyncio.wait(list(self.snapshot_tasks.values()), return_when=asyncio.FIRST_COMPLETED) + for task in done: + for snapshot_id, tracked_task in list(self.snapshot_tasks.items()): + if tracked_task is task: + self.snapshot_tasks.pop(snapshot_id, None) + break + try: + task.result() + except asyncio.CancelledError as err: + if _is_external_task_cancelled(err): + raise + await sync_to_async(recover_orphaned_snapshots, thread_sensitive=True)() + await sync_to_async(recover_orphaned_crawls, thread_sensitive=True)() + except Exception as err: + task_errors.append(err) + if task_errors: + if len(task_errors) == 1: + raise task_errors[0] + raise ExceptionGroup("One or more snapshot tasks failed", task_errors) async def enqueue_pending_snapshots_from_projection(self) -> None: from archivebox.core.models import Snapshot if not isinstance(get_current_event(), CrawlStartEvent): return + if await self.crawl_is_cancelled(): + return + active_snapshot_ids = [snapshot_id for snapshot_id, task in self.snapshot_tasks.items() if not task.done()] + available_slots = max(0, self.max_concurrent_snapshots - len(active_snapshot_ids)) + if available_slots <= 0: + return pending_snapshot_ids = await sync_to_async( - lambda: [ - str(snapshot_id) - for snapshot_id in self.crawl.snapshot_set.exclude(status=Snapshot.StatusChoices.SEALED) + lambda: list( + self.crawl.snapshot_set.exclude(status=Snapshot.StatusChoices.SEALED) + .exclude(id__in=active_snapshot_ids) .filter(retry_at__lte=timezone.now()) .order_by("depth", "created_at") - .values_list("id", flat=True) - ], + .values_list("id", flat=True)[:available_slots], + ), thread_sensitive=True, )() for snapshot_id in pending_snapshot_ids: @@ -361,8 +456,8 @@ class CrawlRunner: current_process.machine = current_iface.machine current_process.save(update_fields=["iface", "machine", "modified_at"]) self.persona = self.crawl.resolve_persona() - self.base_config = get_config(crawl=self.crawl) - self.derived_config = _sanitize_machine_config(Machine.current().config) + self.base_config = get_config(crawl=self.crawl, include_machine=False) + self.derived_config = _sanitize_machine_config(Machine.current().config, lib_dir=self.base_config["LIB_DIR"]) self.crawl_output_dir = str(self.crawl.output_dir) self.base_config["ABX_RUNTIME"] = "archivebox" if self.selected_plugins is None: @@ -463,7 +558,7 @@ class CrawlRunner: from archivebox.config.common import get_config snapshot = Snapshot.objects.select_related("crawl").get(id=snapshot_id) - config = get_config(crawl=self.crawl, snapshot=snapshot) + config = get_config(crawl=self.crawl, snapshot=snapshot, include_machine=False) config.update(self.base_config) config["CRAWL_DIR"] = self.crawl_output_dir config["SNAP_DIR"] = str(snapshot.output_dir) @@ -514,24 +609,13 @@ class CrawlRunner: if parent_snapshot is None: return - for record in discovered_urls: - url = str(record.get("url") or "").strip() - if not url: - continue - child_snapshot = await sync_to_async(self.crawl.create_discovered_snapshot, thread_sensitive=True)( - parent_snapshot, - url=url, - depth=parent_snapshot.depth + 1, - title=str(record.get("title") or "").strip(), - tags=str(record.get("tags") or "").strip(), - ) - if child_snapshot is None: - has_capacity = await sync_to_async(self.crawl.has_remaining_snapshot_capacity, thread_sensitive=True)() - if has_capacity: - continue - break - if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent): - await self.enqueue_snapshot(str(child_snapshot.id)) + await sync_to_async(self.crawl.create_discovered_snapshots, thread_sensitive=True)( + parent_snapshot, + discovered_urls, + depth=parent_snapshot.depth + 1, + ) + if self.process_discovered_snapshots_inline and isinstance(get_current_event(), CrawlStartEvent): + await self.enqueue_pending_snapshots_from_projection() async def run_crawl(self, root_snapshot_id: str, snapshot_ids: list[str]) -> None: snapshot = await sync_to_async(self.load_snapshot_payload, thread_sensitive=True)(root_snapshot_id) @@ -618,6 +702,10 @@ class CrawlRunner: if event.event_id != self.root_crawl_start_event_id: return for snapshot_id in snapshot_ids: + if sum(1 for task in self.snapshot_tasks.values() if not task.done()) >= self.max_concurrent_snapshots: + break + if await self.crawl_is_cancelled(): + break await self.enqueue_snapshot(snapshot_id) await self.wait_for_snapshot_tasks() @@ -652,6 +740,8 @@ class CrawlRunner: self.root_crawl_start_event_id = crawl_start_event.event_id await _run_event_now(event.emit(crawl_start_event), None) finally: + if self.snapshot_tasks: + await self.drain_snapshot_tasks() await _run_event_now( event.emit( CrawlCleanupEvent( @@ -749,7 +839,7 @@ class CrawlRunner: snapshot_hooks = [(plugin, hook) for plugin in plugins.values() for hook in plugin.filter_hooks("Snapshot")] snapshot_phase_timeout = compute_phase_timeout(snapshot_hooks, config) await _emit_machine_config(self.bus, config=config, derived_config=derived_config, parent_event=crawl_start_event) - HookSnapshotService( + snapshot_service = HookSnapshotService( self.bus, url=snapshot["url"], snapshot=abx_snapshot, @@ -760,27 +850,30 @@ class CrawlRunner: snapshot_cleanup_phase_timeout=snapshot_phase_timeout, abort_requested=self.crawl_is_cancelled, ) - snapshot_event = SnapshotEvent( - url=snapshot["url"], - snapshot_id=snapshot["id"], - output_dir=str(output_dir), - depth=int(snapshot["depth"]), - event_timeout=snapshot_phase_timeout, - event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout), - ) - emitted_snapshot_event = crawl_start_event.emit(snapshot_event) - await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout) - completed_snapshot = await self.bus.find( - SnapshotCompletedEvent, - child_of=emitted_snapshot_event, - past=True, - future=snapshot_phase_timeout, - ) - if completed_snapshot is None: - raise RuntimeError(f"Snapshot {snapshot_id} did not complete") - await completed_snapshot.wait(timeout=snapshot_phase_timeout) - await completed_snapshot.event_results_list() - await self.enqueue_discovered_snapshots_from_outputs(snapshot) + try: + snapshot_event = SnapshotEvent( + url=snapshot["url"], + snapshot_id=snapshot["id"], + output_dir=str(output_dir), + depth=int(snapshot["depth"]), + event_timeout=snapshot_phase_timeout, + event_handler_slow_timeout=slow_warning_timeout(snapshot_phase_timeout), + ) + emitted_snapshot_event = crawl_start_event.emit(snapshot_event) + await _run_event_now(emitted_snapshot_event, snapshot_phase_timeout) + completed_snapshot = await self.bus.find( + SnapshotCompletedEvent, + child_of=emitted_snapshot_event, + past=True, + future=snapshot_phase_timeout, + ) + if completed_snapshot is None: + raise RuntimeError(f"Snapshot {snapshot_id} did not complete") + await completed_snapshot.wait(timeout=snapshot_phase_timeout) + await completed_snapshot.event_results_list() + await self.enqueue_discovered_snapshots_from_outputs(snapshot) + finally: + snapshot_service.close() def seal_snapshot_due_to_limit(self, snapshot_id: str) -> None: from archivebox.core.models import Snapshot @@ -815,13 +908,13 @@ def run_crawl( async def _run_binary(binary_id: str) -> None: from archivebox.config.common import get_config - from archivebox.machine.models import Binary, Machine + from archivebox.machine.models import Binary, Machine, _sanitize_machine_config binary = await Binary.objects.aget(id=binary_id) plugins = discover_plugins() - config = get_config() + config = get_config(include_machine=False) machine = await sync_to_async(Machine.current, thread_sensitive=True)() - derived_config = _normalize_runtime_config(dict(machine.config)) + derived_config = _normalize_runtime_config(_sanitize_machine_config(machine.config, lib_dir=config["LIB_DIR"])) config["ABX_RUNTIME"] = "archivebox" config = _normalize_runtime_config(config) bus = create_bus(name=_bus_name("ArchiveBox_binary", str(binary.id)), total_timeout=1800.0) @@ -867,12 +960,12 @@ def run_binary(binary_id: str) -> None: async def _run_install(plugin_names: list[str] | None = None) -> None: from archivebox.config.common import get_config - from archivebox.machine.models import Machine + from archivebox.machine.models import Machine, _sanitize_machine_config plugins = discover_plugins() - config = get_config() + config = get_config(include_machine=False) machine = await sync_to_async(Machine.current, thread_sensitive=True)() - derived_config = _normalize_runtime_config(dict(machine.config)) + derived_config = _normalize_runtime_config(_sanitize_machine_config(machine.config, lib_dir=config["LIB_DIR"])) config["ABX_RUNTIME"] = "archivebox" config = _normalize_runtime_config(config) bus = create_bus(name="ArchiveBox_install", total_timeout=3600.0) diff --git a/archivebox/services/snapshot_service.py b/archivebox/services/snapshot_service.py index ac63cb9f..94f395fd 100644 --- a/archivebox/services/snapshot_service.py +++ b/archivebox/services/snapshot_service.py @@ -1,7 +1,5 @@ from __future__ import annotations -from pathlib import Path - from asgiref.sync import sync_to_async from django.utils import timezone from abx_dl.events import SnapshotCompletedEvent, SnapshotEvent @@ -20,67 +18,18 @@ class SnapshotService(BaseService): self.bus.on(SnapshotEvent, self.on_SnapshotEvent) self.bus.on(SnapshotCompletedEvent, self.on_SnapshotCompletedEvent) - async def _upsert_discovered_snapshot(self, parent_snapshot, *, url: str, depth: int, title: str = "", tags: str = "") -> str | None: - crawl = parent_snapshot.crawl - if depth > crawl.max_depth: - return None - stop_reason = await sync_to_async(self._crawl_limit_stop_reason, thread_sensitive=True)(crawl) - if stop_reason == "crawl_max_size": - return None - snapshot = await sync_to_async(crawl.create_discovered_snapshot, thread_sensitive=True)( - parent_snapshot, - url=url, - depth=depth, - title=title, - tags=tags, - ) - if snapshot is None: - return None - return str(snapshot.id) - async def on_SnapshotEvent(self, event: SnapshotEvent) -> None: from archivebox.core.models import Snapshot - from archivebox.crawls.models import Crawl - crawl = await Crawl.objects.aget(id=self.crawl_id) - snapshot_id: str | None = None - snapshot = await Snapshot.objects.filter(id=event.snapshot_id, crawl=crawl).afirst() + snapshot = await Snapshot.objects.filter(id=event.snapshot_id, crawl_id=self.crawl_id).afirst() if snapshot is not None: snapshot.status = Snapshot.StatusChoices.STARTED snapshot.retry_at = None await snapshot.asave(update_fields=["status", "retry_at", "modified_at"]) - snapshot_id = str(snapshot.id) - elif event.depth > 0: - parent_event = await self.bus.find( - SnapshotEvent, - past=True, - future=False, - where=lambda candidate: candidate.depth == event.depth - 1 and self.bus.event_is_child_of(event, candidate), - ) - parent_snapshot = None - if parent_event is not None: - parent_snapshot = ( - await Snapshot.objects.select_related("crawl", "crawl__created_by") - .filter(id=parent_event.snapshot_id, crawl=crawl) - .afirst() - ) - if parent_snapshot is not None: - snapshot_id = await self._upsert_discovered_snapshot( - parent_snapshot, - url=event.url, - depth=event.depth, - ) - - if snapshot_id: - snapshot = await Snapshot.objects.filter(id=snapshot_id).select_related("crawl", "crawl__created_by").afirst() - if snapshot is not None: - await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)() - if snapshot_id and event.depth > 0: - await self.schedule_snapshot(snapshot_id) + await sync_to_async(snapshot.ensure_crawl_symlink, thread_sensitive=True)() async def on_SnapshotCompletedEvent(self, event: SnapshotCompletedEvent) -> None: - from archivebox.crawls.models import Crawl from archivebox.core.models import Snapshot snapshot = await Snapshot.objects.select_related("crawl", "crawl__created_by").filter(id=event.snapshot_id).afirst() @@ -112,27 +61,8 @@ class SnapshotService(BaseService): await sync_to_async(snapshot.write_index_jsonl, thread_sensitive=True)() await sync_to_async(snapshot.write_json_details, thread_sensitive=True)() await sync_to_async(snapshot.write_html_details, thread_sensitive=True)() - stop_reason = await sync_to_async(self._crawl_limit_stop_reason, thread_sensitive=True)(snapshot.crawl) - if snapshot.depth < snapshot.crawl.max_depth and stop_reason != "crawl_max_size": - from archivebox.hooks import collect_urls_from_plugins - - discovered_urls = await sync_to_async(collect_urls_from_plugins, thread_sensitive=True)(Path(snapshot.output_dir)) - for record in discovered_urls: - discovered_snapshot_id = await self._upsert_discovered_snapshot( - snapshot, - url=str(record.get("url") or "").strip(), - depth=snapshot.depth + 1, - title=str(record.get("title") or "").strip(), - tags=str(record.get("tags") or "").strip(), - ) - if discovered_snapshot_id: - await self.schedule_snapshot(discovered_snapshot_id) finally: - is_finished = await sync_to_async(snapshot.crawl.is_finished, thread_sensitive=True)() - if is_finished and snapshot.crawl.status != Crawl.StatusChoices.SEALED: - snapshot.crawl.status = Crawl.StatusChoices.SEALED - snapshot.crawl.retry_at = None - await snapshot.crawl.asave(update_fields=["status", "retry_at", "modified_at"]) + pass def _crawl_limit_stop_reason(self, crawl) -> str: config = dict(crawl.config or {}) diff --git a/archivebox/templates/admin/base.html b/archivebox/templates/admin/base.html index e00e73e2..2c8b4198 100644 --- a/archivebox/templates/admin/base.html +++ b/archivebox/templates/admin/base.html @@ -1506,7 +1506,7 @@ - +