"""Plugin generator orchestration for wiki generation.
Handles sorting generators by dependency order and running registered
wiki generator plugins.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from local_deepwiki.logging import get_logger
from local_deepwiki.plugins.registry import get_plugin_registry
if TYPE_CHECKING:
from pathlib import Path
from local_deepwiki.config import Config
from local_deepwiki.core.vectorstore import VectorStore
from local_deepwiki.generators.wiki_status import WikiStatusManager
from local_deepwiki.models import IndexStatus, ProgressCallback, WikiPage
from local_deepwiki.plugins.base import WikiGeneratorPlugin
logger = get_logger(__name__)
def sort_generators_by_dependencies(
generators: list[WikiGeneratorPlugin],
) -> list[WikiGeneratorPlugin]:
"""Sort generators respecting run_after dependencies with validation.
Uses topological sort to ensure generators run after their dependencies.
Validates that all dependencies exist and warns about missing ones.
Args:
generators: List of generator plugins to sort.
Returns:
Sorted list of generators respecting dependencies.
"""
if not generators:
return generators
# Build name -> generator mapping
by_name: dict[str, WikiGeneratorPlugin] = {g.generator_name: g for g in generators}
available_names = set(by_name.keys())
# Validate dependencies exist and warn about missing ones
for generator in generators:
missing_deps = set(generator.run_after) - available_names
if missing_deps:
logger.warning(
f"Wiki generator '{generator.generator_name}' has missing dependencies: "
f"{missing_deps}. These generators are not registered and will be skipped."
)
# Build dependency graph for topological sort
# in_degree[name] = number of dependencies that must run first
in_degree: dict[str, int] = {g.generator_name: 0 for g in generators}
# dependents[name] = list of generators that depend on this one
dependents: dict[str, list[str]] = {g.generator_name: [] for g in generators}
for generator in generators:
for dep in generator.run_after:
if dep in available_names:
in_degree[generator.generator_name] += 1
dependents[dep].append(generator.generator_name)
# Kahn's algorithm for topological sort
# Start with generators that have no dependencies
# Sort by priority (higher first) within each level
ready = [g for g in generators if in_degree[g.generator_name] == 0]
ready.sort(key=lambda g: g.priority, reverse=True)
sorted_generators: list[WikiGeneratorPlugin] = []
while ready:
# Take the highest priority generator from ready list
current = ready.pop(0)
sorted_generators.append(current)
# Update dependents
for dep_name in dependents[current.generator_name]:
in_degree[dep_name] -= 1
if in_degree[dep_name] == 0:
# Insert in priority order
dep_gen = by_name[dep_name]
insert_idx = 0
for i, g in enumerate(ready):
if dep_gen.priority > g.priority:
insert_idx = i
break
insert_idx = i + 1
ready.insert(insert_idx, dep_gen)
# Check for cycles (some generators still have unresolved dependencies)
if len(sorted_generators) != len(generators):
unresolved = [
g.generator_name for g in generators if g not in sorted_generators
]
logger.error(
f"Circular dependency detected in wiki generators: {unresolved}. "
f"These generators will not run."
)
return sorted_generators
async def run_plugin_generators(
*,
pages: list[WikiPage],
all_source_files: list[str],
index_status: IndexStatus,
vector_store: VectorStore,
llm: object,
config: Config,
wiki_path: Path,
status_manager: WikiStatusManager,
write_callback: object,
progress_callback: ProgressCallback | None,
) -> tuple[list[WikiPage], int]:
"""Run registered wiki generator plugins.
Args:
pages: Current list of wiki pages (will not be mutated).
all_source_files: List of all source file paths.
index_status: Index status.
vector_store: Vector store for code search.
llm: LLM provider instance.
config: Configuration object.
wiki_path: Path to wiki output directory.
status_manager: Wiki status manager for tracking.
write_callback: Async callback to write pages to disk.
progress_callback: Optional progress callback.
Returns:
Tuple of (new_pages, pages_generated_count).
"""
registry = get_plugin_registry()
generators: list[WikiGeneratorPlugin] = list(registry.wiki_generators.values())
if not generators:
return [], 0
# Validate and sort generators respecting run_after dependencies
generators = sort_generators_by_dependencies(generators)
logger.info(f"Running {len(generators)} wiki generator plugin(s)")
# Build context dict for plugins
plugin_context: dict[str, object] = {
"vector_store": vector_store,
"llm": llm,
"config": config,
"existing_pages": list(pages),
}
new_pages: list[WikiPage] = []
pages_generated = 0
for generator in generators:
try:
logger.debug(f"Running wiki generator plugin: {generator.generator_name}")
result = await generator.generate(
index_status=index_status,
wiki_path=wiki_path,
context=plugin_context,
)
# Add generated pages
for page in result.pages:
new_pages.append(page)
status_manager.record_page_status(page, all_source_files)
await write_callback(page)
pages_generated += 1
# Update existing_pages in context for subsequent plugins
plugin_context["existing_pages"] = list(pages) + list(new_pages)
logger.debug(
f"Plugin '{generator.generator_name}' generated {len(result.pages)} page(s)"
)
except Exception as e:
logger.warning(
f"Wiki generator plugin '{generator.generator_name}' failed: {e}"
)
return new_pages, pages_generated
