diff --git a/examples/desktop_authenticator.py b/examples/desktop_authenticator.py index a3a4205..798cb12 100644 --- a/examples/desktop_authenticator.py +++ b/examples/desktop_authenticator.py @@ -4,7 +4,6 @@ from steampy.guard import generate_confirmation_key, generate_one_time_code - shared_secret = '' identity_secret = '' diff --git a/examples/inventory.py b/examples/inventory.py index c7ab95f..b9ede25 100644 --- a/examples/inventory.py +++ b/examples/inventory.py @@ -1,9 +1,10 @@ import json +import sys +from pathlib import Path -from steampy.client import SteamClient, InvalidCredentials +from steampy.client import InvalidCredentials, SteamClient from steampy.models import GameOptions - # Your Steam username username = '' @@ -30,7 +31,7 @@ steam_client.login(username, password, steam_guard_path) except (ValueError, InvalidCredentials): print('Your login credentials are invalid!') - exit(1) + sys.exit(1) else: print('Finished! Logged in into Steam') @@ -56,6 +57,6 @@ # Dump all the info to inventory_(app_id)_(context_id).json file print('Saving information...') -with open(f'inventory_{app_id}_{context_id}.json', 'w') as file: +with Path(f'inventory_{app_id}_{context_id}.json').open('w') as file: json.dump(item_amounts, file) print(f'Done! Saved to file: inventory_{app_id}_{context_id}.json') diff --git a/examples/storehouse.py b/examples/storehouse.py index cb9ba89..afdd13d 100644 --- a/examples/storehouse.py +++ b/examples/storehouse.py @@ -2,7 +2,6 @@ from steampy.client import SteamClient, TradeOfferState - # Set API key api_key = '' # Set path to SteamGuard file @@ -13,7 +12,7 @@ password = '' -def main(): +def main() -> None: print('This is the donation bot accepting items for free.') if not are_credentials_filled(): diff --git a/setup.py b/setup.py index 11692ad..83c0b7c 100644 --- a/setup.py +++ b/setup.py @@ -1,14 +1,11 @@ -from setuptools import setup -import sys -if not sys.version_info[0] == 3 and sys.version_info[1] < 8: - sys.exit('Python < 3.8 is not supported') +from setuptools import setup version = '1.1.2' setup( name='steampy', - packages=['steampy', 'test', 'examples', ], + packages=['steampy', 'test', 'examples' ], version=version, description='A Steam lib for trade automation', author='Michał Bukowski', @@ -16,11 +13,11 @@ license='MIT', url='https://github.com/bukson/steampy', download_url='https://github.com/bukson/steampy/tarball/' + version, - keywords=['steam', 'trade', ], + keywords=['steam', 'trade' ], classifiers=[], install_requires=[ "requests", "beautifulsoup4", - "rsa" + "rsa", ], ) diff --git a/steampy/client.py b/steampy/client.py index ece1057..9702bfa 100644 --- a/steampy/client.py +++ b/steampy/client.py @@ -1,30 +1,30 @@ -import re -import bs4 +from __future__ import annotations + import json +import re import urllib.parse as urlparse -from typing import List, Union from decimal import Decimal import requests from steampy import guard from steampy.confirmation import ConfirmationExecutor -from steampy.exceptions import SevenDaysHoldException, ApiException -from steampy.login import LoginExecutor, InvalidCredentials +from steampy.exceptions import ApiException, SevenDaysHoldException +from steampy.login import InvalidCredentials, LoginExecutor from steampy.market import SteamMarket -from steampy.models import Asset, TradeOfferState, SteamUrl, GameOptions +from steampy.models import Asset, GameOptions, SteamUrl, TradeOfferState from steampy.utils import ( - text_between, - texts_between, - merge_items_with_descriptions_from_inventory, - steam_id_to_account_id, - merge_items_with_descriptions_from_offers, - get_description_key, - merge_items_with_descriptions_from_offer, account_id_to_steam_id, + get_description_key, get_key_value_from_url, - ping_proxy, login_required, + merge_items_with_descriptions_from_inventory, + merge_items_with_descriptions_from_offer, + merge_items_with_descriptions_from_offers, + ping_proxy, + steam_id_to_account_id, + text_between, + texts_between, ) @@ -32,11 +32,11 @@ class SteamClient: def __init__( self, api_key: str, - username: str = None, - password: str = None, - steam_guard: str = None, - login_cookies: dict = None, - proxies: dict = None, + username: str | None = None, + password: str | None = None, + steam_guard: str | None = None, + login_cookies: dict | None = None, + proxies: dict | None = None, ) -> None: self._api_key = api_key self._session = requests.Session() @@ -62,7 +62,7 @@ def set_proxies(self, proxies: dict) -> dict: if not isinstance(proxies, dict): raise TypeError( 'Proxy must be a dict. Example: ' - '\{"http": "http://login:password@host:port"\, "https": "http://login:password@host:port"\}' + r'\{"http": "http://login:password@host:port"\, "https": "http://login:password@host:port"\}', ) if ping_proxy(proxies): @@ -88,13 +88,13 @@ def get_steam_id(self) -> int: else: raise ValueError(f'Invalid steam_id: {steam_id}') - def login(self, username: str = None, password: str = None, steam_guard: str = None) -> None: + def login(self, username: str | None = None, password: str | None = None, steam_guard: str | None = None) -> None: invalid_client_credentials_is_present = None in (self.username, self._password, self.steam_guard_string) invalid_login_credentials_is_present = None in (username, password, steam_guard) if invalid_client_credentials_is_present and invalid_login_credentials_is_present: raise InvalidCredentials( - 'You have to pass username, password and steam_guard parameters when using "login" method' + 'You have to pass username, password and steam_guard parameters when using "login" method', ) if invalid_client_credentials_is_present: @@ -136,9 +136,9 @@ def is_session_alive(self) -> bool: return steam_login.lower() in main_page_response.text.lower() def api_call( - self, method: str, interface: str, api_method: str, version: str, params: dict = None + self, method: str, interface: str, api_method: str, version: str, params: dict | None = None, ) -> requests.Response: - url = '/'.join((SteamUrl.API_URL, interface, api_method, version)) + url = f'{SteamUrl.API_URL}/{interface}/{api_method}/{version}' response = self._session.get(url, params=params) if method == 'GET' else self._session.post(url, data=params) if self.is_invalid_api_key(response): @@ -158,9 +158,9 @@ def get_my_inventory(self, game: GameOptions, merge: bool = True, count: int = 5 @login_required def get_partner_inventory( - self, partner_steam_id: str, game: GameOptions, merge: bool = True, count: int = 5000 + self, partner_steam_id: str, game: GameOptions, merge: bool = True, count: int = 5000, ) -> dict: - url = '/'.join((SteamUrl.COMMUNITY_URL, 'inventory', partner_steam_id, game.app_id, game.context_id)) + url = f'{SteamUrl.COMMUNITY_URL}/inventory/{partner_steam_id}/{game.app_id}/{game.context_id}' params = {'l': 'english', 'count': count} response_dict = self._session.get(url, params=params).json() @@ -198,10 +198,10 @@ def _filter_non_active_offers(offers_response): offers_sent = offers_response['response'].get('trade_offers_sent', []) offers_response['response']['trade_offers_received'] = list( - filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_received) + filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_received), ) offers_response['response']['trade_offers_sent'] = list( - filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_sent) + filter(lambda offer: offer['trade_offer_state'] == TradeOfferState.Active, offers_sent), ) return offers_response @@ -237,14 +237,12 @@ def get_trade_history( 'include_failed': include_failed, 'include_total': include_total, } - response = self.api_call('GET', 'IEconService', 'GetTradeHistory', 'v1', params).json() - return response + return self.api_call('GET', 'IEconService', 'GetTradeHistory', 'v1', params).json() @login_required def get_trade_receipt(self, trade_id: str): html = self._session.get(f'https://steamcommunity.com/trade/{trade_id}/receipt').content.decode() - items = [json.loads(item) for item in texts_between(html, 'oItem = ', ';\r\n\toItem')] - return items + return [json.loads(item) for item in texts_between(html, 'oItem = ', ';\r\n\toItem')] @login_required def accept_trade_offer(self, trade_offer_id: str) -> dict: @@ -282,23 +280,21 @@ def _fetch_trade_partner_id(self, trade_offer_id: str) -> str: def _confirm_transaction(self, trade_offer_id: str) -> dict: confirmation_executor = ConfirmationExecutor( - self.steam_guard['identity_secret'], self.steam_guard['steamid'], self._session + self.steam_guard['identity_secret'], self.steam_guard['steamid'], self._session, ) return confirmation_executor.send_trade_allow_request(trade_offer_id) def decline_trade_offer(self, trade_offer_id: str) -> dict: url = f'https://steamcommunity.com/tradeoffer/{trade_offer_id}/decline' - response = self._session.post(url, data={'sessionid': self._get_session_id()}).json() - return response + return self._session.post(url, data={'sessionid': self._get_session_id()}).json() def cancel_trade_offer(self, trade_offer_id: str) -> dict: url = f'https://steamcommunity.com/tradeoffer/{trade_offer_id}/cancel' - response = self._session.post(url, data={'sessionid': self._get_session_id()}).json() - return response + return self._session.post(url, data={'sessionid': self._get_session_id()}).json() @login_required def make_offer( - self, items_from_me: List[Asset], items_from_them: List[Asset], partner_steam_id: str, message: str = '' + self, items_from_me: list[Asset], items_from_them: list[Asset], partner_steam_id: str, message: str = '', ) -> dict: offer = self._create_offer_dict(items_from_me, items_from_them) session_id = self._get_session_id() @@ -338,7 +334,7 @@ def get_friend_list(self, steam_id: str, relationship_filter: str = 'all') -> di return data['friendslist']['friends'] @staticmethod - def _create_offer_dict(items_from_me: List[Asset], items_from_them: List[Asset]) -> dict: + def _create_offer_dict(items_from_me: list[Asset], items_from_them: list[Asset]) -> dict: return { 'newversion': True, 'version': 4, @@ -362,8 +358,8 @@ def get_escrow_duration(self, trade_offer_url: str) -> int: @login_required def make_offer_with_url( self, - items_from_me: List[Asset], - items_from_them: List[Asset], + items_from_me: list[Asset], + items_from_them: list[Asset], trade_offer_url: str, message: str = '', case_sensitive: bool = True, @@ -403,7 +399,7 @@ def _get_trade_offer_url(trade_offer_id: str) -> str: @login_required # If convert_to_decimal = False, the price will be returned WITHOUT a decimal point. - def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = False) -> Union[str, Decimal]: + def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = False) -> str | Decimal: response = self._session.get(f'{SteamUrl.COMMUNITY_URL}/market') wallet_info_match = re.search(r'var g_rgWalletInfo = (.*?);', response.text) if wallet_info_match: @@ -414,5 +410,4 @@ def get_wallet_balance(self, convert_to_decimal: bool = True, on_hold: bool = Fa balance_dict_key = 'wallet_delayed_balance' if on_hold else 'wallet_balance' if convert_to_decimal: return Decimal(balance_dict[balance_dict_key]) / 100 - else: - return balance_dict[balance_dict_key] + return balance_dict[balance_dict_key] diff --git a/steampy/confirmation.py b/steampy/confirmation.py index a9836cb..be03a84 100644 --- a/steampy/confirmation.py +++ b/steampy/confirmation.py @@ -1,19 +1,23 @@ +from __future__ import annotations + import enum import json import time -from typing import List from http import HTTPStatus +from typing import TYPE_CHECKING -import requests from bs4 import BeautifulSoup from steampy import guard from steampy.exceptions import ConfirmationExpected from steampy.login import InvalidCredentials +if TYPE_CHECKING: + import requests + class Confirmation: - def __init__(self, data_confid, nonce): + def __init__(self, data_confid, nonce) -> None: self.data_confid = data_confid self.nonce = nonce @@ -52,7 +56,7 @@ def _send_confirmation(self, confirmation: Confirmation) -> dict: headers = {'X-Requested-With': 'XMLHttpRequest'} return self._session.get(f'{self.CONF_URL}/ajaxop', params=params, headers=headers).json() - def _get_confirmations(self) -> List[Confirmation]: + def _get_confirmations(self) -> list[Confirmation]: confirmations = [] confirmations_page = self._fetch_confirmations_page() if confirmations_page.status_code == HTTPStatus.OK: @@ -93,7 +97,7 @@ def _create_confirmation_params(self, tag_string: str) -> dict: 'tag': tag_string, } - def _select_trade_offer_confirmation(self, confirmations: List[Confirmation], trade_offer_id: str) -> Confirmation: + def _select_trade_offer_confirmation(self, confirmations: list[Confirmation], trade_offer_id: str) -> Confirmation: for confirmation in confirmations: confirmation_details_page = self._fetch_confirmation_details_page(confirmation) confirmation_id = self._get_confirmation_trade_offer_id(confirmation_details_page) @@ -101,7 +105,7 @@ def _select_trade_offer_confirmation(self, confirmations: List[Confirmation], tr return confirmation raise ConfirmationExpected - def _select_sell_listing_confirmation(self, confirmations: List[Confirmation], asset_id: str) -> Confirmation: + def _select_sell_listing_confirmation(self, confirmations: list[Confirmation], asset_id: str) -> Confirmation: for confirmation in confirmations: confirmation_details_page = self._fetch_confirmation_details_page(confirmation) confirmation_id = self._get_confirmation_sell_listing_id(confirmation_details_page) diff --git a/steampy/guard.py b/steampy/guard.py index ac35cde..5a80903 100644 --- a/steampy/guard.py +++ b/steampy/guard.py @@ -1,14 +1,15 @@ -import os +from __future__ import annotations + import hmac import json import struct -from time import time -from typing import Dict +from base64 import b64decode, b64encode from hashlib import sha1 -from base64 import b64encode, b64decode +from pathlib import Path +from time import time -def load_steam_guard(steam_guard: str) -> Dict[str, str]: +def load_steam_guard(steam_guard: str) -> dict[str, str]: """Load Steam Guard credentials from json (file or string). Arguments: @@ -17,14 +18,14 @@ def load_steam_guard(steam_guard: str) -> Dict[str, str]: Returns: Dict[str, str]: Parsed json data as a dictionary of strings (both key and value). """ - if os.path.isfile(steam_guard): - with open(steam_guard, 'r') as f: + if Path(steam_guard).is_file(): + with Path(steam_guard).open() as f: return json.loads(f.read(), parse_int=str) else: return json.loads(steam_guard, parse_int=str) -def generate_one_time_code(shared_secret: str, timestamp: int = None) -> str: +def generate_one_time_code(shared_secret: str, timestamp: int | None = None) -> str: if timestamp is None: timestamp = int(time()) time_buffer = struct.pack('>Q', timestamp // 30) # pack as Big endian, uint64 diff --git a/steampy/login.py b/steampy/login.py index ec94b8b..34a0beb 100644 --- a/steampy/login.py +++ b/steampy/login.py @@ -1,13 +1,18 @@ -from http import HTTPStatus +from __future__ import annotations + from base64 import b64encode +from http import HTTPStatus +from typing import TYPE_CHECKING -from rsa import encrypt, PublicKey -from requests import Session, Response +from rsa import PublicKey, encrypt from steampy import guard +from steampy.exceptions import ApiException, CaptchaRequired, InvalidCredentials from steampy.models import SteamUrl from steampy.utils import create_cookie -from steampy.exceptions import InvalidCredentials, CaptchaRequired, ApiException + +if TYPE_CHECKING: + from requests import Response, Session class LoginExecutor: @@ -19,8 +24,8 @@ def __init__(self, username: str, password: str, shared_secret: str, session: Se self.session = session self.refresh_token = '' - def _api_call(self, method: str, service: str, endpoint: str, version: str = 'v1', params: dict = None) -> Response: - url = '/'.join((SteamUrl.API_URL, service, endpoint, version)) + def _api_call(self, method: str, service: str, endpoint: str, version: str = 'v1', params: dict | None = None) -> Response: + url = f'{SteamUrl.API_URL}/{service}/{endpoint}/{version}' # All requests from the login page use the same 'Referer' and 'Origin' values headers = {'Referer': f'{SteamUrl.COMMUNITY_URL}/', 'Origin': SteamUrl.COMMUNITY_URL} if method.upper() == 'GET': @@ -121,7 +126,7 @@ def _update_steam_guard(self, login_response: Response) -> None: update_data = {'client_id': client_id, 'steamid': steamid, 'code_type': code_type, 'code': code} response = self._api_call( - 'POST', 'IAuthenticationService', 'UpdateAuthSessionWithSteamGuardCode', params=update_data + 'POST', 'IAuthenticationService', 'UpdateAuthSessionWithSteamGuardCode', params=update_data, ) if response.status_code == HTTPStatus.OK: self._pool_sessions_steam(client_id, request_id) @@ -137,5 +142,4 @@ def _finalize_login(self) -> Response: sessionid = self.session.cookies['sessionid'] redir = f'{SteamUrl.COMMUNITY_URL}/login/home/?goto=' finalized_data = {'nonce': self.refresh_token, 'sessionid': sessionid, 'redir': redir} - response = self.session.post(SteamUrl.LOGIN_URL + '/jwt/finalizelogin', data=finalized_data) - return response + return self.session.post(SteamUrl.LOGIN_URL + '/jwt/finalizelogin', data=finalized_data) diff --git a/steampy/market.py b/steampy/market.py index a44178c..21e10d4 100644 --- a/steampy/market.py +++ b/steampy/market.py @@ -7,14 +7,14 @@ from steampy.confirmation import ConfirmationExecutor from steampy.exceptions import ApiException, TooManyRequests -from steampy.models import Currency, SteamUrl, GameOptions +from steampy.models import Currency, GameOptions, SteamUrl from steampy.utils import ( - text_between, get_listing_id_to_assets_address_from_html, get_market_listings_from_html, - merge_items_with_descriptions_from_listing, get_market_sell_listings_from_api, login_required, + merge_items_with_descriptions_from_listing, + text_between, ) @@ -31,7 +31,7 @@ def _set_login_executed(self, steamguard: dict, session_id: str) -> None: self.was_login_executed = True def fetch_price( - self, item_hash_name: str, game: GameOptions, currency: Currency = Currency.USD, country='PL' + self, item_hash_name: str, game: GameOptions, currency: Currency = Currency.USD, country='PL', ) -> dict: url = f'{SteamUrl.COMMUNITY_URL}/market/priceoverview/' params = { @@ -68,15 +68,15 @@ def get_my_market_listings(self) -> dict: listing_id_to_assets_address = get_listing_id_to_assets_address_from_html(response.text) listings = get_market_listings_from_html(response.text) listings = merge_items_with_descriptions_from_listing( - listings, listing_id_to_assets_address, assets_descriptions + listings, listing_id_to_assets_address, assets_descriptions, ) if '' in response.text: n_showing = int(text_between(response.text, '', '')) n_total = int( text_between(response.text, '', '').replace( - ',', '' - ) + ',', '', + ), ) if n_showing < n_total < 1000: @@ -89,7 +89,7 @@ def get_my_market_listings(self) -> dict: listing_id_to_assets_address = get_listing_id_to_assets_address_from_html(jresp.get('hovers')) listings_2 = get_market_sell_listings_from_api(jresp.get('results_html')) listings_2 = merge_items_with_descriptions_from_listing( - listings_2, listing_id_to_assets_address, jresp.get('assets') + listings_2, listing_id_to_assets_address, jresp.get('assets'), ) listings['sell_listings'] = {**listings['sell_listings'], **listings_2['sell_listings']} else: @@ -98,13 +98,13 @@ def get_my_market_listings(self) -> dict: response = self._session.get(url) if response.status_code != HTTPStatus.OK: raise ApiException( - f'There was a problem getting the listings. HTTP code: {response.status_code}' + f'There was a problem getting the listings. HTTP code: {response.status_code}', ) jresp = response.json() listing_id_to_assets_address = get_listing_id_to_assets_address_from_html(jresp.get('hovers')) listings_2 = get_market_sell_listings_from_api(jresp.get('results_html')) listings_2 = merge_items_with_descriptions_from_listing( - listings_2, listing_id_to_assets_address, jresp.get('assets') + listings_2, listing_id_to_assets_address, jresp.get('assets'), ) listings['sell_listings'] = {**listings['sell_listings'], **listings_2['sell_listings']} @@ -147,14 +147,14 @@ def create_buy_order( 'quantity': quantity, } headers = { - 'Referer': f'{SteamUrl.COMMUNITY_URL}/market/listings/{game.app_id}/{urllib.parse.quote(market_name)}' + 'Referer': f'{SteamUrl.COMMUNITY_URL}/market/listings/{game.app_id}/{urllib.parse.quote(market_name)}', } response = self._session.post(f'{SteamUrl.COMMUNITY_URL}/market/createbuyorder/', data, headers=headers).json() if (success := response.get('success')) != 1: raise ApiException( - f'There was a problem creating the order. Are you using the right currency? success: {success}' + f'There was a problem creating the order. Are you using the right currency? success: {success}', ) return response @@ -178,16 +178,16 @@ def buy_item( 'quantity': '1', } headers = { - 'Referer': f'{SteamUrl.COMMUNITY_URL}/market/listings/{game.app_id}/{urllib.parse.quote(market_name)}' + 'Referer': f'{SteamUrl.COMMUNITY_URL}/market/listings/{game.app_id}/{urllib.parse.quote(market_name)}', } response = self._session.post( - f'{SteamUrl.COMMUNITY_URL}/market/buylisting/{market_id}', data, headers=headers + f'{SteamUrl.COMMUNITY_URL}/market/buylisting/{market_id}', data, headers=headers, ).json() try: if (success := response['wallet_info']['success']) != 1: raise ApiException( - f'There was a problem buying this item. Are you using the right currency? success: {success}' + f'There was a problem buying this item. Are you using the right currency? success: {success}', ) except Exception: raise ApiException(f'There was a problem buying this item. Message: {response.get("message")}') @@ -217,6 +217,6 @@ def cancel_buy_order(self, buy_order_id) -> dict: def _confirm_sell_listing(self, asset_id: str) -> dict: con_executor = ConfirmationExecutor( - self._steam_guard['identity_secret'], self._steam_guard['steamid'], self._session + self._steam_guard['identity_secret'], self._steam_guard['steamid'], self._session, ) return con_executor.confirm_sell_listing(asset_id) diff --git a/steampy/models.py b/steampy/models.py index e05add7..ca1fe40 100644 --- a/steampy/models.py +++ b/steampy/models.py @@ -1,10 +1,12 @@ from enum import IntEnum -from collections import namedtuple +from typing import NamedTuple -class GameOptions: - PredefinedOptions = namedtuple('PredefinedOptions', ['app_id', 'context_id']) +class PredefinedOptions(NamedTuple): + app_id: str + context_id: str +class GameOptions: STEAM = PredefinedOptions('753', '6') DOTA2 = PredefinedOptions('570', '2') CS = PredefinedOptions('730', '2') diff --git a/steampy/utils.py b/steampy/utils.py index fb4f3bc..f828cf7 100644 --- a/steampy/utils.py +++ b/steampy/utils.py @@ -1,18 +1,22 @@ -import os -import re +from __future__ import annotations + import copy import math +import re import struct -from typing import List from decimal import Decimal -from urllib.parse import urlparse, parse_qs +from pathlib import Path +from typing import TYPE_CHECKING +from urllib.parse import parse_qs, urlparse import requests from bs4 import BeautifulSoup, Tag from requests.structures import CaseInsensitiveDict -from steampy.models import GameOptions -from steampy.exceptions import ProxyConnectionError, LoginRequired +from steampy.exceptions import LoginRequired, ProxyConnectionError + +if TYPE_CHECKING: + from steampy.models import GameOptions def login_required(func): @@ -120,12 +124,8 @@ def merge_items_with_descriptions_from_offers(offers_response: dict) -> dict: descriptions = {get_description_key(offer): offer for offer in offers_response['response'].get('descriptions', [])} received_offers = offers_response['response'].get('trade_offers_received', []) sent_offers = offers_response['response'].get('trade_offers_sent', []) - offers_response['response']['trade_offers_received'] = list( - map(lambda offer: merge_items_with_descriptions_from_offer(offer, descriptions), received_offers) - ) - offers_response['response']['trade_offers_sent'] = list( - map(lambda offer: merge_items_with_descriptions_from_offer(offer, descriptions), sent_offers) - ) + offers_response['response']['trade_offers_received'] = [merge_items_with_descriptions_from_offer(offer, descriptions) for offer in received_offers] + offers_response['response']['trade_offers_sent'] = [merge_items_with_descriptions_from_offer(offer, descriptions) for offer in sent_offers] return offers_response @@ -145,7 +145,7 @@ def merge_items_with_descriptions_from_listing(listings: dict, ids_to_assets_add return listings -def merge_items(items: List[dict], descriptions: dict, **kwargs) -> dict: +def merge_items(items: list[dict], descriptions: dict, **kwargs) -> dict: merged_items = {} for item in items: @@ -181,7 +181,7 @@ def get_market_listings_from_html(html: str) -> dict: def get_sell_listings_from_node(node: Tag) -> dict: - sell_listings_raw = node.findAll('div', {'id': re.compile('mylisting_\d+')}) + sell_listings_raw = node.findAll('div', {'id': re.compile(r'mylisting_\d+')}) sell_listings_dict = {} for listing_raw in sell_listings_raw: @@ -225,7 +225,7 @@ def get_buy_orders_from_node(node: Tag) -> dict: def get_listing_id_to_assets_address_from_html(html: str) -> dict: listing_id_to_assets_address = {} - regex = "CreateItemHoverFromContainer\( [\w]+, 'mylisting_([\d]+)_[\w]+', ([\d]+), '([\d]+)', '([\d]+)', [\d]+ \);" + regex = r"CreateItemHoverFromContainer\( [\w]+, 'mylisting_([\d]+)_[\w]+', ([\d]+), '([\d]+)', '([\d]+)', [\d]+ \);" for match in re.findall(regex, html): listing_id_to_assets_address[match[0]] = [str(match[1]), match[2], match[3]] @@ -243,19 +243,19 @@ def get_key_value_from_url(url: str, key: str, case_sensitive: bool = True) -> s def load_credentials(): - dirname = os.path.dirname(os.path.abspath(__file__)) - with open(f'{dirname}/../secrets/credentials.pwd', 'r') as f: + dirname = Path(__file__).resolve().parent + with Path(f'{dirname}/../secrets/credentials.pwd').open() as f: return [Credentials(line.split()[0], line.split()[1], line.split()[2]) for line in f] class Credentials: - def __init__(self, login: str, password: str, api_key: str): + def __init__(self, login: str, password: str, api_key: str) -> None: self.login = login self.password = password self.api_key = api_key -def ping_proxy(proxies: dict): +def ping_proxy(proxies: dict) -> bool: try: requests.get('https://steamcommunity.com/', proxies=proxies) return True diff --git a/test/test_client.py b/test/test_client.py index 2a7bccc..8717e76 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -1,25 +1,26 @@ -import os +import unittest from decimal import Decimal +from pathlib import Path from unittest import TestCase -import unittest from steampy.client import SteamClient from steampy.exceptions import LoginRequired -from steampy.models import GameOptions, Asset +from steampy.models import Asset, GameOptions from steampy.utils import account_id_to_steam_id, load_credentials + @unittest.skip('Requires secrets/Steamguard.txt') class TestSteamClient(TestCase): @classmethod def setUpClass(cls): cls.credentials = load_credentials()[0] - dirname = os.path.dirname(os.path.abspath(__file__)) + dirname = Path(__file__).resolve().parent cls.steam_guard_file = f'{dirname}/../secrets/Steamguard.txt' def test_get_steam_id(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) - self.assertEqual(client.get_steam_id(), int(self.steam_guard_file['Session']['SteamID'])) + assert client.get_steam_id() == int(self.steam_guard_file['Session']['SteamID']) def test_login(self): client = SteamClient(self.credentials.api_key) @@ -28,26 +29,26 @@ def test_login(self): def test_is_session_alive(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) - self.assertTrue(client.is_session_alive()) + assert client.is_session_alive() def test_logout(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) - self.assertTrue(client.is_session_alive()) + assert client.is_session_alive() client.logout() def test_client_with_statement(self): with SteamClient( - self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file + self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file, ) as client: - self.assertTrue(client.is_session_alive()) + assert client.is_session_alive() def test_send_offer_without_sessionid_cookie(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) client._session.cookies.set('sessionid', None, domain='steamcommunity.com') cookies = client._session.cookies.get_dict('steamcommunity.com') - self.assertNotIn('sessionid', cookies) + assert 'sessionid' not in cookies game = GameOptions.TF2 asset_id = '' my_asset = Asset(asset_id, game) @@ -59,14 +60,14 @@ def test_sessionid_cookie(self): client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) community_cookies = client._session.cookies.get_dict('steamcommunity.com') store_cookies = client._session.cookies.get_dict('store.steampowered.com') - self.assertIn('sessionid', community_cookies) - self.assertIn('sessionid', store_cookies) + assert 'sessionid' in community_cookies + assert 'sessionid' in store_cookies def test_get_my_inventory(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) inventory = client.get_my_inventory(GameOptions.CS) - self.assertIsNotNone(inventory) + assert inventory is not None def test_get_partner_inventory(self): client = SteamClient(self.credentials.api_key) @@ -74,23 +75,23 @@ def test_get_partner_inventory(self): partner_id = '' game = GameOptions.TF2 inventory = client.get_partner_inventory(partner_id, game) - self.assertIsNotNone(inventory) + assert inventory is not None def test_get_trade_offers_summary(self): client = SteamClient(self.credentials.api_key) summary = client.get_trade_offers_summary() - self.assertIsNotNone(summary) + assert summary is not None def test_get_trade_offers(self): client = SteamClient(self.credentials.api_key) offers = client.get_trade_offers() - self.assertIsNotNone(offers) + assert offers is not None def test_get_trade_offer(self): client = SteamClient(self.credentials.api_key) trade_offer_id = '1442685162' offer = client.get_trade_offer(trade_offer_id) - self.assertIsNotNone(offer) + assert offer is not None def test_accept_trade_offer_without_login(self): client = SteamClient(self.credentials.api_key) @@ -101,21 +102,21 @@ def test_accept_trade_offer(self): client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) trade_offer_id = '1451378159' response_dict = client.accept_trade_offer(trade_offer_id) - self.assertIsNotNone(response_dict) + assert response_dict is not None def test_decline_trade_offer(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) trade_offer_id = '1449530707' response_dict = client.decline_trade_offer(trade_offer_id) - self.assertEqual(response_dict['response'], {}) + assert response_dict['response'] == {} def test_cancel_trade_offer(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) trade_offer_id = '1450637835' response_dict = client.cancel_trade_offer(trade_offer_id) - self.assertEqual(response_dict['response'], {}) + assert response_dict['response'] == {} def test_make_offer(self): client = SteamClient(self.credentials.api_key) @@ -129,8 +130,8 @@ def test_make_offer(self): my_asset = Asset(my_first_item['id'], game) partner_asset = Asset(partner_first_item['id'], game) response = client.make_offer([my_asset], [partner_asset], partner_id, 'TESTOWA OFERTA') - self.assertIsNotNone(response) - self.assertIn('tradeofferid', response.keys()) + assert response is not None + assert 'tradeofferid' in response def test_make_offer_url(self): partner_account_id = '32384925' @@ -150,8 +151,8 @@ def test_make_offer_url(self): my_asset = Asset(my_first_item['id'], game) partner_asset = Asset(partner_first_item['id'], game) response = client.make_offer_with_url([my_asset], [partner_asset], sample_trade_url, 'TESTOWA OFERTA') - self.assertIsNotNone(response) - self.assertIn('tradeofferid', response.keys()) + assert response is not None + assert 'tradeofferid' in response def test_get_escrow_duration(self): # A sample trade URL with escrow time of 15 days cause mobile auth not added @@ -159,13 +160,13 @@ def test_get_escrow_duration(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) response = client.get_escrow_duration(sample_trade_url) - self.assertEqual(response, 15) + assert response == 15 def test_get_wallet_balance(self): with SteamClient( - self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file + self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file, ) as client: wallet_balance = client.get_wallet_balance() - self.assertIsInstance(wallet_balance, Decimal) + assert isinstance(wallet_balance, Decimal) wallet_balance = client.get_wallet_balance(convert_to_decimal=False) - self.assertIsInstance(wallet_balance, str) + assert isinstance(wallet_balance, str) diff --git a/test/test_guard.py b/test/test_guard.py index e131e24..83421d0 100644 --- a/test/test_guard.py +++ b/test/test_guard.py @@ -8,23 +8,23 @@ class TestGuard(TestCase): @classmethod def setUpClass(cls): - cls.shared_secret = b64encode('1234567890abcdefghij'.encode('utf-8')) - cls.identity_secret = b64encode('abcdefghijklmnoprstu'.encode('utf-8')) + cls.shared_secret = b64encode(b'1234567890abcdefghij') + cls.identity_secret = b64encode(b'abcdefghijklmnoprstu') def test_one_time_code(self): timestamp = 1469184207 code = guard.generate_one_time_code(self.shared_secret, timestamp) - self.assertEqual(code, 'P2QJN') + assert code == 'P2QJN' def test_confirmation_key(self): timestamp = 1470838334 confirmation_key = guard.generate_confirmation_key(self.identity_secret, Tag.CONF.value, timestamp) - self.assertEqual(confirmation_key, b'pWqjnkcwqni+t/n+5xXaEa0SGeA=') + assert confirmation_key == b'pWqjnkcwqni+t/n+5xXaEa0SGeA=' def test_generate_device_id(self): steam_id = '12341234123412345' device_id = guard.generate_device_id(steam_id) - self.assertEqual(device_id, 'android:677cf5aa-3300-7807-d1e2-c408142742e2') + assert device_id == 'android:677cf5aa-3300-7807-d1e2-c408142742e2' def test_load_steam_guard(self): expected_keys = ('steamid', 'shared_secret', 'identity_secret') @@ -33,5 +33,5 @@ def test_load_steam_guard(self): guard_data = guard.load_steam_guard(guard_json_str) for key in expected_keys: - self.assertIn(key, guard_data) - self.assertIsInstance(guard_data[key], str) + assert key in guard_data + assert isinstance(guard_data[key], str) diff --git a/test/test_market.py b/test/test_market.py index e65e7b0..4670e47 100644 --- a/test/test_market.py +++ b/test/test_market.py @@ -1,25 +1,26 @@ -import os -from unittest import TestCase import unittest +from pathlib import Path +from unittest import TestCase from steampy.client import SteamClient from steampy.exceptions import TooManyRequests -from steampy.models import GameOptions, Currency +from steampy.models import Currency, GameOptions from steampy.utils import load_credentials + @unittest.skip('Requires secrets/Steamguard.txt') class TestMarket(TestCase): @classmethod def setUpClass(cls): cls.credentials = load_credentials()[0] - dirname = os.path.dirname(os.path.abspath(__file__)) + dirname = Path(__file__).resolve().parent cls.steam_guard_file = f'{dirname}/../secrets/Steamguard.txt' def test_get_price(self): client = SteamClient(self.credentials.api_key) item = 'M4A1-S | Cyrex (Factory New)' prices = client.market.fetch_price(item, GameOptions.CS) - self.assertTrue(prices['success']) + assert prices['success'] def test_get_price_to_many_requests(self): def request_loop(): @@ -33,21 +34,21 @@ def request_loop(): def test_get_price_history(self): with SteamClient( - self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file + self.credentials.api_key, self.credentials.login, self.credentials.password, self.steam_guard_file, ) as client: item = 'M4A1-S | Cyrex (Factory New)' response = client.market.fetch_price_history(item, GameOptions.CS) - self.assertTrue(response['success']) - self.assertIn('prices', response) + assert response['success'] + assert 'prices' in response def test_get_all_listings_from_market(self): client = SteamClient(self.credentials.api_key) client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) listings = client.market.get_my_market_listings() - self.assertEqual(len(listings), 2) - self.assertEqual(len(listings.get('buy_orders')), 1) - self.assertEqual(len(listings.get('sell_listings')), 1) - self.assertIsInstance(next(iter(listings.get('sell_listings').values())).get('description'), dict) + assert len(listings) == 2 + assert len(listings.get('buy_orders')) == 1 + assert len(listings.get('sell_listings')) == 1 + assert isinstance(next(iter(listings.get('sell_listings').values())).get('description'), dict) def test_create_and_remove_sell_listing(self): client = SteamClient(self.credentials.api_key) @@ -59,16 +60,16 @@ def test_create_and_remove_sell_listing(self): if item.get('marketable') == 1: asset_id_to_sell = asset_id break - self.assertIsNotNone(asset_id_to_sell, 'You need at least 1 marketable item to pass this test') + assert asset_id_to_sell is not None, 'You need at least 1 marketable item to pass this test' response = client.market.create_sell_order(asset_id_to_sell, game, '10000') - self.assertTrue(response['success']) + assert response['success'] sell_listings = client.market.get_my_market_listings()['sell_listings'] listing_to_cancel = None for listing in sell_listings.values(): if listing['description']['id'] == asset_id_to_sell: listing_to_cancel = listing['listing_id'] break - self.assertIsNotNone(listing_to_cancel) + assert listing_to_cancel is not None client.market.cancel_sell_order(listing_to_cancel) def test_create_and_cancel_buy_order(self): @@ -76,10 +77,10 @@ def test_create_and_cancel_buy_order(self): client.login(self.credentials.login, self.credentials.password, self.steam_guard_file) # PUT THE REAL CURRENCY OF YOUR STEAM WALLET, OTHER CURRENCIES WON'T WORK response = client.market.create_buy_order( - 'AK-47 | Redline (Field-Tested)', '10.34', 2, GameOptions.CS, Currency.EURO + 'AK-47 | Redline (Field-Tested)', '10.34', 2, GameOptions.CS, Currency.EURO, ) buy_order_id = response['buy_orderid'] - self.assertEqual(response['success'], 1) - self.assertIsNotNone(buy_order_id) + assert response['success'] == 1 + assert buy_order_id is not None response = client.market.cancel_buy_order(buy_order_id) - self.assertTrue(response['success']) + assert response['success'] diff --git a/test/test_utils.py b/test/test_utils.py index 06d16e6..262daa2 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -8,47 +8,45 @@ class TestUtils(TestCase): def test_text_between(self): text = 'var a = "dupadupa";' text_between = utils.text_between(text, 'var a = "', '";') - self.assertEqual(text_between, 'dupadupa') + assert text_between == 'dupadupa' def test_texts_between(self): text = '
  • element 1
    • \n
  • some random element
    • ' - items = [] - for el in utils.texts_between(text, '
  • ', '
    • '): - items.append(el) - self.assertEqual(items, ['element 1', 'some random element']) + items = list(utils.texts_between(text, '
  • ', '
    • ')) + assert items == ['element 1', 'some random element'] def test_account_id_to_steam_id(self): account_id = '358617487' steam_id = utils.account_id_to_steam_id(account_id) - self.assertEqual('76561198318883215', steam_id) + assert steam_id == '76561198318883215' def test_steam_id_to_account_id(self): steam_id = '76561198318883215' account_id = utils.steam_id_to_account_id(steam_id) - self.assertEqual(account_id, '358617487') + assert account_id == '358617487' def test_get_key_value_from_url(self): url = 'https://steamcommunity.com/tradeoffer/new/?partner=aaa&token=bbb' - self.assertEqual(utils.get_key_value_from_url(url, 'partner'), 'aaa') - self.assertEqual(utils.get_key_value_from_url(url, 'token'), 'bbb') + assert utils.get_key_value_from_url(url, 'partner') == 'aaa' + assert utils.get_key_value_from_url(url, 'token') == 'bbb' def test_get_key_value_from_url_case_insensitive(self): url = 'https://steamcommunity.com/tradeoffer/new/?Partner=aaa&Token=bbb' - self.assertEqual(utils.get_key_value_from_url(url, 'partner', case_sensitive=False), 'aaa') - self.assertEqual(utils.get_key_value_from_url(url, 'token', case_sensitive=False), 'bbb') + assert utils.get_key_value_from_url(url, 'partner', case_sensitive=False) == 'aaa' + assert utils.get_key_value_from_url(url, 'token', case_sensitive=False) == 'bbb' def test_calculate_gross_price(self): steam_fee = Decimal('0.05') # 5% publisher_fee = Decimal('0.1') # 10% - self.assertEqual(utils.calculate_gross_price(Decimal('0.01'), publisher_fee, steam_fee), Decimal('0.03')) - self.assertEqual(utils.calculate_gross_price(Decimal('0.10'), publisher_fee, steam_fee), Decimal('0.12')) - self.assertEqual(utils.calculate_gross_price(Decimal('100'), publisher_fee, steam_fee), Decimal('115')) + assert utils.calculate_gross_price(Decimal('0.01'), publisher_fee, steam_fee) == Decimal('0.03') + assert utils.calculate_gross_price(Decimal('0.10'), publisher_fee, steam_fee) == Decimal('0.12') + assert utils.calculate_gross_price(Decimal(100), publisher_fee, steam_fee) == Decimal(115) def test_calculate_net_price(self): steam_fee = Decimal('0.05') # 5% publisher_fee = Decimal('0.1') # 10% - self.assertEqual(utils.calculate_net_price(Decimal('0.03'), publisher_fee, steam_fee), Decimal('0.01')) - self.assertEqual(utils.calculate_net_price(Decimal('0.12'), publisher_fee, steam_fee), Decimal('0.10')) - self.assertEqual(utils.calculate_net_price(Decimal('115'), publisher_fee, steam_fee), Decimal('100')) + assert utils.calculate_net_price(Decimal('0.03'), publisher_fee, steam_fee) == Decimal('0.01') + assert utils.calculate_net_price(Decimal('0.12'), publisher_fee, steam_fee) == Decimal('0.10') + assert utils.calculate_net_price(Decimal(115), publisher_fee, steam_fee) == Decimal(100)