Add files via upload

factory.py

@@ -0,0 +1,38 @@
+from typing import Dict, Type
+
+from mediaflow_proxy.extractors.base import BaseExtractor, ExtractorError
+from mediaflow_proxy.extractors.doodstream import DoodStreamExtractor
+from mediaflow_proxy.extractors.livetv import LiveTVExtractor
+from mediaflow_proxy.extractors.mixdrop import MixdropExtractor
+from mediaflow_proxy.extractors.uqload import UqloadExtractor
+from mediaflow_proxy.extractors.streamtape import StreamtapeExtractor
+from mediaflow_proxy.extractors.supervideo import SupervideoExtractor
+from mediaflow_proxy.extractors.vixcloud import VixCloudExtractor
+from mediaflow_proxy.extractors.okru import OkruExtractor
+from mediaflow_proxy.extractors.maxstream import MaxstreamExtractor
+
+
+
+
+class ExtractorFactory:
+    """Factory for creating URL extractors."""
+
+    _extractors: Dict[str, Type[BaseExtractor]] = {
+        "Doodstream": DoodStreamExtractor,
+        "Uqload": UqloadExtractor,
+        "Mixdrop": MixdropExtractor,
+        "Streamtape": StreamtapeExtractor,
+        "Supervideo": SupervideoExtractor,
+        "VixCloud": VixCloudExtractor,
+        "Okru": OkruExtractor,
+        "Maxstream": MaxstreamExtractor,
+        "LiveTV": LiveTVExtractor,
+    }
+
+    @classmethod
+    def get_extractor(cls, host: str, request_headers: dict) -> BaseExtractor:
+        """Get appropriate extractor instance for the given host."""
+        extractor_class = cls._extractors.get(host)
+        if not extractor_class:
+            raise ExtractorError(f"Unsupported host: {host}")
+        return extractor_class(request_headers)

maxstream.py

@@ -0,0 +1,59 @@
+import re
+import string
+from typing import Dict, Any
+
+from mediaflow_proxy.extractors.base import BaseExtractor, ExtractorError
+
+
+class MaxstreamExtractor(BaseExtractor):
+    """Maxstream URL extractor."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.mediaflow_endpoint = "hls_manifest_proxy"
+
+    async def extract(self, url: str, **kwargs) -> Dict[str, Any]:
+        """Extract Maxstream URL."""
+
+        response = await self._make_request(url, headers={"accept-language": "en-US,en;q=0.5"})
+
+        # Extract and decode URL
+        match = re.search(r"\}\('(.+)',.+,'(.+)'\.split", response.text)
+        if not match:
+            raise ExtractorError("Failed to extract URL components")
+
+        s1 = match.group(2)
+        # Extract Terms
+        terms = s1.split("|")
+
+        urlset_index = terms.index("urlset")
+        hls_index = terms.index("hls")
+        sources_index = terms.index("sources")
+        result = terms[urlset_index + 1 : hls_index]
+        reversed_elements = result[::-1]
+        first_part = terms[hls_index + 1 : sources_index]
+        reversed_first_part = first_part[::-1]
+        first_url_part = ""
+        first_url_part += "".join(
+            first_part if "0" in first_part else first_part + "-" for first_part in reversed_first_part
+        )
+        base_url = f"https://{first_url_part}.host-cdn.net/hls/"
+        if len(reversed_elements) == 1:
+            final_url = base_url + "," + reversed_elements[0] + ".urlset/master.m3u8"
+        lenght = len(reversed_elements)
+        i = 1
+        # Join up the elements
+        for i, element in enumerate(reversed_elements):
+            base_url += (
+                element
+                + (", " if i < len(reversed_elements) - 1 else "")
+                + (".urlset/master.m3u8" if lenght == i + 1 else "")
+            )
+        final_url = base_url
+
+        self.base_headers["referer"] = url
+        return {
+            "destination_url": final_url,
+            "request_headers": self.base_headers,
+            "mediaflow_endpoint": self.mediaflow_endpoint,
+        }

okru.py

@@ -0,0 +1,33 @@
+import re
+from bs4 import BeautifulSoup, SoupStrainer
+import json
+
+from typing import Dict, Any
+
+from mediaflow_proxy.extractors.base import BaseExtractor, ExtractorError
+
+
+class OkruExtractor(BaseExtractor):
+    """Okru URL extractor."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.mediaflow_endpoint = "hls_manifest_proxy"
+
+    async def extract(self, url: str, **kwargs) -> Dict[str, Any]:
+        """Extract Okru URL."""
+        response = await self._make_request(url)
+        soup = BeautifulSoup(response.text, "lxml", parse_only=SoupStrainer("div"))
+        if soup:
+            div = soup.find("div", {"data-module": "OKVideo"})
+            data_options = div.get("data-options")
+            data = json.loads(data_options)
+            metadata = json.loads(data["flashvars"]["metadata"])
+            final_url = metadata["hlsMasterPlaylistUrl"]
+
+            self.base_headers["referer"] = url
+            return {
+                "destination_url": final_url,
+                "request_headers": self.base_headers,
+                "mediaflow_endpoint": self.mediaflow_endpoint,
+            }