'Refactored by AI' (#13)

Co-authored-by: Sourcery AI <>

generate_docs.py

@@ -35,8 +35,6 @@
 ]
 
 if __name__ == "__main__":
-    all_modules = {}
-
     render.configure(
         logo_link='https://lara.monster/stlib',
         footer_text='[email protected]',
@@ -47,9 +45,10 @@
     html_path.mkdir()
 
     stlib.__all__ = stlib_modules
-    for module_name in extract.walk_specs(['stlib']):
-        all_modules[module_name] = doc.Module.from_name(module_name)
-
+    all_modules = {
+        module_name: doc.Module.from_name(module_name)
+        for module_name in extract.walk_specs(['stlib'])
+    }
     for module in all_modules.values():
         html = render.html_module(module, all_modules=all_modules)
         file = html_path / f"{module.fullname.replace('.', '/')}.html"

setup.py

@@ -28,19 +28,15 @@
 SDK_PATH = Path(os.getenv("SDK_PATH", Path(SOURCES_PATH, 'sdk')))
 HEADERS_PATH = Path(os.getenv("HEADERS_PATH", Path(SDK_PATH, 'public')))
 
-if sys.maxsize > 2 ** 32:
-    arch = '64'
-else:
-    arch = ''
-
+arch = '64' if sys.maxsize > 2 ** 32 else ''
 if os.name == 'nt':
-    REDIST_PATH = 'win' + arch
-    API_NAME = 'steam_api' + arch
-    EXTRA_NAME = API_NAME + '.dll'
+    REDIST_PATH = f'win{arch}'
+    API_NAME = f'steam_api{arch}'
+    EXTRA_NAME = f'{API_NAME}.dll'
 elif os.name == 'posix':
-    REDIST_PATH = 'linux' + arch if arch else '32'
+    REDIST_PATH = f'linux{arch}' if arch else '32'
     API_NAME = 'steam_api'
-    EXTRA_NAME = 'lib' + API_NAME + '.so'
+    EXTRA_NAME = f'lib{API_NAME}.so'
 else:
     print('Your system is currently not supported.')
     sys.exit(1)
@@ -51,7 +47,7 @@ def run(self):
         bin_path = SDK_PATH / 'redistributable_bin'
         output_dir = Path(self.build_lib) / 'stlib'
         output_dir.mkdir(parents=True, exist_ok=True)
-        compatible = True if platform.machine().lower() in ['x86_64', 'amd64', 'i386', 'x86'] else False
+        compatible = platform.machine().lower() in ['x86_64', 'amd64', 'i386', 'x86']
 
         if compatible and HEADERS_PATH.exists():
             shutil.copy(
@@ -63,11 +59,9 @@ def run(self):
             self.warn("build of steam_api C extension has been disabled")
 
 
-all_sources = []
-for file in SOURCES_PATH.iterdir():
-    if file.suffix == ".cpp":
-        all_sources.append(str(file))
-
+all_sources = [
+    str(file) for file in SOURCES_PATH.iterdir() if file.suffix == ".cpp"
+]
 steamworks = Extension(
     'stlib.steamworks',
     sources=all_sources,

src/stlib/__init__.py

@@ -120,5 +120,4 @@ async def set_default_http_params(session_index: int, *args: Any, **kwargs: Any)
     :param kwargs: extra kwargs when creating a new http session
     :return: `aiohttp.ClientSession`
     """
-    http_session = await _Base.new_http_session(session_index, *args, **kwargs)
-    return http_session
+    return await _Base.new_http_session(session_index, *args, **kwargs)

src/stlib/community.py

@@ -175,7 +175,7 @@ async def _new_mobileconf_query(
         json_data = await self.request_json(f'{self.api_url}/ISteamWebAPIUtil/GetServerInfo/v1')
         server_time = json_data['servertime']
 
-        params = {
+        return {
             'p': deviceid,
             'a': steamid.id64,
             'k': universe.generate_time_hash(server_time, tag, identity_secret),
@@ -184,8 +184,6 @@ async def _new_mobileconf_query(
             'tag': tag,
         }
 
-        return params
-
     async def get_steam_session_id(self) -> str:
         """Get steam session id"""
         response = await self.request(self.community_url)
@@ -221,24 +219,23 @@ async def get_inventory(
         json_data = {}
 
         while True:
-            json_data.update(await self.request_json(
+            json_data |= await self.request_json(
                 f"{self.community_url}/inventory/{steamid.id64}/{appid}/{contextid}",
                 params=params,
                 auto_recovery=False,
-            ))
+            )
 
             if not json_data['success']:
                 raise AttributeError("Unable to get inventory details")
 
             if json_data['total_inventory_count'] == 0:
                 raise InventoryEmptyError(steamid, appid, contextid, "Inventory is empty")
 
-            if 'last_assetid' in json_data:
-                params['start_assetid'] = json_data['last_assetid']
-                await asyncio.sleep(10)
-            else:
+            if 'last_assetid' not in json_data:
                 break
 
+            params['start_assetid'] = json_data['last_assetid']
+            await asyncio.sleep(10)
         items = []
         for item in json_data['descriptions']:
             if 'market_name' in item:
@@ -438,9 +435,9 @@ async def get_confirmations(
                 else:
                     give = [json_data['type']]
 
-                quantity = listing_prices.find(text=lambda element: 'Quantity' in element.text)
-
-                if quantity:
+                if quantity := listing_prices.find(
+                    text=lambda element: 'Quantity' in element.text
+                ):
                     give[0] = f'{quantity.next.next.strip()} {give[0]}'
             elif confirmation['type'] == 5:
                 to = "Steam"
@@ -489,12 +486,11 @@ async def get_card_drops_remaining(self, steamid: universe.SteamId, appid: int)
 
         progress = stats.find('span', class_='progress_info_bold')
 
-        if not progress or "No" in progress.text:
-            cards = 0
-        else:
-            cards = int(progress.text.split(' ', 3)[0])
-
-        return cards
+        return (
+            0
+            if not progress or "No" in progress.text
+            else int(progress.text.split(' ', 3)[0])
+        )
 
     async def get_last_played_game(self, steamid: universe.SteamId) -> Optional[int]:
         """
@@ -521,17 +517,18 @@ async def get_my_orders(self) -> Tuple[List[Order], List[Order]]:
         json_data = {}
 
         while True:
-            json_data.update(await self.request_json(f"{self.community_url}/market/mylistings", params=params))
+            json_data |= await self.request_json(
+                f"{self.community_url}/market/mylistings", params=params
+            )
 
             if not json_data['success']:
                 raise AttributeError("Unable to get order details")
 
-            if json_data['total_count'] > 100:
-                params['start'] += 100
-                await asyncio.sleep(5)
-            else:
+            if json_data['total_count'] <= 100:
                 break
 
+            params['start'] += 100
+            await asyncio.sleep(5)
         my_sell_orders = []
 
         for order in json_data['listings']:
@@ -610,9 +607,9 @@ async def get_item_histogram(self, appid: int, hash_name: str) -> Dict[str, Any]
             'norender': 1,
         }
 
-        json_data = await self.request_json(f"{self.community_url}/market/itemordershistogram", params=params)
-
-        return json_data
+        return await self.request_json(
+            f"{self.community_url}/market/itemordershistogram", params=params
+        )
 
     async def send_trade_offer(
             self,
@@ -631,26 +628,24 @@ async def send_trade_offer(
         :param them: List of them items to trade
         :return: Json data
         """
-        me_assets = []
-
-        for item in me:
-            me_assets.append({
+        me_assets = [
+            {
                 'appid': item[0],
                 'contextid': str(contextid),
                 'amount': item[2],
                 'assetid': str(item[1]),
-            })
-
-        them_assets = []
-
-        for item in them:
-            them_assets.append({
+            }
+            for item in me
+        ]
+        them_assets = [
+            {
                 'appid': item[0],
                 'contextid': str(contextid),
                 'amount': item[2],
                 'assetid': str(item[1]),
-            })
-
+            }
+            for item in them
+        ]
         offer = {
             'newversion': True,
             'version': len(me_assets) + len(them_assets) + 1,
@@ -677,8 +672,9 @@ async def send_trade_offer(
         }
 
         headers = {'referer': f'{self.community_url}/tradeoffer/new/?partner={steamid.id3}&token={token}'}
-        json_data = await self.request_json(f"{self.community_url}/tradeoffer/new/send", data=data, headers=headers)
-        return json_data
+        return await self.request_json(
+            f"{self.community_url}/tradeoffer/new/send", data=data, headers=headers
+        )
 
     async def send_confirmation(
             self,
@@ -701,8 +697,9 @@ async def send_confirmation(
         """
         extra_params = {'cid': trade_id, 'ck': trade_key, 'op': action}
         params = await self._new_mobileconf_query(deviceid, steamid, identity_secret, 'conf')
-        json_data = await self.request_json(f'{self.mobileconf_url}/ajaxop', params={**params, **extra_params})
-        return json_data
+        return await self.request_json(
+            f'{self.mobileconf_url}/ajaxop', params={**params, **extra_params}
+        )
 
     async def revoke_api_key(self) -> None:
         """

src/stlib/login.py

@@ -156,7 +156,7 @@ async def _new_login_data(self, mobile_login: bool = False) -> Dict[str, Any]:
 
         encrypted_password = universe.encrypt_password(steam_key, self.__password)
 
-        data = {
+        return {
             "account_name": self.username,
             "encrypted_password": encrypted_password.decode(),
             "encryption_timestamp": steam_key.timestamp,
@@ -166,8 +166,6 @@ async def _new_login_data(self, mobile_login: bool = False) -> Dict[str, Any]:
             "website_id": "Mobile" if mobile_login else "Community",
         }
 
-        return data
-
     async def get_steam_key(self, username: str) -> universe.SteamKey:
         """
         Get `SteamKey`
@@ -180,13 +178,12 @@ async def get_steam_key(self, username: str) -> universe.SteamKey:
             params=params,
         )
 
-        if json_data['response']:
-            public_mod = int(json_data['response']['publickey_mod'], 16)
-            public_exp = int(json_data['response']['publickey_exp'], 16)
-            timestamp = int(json_data['response']['timestamp'])
-        else:
+        if not json_data['response']:
             raise ValueError('Failed to get public key.')
 
+        public_mod = int(json_data['response']['publickey_mod'], 16)
+        public_exp = int(json_data['response']['publickey_exp'], 16)
+        timestamp = int(json_data['response']['timestamp'])
         return universe.SteamKey(rsa.PublicKey(public_mod, public_exp), timestamp)
 
     async def get_captcha(self, gid: int) -> bytes:
@@ -282,11 +279,7 @@ async def do_login(
             if not auth_data['response']:
                 raise LoginError('SteamGuard code is wrong')
         else:
-            captcha_requested = False
-
-            if len(json_data['response']['allowed_confirmations']) > 1:
-                captcha_requested = True
-
+            captcha_requested = len(json_data['response']['allowed_confirmations']) > 1
             auth_code_type = json_data['response']['allowed_confirmations'][0]['confirmation_type']
 
             if auth_code_type == 2:
@@ -345,7 +338,9 @@ async def is_logged_in(self) -> bool:
         :return: bool
         """
         try:
-            response = await self.request(f'https://store.steampowered.com/account', allow_redirects=False)
+            response = await self.request(
+                'https://store.steampowered.com/account', allow_redirects=False
+            )
         except LoginError:
             return False
 

src/stlib/universe.py

@@ -156,7 +156,7 @@ def generate_steam_code(server_time: int, shared_secret: Union[str, bytes]) -> s
     :param shared_secret: User shared secret
     :return: Steam OTP
     """
-    msg = int(server_time / 30).to_bytes(8, 'big')
+    msg = (server_time // 30).to_bytes(8, 'big')
     key = base64.b64decode(shared_secret)
     auth_code_raw = generate_otp_code(msg, key)
 
@@ -178,9 +178,7 @@ def generate_device_id(base: str) -> str:
     device_id = ['android:']
 
     for start, end in ([0, 8], [8, 12], [12, 16], [16, 20], [20, 32]):
-        device_id.append(digest[start:end])
-        device_id.append('-')
-
+        device_id.extend((digest[start:end], '-'))
     device_id.pop(-1)
     return ''.join(device_id)
 

src/stlib/utils.py

@@ -168,9 +168,8 @@ async def destroy_session(cls, session_index: int, no_fail: bool = False) -> Non
             assert isinstance(http_session, aiohttp.ClientSession), "Wrong http session type"
             await http_session.close()
             del _session_cache['http_session'][session_index]
-        else:
-            if not no_fail:
-                raise IndexError(f"There's no session at {session_index}")
+        elif not no_fail:
+            raise IndexError(f"There's no session at {session_index}")
 
     @classmethod
     def get_session(cls, session_index: int) -> 'Base':

src/stlib/webapi.py

@@ -118,14 +118,12 @@ async def _new_mobile_data(
     ) -> Dict[str, Any]:
         current_time = int(time.time())
 
-        data = {
+        return {
             'steamid': steamid.id64,
             'authenticator_time': current_time,
             'authenticator_type': universe.TOKEN_TYPE[token_type],
         }
 
-        return data
-
     async def get_server_time(self) -> int:
         """Get server time"""
         json_data = await self.request_json(f'{self.api_url}/ISteamWebAPIUtil/GetServerInfo/v1')
@@ -256,7 +254,7 @@ async def new_authenticator(
         if response['status'] == 29:
             raise AuthenticatorExists('An Authenticator is already active for that account.')
 
-        if response['status'] == 84 or response['status'] == 2:
+        if response['status'] in [84, 2]:
             raise PhoneNotRegistered('Phone not registered on Steam Account.')
 
         if response['status'] != 1: