🚧 Get game data

setup.cfg

@@ -23,6 +23,7 @@ install_requires =
     colorama
     typer
     pydantic
+    cloudscraper
 
 
 [options.entry_points]
@@ -79,9 +80,11 @@ format = wemake
 show-source = True
 exclude =
     src/mulefactory/version.py
-ignore = D107,D202,D203,D401,E203,E402,E501,W503,S101,H601
+ignore = D107,D202,D203,D401,E203,E402,E501,W503,S101,H601,D102,WPS431,D100,D101,D104,D400
 inline-quotes = double
 max-line-length = 120
+per-file-ignores =
+    tests/**.py:S106,WPS432,D103,WPS111
 
 [mypy-tests.*]
 ignore_errors = True

src/mulefactory/domain/mfactory.py

@@ -0,0 +1,25 @@
+import cloudscraper
+
+
+class Mulefactory(object):
+    scraper = cloudscraper.create_scraper(debug=True)
+
+    def __init__(self, server_id: int):
+        self.server_id = server_id
+
+    def login(self, username: str, password: str):
+        """Logs a user in to mulefactory.
+
+        Args:
+            username: supply.mulefactory username
+            password: supply.mulefactory password
+
+        Returns:
+            Response
+        """
+        request_data = {"postid": "login", "admin_name": username, "admin_pass": password}
+        url = "http://supply.mulefactory.com/"
+        return self.scraper.post(url=url, data=request_data)
+
+    def game_data(self):
+        raise NotImplementedError()

src/mulefactory/domain/path_of_exile.py

@@ -1,52 +1,29 @@
-from typing import Dict, Any
-import requests
 import datetime
+from typing import Any, Dict
 
-from mulefactory.domain import mulefactory
+from mulefactory.domain import mfactory
+
+
+class POEMulefactory(mfactory.Mulefactory):
+    game_id: int = 14
+    beginning_of_unix_time = datetime.datetime.utcfromtimestamp(0)
 
-class POEMulefactory(mulefactory.Mulefactory):
-    b = datetime.datetime.utcfromtimestamp(0)
     def game_data(self):
-        """curl 'https://supply.mulefactory.com/ajax.php?
-        module=supply&
-        target=ajax&
-        command=check&
-        gameid=14&
-        serverid=1047&
-        _=1619558804510'
-        -H 'User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:88.0) Gecko/20100101 Firefox/88.0' 
-        -H 'Accept: application/json, text/javascript, */*; q=0.01' 
-        -H 'Accept-Language: en-US,en;q=0.5' --compressed 
-        -H 'X-Requested-With: XMLHttpRequest' 
-        -H 'Connection: keep-alive' 
-        -H 'Referer: https://supply.mulefactory.com/supply/?gameid=14&serverid=1047' 
-        -H 'Cookie: __cfduid=dab170aa372b23e24571bc58f20aa34d91619558519; PHPSESSID=55c6400fcab994bee9d1624b5427fd8165b47635; appHeadReferer_v01=YToyOntzOjY6IlN1cHBseSI7czo2MjoiaHR0cHM6Ly9zdXBwbHkubXVsZWZhY3RvcnkuY29tL3N1cHBseS8%2FZ2FtZWlkPTE0JnNlcnZlcmlkPTEwNDciO3M6MDoiIjtzOjMxOiJodHRwczovL3N1cHBseS5tdWxlZmFjdG9yeS5jb20vIjt9' 
-        -H 'Pragma: no-cache' 
-        -H 'Cache-Control: no-cache' 
-        -H 'TE: Trailers'"
+        """Gets the item prices for a specific server_id and game.
+
+        Returns:
+            Response
         """
         url: str = "https://supply.mulefactory.com/ajax.php"
