handler for pin codes, pickles

.gitignore

@@ -7,7 +7,7 @@ dist
 .gitpod.yml
 
 # Credentials
-.session.json
+.session.pkl
 
 # Codesandbox Development
 # .codesandbox/
@@ -42,6 +42,4 @@ package-lock.json
 npm-debug.log*
 yarn-debug.log*
 yarn-error.log*
-session.json
-.gitignore
 .env

example/main.py

@@ -15,6 +15,11 @@
 asset_current = "EURUSD_otc"
 
 
+def pin_code_handler() -> str:
+    code = input("Enter the code sent to your email: ")
+    return code
+
+
 class SingletonDecorator:
     """
     A decorator that turns a class into a singleton.
@@ -76,7 +81,13 @@ def run(y):
     return z
 
 
-client = Quotex(email="", password="", headless=False)
+client = Quotex(
+    email="",
+    password="",
+    on_pin_code=pin_code_handler,
+    headless=False,
+)
+
 client.debug_ws_enable = False
 
 

quotexpy/__init__.py

@@ -26,11 +26,11 @@ def truncate(f, n):
 class Quotex(object):
     __version__ = "1.40.7"
 
-    def __init__(self, email: str, password: str, headless=True):
+    def __init__(self, email: str, password: str, **kwargs):
         self.api = None
         self.email = email
         self.password = password
-        self.headless = headless
+        self.kwargs = kwargs
         self.set_ssid = None
         self.duration = None
         self.suspend = 0.5
@@ -127,7 +127,7 @@ async def get_candle_v2(self, asset: str, period: int):
     async def connect(self) -> bool:
         if self.api and self.api.check_websocket_if_connect:
             self.close()
-        self.api = QuotexAPI(self.email, self.password, self.headless)
+        self.api = QuotexAPI(self.email, self.password, **self.kwargs)
         self.api.trace_ws = self.debug_ws_enable
         check = await self.api.connect()
         if check:

quotexpy/api.py

@@ -4,10 +4,11 @@
 import ssl
 import time
 import json
+import pickle
+import typing
 import certifi
 import logging
 import urllib3
-import typing
 import threading
 
 from quotexpy.exceptions import QuotexTimeout
@@ -72,16 +73,14 @@ class QuotexAPI(object):
     websocket_error_reason = None
     balance_id = None
 
-    def __init__(self, email: str, password: str, headless: bool):
+    def __init__(self, email: str, password: str, **kwargs):
         """
-        :param str host: The hostname or ip address of a Quotex server.
         :param str email: The email of a Quotex server.
         :param str password: The password of a Quotex server.
-        :param proxies: The proxies of a Quotex server.
         """
         self.email = email
         self.password = password
-        self.headless = headless
+        self.kwargs = kwargs
         self._temp_status = ""
         self.settings_list = {}
         self.signal_data = nested_dict(2, dict)
@@ -178,11 +177,14 @@ def get_candles(self):
 
     def check_session(self) -> typing.Tuple[str, str]:
         data = {}
-        if os.path.isfile(".session.json"):
-            with open(".session.json") as file:
-                data = json.loads(file.read())
-            self.user_agent = data.get("user_agent")
-        return data.get("ssid"), data.get("cookies")
+        if os.path.isfile(".session.pkl"):
+            with open(".session.pkl", "rb") as file:
+                data = pickle.load(file)
+
+            sessions = data.get(self.email, [])
+            for session in sessions:
+                return session.get("ssid", ""), session.get("cookies", "")
+        return "", ""
 
     def send_websocket_request(self, data, no_force_send=True) -> None:
         """Send websocket request to Quotex server.
@@ -216,7 +218,7 @@ async def get_ssid(self) -> typing.Tuple[str, str]:
         self.logger.info("authenticating user")
         ssid, cookies = self.check_session()
         if not ssid:
-            ssid, cookies = await self.login(self.email, self.password, self.headless)
+            ssid, cookies = await self.login(self.email, self.password, **self.kwargs)
             self.logger.info("login successful")
         return ssid, cookies
 
