add generic http source providers

pddnsc/sources/http.py

@@ -0,0 +1,75 @@
+import httpx
+
+from pddnsc.base import BaseSourceProvider, filter_ipv4, filter_ipv6
+
+
+class GenericHttpSource(BaseSourceProvider):
+    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):
+    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)