The_Great_Archive/ArchiveBox
1
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox.git synced 2026-09-14 11:06:13 +05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Adopt canonical abx-dl execution plan API

Browse Source
This commit is contained in:
Nick Sweeting 2026-09-02 04:44:55 -07:00
parent 23b91d6840
commit e7b3f1c485
No known key found for this signature in database
11 changed files with 54 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
archivebox/cli/archivebox_add.py
View File

@ -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()]

4
archivebox/cli/archivebox_extract.py
View File

@ -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 = {

7
archivebox/cli/archivebox_install.py
View File

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

4
archivebox/cli/archivebox_update.py
View File

@ -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

7
archivebox/cli/archivebox_version.py
View File

@ -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] = {}

6
archivebox/config/common.py
View File

@ -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,
),
)

4
archivebox/services/process_service.py
View File

@ -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

54
archivebox/services/runner.py
View File

@ -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,

9
archivebox/tests/conftest.py
View File

@ -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}"

8
archivebox/tests/test_archive_result_service.py
View File

@ -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}"

4
archivebox/tests/test_recursive_crawl.py
View File

@ -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",