import os import re import time from pathlib import Path from lupa import LuaRuntime, LuaError from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler from log.Logger import Logger from .PluginFSHandler import PluginFSHandler from .luarails import guardrails_code PLUGINS_DIR = Path(__file__).parent / "plugins" HTML_DIR = Path(__file__).parent / "../html" MARKDOWN_DIR = Path(__file__).parent / ".." / "markdown" class PluginManager: def __init__(self): self.lua = LuaRuntime(unpack_returned_tuples=True) self.plugins = {} # name -> dict{path, lua_module, hooks, routes} self.routes = {} # path -> (plugin_name, lua_fn) self.hooks = {} # hook_name -> list of (plugin_name, lua_fn) # Create directories PLUGINS_DIR.mkdir(exist_ok=True) HTML_DIR.mkdir(exist_ok=True) MARKDOWN_DIR.mkdir(exist_ok=True) # Expose helpers to Lua self._setup_lua_globals() self._start_watcher() def _setup_lua_globals(self): """Set up all Lua global functions and guardrails""" # plugin_manager.py from .Actions import Actions g = self.lua.globals() self.actions = Actions() # Route and hook registration g.add_route = self._expose_add_route g.register_hook = self._expose_register_hook # Logging - using custom Logger # TODO: With Logger do custom Plugin Loading g.log = lambda msg: Logger.log_lua_info(f"PLUGIN => {msg}") g.log_warn = lambda msg: Logger.log_lua_warning(f"PLUGIN => {msg}") g.log_lua_error = lambda msg: Logger.log_lua_error(f"PLUGIN => {msg}") self.actions = Actions g.read_file = self.actions._safe_read_file g.write_file = self.actions._safe_write_file # HTML g.read_html = lambda fn: self.actions._read_content(self.actions.html_dir, fn) g.write_html = lambda fn, c: self.actions._write_content(self.actions.html_dir, fn, c) g.list_html_files = lambda: self.actions._list_files(self.actions.html_dir, ".html") # Markdown g.read_markdown = lambda fn: self.actions._read_content(self.actions.markdown_dir, fn) g.write_markdown = lambda fn, c: self.actions._write_content(self.actions.markdown_dir, fn, c) g.list_markdown_files = lambda: self.actions._list_files(self.actions.markdown_dir, ".md") # HTML helpers g.html_find_tag = self.actions._html_find_tag g.html_replace_tag = self.actions._html_replace_tag g.html_insert_before = self.actions._html_insert_before g.html_insert_after = self.actions._html_insert_after g.html_wrap_content = self.actions._html_wrap_content # Markdown helpers g.md_add_header = self.actions._md_add_header g.md_replace_section = self.actions._md_replace_section g.md_append_content = self.actions._md_append_content # JSON g.table_to_json = self.actions._table_to_json g.json_to_table = self.actions._json_to_table # Guardrails - predefined safe patterns self._setup_lua_guardrails() def _setup_lua_guardrails(self): try: self.lua.execute(guardrails_code) except LuaError as e: Logger.log_lua_error(f"Failed to initialize Lua guardrails: {e}") """ Lifecycle of Plugin """ def load_all(self): for p in PLUGINS_DIR.glob("*.lua"): self.load_plugin(p) def load_plugin(self, path: Path): name = path.name try: code = path.read_text(encoding="utf-8") self._current_plugin = name lua_module = self.lua.execute(code) if lua_module is None: lua_module = {} self.plugins[name] = {"path": path, "module": lua_module} Logger.log_lua_info(f"Loaded plugin: {name}") except LuaError as e: Logger.log_lua_error(f"Lua error while loading {name}: {e}") except Exception as e: Logger.log_lua_error(f"Error loading plugin {name}: {e}") finally: self._current_plugin = None def reload_plugin(self, path: Path): name = path.name self.unload_plugin(name) time.sleep(0.05) if path.exists(): self.load_plugin(path) else: Logger.log_lua_warning(f"Tried to reload {name}, but file no longer exists") def unload_plugin(self, name: str): # Remove routes/hook registrations from this plugin to_remove_routes = [r for r, v in self.routes.items() if v[0] == name] for r in to_remove_routes: del self.routes[r] for hook, lst in list(self.hooks.items()): self.hooks[hook] = [x for x in lst if x[0] != name] if not self.hooks[hook]: del self.hooks[hook] if name in self.plugins: del self.plugins[name] Logger.log_lua_info(f"Unloaded plugin {name}") else: Logger.log_lua_warning(f"Tried to unload {name}, but it was not loaded") """ Expose a new API route """ def _expose_add_route(self, path, lua_fn): """Called from Lua as add_route(path, function(req) ... end)""" p = str(path) plugin_name = self._current_loading_plugin_name() if not plugin_name: plugin_name = "" self.routes[p] = (plugin_name, lua_fn) Logger.log_lua_info(f"Plugin {plugin_name} registered route {p}") def _expose_register_hook(self, hook_name, lua_fn): hook = str(hook_name) plugin_name = self._current_loading_plugin_name() or "" self.hooks.setdefault(hook, []).append((plugin_name, lua_fn)) Logger.log_lua_info(f"Plugin {plugin_name} registered hook {hook}") def _current_loading_plugin_name(self): return getattr(self, "_current_plugin", None) """ Running hooks & handling routes """ def handle_request(self, path, request_info=None): """If a plugin registered a route for this path, call it and return (status, headers, body) or raw str.""" if path in self.routes: plugin_name, lua_fn = self.routes[path] try: lua_req = request_info or {} res = lua_fn(lua_req) if isinstance(res, tuple) and len(res) == 3: return res return (200, {"Content-Type": "text/html"}, str(res)) except LuaError as e: Logger.log_lua_error(f"Lua error in route {path}: {e}") return (500, {"Content-Type": "text/plain"}, f"Plugin error: {e}") return None def run_hook(self, hook_name: str, *args): """Run all registered hook functions for hook_name; return last non-None return value.""" if hook_name not in self.hooks: return None last = None for plugin_name, fn in list(self.hooks[hook_name]): try: out = fn(*args) if out is not None: last = out except LuaError as e: Logger.log_lua_error(f"Lua error in hook {hook_name} from {plugin_name}: {e}") return last """ File watcher """ def _start_watcher(self): self.observer = Observer() self.fs_handler = PluginFSHandler(self) self.observer.schedule(self.fs_handler, str(PLUGINS_DIR), recursive=False) self.observer.start() Logger.log_lua_info("Started plugin folder watcher") def stop(self): self.observer.stop() self.observer.join()