|
|
|
|
|
|
|
|
|
from __future__ import annotations |
|
|
|
import importlib |
|
import os |
|
from abc import ABC, abstractmethod |
|
from typing import Dict, List, Type |
|
|
|
|
|
|
|
|
|
class Plugin(ABC): |
|
""" |
|
Abstract base class for AnyCoder runtime plugins. |
|
|
|
Required attributes (class‑level): |
|
---------------------------------- |
|
name : str – unique key (used to invoke the plugin) |
|
description : str – short human description |
|
""" |
|
|
|
name: str |
|
description: str |
|
|
|
|
|
@abstractmethod |
|
def initialize(self, config: Dict | None = None) -> None: |
|
"""Called once at start‑up. Use for auth / heavy setup.""" |
|
... |
|
|
|
@abstractmethod |
|
def execute(self, **kwargs) -> Dict: |
|
""" |
|
Execute plugin action. Must return a JSON‑serialisable dict. |
|
|
|
`**kwargs` are passed from the caller verbatim. |
|
""" |
|
... |
|
|
|
|
|
|
|
|
|
|
|
class PluginManager: |
|
""" |
|
Discovers *.py files under `plugins_dir`, registers concrete Plugin |
|
subclasses, initialises them (once), and lets the app invoke them. |
|
""" |
|
|
|
def __init__(self, plugins_dir: str = "plugins") -> None: |
|
self.plugins_dir = plugins_dir |
|
self._registry: Dict[str, Type[Plugin]] = {} |
|
self._instances: Dict[str, Plugin] = {} |
|
|
|
|
|
def discover(self) -> None: |
|
"""Import every *.py file in `plugins_dir` (non‑private).""" |
|
if not os.path.isdir(self.plugins_dir): |
|
return |
|
|
|
for filename in os.listdir(self.plugins_dir): |
|
if filename.startswith("_") or not filename.endswith(".py"): |
|
continue |
|
|
|
module_path = f"{self.plugins_dir}.{filename[:-3]}" |
|
try: |
|
module = importlib.import_module(module_path) |
|
except Exception as exc: |
|
print(f"[PLUGIN] Failed to import {module_path}: {exc}") |
|
continue |
|
|
|
|
|
for attr in dir(module): |
|
obj = getattr(module, attr) |
|
if ( |
|
isinstance(obj, type) |
|
and issubclass(obj, Plugin) |
|
and obj is not Plugin |
|
): |
|
self.register(obj) |
|
|
|
|
|
def register(self, plugin_cls: Type[Plugin]) -> None: |
|
key = plugin_cls.name |
|
if not key: |
|
raise ValueError("Plugin class missing `.name` attribute.") |
|
self._registry[key] = plugin_cls |
|
|
|
def initialize_all(self, config: Dict | None = None) -> None: |
|
for name, cls in self._registry.items(): |
|
try: |
|
inst = cls() |
|
inst.initialize(config or {}) |
|
self._instances[name] = inst |
|
except Exception as exc: |
|
print(f"[PLUGIN] Init failed for {name}: {exc}") |
|
|
|
|
|
def list_plugins(self) -> List[str]: |
|
return list(self._registry) |
|
|
|
def execute(self, name: str, **kwargs) -> Dict: |
|
if name not in self._instances: |
|
raise ValueError(f"Plugin '{name}' is not initialised.") |
|
return self._instances[name].execute(**kwargs) |
|
|
|
|
|
|
|
|
|
|
|
class VSCodeSnippetPlugin(Plugin): |
|
"""Generate VSCode snippet JSON for quick copy‑paste.""" |
|
|
|
name = "vscode_snippets" |
|
description = "Produces VS Code snippet templates." |
|
|
|
def initialize(self, config: Dict | None = None) -> None: |
|
cfg = config or {} |
|
self.snippet_dir = cfg.get("snippet_dir", "./snippets") |
|
|
|
def execute(self, *, language: str = "python", snippet_name: str) -> Dict: |
|
path = os.path.join(self.snippet_dir, f"{language}.{snippet_name}.json") |
|
if os.path.isfile(path): |
|
with open(path, "r", encoding="utf-8") as fh: |
|
content = fh.read() |
|
else: |
|
content = '{ "prefix": "todo", "body": ["// add your snippet here"] }' |
|
return {"plugin": self.name, "snippet": content} |
|
|
|
|
|
|
|
|
|
|
|
plugin_manager = PluginManager() |
|
plugin_manager.discover() |
|
plugin_manager.initialize_all() |
|
|