feat(library): enhance service structure and add dev dependencies

freedium-library/pdm.lock

@@ -2,10 +2,10 @@
 # It is not intended for manual editing.
 
 [metadata]
-groups = ["default"]
+groups = ["default", "dev"]
 strategy = ["inherit_metadata"]
 lock_version = "4.5.0"
-content_hash = "sha256:0489285d5cdf4e87cdd78b7cf2b3eabc8e2aa5577fba25dab97984223b9056eb"
+content_hash = "sha256:fd3b813143614f861a04ce0e5eb385121bc26ef1459e91c6c80e50d67437c4fb"
 
 [[metadata.targets]]
 requires_python = "==3.12.*"
@@ -58,8 +58,8 @@ name = "colorama"
 version = "0.4.6"
 requires_python = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7"
 summary = "Cross-platform colored terminal text."
-groups = ["default"]
-marker = "platform_system == \"Windows\""
+groups = ["default", "dev"]
+marker = "sys_platform == \"win32\" or platform_system == \"Windows\""
 files = [
     {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"},
     {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
@@ -138,6 +138,17 @@ files = [
     {file = "idna-3.10.tar.gz", hash = "sha256:12f65c9b470abda6dc35cf8e63cc574b1c52b11df2c86030af0ac09b01b13ea9"},
 ]
 
+[[package]]
+name = "iniconfig"
+version = "2.0.0"
+requires_python = ">=3.7"
+summary = "brain-dead simple config-ini parsing"
+groups = ["dev"]
+files = [
+    {file = "iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374"},
+    {file = "iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3"},
+]
+
 [[package]]
 name = "litestar"
 version = "2.12.1"
@@ -235,6 +246,28 @@ files = [
     {file = "multidict-6.1.0.tar.gz", hash = "sha256:22ae2ebf9b0c69d206c003e2f6a914ea33f0a932d4aa16f236afc049d9958f4a"},
 ]
 
+[[package]]
+name = "packaging"
+version = "24.1"
+requires_python = ">=3.8"
+summary = "Core utilities for Python packages"
+groups = ["dev"]
+files = [
+    {file = "packaging-24.1-py3-none-any.whl", hash = "sha256:5b8f2217dbdbd2f7f384c41c628544e6d52f2d0f53c6d0c3ea61aa5d1d7ff124"},
+    {file = "packaging-24.1.tar.gz", hash = "sha256:026ed72c8ed3fcce5bf8950572258698927fd1dbda10a5e981cdf0ac37f4f002"},
+]
+
+[[package]]
+name = "pluggy"
+version = "1.5.0"
+requires_python = ">=3.8"
+summary = "plugin and hook calling mechanisms for python"
+groups = ["dev"]
+files = [
+    {file = "pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669"},
+    {file = "pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1"},
+]
+
 [[package]]
 name = "polyfactory"
 version = "2.17.0"
@@ -261,6 +294,39 @@ files = [
     {file = "pygments-2.18.0.tar.gz", hash = "sha256:786ff802f32e91311bff3889f6e9a86e81505fe99f2735bb6d60ae0c5004f199"},
 ]
 
+[[package]]
+name = "pytest"
+version = "8.3.3"
+requires_python = ">=3.8"
+summary = "pytest: simple powerful testing with Python"
+groups = ["dev"]
+dependencies = [
+    "colorama; sys_platform == \"win32\"",
+    "exceptiongroup>=1.0.0rc8; python_version < \"3.11\"",
+    "iniconfig",
+    "packaging",
+    "pluggy<2,>=1.5",
+    "tomli>=1; python_version < \"3.11\"",
+]
+files = [
+    {file = "pytest-8.3.3-py3-none-any.whl", hash = "sha256:a6853c7375b2663155079443d2e45de913a911a11d669df02a50814944db57b2"},
+    {file = "pytest-8.3.3.tar.gz", hash = "sha256:70b98107bd648308a7952b06e6ca9a50bc660be218d53c257cc1fc94fda10181"},
+]
+
+[[package]]
+name = "pytest-asyncio"
+version = "0.24.0"
+requires_python = ">=3.8"
+summary = "Pytest support for asyncio"
+groups = ["dev"]
+dependencies = [
+    "pytest<9,>=8.2",
+]
+files = [
+    {file = "pytest_asyncio-0.24.0-py3-none-any.whl", hash = "sha256:a811296ed596b69bf0b6f3dc40f83bcaf341b155a269052d82efa2b25ac7037b"},
+    {file = "pytest_asyncio-0.24.0.tar.gz", hash = "sha256:d081d828e576d85f875399194281e92bf8a68d60d72d1a2faf2feddb6c46b276"},
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"

freedium-library/pyproject.toml

@@ -15,3 +15,8 @@ license = {text = "MIT"}
 
 [tool.pdm]
 distribution = false
+
+[tool.pdm.dev-dependencies]
+dev = [
+    "pytest-asyncio>=0.24.0",
+]

freedium-library/src/freedium_library/container.py

@@ -0,0 +1,7 @@
+from dependency_injector import containers, providers
+
+from freedium_library.models.request import Request
+
+
+class Container(containers.DeclarativeContainer):
+    request = providers.Singleton(Request)

freedium-library/src/freedium_library/models/__init__.py

@@ -0,0 +1,3 @@
+from .request import Request
+
+__all__ = ["Request"]

freedium-library/src/freedium_library/models/request.py

@@ -0,0 +1,99 @@
+from typing import Any, Dict, Optional
+
+from httpx import AsyncClient, Client, Response
+
+
+class Request:
+    def __init__(self):
+        self._client = Client()
+        self._async_client = AsyncClient()
+
+    def __del__(self):
+        self._client.close()
+        # self._async_client.close()
+
+    def get(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return self._client.get(
+            url, params=params, headers=headers, follow_redirects=follow_redirects
+        )
+
+    def post(
+        self,
+        url: str,
+        data: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return self._client.post(
+            url, json=data, headers=headers, follow_redirects=follow_redirects
+        )
+
+    def put(
+        self,
+        url: str,
+        data: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return self._client.put(
+            url, json=data, headers=headers, follow_redirects=follow_redirects
+        )
+
+    def delete(
+        self,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return self._client.delete(
+            url, headers=headers, follow_redirects=follow_redirects
+        )
+
+    async def aget(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return await self._async_client.get(
+            url, params=params, headers=headers, follow_redirects=follow_redirects
+        )
+
+    async def apost(
+        self,
+        url: str,
+        data: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return await self._async_client.post(
+            url, json=data, headers=headers, follow_redirects=follow_redirects
+        )
+
+    async def aput(
+        self,
+        url: str,
+        data: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return await self._async_client.put(
+            url, json=data, headers=headers, follow_redirects=follow_redirects
+        )
+
+    async def adelete(
+        self,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        follow_redirects: bool = True,
+    ) -> Response:
+        return await self._async_client.delete(
+            url, headers=headers, follow_redirects=follow_redirects
+        )

freedium-library/src/freedium_library/services/base.py

@@ -1,5 +1,28 @@
-from abc import ABC
+from abc import ABC, abstractmethod
+from typing import Any
+
+from dependency_injector.wiring import Provide
+
+from freedium_library.container import Container
+from freedium_library.models.request import Request
 
 
 class BaseService(ABC):
-    pass
+    def __init__(self, request: Request = Provide[Container.request]):
+        self.request = request
+
+    @abstractmethod
+    def is_valid(self) -> bool:
+        pass
+
+    @abstractmethod
+    async def ais_valid(self) -> bool:
+        pass
+
+    @abstractmethod
+    def render(self) -> Any:
+        pass
+
+    @abstractmethod
+    async def arender(self) -> Any:
+        pass

freedium-library/src/freedium_library/services/exceptions.py

@@ -0,0 +1,6 @@
+class BaseServiceError(Exception):
+    pass
+
+
+class InvalidServiceError(BaseServiceError):
+    pass

freedium-library/src/freedium_library/services/medium/__init__.py

@@ -1 +1,3 @@
 from .medium import MediumService
+
+__all__ = ["MediumService"]

freedium-library/src/freedium_library/services/medium/api.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from freedium_library.models import Request
+
+
+class MediumApiService:
+    def __init__(self, request: Request):
+        self.request = request

freedium-library/src/freedium_library/services/medium/container.py

@@ -0,0 +1,9 @@
+from dependency_injector import containers, providers
+
+from .api import MediumApiService
+from .validators import MediumServicePathValidator
+
+
+class MediumContainer(containers.DeclarativeContainer):
+    medium_api_service = providers.Singleton(MediumApiService)
+    medium_path_validator = providers.Singleton(MediumServicePathValidator)

freedium-library/src/freedium_library/services/medium/exceptions.py

@@ -0,0 +1,9 @@
+from freedium_library.services.exceptions import BaseServiceError
+
+
+class MediumServiceError(BaseServiceError):
+    pass
+
+
+class InvalidMediumServicePathError(MediumServiceError):
+    pass

freedium-library/src/freedium_library/services/medium/medium.py

@@ -1,2 +1,59 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
+from dependency_injector.wiring import Provide
+
+from freedium_library.container import Container
+from freedium_library.services.base import BaseService
+
+from .api import MediumApiService
+from .container import MediumContainer
+from .validators import MediumServicePathValidator
+
+if TYPE_CHECKING:
+    from freedium_library.models.request import Request
+
+
 class MediumService(BaseService):
-    pass
+    def __init__(
+        self,
+        request: Request = Provide[Container.request],
+        api_service: MediumApiService = Provide[MediumContainer.medium_api_service],
+        path_validator: MediumServicePathValidator = Provide[
+            MediumContainer.medium_path_validator
+        ],
+    ):
+        self.request = request
+        self.api_service = api_service
+        self.path_validator = path_validator
+        self._content: Optional[str] = None
+
+    def is_valid(self, path: str) -> bool:
+        return self.path_validator.is_valid(path)
+
+    async def ais_valid(self, path: str) -> bool:
+        return await self.path_validator.ais_valid(path)
+
+    def render(self, path: str) -> str:
+        if not self.is_valid():
+            raise ValueError("Invalid Medium URL")
+        if self._content is None:
+            response = self.request.get(self._url)
+            self._content = self._process_content(response.text)
+        return self._content
+
+    async def arender(self) -> str:
+        if not await self.ais_valid():
+            raise ValueError("Invalid Medium URL")
+        if self._content is None:
+            response = await self.request.aget(self._url)
+            self._content = self._process_content(response.text)
+        return self._content
+
+    def _process_content(self, content: str) -> str:
+        return content
+
+    def set_url(self, url: str) -> None:
+        self._url = url
+        self._content = None

freedium-library/src/freedium_library/services/medium/validators.py

@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any, Optional
+from urllib.parse import parse_qs, urlparse
+
+from dependency_injector.wiring import Provide
+from loguru import logger
+
+from freedium_library.container import Container
+from freedium_library.utils import URLProcessor
+
+if TYPE_CHECKING:
+    from freedium_library.models.request import Request
+
+    from .api import MediumApiService
+
+
+class MediumServicePathValidator:
+    def __init__(
+        self,
+        api_service: MediumApiService,
+    ):
+        self.api_service = api_service
+        self.hash_validator = _MediumServiceHashesValidator()
+        self.url_validator = _MediumServiceURLValidator(
+            self.api_service, self.hash_validator
+        )
+
+    def is_valid(self, path: str) -> bool:
+        return self.url_validator.is_valid(path) or self.hash_validator.extract_hashes(
+            path
+        )
+
+
+class _MediumServiceURLValidator:
+    def __init__(
+        self,
+        api_service: MediumApiService,
+        hash_validator: _MediumServiceHashesValidator,
+        request: Request = Provide[Container.request],
+    ):
+        self.api_service = api_service
+        self.request = request
+        self.hash_validator = hash_validator
+
+    def is_valid(self, url: str) -> bool: ...
+
+    def _get_short_link_request_params(
+        self, short_url_id: str
+    ) -> tuple[str, dict[str, Any]]:
+        req_url = f"https://rsci.app.link/{short_url_id}"
+        req_headers = {
+            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36"
+        }  # TODO: use random user agent generator library
+        return req_url, {"headers": req_headers, "follow_redirects": False}
+
+    async def resolve_medium_url(self, url: str) -> Optional[str]:
+        logger.debug(f"Trying resolve {url=}")
+        parsed_url = urlparse(url)
+        parsed_netloc = URLProcessor.un_wwwify(parsed_url.netloc)
+
+        if parsed_url.path.startswith("/p/"):
+            logger.debug("URL is Medium 'mobile' link")
+            post_id = parsed_url.path.rsplit("/p/")[1]
+
+        elif parsed_netloc == "l.facebook.com" and parsed_url.path.startswith("/l.php"):
+            logger.debug("URL seems like is Facebook redirect (tracking) link")
+
+            parsed_query = parse_qs(parsed_url.query)
+            if parsed_query.get("u") and len(parsed_query["u"]) == 1:
+                post_url = parsed_query["u"][0]
+                return await self.resolve_medium_url(post_url)
+
+            logger.debug("...but we get fucked up...")
+            return None
+
+        elif (
+            parsed_netloc == "webcache.googleusercontent.com"
+            and parsed_url.path.startswith("/search")
+        ):
+            logger.debug("URL seems like is Google Web Archive page link")
+
+            parsed_query = parse_qs(parsed_url.query)
+            if parsed_query.get("q") and len(parsed_query["q"]) == 1:
+                post_url = parsed_query["q"][0].removeprefix("cache:")
+                return await self.resolve_medium_url(post_url)
+
+            logger.debug("...but we get fucked up...")
+            return None
+
+        elif parsed_netloc == "google.com" and parsed_url.path.startswith("/url"):
+            logger.debug("URL seems like is Google redirect (tracking) link")
+
+            parsed_query = parse_qs(parsed_url.query)
+            if parsed_query.get("url") and len(parsed_query["url"]) == 1:
+                logger.debug("..and we got 'url' passed param. Make resolve them....")
+                post_url = parsed_query["url"][0]
+                return await self.resolve_medium_url(post_url)
+            elif parsed_query.get("q") and len(parsed_query["q"]) == 1:
+                logger.debug("..and we got 'q' passed param. Make resolve them....")
+                post_url = parsed_query["q"][0]
+                return await self.resolve_medium_url(post_url)
+
+            logger.debug("...but we get fucked up...")
+            return None
+
+        elif parsed_netloc == "12ft.io":
+            logger.debug("URL seems like is from our partner named 12ft.io")
+
+            parsed_query = parse_qs(parsed_url.query)
+            if parsed_query.get("q") and len(parsed_query["q"]) == 1:
+                logger.debug("..and we got 'q' passed param. Make resolve them....")
+                post_url = parsed_query["q"][0]
+                return await self.resolve_medium_url(post_url)
+
+            logger.debug("...but we get fucked up...")
+            return None
+
+        elif parsed_url.path.startswith("/m/global-identity-2"):
+            logger.debug(
+                "URL seems like is Medium redirect (tracking) link. Possibly from email subscription"
+            )
+
+            parsed_query = parse_qs(parsed_url.query)
+            if (
+                parsed_query.get("redirectUrl")
+                and len(parsed_query["redirectUrl"]) == 1
+            ):
+                logger.debug(
+                    "..and we got 'redirectUrl' passed param. Make resolve them...."
+                )
+                post_url = parsed_query["redirectUrl"][0]
+                return await self.resolve_medium_url(post_url)
+
+            logger.debug("...but we get fucked up...")
+            return None
+
+        elif parsed_netloc == "link.medium.com":
+            logger.debug(
+                "URL seems like is Medium short (SHORT) redirect (tracking) link. Make resolve them..."
+            )
+            short_url_id = parsed_url.path.removeprefix("/")
+            post_url = await self.aresolve_medium_short_link(short_url_id)
+            return await self.resolve_medium_url(post_url)
+
+        else:
+            logger.debug(
+                "We can't determine the URL type. Let's just try to extract the post_id..."
+            )
+            post_url = parsed_url.path.split("/")[-1]
+            post_id = post_url.split("-")[-1]
+
+        if not self.hash_validator.is_valid(post_id):
+            logger.warning(f"...but hoops, that's invalid post_id: {post_id}")
+            return None
+
+        return post_id
+
+    def resolve_medium_short_link(self, short_url_id: str) -> str:
+        req_url, params = self._get_short_link_request_params(short_url_id)
+        response = self.request.get(req_url, **params)
+        return response.headers["Location"]
+
+    async def aresolve_medium_short_link(self, short_url_id: str) -> str:
+        req_url, params = self._get_short_link_request_params(short_url_id)
+        response = await self.request.aget(req_url, **params)
+        return response.headers["Location"]
+
+
+class _MediumServiceHashesValidator:
+    def is_valid(self, hash: str) -> bool:
+        return bool(self.extract_hashes(hash))
+
+    def extract_hashes(self, path: str) -> list[str]:
+        logger.debug(f"Extracting hashes from path: {path}")
+        if not path:
+            return []
+
+        # we ignore ? symbols, because they can be in the end of the path right before the query params
+        logger.trace("Stage 1: finding hashes preceded by '-'")
+        match = re.findall(r"-(\b[a-fA-F0-9]{8,12}\b)\??", path)
+        if not match:
+            logger.trace("Stage 2: finding hashes without '-'")
+            match = re.findall(r"(\b[a-fA-F0-9]{8,12}\b)\??", path)
+
+        return match if match else []

freedium-library/src/freedium_library/services/medium/validators_hashes_test.py

@@ -0,0 +1,93 @@
+import pytest
+
+from freedium_library.container import Container
+from freedium_library.services.medium.container import MediumContainer
+from freedium_library.services.medium.validators import (
+    _MediumServiceHashesValidator,  # type: ignore
+)
+
+
+class TestMediumServiceHashesValidator:
+    @pytest.fixture
+    def validator(self) -> _MediumServiceHashesValidator:
+        Container().wire(modules=[__name__])
+        MediumContainer().wire(modules=[__name__])
+        return _MediumServiceHashesValidator()
+
+    def test_is_valid(self, validator: _MediumServiceHashesValidator):
+        assert validator.is_valid("515dd5a43948")
+        assert validator.is_valid("c81e00f6320d")
+        assert validator.is_valid("-515dd5a43948")
+        assert not validator.is_valid("")
+        assert not validator.is_valid("invalid")
+        assert not validator.is_valid("12345")
+        assert not validator.is_valid("zzzzzzzzzzz")
+
+    def test_extract_hashes(self, validator: _MediumServiceHashesValidator):
+        assert validator.extract_hashes("stop-wasting-your-life-27832c8f6644") == [
+            "27832c8f6644"
+        ]
+        assert validator.extract_hashes("some-linux-commands-9dc802a10618") == [
+            "9dc802a10618"
+        ]
+
+        assert validator.extract_hashes("515dd5a43948") == ["515dd5a43948"]
+        assert validator.extract_hashes("c81e00f6320d") == ["c81e00f6320d"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/macoclock/12-macos-apps-515dd5a43948"
+        ) == ["515dd5a43948"]
+        assert validator.extract_hashes("https://www.freedium.cfd/c81e00f6320d") == [
+            "c81e00f6320d"
+        ]
+
+        assert validator.extract_hashes("") == []
+        assert validator.extract_hashes("no-hash-here") == []
+        assert validator.extract_hashes("invalid-hash-123") == []
+        assert validator.extract_hashes("zzzzzzzzzzz") == []
+
+        assert validator.extract_hashes(
+            "https://medium.com/macoclock/the-11-craziest-and-most-advanced-macos-tips-tricks-ive-ever-seen-cd842ce3f0a3"
+        ) == ["cd842ce3f0a3"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/calendar/practicing-purposeful-productivity-ef976b89890e"
+        ) == ["ef976b89890e"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/@andreafeccomandi/how-to-use-the-self-fulfilling-prophecy-narrative-technique-2c7bf44edbdc"
+        ) == ["2c7bf44edbdc"]
+
+        assert validator.extract_hashes(
+            "https://grigorkh.medium.com/native-lazy-loading-of-images-and-iframes-6931fe455632"
+        ) == ["6931fe455632"]
+
+        assert validator.extract_hashes(
+            "codeium-best-free-alternative-for-github-copilot-5a010f74d3e1?subid1=20241023-0155-4648-992f-1316e4a9d771"
+        ) == ["5a010f74d3e1", "1316e4a9d771"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/dataman-in-ai/temporal-fusion-transformer-for-interpretable-time-series-predictions-4b439aa3d9bd"
+        ) == ["4b439aa3d9bd"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/dataman-in-ai/temporal-fusion-transformer-for-interpretable-time-series-predictions-4b439aa3d9bd+"
+        ) == ["4b439aa3d9bd"]
+
+        assert validator.extract_hashes(
+            "https://medium.com/dataman-in-ai/temporal-fusion-transformer-for-interpretable-time-series-predictions-4b439aa3d9bd+fds"
+        ) == ["4b439aa3d9bd"]
+
+        assert validator.extract_hashes(
+            "https://12ft.io/api/proxy?q=https%3A%2F%2Fednico.medium.com%2Fpkm-weekly-2024-05-26-6601694e4147"
+        ) == ["6601694e4147"]
+
+    def test_extract_multiple_hashes(self, validator: _MediumServiceHashesValidator):
+        text = "first-515dd5a43948 second-c81e00f6320d"
+        assert set(validator.extract_hashes(text)) == {"515dd5a43948", "c81e00f6320d"}
+
+    def test_edge_cases(self, validator: _MediumServiceHashesValidator):
+        assert validator.extract_hashes(None) == []  # type: ignore
+        assert validator.extract_hashes("   ") == []
+        assert validator.extract_hashes("abc123") == []
+        assert validator.extract_hashes("12345678901234567890") == []

freedium-library/src/freedium_library/services/medium/validators_url_test.py

@@ -0,0 +1,203 @@
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from freedium_library.models.request import Request
+from freedium_library.services.medium.api import MediumApiService
+from freedium_library.services.medium.validators import (
+    _MediumServiceHashesValidator,  # type: ignore
+    _MediumServiceURLValidator,  # type: ignore
+)
+
+
+@pytest.fixture
+def mock_request() -> Mock:
+    request = Mock()
+    request.get = Mock()
+    request.aget = AsyncMock()
+    return request
+
+
+@pytest.fixture
+def mock_api_service() -> Mock:
+    return Mock()
+
+
+@pytest.fixture
+def url_validator(
+    mock_api_service: Mock, mock_request: Mock
+) -> _MediumServiceURLValidator:
+    hash_validator = _MediumServiceHashesValidator()
+    return _MediumServiceURLValidator(
+        api_service=mock_api_service,
+        hash_validator=hash_validator,
+        request=mock_request,
+    )
+
+
+def test_resolve_medium_short_link(
+    url_validator: _MediumServiceURLValidator, mock_request: Mock
+) -> None:
+    mock_request.get.return_value = Mock(
+        headers={"Location": "https://medium.com/story-123456789abc"}
+    )
+
+    result = url_validator.resolve_medium_short_link("vYe3nWA8wBb")
+
+    assert result == "https://medium.com/story-123456789abc"
+    mock_request.get.assert_called_once()
+    assert "rsci.app.link/vYe3nWA8wBb" in mock_request.get.call_args[0][0]
+
+
+@pytest.mark.asyncio
+async def test_aresolve_medium_short_link(
+    url_validator: _MediumServiceURLValidator, mock_request: Mock
+) -> None:
+    mock_request.aget.return_value = Mock(
+        headers={"Location": "https://medium.com/story-123456789abc"}
+    )
+
+    result = await url_validator.aresolve_medium_short_link("vYe3nWA8wBb")
+
+    assert result == "https://medium.com/story-123456789abc"
+    mock_request.aget.assert_called_once()
+    assert "rsci.app.link/vYe3nWA8wBb" in mock_request.aget.call_args[0][0]
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_with_short_link(
+    url_validator: _MediumServiceURLValidator, mock_request: Mock
+) -> None:
+    mock_request.aget.return_value = Mock(
+        headers={"Location": "https://medium.com/story-123456789abc"}
+    )
+
+    result = await url_validator.resolve_medium_url(
+        "https://link.medium.com/vYe3nWA8wBb"
+    )
+
+    assert result == "123456789abc"
+    mock_request.aget.assert_called_once()
+    assert "rsci.app.link/vYe3nWA8wBb" in mock_request.aget.call_args[0][0]
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_invalid_hash(
+    url_validator: _MediumServiceURLValidator, mock_request: Mock
+) -> None:
+    mock_request.aget.return_value = Mock(
+        headers={"Location": "https://medium.com/invalid-story"}
+    )
+
+    result = await url_validator.resolve_medium_url(
+        "https://link.medium.com/vYe3nWA8wBb"
+    )
+
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_with_real_short_link(
+    url_validator: _MediumServiceURLValidator, mock_request: Mock
+) -> None:
+    mock_request.aget.return_value = Mock(
+        headers={"Location": "https://medium.com/story-77ae792a1a43"}
+    )
+
+    result = await url_validator.resolve_medium_url(
+        "https://link.medium.com/vYe3nWA8wBb"
+    )
+
+    assert result == "77ae792a1a43"
+    mock_request.aget.assert_called_once()
+    assert "rsci.app.link/vYe3nWA8wBb" in mock_request.aget.call_args[0][0]
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_resolve_medium_url_with_real_short_link_integration() -> None:
+    request = Request()
+    api_service = MediumApiService(request=request)
+    hash_validator = _MediumServiceHashesValidator()
+    url_validator = _MediumServiceURLValidator(api_service, hash_validator, request)
+
+    result = await url_validator.resolve_medium_url(
+        "https://link.medium.com/vYe3nWA8wBb"
+    )
+
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_facebook_redirect(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url(
+        "https://l.facebook.com/l.php?u=https://medium.com/story-77ae792a1a43"
+    )
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_google_cache(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url(
+        "https://webcache.googleusercontent.com/search?q=https://medium.com/story-77ae792a1a43"
+    )
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_google_redirect(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url(
+        "https://google.com/url?url=https://medium.com/story-77ae792a1a43"
+    )
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_12ft(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url(
+        "https://12ft.io?q=https://medium.com/story-77ae792a1a43"
+    )
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_global_identity(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url(
+        "https://medium.com/m/global-identity-2?redirectUrl=https://medium.com/story-77ae792a1a43"
+    )
+    assert result == "77ae792a1a43"
+
+
+@pytest.mark.asyncio
+async def test_resolve_medium_url_mobile_link(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = await url_validator.resolve_medium_url("https://medium.com/p/77ae792a1a43")
+    assert result == "77ae792a1a43"
+
+
+def test_hash_validator_valid_hash(url_validator: _MediumServiceURLValidator) -> None:
+    result = url_validator.hash_validator.is_valid("77ae792a1a43")
+    assert result is True
+
+
+def test_hash_validator_invalid_hash(url_validator: _MediumServiceURLValidator) -> None:
+    result = url_validator.hash_validator.is_valid("invalid")
+    assert result is False
+
+
+def test_hash_validator_extract_hashes(
+    url_validator: _MediumServiceURLValidator,
+) -> None:
+    result = url_validator.hash_validator.extract_hashes("story-77ae792a1a43-something")
+    assert result == ["77ae792a1a43"]

freedium-library/src/freedium_library/utils.py

@@ -0,0 +1,59 @@
+from urllib.parse import urlparse, urlunparse
+
+
+class URLProcessor:
+    """A class for processing and sanitizing URLs."""
+
+    @staticmethod
+    def un_wwwify(url: str) -> str:
+        """Remove 'www.' prefix from URLs while preserving URL structure.
+
+        Args:
+            url: The URL string to process
+
+        Returns:
+            str: URL with 'www.' prefix removed if present, otherwise unchanged
+
+        Raises:
+            TypeError: If url is not a string
+            ValueError: If url is empty or malformed
+        """
+        if not isinstance(url, str):
+            raise TypeError("url must be a string")
+
+        if not url:
+            raise ValueError("url must not be empty")
+
+        try:
+            parsed = urlparse(url)
+            netloc = parsed.netloc or parsed.path
+
+            if netloc.startswith("www."):
+                netloc = netloc[4:]
+
+            if parsed.netloc:
+                return urlunparse(parsed._replace(netloc=netloc))
+            elif parsed.path:
+                return urlunparse(parsed._replace(path=netloc))
+
+            return netloc
+        except Exception as e:
+            raise ValueError(f"Invalid URL format: {str(e)}")
+
+    @staticmethod
+    def unquerify_url(url: str) -> str:
+        """Sanitizes a URL by removing all query parameters.
+
+        Args:
+            url: The URL to sanitize.
+
+        Returns:
+            str: A sanitized URL without query parameters and trailing slash.
+        """
+        parsed_url = urlparse(url)
+        query = parsed_url.query
+        if query:
+            parsed_url = parsed_url._replace(query="")
+
+        sanitized_url = urlunparse(parsed_url)
+        return sanitized_url.removesuffix("/")

freedium-library/src/freedium_library/utils_test.py

@@ -0,0 +1,112 @@
+import pytest
+
+from freedium_library.utils import URLProcessor
+
+
+class TestUnWwwify:
+    @pytest.fixture
+    def url_processor(self):
+        return URLProcessor()
+
+    def test_basic_www_removal(self, url_processor: URLProcessor):
+        assert url_processor.un_wwwify("www.example.com") == "example.com"
+        assert (
+            url_processor.un_wwwify("https://www.example.com") == "https://example.com"
+        )
+        assert url_processor.un_wwwify("http://www.example.com") == "http://example.com"
+
+    def test_non_www_urls(self, url_processor: URLProcessor):
+        assert url_processor.un_wwwify("example.com") == "example.com"
+        assert url_processor.un_wwwify("https://example.com") == "https://example.com"
+        assert url_processor.un_wwwify("sub.example.com") == "sub.example.com"
+
+    def test_urls_with_paths(self, url_processor: URLProcessor):
+        assert url_processor.un_wwwify("www.example.com/path") == "example.com/path"
+        assert (
+            url_processor.un_wwwify("https://www.example.com/path")
+            == "https://example.com/path"
+        )
+        assert (
+            url_processor.un_wwwify("https://www.example.com/path/to/resource")
+            == "https://example.com/path/to/resource"
+        )
+
+    def test_urls_with_query_params(self, url_processor: URLProcessor):
+        assert (
+            url_processor.un_wwwify("www.example.com?key=value")
+            == "example.com?key=value"
+        )
+        assert (
+            url_processor.un_wwwify("https://www.example.com?key=value")
+            == "https://example.com?key=value"
+        )
+        assert (
+            url_processor.un_wwwify("https://www.example.com/path?key=value&other=123")
+            == "https://example.com/path?key=value&other=123"
+        )
+
+    def test_error_handling(self, url_processor: URLProcessor):
+        with pytest.raises(ValueError):
+            url_processor.un_wwwify("")
+
+        with pytest.raises(TypeError):
+            url_processor.un_wwwify(None)  # type: ignore
+
+    def test_special_cases(self, url_processor: URLProcessor):
+        assert url_processor.un_wwwify("www.www.example.com") == "www.example.com"
+        assert url_processor.un_wwwify("wwww.example.com") == "wwww.example.com"
+        assert url_processor.un_wwwify("example.www.com") == "example.www.com"
+
+
+class TestUnquerifyUrl:
+    @pytest.fixture
+    def url_processor(self):
+        return URLProcessor()
+
+    def test_basic_query_removal(self, url_processor: URLProcessor):
+        assert url_processor.unquerify_url("example.com?param=value") == "example.com"
+        assert (
+            url_processor.unquerify_url("https://example.com?param=value")
+            == "https://example.com"
+        )
+
+    def test_multiple_query_params(self, url_processor: URLProcessor):
+        assert url_processor.unquerify_url("example.com?p1=v1&p2=v2") == "example.com"
+        assert (
+            url_processor.unquerify_url("https://example.com?p1=v1&p2=v2")
+            == "https://example.com"
+        )
+
+    def test_urls_without_query(self, url_processor: URLProcessor):
+        assert url_processor.unquerify_url("example.com") == "example.com"
+        assert (
+            url_processor.unquerify_url("https://example.com") == "https://example.com"
+        )
+
+    def test_urls_with_paths(self, url_processor: URLProcessor):
+        assert (
+            url_processor.unquerify_url("example.com/path?param=value")
+            == "example.com/path"
+        )
+        assert (
+            url_processor.unquerify_url("https://example.com/path?param=value")
+            == "https://example.com/path"
+        )
+
+    def test_trailing_slash_removal(self, url_processor: URLProcessor):
+        assert (
+            url_processor.unquerify_url("example.com/path/?param=value")
+            == "example.com/path"
+        )
+        assert (
+            url_processor.unquerify_url("https://example.com/?param=value")
+            == "https://example.com"
+        )
+
+    def test_error_handling(self, url_processor: URLProcessor):
+        ...
+        # with pytest.raises(ValueError):
+        #     url_processor.unquerify_url("")
+
+        # with pytest.raises(TypeError):
+        #     url_processor.unquerify_url(None)

medium-parser/medium_parser/utils.py

@@ -65,7 +65,7 @@ KNOWN_MEDIUM_DOMAINS = (
     "entrepreneurshandbook.co",
     "prototypr.io",
     "theascent.pub",
-    "storiusmag.com"
+    "storiusmag.com",
 )
 NOT_MEDIUM_DOMAINS = (
     "github.com",
@@ -151,6 +151,7 @@ def unquerify_url(url: str) -> str:
     query = parsed_url.query
     if query:
         parsed_url = parsed_url._replace(query="")
+
     sanitized_url = urllib.parse.urlunparse(parsed_url)
     return sanitized_url.removesuffix("/")
 
@@ -229,20 +230,24 @@ def basic_hex_check(hex_string: str) -> bool:
 @lru_cache(maxsize=100)
 def extract_hex_string(input_string: str) -> str:
     # First try to find a hexadecimal string preceded by a '-'
-    match = re.findall(r'-(\b[a-fA-F0-9]{8,12}\b)', input_string)
+    match = re.findall(r"-(\b[a-fA-F0-9]{8,12}\b)", input_string)
     if not match:
         # If no match, try to find a hexadecimal string without the '-'
-        match = re.findall(r'(\b[a-fA-F0-9]{8,12}\b)', input_string)
+        match = re.findall(r"(\b[a-fA-F0-9]{8,12}\b)", input_string)
     return match[-1] if match else None
 
 
 async def resolve_medium_short_link(short_url_id: str, timeout: int = 5) -> str:
     async with aiohttp.ClientSession() as session:
-        retry_client = RetryClient(client_session=session, raise_for_status=False, retry_options=retry_options)
+        retry_client = RetryClient(
+            client_session=session, raise_for_status=False, retry_options=retry_options
+        )
         request = await retry_client.get(
             f"https://rsci.app.link/{short_url_id}",
             timeout=timeout,
-            headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36"},
+            headers={
+                "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36"
+            },
             allow_redirects=False,
         )
         post_url = request.headers["Location"]
@@ -271,7 +276,10 @@ async def resolve_medium_url(url: str, timeout: int = 5) -> str:
         logger.debug("...but we get fucked up...")
         return False
 
-    elif parsed_netloc == "webcache.googleusercontent.com" and parsed_url.path.startswith("/search"):
+    elif (
+        parsed_netloc == "webcache.googleusercontent.com"
+        and parsed_url.path.startswith("/search")
+    ):
         logger.debug("URL seems like is Google Web Archive page link")
 
         parsed_query = parse_qs(parsed_url.query)
@@ -311,11 +319,15 @@ async def resolve_medium_url(url: str, timeout: int = 5) -> str:
         return False
 
     elif parsed_url.path.startswith("/m/global-identity-2"):
-        logger.debug("URL seems like is Medium redirect (tracking) link. Possibly from email subscription")
+        logger.debug(
+            "URL seems like is Medium redirect (tracking) link. Possibly from email subscription"
+        )
 
         parsed_query = parse_qs(parsed_url.query)
         if parsed_query.get("redirectUrl") and len(parsed_query["redirectUrl"]) == 1:
-            logger.debug("..and we got 'redirectUrl' passed param. Make resolve them....")
+            logger.debug(
+                "..and we got 'redirectUrl' passed param. Make resolve them...."
+            )
             post_url = parsed_query["redirectUrl"][0]
             return await resolve_medium_url(post_url)
 
@@ -323,13 +335,17 @@ async def resolve_medium_url(url: str, timeout: int = 5) -> str:
         return False
 
     elif parsed_netloc == "link.medium.com":
-        logger.debug("URL seems like is Medium short (SHORT) redirect (tracking) link. Make resolve them...")
+        logger.debug(
+            "URL seems like is Medium short (SHORT) redirect (tracking) link. Make resolve them..."
+        )
         short_url_id = parsed_url.path.removeprefix("/")
         post_url = await resolve_medium_short_link(short_url_id, timeout)
         return await resolve_medium_url(post_url)
 
     else:
-        logger.debug("We can't determine the URL type. Let's just try to extract the post_id...")
+        logger.debug(
+            "We can't determine the URL type. Let's just try to extract the post_id..."
+        )
         post_url = parsed_url.path.split("/")[-1]
         post_id = post_url.split("-")[-1]
 
@@ -342,7 +358,9 @@ async def resolve_medium_url(url: str, timeout: int = 5) -> str:
 
 async def resolve_medium_url_old(url: str, timeout: int = 5) -> str:
     async with aiohttp.ClientSession() as session:
-        retry_client = RetryClient(client_session=session, raise_for_status=False, retry_options=retry_options)
+        retry_client = RetryClient(
+            client_session=session, raise_for_status=False, retry_options=retry_options
+        )
         request = await retry_client.get(url, timeout=timeout)
         response = await request.text()
 
@@ -363,7 +381,9 @@ async def resolve_medium_url_old(url: str, timeout: int = 5) -> str:
 
 async def is_valid_medium_url_old(url: str, timeout: int = 5):
     async with aiohttp.ClientSession() as session:
-        retry_client = RetryClient(client_session=session, raise_for_status=False, retry_options=retry_options)
+        retry_client = RetryClient(
+            client_session=session, raise_for_status=False, retry_options=retry_options
+        )
 
         try:
             request = await retry_client.get(url, timeout=timeout)