File size: 5,168 Bytes
9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 9fb92de 4179782 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
# plugins.py
# ------------------------------------------------------------------
# Generic plugin framework for AnyCoder AI
# ------------------------------------------------------------------
from __future__ import annotations
import importlib
import os
from abc import ABC, abstractmethod
from typing import Dict, List, Type
# ------------------------------------------------------------------ #
# 1 Base plugin interface
# ------------------------------------------------------------------ #
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
# ---- lifecycle ----------------------------------------------------
@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.
"""
...
# ------------------------------------------------------------------ #
# 2 Plugin manager
# ------------------------------------------------------------------ #
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] = {}
# ---------- discovery ---------------------------------------------
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: # pragma: no cover
print(f"[PLUGIN] Failed to import {module_path}: {exc}")
continue
# Register any subclasses of Plugin
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)
# ---------- registry ----------------------------------------------
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: # pragma: no cover
print(f"[PLUGIN] Init failed for {name}: {exc}")
# ---------- public API --------------------------------------------
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)
# ------------------------------------------------------------------ #
# 3 Example built‑in plugin
# ------------------------------------------------------------------ #
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}
# ------------------------------------------------------------------ #
# 4 Global manager instance (auto‑discover on import)
# ------------------------------------------------------------------ #
plugin_manager = PluginManager()
plugin_manager.discover()
plugin_manager.initialize_all()
|