diff --git a/akipy/akinator.py b/akipy/akinator.py index 5591392..67804ac 100644 --- a/akipy/akinator.py +++ b/akipy/akinator.py @@ -22,6 +22,7 @@ SOFTWARE. """ + # Akinator API wrapper for interacting with the Akinator game. # This module provides a class `Akinator` that allows users to play the Akinator game programmatically. import html @@ -108,13 +109,10 @@ def __initialise(self): ValueError: If the response does not contain expected data. """ url = f"{self.uri}/game" - data = { - "sid": self.theme, - "cm": str(self.child_mode).lower() - } + data = {"sid": self.theme, "cm": str(self.child_mode).lower()} self.client = httpx.Client() try: - req = request_handler(url=url, method='POST', data=data, client=self.client) + req = request_handler(url=url, method="POST", data=data, client=self.client) req.raise_for_status() # Raise an HTTPError for bad responses (4xx and 5xx) text = req.text @@ -124,7 +122,9 @@ def __initialise(self): self.identifiant = re.search(r"#identifiant'\).val\('(.+?)'\)", text) if not self.session or not self.signature or not self.identifiant: - raise ValueError("Response does not contain expected data: session, signature, or identifiant") + raise ValueError( + "Response does not contain expected data: session, signature, or identifiant" + ) self.session = self.session.group(1) self.signature = self.signature.group(1) @@ -146,14 +146,16 @@ def __initialise(self): text, ) if not proposition_match: - raise ValueError("Response does not contain expected data: proposition message") + raise ValueError( + "Response does not contain expected data: proposition message" + ) self.proposition_message = html.unescape(proposition_match.group(1)) # Initialize other attributes self.progression = "0.00000" self.step = "0" - self.akitude = 'defi.png' + self.akitude = "defi.png" except httpx.HTTPError as e: raise httpx.HTTPStatusError(f"Failed to connect to Akinator server: {e}") except ValueError as e: @@ -173,20 +175,20 @@ def __update(self, action: str, resp: dict): NotImplementedError: If the action is not recognized. """ if action == "answer": - self.akitude = resp['akitude'] - self.step = resp['step'] - self.progression = resp['progression'] - self.question = resp['question'] + self.akitude = resp["akitude"] + self.step = resp["step"] + self.progression = resp["progression"] + self.question = resp["question"] elif action == "win": self.win = True - self.id_proposition = resp['id_proposition'] - self.name_proposition = resp['name_proposition'] - self.description_proposition = resp['description_proposition'] + self.id_proposition = resp["id_proposition"] + self.name_proposition = resp["name_proposition"] + self.description_proposition = resp["description_proposition"] # This is necessary to prevent Akinator from immediately proposing a new character after an exclusion self.step_last_proposition = self.step - self.pseudo = resp['pseudo'] - self.flag_photo = resp['flag_photo'] - self.photo = resp['photo'] + self.pseudo = resp["pseudo"] + self.flag_photo = resp["flag_photo"] + self.photo = resp["photo"] else: raise NotImplementedError(f"Unable to handle action: {action}") @@ -216,10 +218,12 @@ def __get_region(self, lang): try: # Make a GET request to the Akinator server - req = request_handler(url=url, method='GET') + req = request_handler(url=url, method="GET") # Check if the request was successful if req.status_code != 200: - raise httpx.HTTPStatusError(f"Failed to connect to Akinator server: {req.status_code}") + raise httpx.HTTPStatusError( + f"Failed to connect to Akinator server: {req.status_code}" + ) else: # Update the instance variables with the response data self.uri = url @@ -253,12 +257,12 @@ def handle_response(self, resp: httpx.Response): if "A technical problem has occurred." in text: raise RuntimeError("A technical problem has occurred.") raise RuntimeError(f"Unexpected response: {text}") - if 'completion' not in data: + if "completion" not in data: # Assume the completion key is missing because a step has been undone or skipped - data['completion'] = self.completion - if data['completion'] == "KO - TIMEOUT": + data["completion"] = self.completion + if data["completion"] == "KO - TIMEOUT": raise TimeoutError("The session has timed out.") - if data['completion'] == "SOUNDLIKE": + if data["completion"] == "SOUNDLIKE": self.finished = True self.win = True if not self.id_proposition: @@ -267,7 +271,7 @@ def handle_response(self, resp: httpx.Response): self.__update(action="win", resp=data) else: self.__update(action="answer", resp=data) - self.completion = data['completion'] + self.completion = data["completion"] def answer(self, option: str | int): if self.win: @@ -277,7 +281,9 @@ def answer(self, option: str | int): return self.choose() if answer == 1: return self.exclude() - raise InvalidChoiceError("Only 'yes' or 'no' can be answered when Akinator has proposed a win") + raise InvalidChoiceError( + "Only 'yes' or 'no' can be answered when Akinator has proposed a win" + ) url = f"{self.uri}/answer" data = { "step": self.step, @@ -291,7 +297,9 @@ def answer(self, option: str | int): } try: - resp = request_handler(url=url, method='POST', data=data, client=self.client) + resp = request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -312,7 +320,9 @@ def back(self): self.win = False try: - resp = request_handler(url=url, method='POST', data=data, client=self.client) + resp = request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -320,7 +330,9 @@ def back(self): def exclude(self): if not self.win: - raise InvalidChoiceError("You can only exclude when Akinator has proposed a win") + raise InvalidChoiceError( + "You can only exclude when Akinator has proposed a win" + ) if self.finished: return self.defeat() url = f"{self.uri}/exclude" @@ -336,7 +348,9 @@ def exclude(self): self.id_proposition = "" try: - resp = request_handler(url=url, method='POST', data=data, client=self.client) + resp = request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -344,7 +358,9 @@ def exclude(self): def choose(self): if not self.win: - raise InvalidChoiceError("You can only choose when Akinator has proposed a win") + raise InvalidChoiceError( + "You can only choose when Akinator has proposed a win" + ) url = f"{self.uri}/choice" data = { "step": self.step, @@ -359,37 +375,53 @@ def choose(self): } try: - resp = request_handler(url=url, method='POST', data=data, client=self.client, follow_redirects=True) + resp = request_handler( + url=url, + method="POST", + data=data, + client=self.client, + follow_redirects=True, + ) if resp.status_code not in range(200, 400): resp.raise_for_status() except Exception as e: raise e self.finished = True self.win = True - self.akitude = 'triomphe.png' + self.akitude = "triomphe.png" self.id_proposition = "" try: text = resp.text # The response for this request is always HTML+JS, so we need to parse it to get the number of times the character has been played, and the win message in the correct language - win_message = html.unescape(re.search(r'(.+?)<\/span>', text).group(1)) - already_played = html.unescape(re.search(r'let tokenDejaJoue = "([\w\s]+)";', text).group(1)) + win_message = html.unescape( + re.search(r'(.+?)<\/span>', text).group(1) + ) + already_played = html.unescape( + re.search(r'let tokenDejaJoue = "([\w\s]+)";', text).group(1) + ) times_selected = re.search(r'let timesSelected = "(\d+)";', text).group(1) - times = html.unescape(re.search(r'<\/span>\s+([\w\s]+)<\/span>', text).group(1)) + times = html.unescape( + re.search( + r'<\/span>\s+([\w\s]+)<\/span>', text + ).group(1) + ) self.question = f"{win_message}\n{already_played} {times_selected} {times}" except Exception: pass - self.progression = '100.00000' + self.progression = "100.00000" return self def defeat(self): # The Akinator website normally displays the defeat screen directly using HTML; we replicate here what the user would see self.finished = True self.win = False - self.akitude = 'deception.png' + self.akitude = "deception.png" self.id_proposition = "" # TODO: Get the correct defeat message in the user's language - self.question = "Bravo, you have defeated me !\nShare your feat with your friends" - self.progression = '100.00000' + self.question = ( + "Bravo, you have defeated me !\nShare your feat with your friends" + ) + self.progression = "100.00000" return self @property diff --git a/akipy/async_akipy.py b/akipy/async_akipy.py index 3b3cde9..a92e81b 100644 --- a/akipy/async_akipy.py +++ b/akipy/async_akipy.py @@ -22,6 +22,7 @@ SOFTWARE. """ + import html import re @@ -45,30 +46,35 @@ class Akinator(SyncAkinator): async def __initialise(self): url = f"{self.uri}/game" - data = { - "sid": self.theme, - "cm": str(self.child_mode).lower() - } + data = {"sid": self.theme, "cm": str(self.child_mode).lower()} self.client = httpx.AsyncClient() try: - req = (await async_request_handler(url=url, method="POST", data=data, client=self.client)).text - + req = ( + await async_request_handler( + url=url, method="POST", data=data, client=self.client + ) + ).text + self.session = re.search(r"#session'\).val\('(.+?)'\)", req).group(1) self.signature = re.search(r"#signature'\).val\('(.+?)'\)", req).group(1) - self.identifiant = re.search(r"#identifiant'\).val\('(.+?)'\)", req).group(1) + self.identifiant = re.search(r"#identifiant'\).val\('(.+?)'\)", req).group( + 1 + ) match = re.search( r'

