Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_text_between(self):
text = 'var a = "dupadupa";'
text_between = utils.text_between(text, 'var a = "', '";')
self.assertEquals(text_between, 'dupadupa')
def _fetch_trade_partner_id(self, trade_offer_id: str) -> str:
url = self._get_trade_offer_url(trade_offer_id)
offer_response_text = self._session.get(url).text
if 'You have logged in from a new device. In order to protect the items' in offer_response_text:
raise SevenDaysHoldException("Account has logged in a new device and can't trade for 7 days")
return text_between(offer_response_text, "var g_ulTradePartnerSteamID = '", "';")
def get_my_market_listings(self) -> dict:
response = self._session.get("%s/market" % SteamUrl.COMMUNITY_URL)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % response.status_code)
assets_descriptions = json.loads(text_between(response.text, "var g_rgAssets = ", ";\r\n"))
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)
if '<span id="tabContentsMyActiveMarketListings_end">' in response.text:
n_showing = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_end">', '</span>'))
n_total = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_total">', '</span>').replace(',',''))
if n_showing < n_total < 1000:
url = "%s/market/mylistings/render/?query=&start=%s&count=%s" % (SteamUrl.COMMUNITY_URL, n_showing, -1)
response = self._session.get(url)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % 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["sell_listings"] = {**listings["sell_listings"], **listings_2["sell_listings"]}
else:
for i in range(0, n_total, 100):
url = "%s/market/mylistings/?query=&start=%s&count=%s" % (SteamUrl.COMMUNITY_URL, n_showing + i, 100)
response = self._session.get(url)</span>
def get_escrow_duration(self, trade_offer_url: str) -> int:
headers = {'Referer': SteamUrl.COMMUNITY_URL + urlparse.urlparse(trade_offer_url).path,
'Origin': SteamUrl.COMMUNITY_URL}
response = self._session.get(trade_offer_url, headers=headers).text
my_escrow_duration = int(text_between(response, "var g_daysMyEscrow = ", ";"))
their_escrow_duration = int(text_between(response, "var g_daysTheirEscrow = ", ";"))
return max(my_escrow_duration, their_escrow_duration)
def get_my_market_listings(self) -> dict:
response = self._session.get("%s/market" % SteamUrl.COMMUNITY_URL)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % response.status_code)
assets_descriptions = json.loads(text_between(response.text, "var g_rgAssets = ", ";\r\n"))
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)
if '<span id="tabContentsMyActiveMarketListings_end">' in response.text:
n_showing = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_end">', '</span>'))
n_total = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_total">', '</span>').replace(',',''))
if n_showing < n_total < 1000:
url = "%s/market/mylistings/render/?query=&start=%s&count=%s" % (SteamUrl.COMMUNITY_URL, n_showing, -1)
response = self._session.get(url)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % 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["sell_listings"] = {**listings["sell_listings"], **listings_2["sell_listings"]}
else:
for i in range(0, n_total, 100):
url = "%s/market/mylistings/?query=&start=%s&count=%s" % (SteamUrl.COMMUNITY_URL, n_showing + i, 100)
response = self._session.get(url)
if response.status_code != 200:</span>
def get_my_market_listings(self) -> dict:
response = self._session.get("%s/market" % SteamUrl.COMMUNITY_URL)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % response.status_code)
assets_descriptions = json.loads(text_between(response.text, "var g_rgAssets = ", ";\r\n"))
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)
if '<span id="tabContentsMyActiveMarketListings_end">' in response.text:
n_showing = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_end">', '</span>'))
n_total = int(text_between(response.text, '<span id="tabContentsMyActiveMarketListings_total">', '</span>').replace(',',''))
if n_showing < n_total < 1000:
url = "%s/market/mylistings/render/?query=&start=%s&count=%s" % (SteamUrl.COMMUNITY_URL, n_showing, -1)
response = self._session.get(url)
if response.status_code != 200:
raise ApiException("There was a problem getting the listings. http code: %s" % 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,</span>
def get_escrow_duration(self, trade_offer_url: str) -> int:
headers = {'Referer': SteamUrl.COMMUNITY_URL + urlparse.urlparse(trade_offer_url).path,
'Origin': SteamUrl.COMMUNITY_URL}
response = self._session.get(trade_offer_url, headers=headers).text
my_escrow_duration = int(text_between(response, "var g_daysMyEscrow = ", ";"))
their_escrow_duration = int(text_between(response, "var g_daysTheirEscrow = ", ";"))
return max(my_escrow_duration, their_escrow_duration)