71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
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() |