bnovo plugin

scheduller

scheduler/utils.py

@@ -0,0 +1,71 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+import os
+import inspect
+import importlib
+from typing import List, Tuple
+from pathspec import PathSpec
+
+def reload_tasks_periodically(scheduler: AsyncIOScheduler):
+    """Перезагрузка задач из базы данных каждые 5 минут."""
+    from scheduler.tasks import load_tasks_to_scheduler
+    scheduler.add_job(lambda: load_tasks_to_scheduler(scheduler), "interval", minutes=5)
+
+
+
+def load_gitignore_patterns(project_root: str) -> PathSpec:
+    """
+    Загружает паттерны из файла .gitignore.
+    """
+    gitignore_path = os.path.join(project_root, ".gitignore")
+    if os.path.exists(gitignore_path):
+        with open(gitignore_path, "r", encoding="utf-8") as f:
+            patterns = f.readlines()
+        return PathSpec.from_lines("gitwildmatch", patterns)
+    return PathSpec.from_lines("gitwildmatch", [])  # Пустой PathSpec
+
+def get_project_functions() -> List[Tuple[str, str]]:
+    """
+    Сканирует проект и возвращает список всех функций в формате (путь, имя функции),
+    исключая файлы и папки, указанные в .gitignore.
+    """
+    functions = []
+    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+    # Загружаем паттерны из .gitignore
+    gitignore_spec = load_gitignore_patterns(project_root)
+
+    for root, dirs, files in os.walk(project_root):
+        # Исключаем директории, указанные в .gitignore
+        dirs[:] = [d for d in dirs if not gitignore_spec.match_file(os.path.relpath(os.path.join(root, d), project_root))]
+        
+        for file in files:
+            file_path = os.path.relpath(os.path.join(root, file), project_root)
+            if (
+                file.endswith(".py") and 
+                not file.startswith("__") and 
+                not gitignore_spec.match_file(file_path)
+            ):
+                module_path = os.path.relpath(os.path.join(root, file), project_root)
+                module_name = module_path.replace(os.sep, ".").replace(".py", "")
+
+                try:
+                    module = importlib.import_module(module_name)
+                    for name, func in inspect.getmembers(module, inspect.isfunction):
+                        functions.append((f"{module_name}.{name}", name))
+                except Exception as e:
+                    print(f"Ошибка загрузки модуля {module_name}: {e}")
+
+    return functions
+
+
+
+import importlib
+
+def execute_function(function_path):
+    """
+    Выполняет функцию по указанному пути.
+    """
+    module_name, func_name = function_path.rsplit(".", 1)
+    module = importlib.import_module(module_name)
+    func = getattr(module, func_name)
+    return func()