Update plugins.py
Browse files- plugins.py +86 -62
plugins.py
CHANGED
@@ -1,115 +1,139 @@
|
|
1 |
-
|
|
|
|
|
|
|
|
|
|
|
2 |
import importlib
|
3 |
import os
|
|
|
4 |
from typing import Dict, List, Type
|
5 |
|
6 |
-
#
|
7 |
-
|
|
|
8 |
class Plugin(ABC):
|
9 |
"""
|
10 |
-
|
|
|
|
|
|
|
|
|
|
|
11 |
"""
|
|
|
12 |
name: str
|
13 |
description: str
|
14 |
|
|
|
15 |
@abstractmethod
|
16 |
-
def initialize(self, config: Dict) -> None:
|
17 |
-
"""
|
18 |
-
|
19 |
|
20 |
@abstractmethod
|
21 |
def execute(self, **kwargs) -> Dict:
|
22 |
-
"""
|
23 |
-
|
24 |
|
|
|
|
|
|
|
25 |
|
|
|
|
|
|
|
|
|
26 |
class PluginManager:
|
27 |
"""
|
28 |
-
|
|
|
29 |
"""
|
30 |
-
|
|
|
31 |
self.plugins_dir = plugins_dir
|
32 |
self._registry: Dict[str, Type[Plugin]] = {}
|
33 |
self._instances: Dict[str, Plugin] = {}
|
34 |
|
|
|
35 |
def discover(self) -> None:
|
36 |
-
"""
|
37 |
-
Auto-import all modules in the plugins directory.
|
38 |
-
"""
|
39 |
if not os.path.isdir(self.plugins_dir):
|
40 |
return
|
|
|
41 |
for filename in os.listdir(self.plugins_dir):
|
42 |
-
if
|
43 |
continue
|
44 |
-
|
45 |
-
module_path = f"{self.plugins_dir}.{
|
46 |
try:
|
47 |
module = importlib.import_module(module_path)
|
48 |
-
|
49 |
-
|
50 |
-
|
51 |
-
|
52 |
-
|
53 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
54 |
|
|
|
55 |
def register(self, plugin_cls: Type[Plugin]) -> None:
|
56 |
-
"""
|
57 |
-
Register a plugin class by its .name attribute.
|
58 |
-
"""
|
59 |
key = plugin_cls.name
|
|
|
|
|
60 |
self._registry[key] = plugin_cls
|
61 |
|
62 |
-
def initialize_all(self, config: Dict = None) -> None:
|
63 |
-
"""
|
64 |
-
Instantiate and initialize all registered plugins.
|
65 |
-
"""
|
66 |
for name, cls in self._registry.items():
|
67 |
try:
|
68 |
-
|
69 |
-
|
70 |
-
self._instances[name] =
|
71 |
-
except Exception as
|
72 |
-
print(f"
|
73 |
|
|
|
74 |
def list_plugins(self) -> List[str]:
|
75 |
-
|
76 |
-
Return the list of registered plugin names.
|
77 |
-
"""
|
78 |
-
return list(self._registry.keys())
|
79 |
|
80 |
def execute(self, name: str, **kwargs) -> Dict:
|
81 |
-
|
82 |
-
|
83 |
-
|
84 |
-
plugin = self._instances.get(name)
|
85 |
-
if not plugin:
|
86 |
-
raise ValueError(f"Plugin '{name}' is not initialized.")
|
87 |
-
return plugin.execute(**kwargs)
|
88 |
|
89 |
|
90 |
-
#
|
|
|
|
|
91 |
class VSCodeSnippetPlugin(Plugin):
|
|
|
|
|
92 |
name = "vscode_snippets"
|
93 |
-
description = "
|
94 |
|
95 |
-
def initialize(self, config: Dict) -> None:
|
96 |
-
|
97 |
-
self.snippet_dir =
|
98 |
|
99 |
-
def execute(self,
|
100 |
-
# Generate or fetch a snippet
|
101 |
-
language = kwargs.get('language', 'python')
|
102 |
-
snippet_name = kwargs.get('snippet_name')
|
103 |
-
# Load snippet from disk or generate dynamically
|
104 |
path = os.path.join(self.snippet_dir, f"{language}.{snippet_name}.json")
|
105 |
-
if os.path.
|
106 |
-
with open(path,
|
107 |
-
content =
|
108 |
else:
|
109 |
-
content = '{ "prefix": "todo", "body": ["//
|
110 |
-
return {
|
|
|
111 |
|
112 |
-
#
|
|
|
|
|
113 |
plugin_manager = PluginManager()
|
114 |
plugin_manager.discover()
|
115 |
plugin_manager.initialize_all()
|
|
|
1 |
+
# plugins.py
|
2 |
+
# ------------------------------------------------------------------
|
3 |
+
# Generic plugin framework for AnyCoder AI
|
4 |
+
# ------------------------------------------------------------------
|
5 |
+
from __future__ import annotations
|
6 |
+
|
7 |
import importlib
|
8 |
import os
|
9 |
+
from abc import ABC, abstractmethod
|
10 |
from typing import Dict, List, Type
|
11 |
|
12 |
+
# ------------------------------------------------------------------ #
|
13 |
+
# 1 Base plugin interface
|
14 |
+
# ------------------------------------------------------------------ #
|
15 |
class Plugin(ABC):
|
16 |
"""
|
17 |
+
Abstract base class for AnyCoder runtime plugins.
|
18 |
+
|
19 |
+
Required attributes (class‑level):
|
20 |
+
----------------------------------
|
21 |
+
name : str – unique key (used to invoke the plugin)
|
22 |
+
description : str – short human description
|
23 |
"""
|
24 |
+
|
25 |
name: str
|
26 |
description: str
|
27 |
|
28 |
+
# ---- lifecycle ----------------------------------------------------
|
29 |
@abstractmethod
|
30 |
+
def initialize(self, config: Dict | None = None) -> None:
|
31 |
+
"""Called once at start‑up. Use for auth / heavy setup."""
|
32 |
+
...
|
33 |
|
34 |
@abstractmethod
|
35 |
def execute(self, **kwargs) -> Dict:
|
36 |
+
"""
|
37 |
+
Execute plugin action. Must return a JSON‑serialisable dict.
|
38 |
|
39 |
+
`**kwargs` are passed from the caller verbatim.
|
40 |
+
"""
|
41 |
+
...
|
42 |
|
43 |
+
|
44 |
+
# ------------------------------------------------------------------ #
|
45 |
+
# 2 Plugin manager
|
46 |
+
# ------------------------------------------------------------------ #
|
47 |
class PluginManager:
|
48 |
"""
|
49 |
+
Discovers *.py files under `plugins_dir`, registers concrete Plugin
|
50 |
+
subclasses, initialises them (once), and lets the app invoke them.
|
51 |
"""
|
52 |
+
|
53 |
+
def __init__(self, plugins_dir: str = "plugins") -> None:
|
54 |
self.plugins_dir = plugins_dir
|
55 |
self._registry: Dict[str, Type[Plugin]] = {}
|
56 |
self._instances: Dict[str, Plugin] = {}
|
57 |
|
58 |
+
# ---------- discovery ---------------------------------------------
|
59 |
def discover(self) -> None:
|
60 |
+
"""Import every *.py file in `plugins_dir` (non‑private)."""
|
|
|
|
|
61 |
if not os.path.isdir(self.plugins_dir):
|
62 |
return
|
63 |
+
|
64 |
for filename in os.listdir(self.plugins_dir):
|
65 |
+
if filename.startswith("_") or not filename.endswith(".py"):
|
66 |
continue
|
67 |
+
|
68 |
+
module_path = f"{self.plugins_dir}.{filename[:-3]}"
|
69 |
try:
|
70 |
module = importlib.import_module(module_path)
|
71 |
+
except Exception as exc: # pragma: no cover
|
72 |
+
print(f"[PLUGIN] Failed to import {module_path}: {exc}")
|
73 |
+
continue
|
74 |
+
|
75 |
+
# Register any subclasses of Plugin
|
76 |
+
for attr in dir(module):
|
77 |
+
obj = getattr(module, attr)
|
78 |
+
if (
|
79 |
+
isinstance(obj, type)
|
80 |
+
and issubclass(obj, Plugin)
|
81 |
+
and obj is not Plugin
|
82 |
+
):
|
83 |
+
self.register(obj)
|
84 |
|
85 |
+
# ---------- registry ----------------------------------------------
|
86 |
def register(self, plugin_cls: Type[Plugin]) -> None:
|
|
|
|
|
|
|
87 |
key = plugin_cls.name
|
88 |
+
if not key:
|
89 |
+
raise ValueError("Plugin class missing `.name` attribute.")
|
90 |
self._registry[key] = plugin_cls
|
91 |
|
92 |
+
def initialize_all(self, config: Dict | None = None) -> None:
|
|
|
|
|
|
|
93 |
for name, cls in self._registry.items():
|
94 |
try:
|
95 |
+
inst = cls()
|
96 |
+
inst.initialize(config or {})
|
97 |
+
self._instances[name] = inst
|
98 |
+
except Exception as exc: # pragma: no cover
|
99 |
+
print(f"[PLUGIN] Init failed for {name}: {exc}")
|
100 |
|
101 |
+
# ---------- public API --------------------------------------------
|
102 |
def list_plugins(self) -> List[str]:
|
103 |
+
return list(self._registry)
|
|
|
|
|
|
|
104 |
|
105 |
def execute(self, name: str, **kwargs) -> Dict:
|
106 |
+
if name not in self._instances:
|
107 |
+
raise ValueError(f"Plugin '{name}' is not initialised.")
|
108 |
+
return self._instances[name].execute(**kwargs)
|
|
|
|
|
|
|
|
|
109 |
|
110 |
|
111 |
+
# ------------------------------------------------------------------ #
|
112 |
+
# 3 Example built‑in plugin
|
113 |
+
# ------------------------------------------------------------------ #
|
114 |
class VSCodeSnippetPlugin(Plugin):
|
115 |
+
"""Generate VSCode snippet JSON for quick copy‑paste."""
|
116 |
+
|
117 |
name = "vscode_snippets"
|
118 |
+
description = "Produces VS Code snippet templates."
|
119 |
|
120 |
+
def initialize(self, config: Dict | None = None) -> None:
|
121 |
+
cfg = config or {}
|
122 |
+
self.snippet_dir = cfg.get("snippet_dir", "./snippets")
|
123 |
|
124 |
+
def execute(self, *, language: str = "python", snippet_name: str) -> Dict:
|
|
|
|
|
|
|
|
|
125 |
path = os.path.join(self.snippet_dir, f"{language}.{snippet_name}.json")
|
126 |
+
if os.path.isfile(path):
|
127 |
+
with open(path, "r", encoding="utf-8") as fh:
|
128 |
+
content = fh.read()
|
129 |
else:
|
130 |
+
content = '{ "prefix": "todo", "body": ["// add your snippet here"] }'
|
131 |
+
return {"plugin": self.name, "snippet": content}
|
132 |
+
|
133 |
|
134 |
+
# ------------------------------------------------------------------ #
|
135 |
+
# 4 Global manager instance (auto‑discover on import)
|
136 |
+
# ------------------------------------------------------------------ #
|
137 |
plugin_manager = PluginManager()
|
138 |
plugin_manager.discover()
|
139 |
plugin_manager.initialize_all()
|