mkdocs.yml

@@ -1,11 +1,21 @@
 site_name: pddnsc
+site_description: Документация pddnsc
+site_author: b4tman
+repo_url: https://gitea.b4tman.ru/b4tman/pddnsc
+repo_name: pddnsc
+copyright: (c) 2024 Дмитрий Беляев
 
-theme: material
+theme:
+   name: material
+   language: ru
+   locale: ru
+   highlightjs: true
 markdown_extensions:
   - pymdownx.snippets
 
 plugins:
-  - search
+  - search:
+      lang: ru
   - mkapi:
       config: mkapi_conf.py
 

pddnsc/base.py

@@ -2,6 +2,7 @@ import asyncio
 from abc import ABC, abstractmethod
 from typing import NamedTuple, Optional
 from netaddr import valid_ipv4, valid_ipv6
+from httpx import AsyncHTTPTransport
 
 
 class IPAddreses(NamedTuple):
@@ -18,7 +19,13 @@ class BaseSourceProvider(ABC):
     _childs = {}
     registred = {}
 
-    def __init__(self, name, config, ipv4t, ipv6t):
+    def __init__(
+        self,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
         self.name, self.config, self.ipv4t, self.ipv6t = (
             name,
             config,
@@ -35,6 +42,7 @@ class BaseSourceProvider(ABC):
         return f"{self.__class__.__name__}: {self.name}"
 
     async def fetch_all(self) -> IPAddreses:
+        """метод для получения всех ip адресов сразу"""
         results = await asyncio.gather(
             self.fetch_v4(), self.fetch_v6(), return_exceptions=True
         )
@@ -48,7 +56,8 @@ class BaseSourceProvider(ABC):
         return super().__init_subclass__()
 
     @classmethod
-    def validate_source_config(cls, name, config):
+    def validate_source_config(cls, name: str, config: dict):
+        """метод валидации конфигурации для провайдера"""
         if "provider" not in config:
             return False
         prov_name = config["provider"]
@@ -57,7 +66,15 @@ class BaseSourceProvider(ABC):
         return True
 
     @classmethod
-    def register_provider(cls, name, config, ipv4t, ipv6t):
+    def register_provider(
+        cls,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
+        """метод регистрации провайдера по конфигурации"""
+
         if not cls.validate_source_config(name, config):
             return
         provider = cls._childs[config["provider"]]
@@ -80,7 +97,13 @@ class BaseOutputProvider(ABC):
     _childs = {}
     registred = {}
 
-    def __init__(self, name, config, ipv4t, ipv6t):
+    def __init__(
+        self,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
         self.name, self.config = name, config
         self.ipv4t, self.ipv6t = ipv4t, ipv6t
         self.post_init()
@@ -96,13 +119,16 @@ class BaseOutputProvider(ABC):
     def __str__(self):
         return f"{self.__class__.__name__}: {self.name}"
 
-    def best_transport(self, addr_v4, addr_v6):
+    def best_transport(self, addr_v4: str, addr_v6: str) -> AsyncHTTPTransport:
+        """метод выбирает лучший транспорт для отправки адресов (либо ipv4 либо ipv6)"""
         if addr_v6:
             return self.ipv6t
         return self.ipv4t
 
     @classmethod
-    def validate_source_config(cls, name, config):
+    def validate_source_config(cls, name: str, config: dict):
+        """метод валидации конфигурации для провайдера"""
+
         if "provider" not in config:
             return False
         prov_name = config["provider"]
@@ -111,17 +137,25 @@ class BaseOutputProvider(ABC):
         return True
 
     @classmethod
-    def register_provider(cls, name, config, ipv4t, ipv6t):
+    def register_provider(
+        cls,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
+        """метод регистрации провайдера по конфигурации"""
         if not cls.validate_source_config(name, config):
             return
         provider = cls._childs[config["provider"]]
         cls.registred[name] = provider(name, config, ipv4t, ipv6t)
 
-    async def set_addrs(self, source_provider, addr_v4, addr_v6):
+    async def set_addrs(self, source_provider: str, addr_v4: str, addr_v6: str):
+        """метод внешнего интерфейса для отправки адресов"""
         return await self.set_addrs_imp(source_provider, addr_v4, addr_v6)
 
     @abstractmethod
-    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+    async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
         """необходимый метод для реализации отправки/вывода IP аддресов"""
         ...
 
@@ -132,7 +166,13 @@ class BaseFilterProvider(ABC):
     _childs = {}
     registred = {}
 
-    def __init__(self, name, config, ipv4t, ipv6t):
+    def __init__(
+        self,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
         self.name, self.config = name, config
         self.ipv4t, self.ipv6t = ipv4t, ipv6t
         self.post_init()
@@ -148,13 +188,10 @@ class BaseFilterProvider(ABC):
     def __str__(self):
         return f"{self.__class__.__name__}: {self.name}"
 
-    def best_client(self, addr_v4, addr_v6):
-        if addr_v6 is None and addr_v4 is not None:
-            return self.ipv4t
-        return self.ipv6t
-
     @classmethod
-    def validate_source_config(cls, name, config):
+    def validate_source_config(cls, name: str, config: dict) -> bool:
+        """метод валидации конфигурации для провайдера"""
+
         if "provider" not in config:
             return False
         prov_name = config["provider"]
@@ -163,17 +200,26 @@ class BaseFilterProvider(ABC):
         return True
 
     @classmethod
-    def register_provider(cls, name, config, ipv4t, ipv6t):
+    def register_provider(
+        cls,
+        name: str,
+        config: dict,
+        ipv4t: AsyncHTTPTransport,
+        ipv6t: AsyncHTTPTransport,
+    ):
+        """метод регистрации провайдера по конфигурации"""
+
         if not cls.validate_source_config(name, config):
             return
         provider = cls._childs[config["provider"]]
         cls.registred[name] = provider(name, config, ipv4t, ipv6t)
 
-    async def check(self, source_provider, addr_v4, addr_v6):
+    async def check(self, source_provider: str, addr_v4: str, addr_v6: str) -> bool:
+        """метод внешнего интерфейса для проверки адресов"""
         return await self.check_imp(source_provider, addr_v4, addr_v6)
 
     @abstractmethod
-    async def check_imp(self, source_provider, addr_v4, addr_v6):
+    async def check_imp(self, source_provider: str, addr_v4: str, addr_v6: str) -> bool:
         """необходимый метод реализации проверки"""
         ...
 

pddnsc/cli.py

@@ -8,7 +8,16 @@ from .plugins import use_plugins
 from typing import Optional
 
 
-def is_valid_addreses(addrs: IPAddreses, config) -> bool:
+def is_valid_addreses(addrs: IPAddreses, config: dict) -> bool:
+    """Проверка валидности IP адресов
+
+    Args:
+        addrs (IPAddreses): IP адреса - результат одного из источников
+        config (dict): общая конфигурация
+
+    Returns:
+        bool: валиден или нет
+    """
     result = addrs.ipv4 or addrs.ipv6
     if config.get("require_ipv4"):
         result = result and addrs.ipv4
@@ -17,7 +26,15 @@ def is_valid_addreses(addrs: IPAddreses, config) -> bool:
     return result
 
 
-async def get_ip_addresses(config) -> Optional[IPAddreses]:
+async def get_ip_addresses(config: dict) -> Optional[IPAddreses]:
+    """Получение всех IP адресов из всех источников
+
+    Args:
+        config (dict): общая конфигурация
+
+    Returns:
+        Optional[IPAddreses]: результат получения, либо None
+    """
     providers = BaseSourceProvider.registred.values()
     ip_addresses = None
     is_done = False
@@ -41,7 +58,16 @@ async def get_ip_addresses(config) -> Optional[IPAddreses]:
     return ip_addresses
 
 
-async def check_ip_addresses(ip_addresses):
+async def check_ip_addresses(ip_addresses: IPAddreses) -> bool:
+    """Проверка результата получения IP адресов с помощью фильтров
+
+    Args:
+        ip_addresses (IPAddreses): IP адреса
+
+    Returns:
+        bool: корректны ли адреса (изменились ли они),
+              надо ли продолжать обработку и отправлять их на сервер
+    """
     providers = BaseFilterProvider.registred.values()
     result = True
     failed = ""
@@ -69,7 +95,12 @@ async def check_ip_addresses(ip_addresses):
     return result
 
 
-async def send_ip_addreses(ip_addresses):
+async def send_ip_addreses(ip_addresses: IPAddreses):
+    """Отправка адресов на все плагины вывода
+
+    Args:
+        ip_addresses (IPAddreses): IP адреса
+    """
     providers = BaseOutputProvider.registred.values()
     await asyncio.gather(
         *(
@@ -79,7 +110,8 @@ async def send_ip_addreses(ip_addresses):
     )
 
 
-def print_debug_info(config):
+def print_debug_info(config: dict):
+    """Вывод всех зарегистрированных плагинов и другой отладочной информации"""
     debug = config.get("debug", False)
     if debug:
         print("DEBUG info:")
@@ -103,7 +135,16 @@ def print_debug_info(config):
         )
 
 
-async def app(config, ipv4t, ipv6t):
+async def app(
+    config: dict, ipv4t: httpx.AsyncHTTPTransport, ipv6t: httpx.AsyncHTTPTransport
+):
+    """Запуск приложения
+
+    Args:
+        config (dict): общая конфигурация
+        ipv4t (httpx.AsyncHTTPTransport): транспорт IPv4
+        ipv6t (httpx.AsyncHTTPTransport): транспорт IPv6
+    """
     use_plugins(config, ipv4t, ipv6t)
 
     print_debug_info(config)
@@ -120,6 +161,9 @@ async def app(config, ipv4t, ipv6t):
 
 
 async def main():
+    """Точка входа программы
+    загрузка конфигурации и создание транспортов IPv4 и IPv6
+    """
     config = toml.load("settings/config.toml")
     async with httpx.AsyncHTTPTransport(
         local_address="0.0.0.0", proxy=config.get("proxy_v4")

pddnsc/filters/files.py

@@ -8,7 +8,14 @@ from pddnsc.base import BaseFilterProvider
 
 
 class StateHashFilter(BaseFilterProvider):
-    async def check_imp(self, source_provider, addr_v4, addr_v6):
+    """Проверка на то что хотябы один IP адрес изменился по хешу сохраненному в файле
+
+    Конфигурация:
+
+      - filepath: имя файла
+    """
+
+    async def check_imp(self, source_provider: str, addr_v4: str, addr_v6: str) -> bool:
         if not isfile(self.config["filepath"]):
             return True
 
@@ -23,13 +30,24 @@ class StateHashFilter(BaseFilterProvider):
 
 
 class StateFileFilter(BaseFilterProvider):
-    async def check_imp(self, source_provider, addr_v4, addr_v6):
+    """Проверка на то что хотябы один IP адрес изменился по сравнению с данными в json файле
+
+    Конфигурация:
+
+      - filepath: имя файла
+      - check_ipv4: проверка ipv4 адреса
+      - check_ipv6: проверка ipv6 адреса
+
+      если нет ни одного параметра то проверка выполняется для всех адресов
+    """
+
+    async def check_imp(self, source_provider: str, addr_v4: str, addr_v6: str) -> bool:
         if not isfile(self.config["filepath"]):
             return True
 
         new_state = {
-            "ipv4": addr_v4 or "",
-            "ipv6": addr_v6 or "",
+            "ipv4": addr_v4,
+            "ipv6": addr_v6,
         }
 
         async with aiofiles.open(

pddnsc/loaders.py

@@ -5,7 +5,12 @@ import traceback
 from importlib import util
 
 
-def load_module(path):
+def load_module(path: str):
+    """загрузка python модуля
+
+    Args:
+        path (str): имя файла
+    """
     name = os.path.split(path)[-1]
     spec = util.spec_from_file_location(name, path)
     module = util.module_from_spec(spec)
@@ -13,7 +18,12 @@ def load_module(path):
     return module
 
 
-def load_plugins(init_filepath):
+def load_plugins(init_filepath: str):
+    """Загрузка плагинов из пакета
+
+    Args:
+        init_filepath (str): имя `__init__.py` файла пакета
+    """
     dirpath = os.path.dirname(os.path.abspath(init_filepath))
     for fname in os.listdir(dirpath):
         if (

pddnsc/outputs/console.py

@@ -4,7 +4,9 @@ from pddnsc.base import BaseOutputProvider
 
 
 class JustPrint(BaseOutputProvider):
-    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+    """Вывод IP адресов в консоль"""
+
+    async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
         print(f">> {self.name}")
         print(f"addresses from: {source_provider}")
         print(f"IPv4: {addr_v4}")

pddnsc/outputs/files.py

@@ -6,7 +6,14 @@ from pddnsc.base import BaseOutputProvider
 
 
 class StateFile(BaseOutputProvider):
-    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+    """Схранение всех IP адресов в json файл
+
+    Конфигурация:
+
+      - filepath: имя файла
+    """
+
+    async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
         state = {
             "ipv4": addr_v4 or "",
             "ipv6": addr_v6 or "",
@@ -19,7 +26,14 @@ class StateFile(BaseOutputProvider):
 
 
 class StateHashFile(BaseOutputProvider):
-    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+    """Сохранение хеша от всех IP адресов в файл
+
+    Конфигурация:
+
+      - filepath: имя файла
+    """
+
+    async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
         state_str = (addr_v4 or "") + (addr_v6 or "")
         sha = hashlib.sha256(state_str.encode(encoding="utf-8"))
         async with aiofiles.open(

pddnsc/outputs/vscale.py

@@ -7,13 +7,28 @@ from pddnsc.base import BaseOutputProvider
 
 
 class VscaleDomains(BaseOutputProvider):
+    """Домены на vds.selectel.ru (ранее vscale.io)
+
+    Конфигурация:
+
+      - api_token_env: имя переменной окружения, в которой находится токен доступа, по умолчанию `VSCALE_API_TOKEN`
+      - api_base: базовый адрес API, по умолчанию `https://api.vscale.io/v1/`
+      - domain: имя основного домена, например `example.com`
+      - target: имя изменяемой записи, например `www` (чтобы изменить `www.example.com`)
+      - ttl: время жизни записи, по умолчанию 3600
+      - save_ipv4: сохранять ли ipv4 адрес (A запись)
+      - save_ipv6: сохранять ли ipv6 адрес (AAAA запись)
+
+      (если нет ни одного из [save_ipv4, save_ipv6] то сохраняются оба адреса)
+    """
+
     def post_init(self):
         token = self.get_api_token()
         if not token:
             raise KeyError("no api token, use env VSCALE_API_TOKEN")
         self.__headers = {"X-Token": token}
         self.api_base = self.config.get("api_base", "https://api.vscale.io/v1/")
-        self.ttl = self.config.get("ttl", 300)
+        self.ttl = self.config.get("ttl", 3600)
         self.domain = self.config["domain"]
         target = self.config["target"]
         self.target = f"{target}.{self.domain}"
@@ -24,10 +39,12 @@ class VscaleDomains(BaseOutputProvider):
             self.save_ipv4 = self.save_ipv6 = True
 
     def get_api_token(self) -> str:
+        """получение API токена"""
         token_env = self.config.get("api_token_env", "VSCALE_API_TOKEN")
         return os.environ[token_env]
 
-    async def find_domain_id(self, client) -> Optional[int]:
+    async def find_domain_id(self, client: httpx.AsyncClient) -> Optional[int]:
+        """поиск ID домена"""
         response = await client.get("/domains/")
         if response.is_success:
             data = response.json()
@@ -40,7 +57,10 @@ class VscaleDomains(BaseOutputProvider):
         else:
             raise ValueError(f"failed to find domain id, code: {response.status_code}")
 
-    async def find_record(self, client, domain_id, record_type) -> Optional[int]:
+    async def find_record(
+        self, client: httpx.AsyncClient, domain_id: int, record_type: str
+    ) -> Optional[int]:
+        """поиск ID записи"""
         response = await client.get(
             f"/domains/{domain_id}/records/",
         )
@@ -55,14 +75,25 @@ class VscaleDomains(BaseOutputProvider):
                 f"error list records {domain_id=}: ", response.status_code
             )
 
-    async def get_record_value(self, client, domain_id, record_id) -> str:
+    async def get_record_value(
+        self, client: httpx.AsyncClient, domain_id: int, record_id: int
+    ) -> str:
+        """получение действующего значения записи"""
         response = await client.get(f"/domains/{domain_id}/records/{record_id}")
         if response.is_success:
             data = response.json()
             if isinstance(data, dict):
                 return data["content"]
 
-    async def change_record(self, client, domain_id, record_id, record_type, value):
+    async def change_record(
+        self,
+        client: httpx.AsyncClient,
+        domain_id: int,
+        record_id: int,
+        record_type: str,
+        value: str,
+    ):
+        """изменение записи"""
         data = {
             "content": value,
             "name": self.target,
@@ -80,7 +111,10 @@ class VscaleDomains(BaseOutputProvider):
                 f"failed to change record: {self.target=},{domain_id=}, {record_id=}, {record_type=}, {value=}"
             )
 
-    async def create_record(self, client, domain_id, record_type, value):
+    async def create_record(
+        self, client: httpx.AsyncClient, domain_id: int, record_type: str, value: str
+    ):
+        """создание новой записи"""
         data = {
             "content": value,
             "name": self.target,
@@ -96,7 +130,7 @@ class VscaleDomains(BaseOutputProvider):
                 f"failed to create record: {self.target=},{domain_id=}, {record_type=}, {value=}, {response.status_code=}"
             )
 
-    async def set_addrs_imp(self, source_provider, addr_v4, addr_v6):
+    async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
         save_addrs = []
         if addr_v4 and self.save_ipv4:
             save_addrs.append(("A", addr_v4))

pddnsc/plugins.py

@@ -1,12 +1,14 @@
 """ модуль взаимодействия и регистрации плагинов """
 
+from httpx import AsyncHTTPTransport
 from .base import BaseSourceProvider, BaseFilterProvider, BaseOutputProvider
 from . import sources
 from . import outputs
 from . import filters
 
 
-def use_plugins(config, ipv4t, ipv6t):
+def use_plugins(config: dict, ipv4t: AsyncHTTPTransport, ipv6t: AsyncHTTPTransport):
+    """Регистрация всех плагинов указаных в конфигурации"""
     for source_name in config["sources"]:
         BaseSourceProvider.register_provider(
             source_name, config["sources"][source_name], ipv4t, ipv6t

pddnsc/sources/fake.py

@@ -9,11 +9,11 @@ class DummySource(BaseSourceProvider):
     """имитация получения пустых адресов"""
 
     async def fetch_v4(self) -> str:
-        result = await asyncio.sleep(self.config.get("delay", 1), result=None)
+        result = await asyncio.sleep(self.config.get("delay", 1), result="")
         return result
 
     async def fetch_v6(self) -> str:
-        result = await asyncio.sleep(self.config.get("delay", 1), result=None)
+        result = await asyncio.sleep(self.config.get("delay", 1), result="")
         return result
 
 

pddnsc/sources/http.py

@@ -4,7 +4,14 @@ from pddnsc.base import BaseSourceProvider, filter_ipv4, filter_ipv6
 
 
 class GenericHttpSource(BaseSourceProvider):
-    """базовый провайдер получения IP адресов по http/https ссылкам в виде текста"""
+    """Базовый провайдер получения IP адресов по http/https ссылкам в виде текста.
+
+    Конфигурация:
+
+      - url_v4: *URL* для получения адреса *IPv4*
+      - url_v6: *URL* для получения адреса *IPv6*
+      - headers (`dict`): словарь дополнительных заголовков (*необязательно*)
+    """
 
     def post_init(self):
         self.url_v4 = self.config.get("url_v4")
@@ -39,7 +46,16 @@ class GenericHttpSource(BaseSourceProvider):
 
 
 class GenericHttpJsonSource(BaseSourceProvider):
-    """базовый провайдер получения IP адресов по http/https ссылкам в виде json"""
+    """Базовый провайдер получения IP адресов по http/https ссылкам в виде json.
+
+    Конфигурация:
+
+      - url_v4: *URL* для получения адреса *IPv4*
+      - url_v6: *URL* для получения адреса *IPv6*
+      - key_v4: ключ *json* (`int` для списка, `str` для словаря) для *IPv4*
+      - key_v6: ключ *json* (`int` для списка, `str` для словаря) для *IPv6*
+      - headers (`dict`): словарь дополнительных заголовков (*необязательно*)
+    """
 
     def post_init(self):
         self.url_v4 = self.config.get("url_v4")