From e7b3f1c4855d05b7c1e5dff7ff4fdbb22132a774 Mon Sep 17 00:00:00 2001 From: Nick Sweeting Date: Wed, 2 Sep 2026 04:44:55 -0700 Subject: [PATCH] Adopt canonical abx-dl execution plan API --- archivebox/cli/archivebox_add.py | 7 +-- archivebox/cli/archivebox_extract.py | 4 +- archivebox/cli/archivebox_install.py | 7 +-- archivebox/cli/archivebox_update.py | 4 +- archivebox/cli/archivebox_version.py | 7 ++- archivebox/config/common.py | 6 +-- archivebox/services/process_service.py | 4 +- archivebox/services/runner.py | 54 +++++++++---------- archivebox/tests/conftest.py | 9 +--- .../tests/test_archive_result_service.py | 8 +-- archivebox/tests/test_recursive_crawl.py | 4 +- 11 files changed, 54 insertions(+), 60 deletions(-) diff --git a/archivebox/cli/archivebox_add.py b/archivebox/cli/archivebox_add.py index 1f5152ae..b8225a76 100644 --- a/archivebox/cli/archivebox_add.py +++ b/archivebox/cli/archivebox_add.py @@ -421,13 +421,14 @@ def main(**kwargs): if only_new is not None: kwargs["config"] = {"ONLY_NEW": bool(only_new)} if extract: - from abx_dl.models import discover_plugins, plugins_matching_output + from archivebox.plugins.discovery import get_plugin_catalog - all_plugins = discover_plugins() + catalog = get_plugin_catalog() + all_plugins = catalog.plugins tokens = [token.strip() for token in extract.split(",") if token.strip()] plugin_names = {name.lower(): name for name in all_plugins} selected = [plugin_names[token.lower()] for token in tokens if token.lower() in plugin_names] - selected += plugins_matching_output(all_plugins, tokens) + selected += catalog.matching_output(tokens) if not selected: raise click.UsageError(f"No plugins found matching extract types: {extract}") existing = [token.strip() for token in (kwargs.get("plugins") or "").split(",") if token.strip()] diff --git a/archivebox/cli/archivebox_extract.py b/archivebox/cli/archivebox_extract.py index 88a6351c..289b9274 100644 --- a/archivebox/cli/archivebox_extract.py +++ b/archivebox/cli/archivebox_extract.py @@ -127,7 +127,7 @@ def run_plugins( from archivebox.core.models import Snapshot from archivebox.core.models import ArchiveResult from archivebox.services.runner import run_crawl - from abx_dl.models import discover_plugins + from archivebox.plugins.discovery import get_plugin_catalog is_tty = sys.stdout.isatty() @@ -207,7 +207,7 @@ def run_plugins( if snapshot_id in existing_snapshot_ids for plugin_name in plugin_names ) - plugins_by_name = discover_plugins(runtime="archivebox") + plugins_by_name = get_plugin_catalog().plugins requested_rows: set[tuple[str, str, str]] = set() for snapshot_id, plugin_name in requested_pairs: exact_hook_names = { diff --git a/archivebox/cli/archivebox_install.py b/archivebox/cli/archivebox_install.py index 6dfc4796..d3c0acf8 100755 --- a/archivebox/cli/archivebox_install.py +++ b/archivebox/cli/archivebox_install.py @@ -17,9 +17,10 @@ def _resolve_install_targets( requested_names: tuple[str, ...], ) -> tuple[list[str], list[str]]: """Resolve plugin names and declared binary aliases, leaving unknown tokens raw.""" - from abx_dl.models import discover_plugins, filter_plugins + from archivebox.plugins.discovery import get_plugin_catalog - plugins = discover_plugins(runtime="archivebox") + catalog = get_plugin_catalog() + plugins = catalog.plugins plugin_names_by_lower = {plugin_name.lower(): plugin_name for plugin_name in plugins} plugin_names_by_binary_alias: dict[str, set[str]] = {} for plugin_name, plugin in plugins.items(): @@ -47,7 +48,7 @@ def _resolve_install_targets( else: raw_binary_names.append(name) - selected_plugins = filter_plugins(plugins, requested_plugins, include_providers=True) if requested_plugins else {} + selected_plugins = catalog.select(requested_plugins).plugins if requested_plugins else {} selected_plugin_names = {name.lower() for name in selected_plugins} raw_binary_names = [name for name in raw_binary_names if name.lower() not in selected_plugin_names] return sorted(selected_plugins), sorted(set(raw_binary_names)) diff --git a/archivebox/cli/archivebox_update.py b/archivebox/cli/archivebox_update.py index 0b98f5ee..39cc71b2 100644 --- a/archivebox/cli/archivebox_update.py +++ b/archivebox/cli/archivebox_update.py @@ -77,7 +77,7 @@ def reindex_snapshots( ) -> dict[str, Any]: from archivebox.cli.archivebox_extract import run_plugins from archivebox.core.models import ArchiveResult, Snapshot - from abx_dl.models import discover_plugins + from archivebox.plugins.discovery import get_plugin_catalog # Search backfill is the one maintenance hook allowed to execute without # reopening a Snapshot. Restrict that exception to already-sealed rows; @@ -86,7 +86,7 @@ def reindex_snapshots( stats: dict[str, Any] = {"processed": 0, "requested": 0, "queued": 0, "skipped_queued": 0, "reindexed": 0, "snapshot_ids": []} records: list[dict[str, str]] = [] - plugins_by_name = discover_plugins(runtime="archivebox") + plugins_by_name = get_plugin_catalog().plugins required_hooks_by_plugin = { plugin_name: frozenset(hook.name for hook in plugins_by_name[plugin_name].filter_hooks("Snapshot")) for plugin_name in search_plugins diff --git a/archivebox/cli/archivebox_version.py b/archivebox/cli/archivebox_version.py index 7cd8a92e..38376bc6 100755 --- a/archivebox/cli/archivebox_version.py +++ b/archivebox/cli/archivebox_version.py @@ -265,15 +265,14 @@ def version( seen_failures: set[str] = set() seen_rows: set[tuple[str, str, str, str]] = set() - from archivebox.plugins.discovery import get_enabled_plugins + from archivebox.plugins.discovery import get_enabled_plugins, get_plugin_catalog from abx_dl.config import get_required_binary_requests from abx_dl.dependencies import resolve_binary_requests - from abx_dl.models import discover_plugins, filter_plugins from abx_dl.orchestrator import create_bus from abxpkg.binary_service import BinaryEvent, BinaryService - plugins = discover_plugins(runtime="archivebox") - enabled_plugins = filter_plugins(plugins, get_enabled_plugins(config=config), include_providers=True) + plugins = get_plugin_catalog() + enabled_plugins = plugins.select(get_enabled_plugins(config=config)).plugins enabled_plugin_names = set(enabled_plugins) runtime_config = normalize_runtime_config(config.for_crawl(), json_safe=False) derived_config: dict[str, object] = {} diff --git a/archivebox/config/common.py b/archivebox/config/common.py index bc83a5fe..b1a4aa12 100644 --- a/archivebox/config/common.py +++ b/archivebox/config/common.py @@ -852,13 +852,11 @@ class ArchiveBoxBaseConfig( disabled_plugins = [plugin_name for plugin_name, enabled_key in enabled_config_keys.items() if not getattr(self, enabled_key)] selected_plugin_roots = plugin_names if selected_plugin_roots: - from abx_dl.models import discover_plugins, filter_plugins + from archivebox.plugins.discovery import get_plugin_catalog selected_plugins = set( - filter_plugins( - discover_plugins(runtime="archivebox"), + get_plugin_catalog().select( sorted(selected_plugin_roots), - include_providers=True, disabled_names=disabled_plugins, ), ) diff --git a/archivebox/services/process_service.py b/archivebox/services/process_service.py index 870550ce..5bb1c48c 100644 --- a/archivebox/services/process_service.py +++ b/archivebox/services/process_service.py @@ -43,10 +43,10 @@ def normalize_process_env(env: dict) -> dict: if is_sensitive_config_key(key) or (key in config_input_names and key not in allowed_config_keys): normalized.pop(key, None) if selected_plugins: - from abx_dl.models import discover_plugins, filter_plugins from archivebox.config.common import _plugin_enabled_config_keys + from archivebox.plugins.discovery import get_plugin_catalog - selected_plugins = set(filter_plugins(discover_plugins(runtime="archivebox"), sorted(selected_plugins), include_providers=True)) + selected_plugins = set(get_plugin_catalog().select(sorted(selected_plugins))) for plugin_name, enabled_key in _plugin_enabled_config_keys().items(): normalized.setdefault(enabled_key, "True" if plugin_name in selected_plugins else "False") return normalized diff --git a/archivebox/services/runner.py b/archivebox/services/runner.py index 1d61a43a..3aefd7d3 100644 --- a/archivebox/services/runner.py +++ b/archivebox/services/runner.py @@ -37,12 +37,11 @@ from abx_dl.events import ( ) from abx_dl.limits import CrawlLimitState from abx_dl.catalog import PluginCatalog -from abx_dl.models import Plugin, Snapshot as AbxSnapshot, filter_plugins +from abx_dl.models import Plugin, Snapshot as AbxSnapshot from abx_dl.orchestrator import ( ExecutionPlan, create_bus, install_plugins as abx_install_plugins, - setup_services as setup_abx_services, ) from abx_dl.services.process_service import ProcessService as HookProcessService from abx_dl.services.snapshot_service import SnapshotService as HookSnapshotService @@ -119,7 +118,7 @@ def _runner_console_line(*, crawl=None, crawl_id=None, snapshot=None, status: st def _count_selected_hooks(plugins: dict[str, Plugin], selected_plugins: list[str] | None) -> int: - selected = filter_plugins(plugins, selected_plugins) if selected_plugins else plugins + selected = PluginCatalog(plugins).select(selected_plugins).plugins if selected_plugins else plugins return sum(1 for plugin in selected.values() for hook in plugin.hooks if "CrawlSetup" in hook.name or "Snapshot" in hook.name) @@ -805,7 +804,7 @@ class CrawlRunner: normalized_config = normalize_runtime_config(config) configured_plugins = [name.strip().lower() for name in str(normalized_config.get("PLUGINS") or "").split(",") if name.strip()] if configured_plugins: - selected_plugin_names = set(filter_plugins(self.plugins, configured_plugins, include_providers=True)) + selected_plugin_names = set(self.catalog.select(configured_plugins)) for plugin_name, enabled_key in _plugin_enabled_config_keys().items(): normalized_config.setdefault(enabled_key, plugin_name in selected_plugin_names) return { @@ -840,7 +839,7 @@ class CrawlRunner: derived_config=derived_config, runtime="archivebox", ) - setup_hooks = [(plugin, hook) for plugin in plan.plugins.values() for hook in plugin.filter_hooks("CrawlSetup")] + setup_hooks = [(plugin, hook) for plugin in plan.catalog.values() for hook in plugin.filter_hooks("CrawlSetup")] abx_snapshot = AbxSnapshot( id=snapshot["id"], url=snapshot["url"], @@ -1034,9 +1033,7 @@ class CrawlRunner: def queued_plugins_selected_by_config(queued_plugins: list[str]) -> list[str]: if not snapshot_selected_plugins: return queued_plugins - expanded_selected_plugins = set( - filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True).keys(), - ) + expanded_selected_plugins = set(self.catalog.select(snapshot_selected_plugins)) return [plugin for plugin in queued_plugins if plugin in expanded_selected_plugins] selected_hooks_by_plugin = None @@ -1102,11 +1099,7 @@ class CrawlRunner: return derived_config = normalize_runtime_config(self.derived_config) output_dir = Path(snapshot["output_dir"]) - plugins = ( - filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True) - if snapshot_selected_plugins - else self.plugins - ) + plugins = self.catalog.select(snapshot_selected_plugins).plugins if snapshot_selected_plugins else self.plugins if selected_hooks_by_plugin is not None: await sync_to_async(fail_unavailable_queued_hooks, thread_sensitive=True)( snapshot["id"], @@ -1129,7 +1122,7 @@ class CrawlRunner: await sync_to_async(run_snapshot_maintenance, thread_sensitive=True)(snapshot_id, output_dir=output_dir) return snapshot_selected_plugins = remaining_queued_plugins - plugins = filter_plugins(self.plugins, snapshot_selected_plugins, include_providers=True) + plugins = self.catalog.select(snapshot_selected_plugins).plugins selected_hooks_by_plugin = include_background_prerequisite_hooks(selected_hooks_by_plugin, plugins) abx_snapshot = AbxSnapshot( id=snapshot["id"], @@ -1144,7 +1137,7 @@ class CrawlRunner: derived_config=derived_config, runtime="archivebox", ) - plugins = plan.plugins + plugins = plan.catalog.plugins snapshot_phase_timeout = plan.snapshot_timeout + 120.0 await plan.seed_config(self.bus, parent_event=crawl_start_event) snapshot_service = plan.attach_snapshot_service( @@ -1311,7 +1304,6 @@ async def _run_binary(binary_id: str) -> None: from archivebox.machine.models import Binary, Machine binary = await Binary.objects.aget(id=binary_id) - plugins = _discover_archivebox_plugins() config = get_config(include_machine=False) machine = await sync_to_async(Machine.current, thread_sensitive=True)() derived_config = normalize_runtime_config(machine.config) @@ -1324,9 +1316,14 @@ async def _run_binary(binary_id: str) -> None: TagService(bus) ArchiveResultService(bus) MachineService(bus) - setup_abx_services( + plan = ExecutionPlan.build( + _discover_archivebox_catalog(), + config=config, + derived_config=derived_config, + runtime="archivebox", + ) + plan.attach_services( bus, - plugins=plugins, install_enabled=False, crawl_setup_enabled=False, crawl_start_enabled=False, @@ -1399,9 +1396,7 @@ def queued_plugins_for_snapshot(snapshot_id: str) -> list[str] | None: def config_overrides_for_queued_plugins(selected_plugins: list[str], **overrides: Any) -> dict[str, Any]: config_overrides = dict(overrides) config_overrides["PLUGINS"] = ",".join(selected_plugins) - selected_plugin_names = set( - filter_plugins(_discover_archivebox_plugins(), [plugin_name.lower() for plugin_name in selected_plugins], include_providers=True), - ) + selected_plugin_names = set(_discover_archivebox_catalog().select(plugin_name.lower() for plugin_name in selected_plugins)) for plugin_name, enabled_key in _plugin_enabled_config_keys().items(): config_overrides[enabled_key] = plugin_name in selected_plugin_names return config_overrides @@ -1840,10 +1835,11 @@ async def _run_install(plugin_names: list[str] | None = None) -> None: bus_destroyed = False try: + catalog = _discover_archivebox_catalog() if plugin_names: - selected_plugins = filter_plugins(plugins, list(plugin_names), include_providers=True) + selected_plugins = catalog.select(plugin_names).plugins else: - selected_plugins = filter_plugins(plugins, get_enabled_plugins(config=config), include_providers=True) + selected_plugins = catalog.select(get_enabled_plugins(config=config)).plugins if not selected_plugins: return plugins_label = ", ".join(plugin_names) if plugin_names else f"enabled ({len(selected_plugins)} of {len(plugins)} available)" @@ -1898,12 +1894,16 @@ async def _run_install(plugin_names: list[str] | None = None) -> None: ) with live_ui if live_ui is not None else nullcontext(): try: + plan = ExecutionPlan.build( + catalog, + selected_plugins=selected_plugins, + config=config, + derived_config=derived_config, + runtime="archivebox", + ) await abx_install_plugins( - plugin_names=selected_plugins, - plugins=plugins, + plan, output_dir=output_dir, - config_overrides=config, - derived_config_overrides=derived_config, emit_jsonl=False, bus=bus, BinaryService=None, diff --git a/archivebox/tests/conftest.py b/archivebox/tests/conftest.py index 4432bc21..76c6146b 100644 --- a/archivebox/tests/conftest.py +++ b/archivebox/tests/conftest.py @@ -1671,19 +1671,14 @@ def run_test_hook( import asyncio from abx_dl.execution import execute_hook - from abx_dl.models import discover_plugins from abx_dl.orchestrator import create_bus from archivebox.machine.models import Process + from archivebox.plugins.discovery import get_plugin_catalog from archivebox.services.process_service import ProcessService, parse_event_datetime resolved_script = script.resolve() hook = next( - ( - hook - for plugin in discover_plugins(runtime="archivebox").values() - for hook in plugin.hooks - if hook.path.resolve() == resolved_script - ), + (hook for plugin in get_plugin_catalog().values() for hook in plugin.hooks if hook.path.resolve() == resolved_script), None, ) assert hook is not None, f"shipped hook is not in the plugin catalog: {script}" diff --git a/archivebox/tests/test_archive_result_service.py b/archivebox/tests/test_archive_result_service.py index 7dc21ff6..57e2d16e 100644 --- a/archivebox/tests/test_archive_result_service.py +++ b/archivebox/tests/test_archive_result_service.py @@ -19,9 +19,9 @@ pytestmark = pytest.mark.django_db(transaction=True) def _snapshot_hook_name(plugin_name: str) -> str: - from abx_dl.models import discover_plugins + from archivebox.plugins.discovery import get_plugin_catalog - plugin = discover_plugins().get(plugin_name) + plugin = get_plugin_catalog().plugins.get(plugin_name) assert plugin is not None, f"missing test plugin {plugin_name}" hooks = plugin.filter_hooks("Snapshot") assert hooks, f"missing Snapshot hooks for {plugin_name}" @@ -47,15 +47,15 @@ def _run_shipped_snapshot_hook( """Run one shipped hook through the production process/result bus services.""" import asyncio - from abx_dl.models import discover_plugins from abx_dl.services.process_service import ProcessService as HookProcessService from abx_plugins.plugins.base.utils import get_hydrated_required_binaries from archivebox.core.models import ArchiveResult from archivebox.machine.models import Process + from archivebox.plugins.discovery import get_plugin_catalog from archivebox.services.archive_result_service import ArchiveResultService from archivebox.services.process_service import ProcessService as PersistedProcessService - discovered_plugin = discover_plugins().get(plugin) + discovered_plugin = get_plugin_catalog().plugins.get(plugin) assert discovered_plugin is not None, f"missing test plugin {plugin}" matching_hooks = [hook for hook in discovered_plugin.filter_hooks("Snapshot") if hook.name == hook_name or hook.path.name == hook_name] assert len(matching_hooks) == 1, f"missing or ambiguous Snapshot hook {plugin}:{hook_name}" diff --git a/archivebox/tests/test_recursive_crawl.py b/archivebox/tests/test_recursive_crawl.py index 0db228e8..5627de72 100644 --- a/archivebox/tests/test_recursive_crawl.py +++ b/archivebox/tests/test_recursive_crawl.py @@ -493,10 +493,10 @@ def test_recursive_crawl_depth_two_all_plugins_runs_snapshots_in_parallel( ): """Run a bounded real depth=2 crawl with all plugins enabled and verify parallel snapshot execution.""" - from abx_dl.models import discover_plugins + from archivebox.plugins.discovery import get_plugin_catalog root_url = recursive_test_site["root_url"] - plugin_selection = ",".join(sorted(plugin for plugin in discover_plugins().keys() if not plugin.startswith("claude"))) + plugin_selection = ",".join(sorted(plugin for plugin in get_plugin_catalog() if not plugin.startswith("claude"))) env = os.environ.copy() for preinstalled_path_key in ( "CHROME_BINARY",