/
azathd
/
mutiagent
Обзор
Документация
Войти
/
azathd
/
mutiagent
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/agents/security/rules_engine.py
42 строки
1 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47
359c81e
Код
Авторство
О чём код?
"""YAML rules loader with watchdog hot-reload.""" from __future__ import annotations from pathlib import Path from typing import Any import yaml from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer from src.common.settings import settings class RulesEngine: def __init__(self, path: str | None = None) -> None: self.path = Path(path or settings.rules_path) self.rules: list[dict[str, Any]] = [] self.blacklist_ips: set[str] = set() self.blacklist_asns: set[str] = set() self.reload() def reload(self) -> None: if not self.path.exists(): return data = yaml.safe_load(self.path.read_text(encoding="utf-8")) or {} self.rules = data.get("rules", []) self.blacklist_ips = set(data.get("blacklist_ips", [])) self.blacklist_asns = set(data.get("blacklist_asns", [])) def watch(self) -> Observer: engine = self class Handler(FileSystemEventHandler): def on_modified(self, event): # type: ignore[no-untyped-def] if Path(event.src_path) == engine.path: engine.reload() obs = Observer() obs.schedule(Handler(), str(self.path.parent), recursive=False) obs.start() return obs