@@ -265,8 +267,8 @@ def send_ssid(self, max_attemps=10) -> bool:
         """
         self.profile.msg = None
         if not self.SSID:
-            if os.path.exists(os.path.join(".session.json")):
-                os.remove(".session.json")
+            if os.path.exists(os.path.join(".session.pkl")):
+                os.remove(".session.pkl")
             return False
 
         self.ssid(self.SSID)

quotexpy/http/login.py

@@ -10,17 +10,17 @@ class Login(Browser):
     base_url = "qxbroker.com"
     https_base_url = f"https://{base_url}"
 
-    async def __call__(self, email: str, password: str, headless: bool):
+    async def __call__(self, email: str, password: str, **kwargs):
         """
         Method to get Quotex API login http request.
         :param str username: The username of a Quotex server.
         :param str password: The password of a Quotex server.
-        :param bool headless: Use WebDriver headless or headful
         """
 
         self.email = email
         self.password = password
-        self.headless = headless
+        self.headless = kwargs.get("headless", False)
+        self.on_pin_code = kwargs.get("on_pin_code", None)
 
         self.ssid, self.cookies = self.get_cookies_and_ssid()
         return self.ssid, self.cookies

quotexpy/http/qxbroker.py

@@ -1,6 +1,7 @@
 import re
 import json
 import time
+import pickle
 import requests
 from pathlib import Path
 from bs4 import BeautifulSoup
@@ -12,6 +13,7 @@
 class Browser(object):
     email = None
     password = None
+    on_ping_code = None
     headless = None
 
     base_url = "qxbroker.com"
@@ -36,6 +38,17 @@ def get_cookies_and_ssid(self) -> Tuple[Any, str]:
             )
 
             time.sleep(5)
+
+        try:
+            code_input = self.browser.find_element(uc.By.NAME, "code")
+            if code_input.is_displayed():
+                code = self.on_ping_code()
+                code_input.send_keys(code)
+                btn = self.browser.find_element(uc.By.XPATH, "//button[@type='submit']")
+                btn.click()
+        except:
+            pass
+
         cookies = self.browser.get_cookies()
         self.api.cookies = cookies
         soup = BeautifulSoup(self.browser.page_source, "html.parser")
@@ -51,11 +64,20 @@ def get_cookies_and_ssid(self) -> Tuple[Any, str]:
 
         dx: dict = json.loads(match)
         ssid = dx.get("token")
-        output_file = Path(".session.json")
-        output_file.parent.mkdir(exist_ok=True, parents=True)
+
         cookiejar = requests.utils.cookiejar_from_dict({c["name"]: c["value"] for c in cookies})
         cookie_string = "; ".join([f"{c.name}={c.value}" for c in cookiejar])
-        output_file.write_text(json.dumps({"cookies": cookie_string, "ssid": ssid, "user_agent": user_agent}, indent=4))
+        output_file = Path(".session.pkl")
+        output_file.parent.mkdir(exist_ok=True, parents=True)
+
+        data = {}
+        if output_file.is_file():
+            with output_file.open("rb") as file:
+                data = pickle.load(file)
+
+        data[self.email] = [{"cookies": cookie_string, "ssid": ssid, "user_agent": user_agent}]
+        with output_file.open("wb") as file:
+            pickle.dump(data, file)
 
         return ssid, cookie_string
 

quotexpy/ws/client.py

@@ -58,8 +58,8 @@ def on_message(self, wss, wm):
         if current_time.tm_sec in [0, 20, 40]:
             self.wss.send('42["tick"]')
         if "authorization/reject" in wm:
-            if os.path.isfile(".session.json"):
-                os.remove(".session.json")
+            if os.path.isfile(".session.pkl"):
+                os.remove(".session.pkl")
             self.api.SSID = None
             self.api.check_rejected_connection = 1
         elif "s_authorization" in wm: