Add async code for get_about and get_socials methods

Author: hnrkcode

loading_sdk/async_api/client.py

@@ -1,14 +1,14 @@
 import math
 
 import aiohttp
-from loading_sdk.async_api.extractors import AboutPageExtractor
 from loading_sdk.settings import (
     API_URL,
     API_VERSION,
     EDITORIAL_POST_TYPES,
     EDITORIAL_SORT,
     USER_AGENT,
 )
+from loading_sdk.async_api.extractors import extract_data
 
 
 async def async_loading_api_client(email=None, password=None):
@@ -497,7 +497,23 @@ async def get_about(self):
 
         :rtype dict
         """
-        about_page = AboutPageExtractor()
-        about_data = await about_page.extract_about_data()
 
-        return about_data
+        data = await extract_data("about")
+
+        if not data:
+            return {"code": 404, "message": "No data found", "data": None}
+
+        return {"code": 200, "message": "OK", "data": data}
+
+    async def get_socials(self):
+        """Get social media links
+
+        :rtype dict
+        """
+
+        data = await extract_data("socials")
+
+        if not data:
+            return {"code": 404, "message": "No results found", "data": None}
+
+        return {"code": 200, "message": "OK", "data": data}

loading_sdk/async_api/extractors.py

@@ -1,33 +1,30 @@
 import json
 import re
+from abc import ABC, abstractmethod
 
 import aiohttp
 from bs4 import BeautifulSoup
 from loading_sdk.settings import BASE_URL, USER_AGENT
 
 
-class AboutPageExtractor:
-    async def extract_about_data(self):
-        about_page_source = await self._get_source(f"{BASE_URL}/om")
-        main_script_url = self._extract_main_script_url(about_page_source)
-        main_script_source = await self._get_source(f"{BASE_URL}/{main_script_url}")
-        about_script_url = self._get_about_script_url(main_script_source)
-        about_script_source = await self._get_source(about_script_url)
-
-        return self._get_about_data(about_script_source)
-
-    async def _get_source(self, url):
+class Extractor(ABC):
+    async def get_source(self, url: str) -> str:
         headers = {"User-Agent": USER_AGENT}
-
         async with aiohttp.ClientSession() as session:
             async with session.get(url, headers=headers) as response:
                 return await response.text()
 
-    def _get_about_script_url(self, source_code):
+    def get_script(self, source: str) -> str:
+        soup = BeautifulSoup(source, "html.parser")
+        main_script = soup.find(src=re.compile(r"/static/js/main\.[0-9a-zA-Z]+\.js"))
+
+        return main_script["src"][1:]
+
+    def get_chunks(self, source: str) -> list:
         chunk_urls = []
 
         # Extracts the code with the javascript chunks.
-        match = re.search(r"(static/js/).+?(?=\{)(.+?(?=\[)).+(.chunk.js)", source_code)
+        match = re.search(r"(static/js/).+?(?=\{)(.+?(?=\[)).+(.chunk.js)", source)
 
         if match:
             # Transform the code into valid JSON so the chunk ids can be stored in a python dict.
@@ -38,10 +35,25 @@ def _get_about_script_url(self, source_code):
                 chunk_url = f"{BASE_URL}/{match.group(1)}{key}.{value}{match.group(3)}"
                 chunk_urls.append(chunk_url)
 
-        return chunk_urls[-1]
+        return chunk_urls
+
+    @abstractmethod
+    def get_data(self):
+        pass
 
-    def _get_about_data(self, source_code):
-        match = re.search(r"var.e=(.+?)(?=\.map).+a=(.+?)(?=\.map)", source_code)
+
+class AboutExtractor(Extractor):
+    async def get_data(self):
+        about_page_source = await self.get_source(f"{BASE_URL}/om")
+        main_script_url = self.get_script(about_page_source)
+        main_script_source = await self.get_source(f"{BASE_URL}/{main_script_url}")
+        chunk_urls = self.get_chunks(main_script_source)
+        about_script_url = chunk_urls[-1]
+        about_script_source = await self.get_source(about_script_url)
+
+        match = re.search(
+            r"var.e=(.+?)(?=\.map).+a=(.+?)(?=\.map)", about_script_source
+        )
 
         if not match:
             return None
@@ -58,13 +70,63 @@ def _get_about_data(self, source_code):
         moderators = moderators.replace("\\n", "")
         moderators = moderators.encode("utf-8").decode("unicode_escape")
 
-        return {
+        data = {
             "people": json.loads(people),
             "moderators": json.loads(moderators),
         }
 
-    def _extract_main_script_url(self, html):
-        soup = BeautifulSoup(html, "html.parser")
-        main_script = soup.find(src=re.compile(r"/static/js/main\.[0-9a-zA-Z]+\.js"))
+        return data
 
-        return main_script["src"][1:]
+
+class SocialsExtractor(Extractor):
+    async def get_data(self):
+        page_source = await self.get_source(BASE_URL)
+        main_script_url = self.get_script(page_source)
+        main_script_source = await self.get_source(f"{BASE_URL}/{main_script_url}")
+
+        match = re.findall(
+            r"(?:href:\")"
+            + r"(https:\/\/|https:\/\/www.(.*?)\..*?\/.*?)"
+            + r"(?:\",target:\"_blank\",rel:\"noreferrer noopener\",className:)"
+            + r"(?:\"Footer-(?:icon|patreon)\")",
+            main_script_source,
+        )
+
+        if not match:
+            return None
+
+        data = [{"name": social[1], "link": social[0]} for social in match]
+
+        return data
+
+
+class ExtractorFactory(ABC):
+    @abstractmethod
+    def get_extractor(self) -> Extractor:
+        pass
+
+
+class AboutExtractorFactory(ExtractorFactory):
+    def get_extractor(self) -> Extractor:
+        return AboutExtractor()
+
+
+class SocialsExtractorFactory(ExtractorFactory):
+    def get_extractor(self) -> Extractor:
+        return SocialsExtractor()
+
+
+async def extract_data(extractor_name):
+    factories = {
+        "about": AboutExtractorFactory(),
+        "socials": SocialsExtractorFactory(),
+    }
+
+    if extractor_name in factories:
+        factory = factories[extractor_name]
+        extractor = factory.get_extractor()
+        data = await extractor.get_data()
+
+        return data
+
+    return None