Peng_ACL_AutoProtect/protector_list.py

"""Simple manager for the protector list stored in protector_list.json.

Each entry is a dict with keys:
- id: int (unique)
- target_ip: str
- src_port: int
- threshold: int | None

This module provides simple load/save/add/remove/update helpers.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import List, Optional, Dict

ROOT = Path(__file__).resolve().parent
LIST_PATH = ROOT / "protector_list.json"


def load_list() -> List[Dict]:
    if not LIST_PATH.exists():
        return []
    with open(LIST_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


def save_list(items: List[Dict]) -> None:
    with open(LIST_PATH, "w", encoding="utf-8") as f:
        json.dump(items, f, ensure_ascii=False, indent=2)


def _next_id(items: List[Dict]) -> int:
    if not items:
        return 1
    return max(int(i.get("id", 0)) for i in items) + 1


def add_entry(target_ip: str, src_port: int, threshold: Optional[int] = None) -> Dict:
    items = load_list()
    entry = {"id": _next_id(items), "target_ip": target_ip, "src_port": int(src_port), "threshold": (int(threshold) if threshold is not None else None)}
    items.append(entry)
    save_list(items)
    return entry


def remove_entry(entry_id: int) -> bool:
    items = load_list()
    new_items = [i for i in items if int(i.get("id", -1)) != int(entry_id)]
    if len(new_items) == len(items):
        return False
    save_list(new_items)
    return True


def update_entry(entry_id: int, **kwargs) -> Optional[Dict]:
    items = load_list()
    found = None
    for i in items:
        if int(i.get("id", -1)) == int(entry_id):
            i.update({k: v for k, v in kwargs.items() if v is not None})
            found = i
            break
    if found is None:
        return None
    save_list(items)
    return found


if __name__ == "__main__":
    # quick demo
    print("Current protector list:")
    print(load_list())