(.+)

', req, ) self.question = html.unescape(match.group(1)) - self.proposition_message = html.unescape(re.search( - r'

([\w\s]+)

', - req, - ).group(1)) + self.proposition_message = html.unescape( + re.search( + r'

([\w\s]+)

', + req, + ).group(1) + ) self.progression = "0.00000" self.step = "0" - self.akitude = 'defi.png' + self.akitude = "defi.png" except Exception: raise httpx.HTTPStatusError @@ -77,12 +83,12 @@ async def __get_region(self, lang): if len(lang) > 2: lang = LANG_MAP[lang] else: - assert (lang in LANG_MAP.values()) + assert lang in LANG_MAP.values() except Exception: raise InvalidLanguageError(lang) url = f"https://{lang}.akinator.com" try: - req = await async_request_handler(url=url, method='GET') + req = await async_request_handler(url=url, method="GET") if req.status_code != 200: raise httpx.HTTPStatusError else: @@ -116,7 +122,9 @@ async def answer(self, option: str | int): return await self.choose() if answer == 1: return await self.exclude() - raise InvalidChoiceError("Only 'yes' or 'no' can be answered when Akinator has proposed a win") + raise InvalidChoiceError( + "Only 'yes' or 'no' can be answered when Akinator has proposed a win" + ) url = f"{self.uri}/answer" data = { "step": self.step, @@ -130,7 +138,9 @@ async def answer(self, option: str | int): } try: - resp = await async_request_handler(url=url, method='POST', data=data, client=self.client) + resp = await async_request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -151,7 +161,9 @@ async def back(self): self.win = False try: - resp = await async_request_handler(url=url, method='POST', data=data, client=self.client) + resp = await async_request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -159,7 +171,9 @@ async def back(self): async def exclude(self): if not self.win: - raise InvalidChoiceError("You can only exclude when Akinator has proposed a win") + raise InvalidChoiceError( + "You can only exclude when Akinator has proposed a win" + ) if self.finished: return self.defeat() url = f"{self.uri}/exclude" @@ -175,7 +189,9 @@ async def exclude(self): self.id_proposition = "" try: - resp = await async_request_handler(url=url, method='POST', data=data, client=self.client) + resp = await async_request_handler( + url=url, method="POST", data=data, client=self.client + ) self.handle_response(resp) except Exception as e: raise e @@ -183,7 +199,9 @@ async def exclude(self): async def choose(self): if not self.win: - raise InvalidChoiceError("You can only choose when Akinator has proposed a win") + raise InvalidChoiceError( + "You can only choose when Akinator has proposed a win" + ) url = f"{self.uri}/choice" data = { "step": self.step, @@ -198,24 +216,38 @@ async def choose(self): } try: - resp = await async_request_handler(url=url, method='POST', data=data, client=self.client, follow_redirects=True) + resp = await async_request_handler( + url=url, + method="POST", + data=data, + client=self.client, + follow_redirects=True, + ) if resp.status_code not in range(200, 400): resp.raise_for_status() except Exception as e: raise e self.finished = True self.win = True - self.akitude = 'triomphe.png' + self.akitude = "triomphe.png" self.id_proposition = "" try: text = resp.text # The response for this request is always HTML+JS, so we need to parse it to get the number of times the character has been played, and the win message in the correct language - win_message = html.unescape(re.search(r'(.+?)<\/span>', text).group(1)) - already_played = html.unescape(re.search(r'let tokenDejaJoue = "([\w\s]+)";', text).group(1)) + win_message = html.unescape( + re.search(r'(.+?)<\/span>', text).group(1) + ) + already_played = html.unescape( + re.search(r'let tokenDejaJoue = "([\w\s]+)";', text).group(1) + ) times_selected = re.search(r'let timesSelected = "(\d+)";', text).group(1) - times = html.unescape(re.search(r'<\/span>\s+([\w\s]+)<\/span>', text).group(1)) + times = html.unescape( + re.search( + r'<\/span>\s+([\w\s]+)<\/span>', text + ).group(1) + ) self.question = f"{win_message}\n{already_played} {times_selected} {times}" except Exception: pass - self.progression = '100.00000' - return self \ No newline at end of file + self.progression = "100.00000" + return self diff --git a/akipy/dicts.py b/akipy/dicts.py index b33b9f2..e794f1a 100644 --- a/akipy/dicts.py +++ b/akipy/dicts.py @@ -3,7 +3,7 @@ "Accept-Encoding": "gzip, deflate", "Accept-Language": "en-US,en;q=0.9", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) snap Chromium/81.0.4044.92 " - "Chrome/81.0.4044.92 Safari/537.36", + "Chrome/81.0.4044.92 Safari/537.36", "x-requested-with": "XMLHttpRequest", } @@ -23,14 +23,10 @@ "portuguese": "pt", "russian": "ru", "turkish": "tr", - "indonesian": "id" + "indonesian": "id", } -THEME_ID = { - "c": 1, - "a": 14, - "o": 2 -} +THEME_ID = {"c": 1, "a": 14, "o": 2} """ c - characters @@ -58,10 +54,10 @@ } ANSWERS = { - 0: ["yes", "y", '0'], - 1: ["no", "n", '1'], - 2: ["i", "idk", "i dont know", "i don't know", '2'], - 3: ["p", "probably", '3'], - 4: ["pn", "probably not", '4'], + 0: ["yes", "y", "0"], + 1: ["no", "n", "1"], + 2: ["i", "idk", "i dont know", "i don't know", "2"], + 3: ["p", "probably", "3"], + 4: ["pn", "probably not", "4"], } -ANSWER_MAP = {a: key for key, values in ANSWERS.items() for a in values} \ No newline at end of file +ANSWER_MAP = {a: key for key, values in ANSWERS.items() for a in values} diff --git a/akipy/exceptions.py b/akipy/exceptions.py index 8da9243..956207a 100644 --- a/akipy/exceptions.py +++ b/akipy/exceptions.py @@ -1,13 +1,16 @@ class InvalidLanguageError(ValueError): """Raise when the user input language is invalid or not supported by Akinator""" + pass class CantGoBackAnyFurther(Exception): """Raise when the user is in the first question and tries to go back further""" + pass class InvalidChoiceError(ValueError): """Raise when the user input is not a valid answer for the current question""" - pass \ No newline at end of file + + pass diff --git a/akipy/utils.py b/akipy/utils.py index aceca80..2689848 100644 --- a/akipy/utils.py +++ b/akipy/utils.py @@ -4,7 +4,13 @@ from .exceptions import InvalidChoiceError -def request_handler(url: str, method: str, data: dict | None = None, client: httpx.Client | None = None, **kwargs) -> httpx.Response: +def request_handler( + url: str, + method: str, + data: dict | None = None, + client: httpx.Client | None = None, + **kwargs, +) -> httpx.Response: """ Sends an HTTP request to the specified URL using the provided method and data. @@ -32,7 +38,13 @@ def request_handler(url: str, method: str, data: dict | None = None, client: htt raise httpx.HTTPError(f"Request failed: {e}") -async def async_request_handler(url: str, method: str, data: dict | None = None, client: httpx.AsyncClient | None = None, **kwargs) -> httpx.Response: +async def async_request_handler( + url: str, + method: str, + data: dict | None = None, + client: httpx.AsyncClient | None = None, + **kwargs, +) -> httpx.Response: """ Sends an asynchronous HTTP request to the specified URL using the provided method and data. @@ -53,7 +65,9 @@ async def async_request_handler(url: str, method: str, data: dict | None = None, if data: kwargs["data"] = data try: - response = await client.request(method, url, headers=HEADERS, timeout=30, **kwargs) + response = await client.request( + method, url, headers=HEADERS, timeout=30, **kwargs + ) response.raise_for_status() # Raise an HTTPError for bad responses (4xx and 5xx) return response except httpx.HTTPError as e: diff --git a/examples/async_aki.py b/examples/async_aki.py index 98af932..bb3ec24 100644 --- a/examples/async_aki.py +++ b/examples/async_aki.py @@ -22,10 +22,11 @@ async def main(): except akipy.InvalidChoiceError: pass + asyncio.run(main()) print(aki) print(aki.name_proposition) print(aki.description_proposition) print(aki.pseudo) -print(aki.photo) \ No newline at end of file +print(aki.photo) diff --git a/examples/nonasync-aki.py b/examples/nonasync-aki.py index 7c87b25..98614ae 100644 --- a/examples/nonasync-aki.py +++ b/examples/nonasync-aki.py @@ -20,4 +20,4 @@ print(aki.name_proposition) print(aki.description_proposition) print(aki.pseudo) -print(aki.photo) \ No newline at end of file +print(aki.photo)