75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
import aiofiles
|
|
import json
|
|
import hashlib
|
|
|
|
from pddnsc.base import BaseOutputProvider
|
|
|
|
|
|
class GenericTextFile(BaseOutputProvider):
|
|
"""Сохранение IP адресов в текстовый файл
|
|
|
|
Конфигурация:
|
|
|
|
- filepath: имя файла
|
|
- encoding: кодировка, по умолчанию "utf-8"
|
|
- mode: режим открытия, по умолчанию "w"
|
|
- save_ipv4: сохранять ли IPv4, по умолчанию нет если save_ipv6 есть в конфиге иначе да
|
|
- save_ipv6: сохранять ли IPv6, по умолчанию нет если save_ipv4 есть в конфиге иначе да
|
|
"""
|
|
|
|
def post_init(self):
|
|
super().post_init()
|
|
self.filepath = self.config["filepath"]
|
|
self.encoding = self.config.get("encoding", "utf-8")
|
|
self.mode = self.config.get("mode", "w")
|
|
self.save_ipv4 = self.config.get("save_ipv4", False)
|
|
self.save_ipv6 = self.config.get("save_ipv6", False)
|
|
if "save_ipv4" not in self.config and "save_ipv4" not in self.config:
|
|
self.save_ipv4 = self.save_ipv6 = True
|
|
self.content = ""
|
|
|
|
async def read(self):
|
|
async with aiofiles.open(self.filepath, mode="r", encoding=self.encoding) as f:
|
|
self.content = await f.read()
|
|
|
|
def set_content(self, ipv4: str, ipv6: str):
|
|
lst = []
|
|
if self.save_ipv4:
|
|
lst.append(ipv4)
|
|
if self.save_ipv6:
|
|
lst.append(ipv6)
|
|
self.content = "\n".join(lst)
|
|
|
|
async def write(self):
|
|
async with aiofiles.open(
|
|
self.filepath, mode=self.mode, encoding=self.encoding
|
|
) as f:
|
|
await f.write(self.content)
|
|
|
|
async def set_addrs_imp(self, source_provider: str, addr_v4: str, addr_v6: str):
|
|
await self.set_content(addr_v4, addr_v6)
|
|
await self.write()
|
|
|
|
|
|
class StateFile(GenericTextFile):
|
|
"""Сохранение всех IP адресов в json файл"""
|
|
|
|
async def set_content(self, addr_v4: str, addr_v6: str):
|
|
state = {}
|
|
if self.save_ipv4:
|
|
state["ipv4"] = addr_v4 or ""
|
|
if self.save_ipv6:
|
|
state["ipv6"] = addr_v6 or ""
|
|
self.content = json.dumps(state)
|
|
|
|
|
|
class StateHashFile(GenericTextFile):
|
|
"""Сохранение хеша от всех IP адресов в файл"""
|
|
|
|
async def set_content(self, addr_v4: str, addr_v6: str):
|
|
state_str = (self.save_ipv4 and addr_v4 or "") + (
|
|
self.save_ipv6 and addr_v6 or ""
|
|
)
|
|
sha = hashlib.sha256(state_str.encode(encoding=self.encoding))
|
|
self.content = sha.hexdigest()
|