's'

http_clients/async_http.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+import httpx
+from typing import Dict, Any
+import utils
+
+OptionalStr = str | None
+OptionalDict = Dict[str, Any] | None
+
+
+async def async_req(
+        url: str,
+        proxy_addr: OptionalStr = None,
+        headers: OptionalDict = None,
+        data: dict | bytes | None = None,
+        json_data: dict | list | None = None,
+        timeout: int = 20,
+        redirect_url: bool = False,
+        return_cookies: bool = False,
+        include_cookies: bool = False,
+        abroad: bool = False,
+        content_conding: str = 'utf-8',
+        verify: bool = False,
+        http2: bool = True
+) -> OptionalDict | OptionalStr | tuple:
+    if headers is None:
+        headers = {}
+    try:
+        proxy_addr = utils.handle_proxy_addr(proxy_addr)
+        if data or json_data:
+            async with httpx.AsyncClient(proxy=proxy_addr, timeout=timeout, verify=verify, http2=http2) as client:
+                response = await client.post(url, data=data, json=json_data, headers=headers)
+        else:
+            async with httpx.AsyncClient(proxy=proxy_addr, timeout=timeout, verify=verify, http2=http2) as client:
+                response = await client.get(url, headers=headers, follow_redirects=True)
+
+        if redirect_url:
+            return str(response.url)
+        elif return_cookies:
+            cookies_dict = {name: value for name, value in response.cookies.items()}
+            return (response.text, cookies_dict) if include_cookies else cookies_dict
+        else:
+            resp_str = response.text
+    except Exception as e:
+        resp_str = str(e)
+
+    return resp_str
+
+
+async def get_response_status(url: str, proxy_addr: OptionalStr = None, headers: OptionalDict = None,
+                              timeout: int = 10, abroad: bool = False, verify: bool = False, http2=False) -> bool:
+
+    try:
+        proxy_addr = utils.handle_proxy_addr(proxy_addr)
+        async with httpx.AsyncClient(proxy=proxy_addr, timeout=timeout, verify=verify) as client:
+            response = await client.head(url, headers=headers, follow_redirects=True)
+            return response.status_code == 200
+    except Exception as e:
+        print(e)
+    return False

http_clients/sync_http.py

@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+import gzip
+import urllib.parse
+import urllib.error
+import requests
+import ssl
+import json
+import urllib.request
+
+no_proxy_handler = urllib.request.ProxyHandler({})
+opener = urllib.request.build_opener(no_proxy_handler)
+
+ssl_context = ssl.create_default_context()
+ssl_context.check_hostname = False
+ssl_context.verify_mode = ssl.CERT_NONE
+OptionalStr = str | None
+OptionalDict = dict | None
+
+
+def sync_req(
+        url: str,
+        proxy_addr: OptionalStr = None,
+        headers: OptionalDict = None,
+        data: dict | bytes | None = None,
+        json_data: dict | list | None = None,
+        timeout: int = 20,
+        redirect_url: bool = False,
+        abroad: bool = False,
+        content_conding: str = 'utf-8'
+) -> str:
+    if headers is None:
+        headers = {}
+    try:
+        if proxy_addr:
+            proxies = {
+                'http': proxy_addr,
+                'https': proxy_addr
+            }
+            if data or json_data:
+                response = requests.post(
+                    url, data=data, json=json_data, headers=headers, proxies=proxies, timeout=timeout
+                )
+            else:
+                response = requests.get(url, headers=headers, proxies=proxies, timeout=timeout)
+            if redirect_url:
+                return response.url
+            resp_str = response.text
+        else:
+            if data and not isinstance(data, bytes):
+                data = urllib.parse.urlencode(data).encode(content_conding)
+            if json_data and isinstance(json_data, (dict, list)):
+                data = json.dumps(json_data).encode(content_conding)
+
+            req = urllib.request.Request(url, data=data, headers=headers)
+
+            try:
+                if abroad:
+                    response = urllib.request.urlopen(req, timeout=timeout)
+                else:
+                    response = opener.open(req, timeout=timeout)
+                if redirect_url:
+                    return response.url
+                content_encoding = response.info().get('Content-Encoding')
+                try:
+                    if content_encoding == 'gzip':
+                        with gzip.open(response, 'rt', encoding=content_conding) as gzipped:
+                            resp_str = gzipped.read()
+                    else:
+                        resp_str = response.read().decode(content_conding)
+                finally:
+                    response.close()
+
+            except urllib.error.HTTPError as e:
+                if e.code == 400:
+                    resp_str = e.read().decode(content_conding)
+                else:
+                    raise
+            except urllib.error.URLError as e:
+                print(f"URL Error: {e}")
+                raise
+            except Exception as e:
+                print(f"An error occurred: {e}")
+                raise
+
+    except Exception as e:
+        resp_str = str(e)
+
+    return resp_str