-        parameters: Dict[str,Any] = {
+        request_parameters: Dict[str, Any] = {
             "module": "supply",
-            "target":"ajax",
-            "command":"check",
-            "gameid":14,
-            "serverid":1047,
-            "_":self.get_time()
-        }
-        headers = {
-            'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:88.0) Gecko/20100101 Firefox/88.0',
-            'Accept': 'application/json, text/javascript, */*; q=0.01',
-            'Accept-Language': 'en-US,en;q=0.5', 
-            'X-Requested-With': 'XMLHttpRequest',
-            'Connection': 'keep-alive',
-            'Referer': 'https://supply.mulefactory.com/supply/?gameid=14&serverid=1047',
-            'Cookie': f'__cfduid={self.cfduid}; PHPSESSID={self.session_id};',
-            'Pragma': 'no-cache',
-            'Cache-Control': 'no-cache',
-            'TE': 'Trailers'
+            "target": "ajax",
+            "command": "check",
+            "gameid": self.game_id,
+            "serverid": self.server_id,
+            "_": self.get_time(),
         }
-        return requests.get(url=url, params=parameters, headers=headers)
+        return self.scraper.get(url=url, params=request_parameters)
 
     def get_time(self):
-        return int((datetime.datetime.utcnow() - self.b).total_seconds())
+        return int((datetime.datetime.utcnow() - self.beginning_of_unix_time).total_seconds())

src/mulefactory/domain/schemas.py

@@ -1,15 +1,33 @@
 import enum
-class Server(str, enum.Enum):
-    standard_sc = "standard_sc" 
-    standard_hc = "standard_hc" 
-    ultimatum_sc = "ultimatum_sc" 
-    ultimatum_hc = "ultimatum_hc" 
+
+from pydantic import BaseModel
+from pydantic import utils
+
+
+class Server(str, enum.Enum):  # noqa: WPS600
+    standard_sc = "standard_sc"
+    standard_hc = "standard_hc"
+    ultimatum_sc = "ultimatum_sc"
+    ultimatum_hc = "ultimatum_hc"
 
     def get_id(self) -> int:
         server_map = {
-            "standard_sc" : 1047,
+            "standard_sc": 1047,
             "standard_hc": 1048,
-            "ultimatum_sc" : 1742,
-            "ultimatum_hc" : 1743,
+            "ultimatum_sc": 1742,
+            "ultimatum_hc": 1743,
         }
         return server_map[self.value]
+
+
+class PoeItem(BaseModel):
+    identifier: int
+    limit: str
+    price: str
+    update: bool
+    exchange_rate: float
+
+    class Config(object):
+        """Pydantic configurator."""
+
+        alias_generator = utils.to_camel

src/mulefactory/main.py

@@ -1,25 +1,34 @@
 """This module is the main entrypoint of the application."""
 import typer
-from mulefactory.domain import schemas, path_of_exile
+
+from mulefactory.domain import path_of_exile
+from mulefactory.domain import schemas
 
 app = typer.Typer()
 
 
 @app.command()
-def path_of_exile(username: str, password: str, game_id: int = 14, server: schemas.Server = schemas.Server.standard_sc):
-    """
+def run_path_of_exile(
+    username: str,
+    password: str,
+    game_id: int = 14,
+    server: schemas.Server = schemas.Server.standard_sc,
+):
+    """Run mulefactory CLI for Path of Exile.
 
     Args:
+        username: spply.mulefactory.com username
+        password: spply.mulefactory.com password
         game_id: https://supply.mulefactory.com/supply/?gameid=14
-        server_id: Poe League, defaults to standard SC. See above link for options
+        server: Poe League, defaults to standard SC. See above link for options
     """
     server_id: int = server.get_id()
     typer.echo("Started Mulefactory POE scanner for '{server}' ({id})!".format(server=server, id=server_id))
-    m = path_of_exile.POEMulefactory()
-    m.login(username=username, password=password)
-    response = m.get_data()
-    typer.echo(response.json())
-
+    mule = path_of_exile.POEMulefactory(server_id=server_id)
+    mule.login(username=username, password=password)
+    response = mule.game_data()
+    typer.echo(response.text)
+    typer.echo(response.raw)
 
 
 if __name__ == "__main__":

tests/test_login.py

@@ -1,14 +1,13 @@
 import os
+
 from mulefactory.domain.mulefactory import Mulefactory
 
+
 def test_login():
     m = Mulefactory()
     username = os.environ["MULEFACTORY_USER"]
     password = os.environ["MULEFACTORY_PASS"]
     response = m.login(username=username, password=password)
-    c = dict(response.cookies)
-    assert "PHPSESSID" in c
-    assert "__cfduid" in c
-    assert m.session_id != ""
-    assert m.cfduid != ""
-
+    assert response.status_code == 200
+    assert "PHPSESSID" in m.scraper.cookies
+    assert "__cfduid" in m.scraper.cookies

tests/test_poe_get_data.py

@@ -0,0 +1,11 @@
+import os
+
+from mulefactory.domain.path_of_exile import POEMulefactory
+
+
+def test_get_poe_data():
+    poe = POEMulefactory()
+    poe.login(username=os.environ["MULEFACTORY_USER"], password=os.environ["MULEFACTORY_PASS"])
+    response = poe.game_data()
+    assert response.status_code == 200
+    assert response.json() is not None