80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
import httpx
|
|
|
|
from pddnsc.base import BaseSourceProvider, filter_ipv4, filter_ipv6
|
|
|
|
|
|
class GenericHttpSource(BaseSourceProvider):
|
|
"""базовый провайдер получения IP адресов по http/https ссылкам в виде текста"""
|
|
|
|
def post_init(self):
|
|
self.url_v4 = self.config.get("url_v4")
|
|
self.url_v6 = self.config.get("url_v6")
|
|
self.headers = self.config.get("headers", {})
|
|
|
|
async def fetch_v4(self) -> str:
|
|
if not self.url_v4:
|
|
return None
|
|
|
|
result = ""
|
|
async with httpx.AsyncClient(
|
|
transport=self.ipv4t, headers=self.headers
|
|
) as client:
|
|
response = await client.get(self.url_v4)
|
|
if response.is_success:
|
|
result = response.text.strip()
|
|
return filter_ipv4(result)
|
|
|
|
async def fetch_v6(self) -> str:
|
|
if not self.url_v6:
|
|
return None
|
|
|
|
result = ""
|
|
async with httpx.AsyncClient(
|
|
transport=self.ipv6t, headers=self.headers
|
|
) as client:
|
|
response = await client.get(self.url_v6)
|
|
if response.is_success:
|
|
result = response.text.strip()
|
|
return filter_ipv6(result)
|
|
|
|
|
|
class GenericHttpJsonSource(BaseSourceProvider):
|
|
"""базовый провайдер получения IP адресов по http/https ссылкам в виде json"""
|
|
|
|
def post_init(self):
|
|
self.url_v4 = self.config.get("url_v4")
|
|
self.url_v6 = self.config.get("url_v6")
|
|
self.key_v4 = self.config.get("key_v4")
|
|
self.key_v6 = self.config.get("key_v6")
|
|
if self.config.get("use_accept", True):
|
|
headers = {"Accept": self.config.get("acept_type", "application/json")}
|
|
else:
|
|
headers = {}
|
|
self.headers = headers.update(self.config.get("headers", {}))
|
|
|
|
async def fetch_v4(self) -> str:
|
|
if not self.url_v4 or self.key_v4 is None:
|
|
return None
|
|
|
|
result = ""
|
|
async with httpx.AsyncClient(
|
|
transport=self.ipv4t, headers=self.headers
|
|
) as client:
|
|
response = await client.get(self.url_v4)
|
|
if response.is_success:
|
|
result = response.json()[self.key_v4].strip()
|
|
return filter_ipv4(result)
|
|
|
|
async def fetch_v6(self) -> str:
|
|
if not self.url_v6 or self.key_v6 is None:
|
|
return None
|
|
|
|
result = ""
|
|
async with httpx.AsyncClient(
|
|
transport=self.ipv6t, headers=self.headers
|
|
) as client:
|
|
response = await client.get(self.url_v6)
|
|
if response.is_success:
|
|
result = response.json()[self.key_v6].strip()
|
|
return filter_ipv6(result)
|