96 lines
4.0 KiB
Python
96 lines
4.0 KiB
Python
import os
|
|
import inspect
|
|
import importlib
|
|
import asyncio
|
|
import logging
|
|
from typing import List, Tuple
|
|
from pathspec import PathSpec
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
|
|
# Настройка логирования
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def reload_tasks_periodically(scheduler: AsyncIOScheduler):
|
|
"""Перезагрузка задач из базы данных каждые 5 минут."""
|
|
async def reload():
|
|
from scheduler.tasks import load_tasks_to_scheduler
|
|
try:
|
|
await load_tasks_to_scheduler(scheduler)
|
|
logger.info("Задачи успешно перезагружены.")
|
|
except Exception as e:
|
|
logger.error(f"Ошибка перезагрузки задач: {e}")
|
|
|
|
scheduler.add_job(lambda: asyncio.run(reload()), "interval", minutes=5)
|
|
|
|
def load_gitignore_patterns(project_root: str) -> PathSpec:
|
|
"""
|
|
Загружает паттерны из файла .gitignore.
|
|
"""
|
|
gitignore_path = os.path.join(project_root, ".gitignore")
|
|
try:
|
|
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)
|
|
except Exception as e:
|
|
logger.warning(f"Ошибка загрузки .gitignore: {e}")
|
|
return PathSpec.from_lines("gitwildmatch", [])
|
|
|
|
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:
|
|
spec = importlib.util.find_spec(module_name)
|
|
if spec is not None:
|
|
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:
|
|
logger.error(f"Ошибка при обработке модуля {module_name}: {e}")
|
|
|
|
return functions
|
|
|
|
def execute_function(function_path: str):
|
|
"""
|
|
Выполняет функцию по указанному пути.
|
|
"""
|
|
try:
|
|
module_name, func_name = function_path.rsplit(".", 1)
|
|
spec = importlib.util.find_spec(module_name)
|
|
if spec is None:
|
|
raise ImportError(f"Модуль {module_name} не найден")
|
|
|
|
module = importlib.import_module(module_name)
|
|
if not hasattr(module, func_name):
|
|
raise AttributeError(f"Функция {func_name} отсутствует в модуле {module_name}")
|
|
|
|
func = getattr(module, func_name)
|
|
logger.info(f"Выполняется функция: {function_path}")
|
|
return func()
|
|
except Exception as e:
|
|
logger.error(f"Ошибка выполнения функции {function_path}: {e}")
|
|
return None
|