steffo/steamleaderboards
steffo/steamleaderboards
1
Fork 0
mirror of https://github.com/Steffo99/steamleaderboards.git synced 2024-11-21 23:34:17 +00:00
Code Activity

Add delay-based rate limiting

Browse source 
As mentioned in #1.
This commit is contained in:
Steffo 2024-05-26 10:19:56 +02:00
parent 428b11fae4
commit 446b715daa
Signed by: steffo
GPG key ID: 5ADA3868646C3FC0
2 changed files with 10 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
steamleaderboards/__init__.py
View file

@ -1,6 +1,7 @@
import requests import requests
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
import typing import typing
import time
class LeaderboardGroup: class LeaderboardGroup:
@ -52,13 +53,13 @@ class ProtoLeaderboard:
self.display_type = int(soup.displaytype.text) self.display_type = int(soup.displaytype.text)
self.app_id = app_id self.app_id = app_id
def full(self) -> "Leaderboard": def full(self, *args, **kwargs) -> "Leaderboard":
return Leaderboard(protoleaderboard=self) return Leaderboard(*args, **kwargs, protoleaderboard=self)
class Leaderboard: class Leaderboard:
# noinspection PyMissingConstructor # noinspection PyMissingConstructor
def __init__(self, app_id=None, lbid=None, *, protoleaderboard=None): def __init__(self, app_id=None, lbid=None, *, protoleaderboard=None, limit=5000, delay=None):
if protoleaderboard: if protoleaderboard:
self.url = protoleaderboard.url self.url = protoleaderboard.url
self.lbid = protoleaderboard.lbid self.lbid = protoleaderboard.lbid
@ -79,6 +80,8 @@ class Leaderboard:
self.display_type = None self.display_type = None
else: else:
raise ValueError("No app_id, lbid or protoleaderboard specified") raise ValueError("No app_id, lbid or protoleaderboard specified")
if delay is None:
delay = 0.5
next_request_url = self.url next_request_url = self.url
self.entries = [] self.entries = []
while next_request_url: while next_request_url:
@ -90,6 +93,8 @@ class Leaderboard:
next_request_url = _bs.find_all("nextRequestURL")[0].text next_request_url = _bs.find_all("nextRequestURL")[0].text
except IndexError: except IndexError:
next_request_url = None next_request_url = None
else:
time.sleep(delay)
def __repr__(self): def __repr__(self):
if self.name: if self.name:

3
steamleaderboards/__main__.py
View file

@ -11,6 +11,7 @@ parser = argparse.ArgumentParser(
parser.add_argument("-o", "--output-dir", dest="output_dir", help="The directory where downloaded leaderboards should be stored in.", type=pathlib.Path) parser.add_argument("-o", "--output-dir", dest="output_dir", help="The directory where downloaded leaderboards should be stored in.", type=pathlib.Path)
parser.add_argument("app_id", type=int, nargs="+") parser.add_argument("app_id", type=int, nargs="+")
parser.add_argument("-d", "--request-delay", dest="request_delay", help="How long to wait between two requests to the scoreboard API.", type=float)
parser.add_argument("-V", "--version", action="version", version=importlib.metadata.version("steamleaderboards")) parser.add_argument("-V", "--version", action="version", version=importlib.metadata.version("steamleaderboards"))
def main(): def main():
@ -31,7 +32,7 @@ def main():
lg: LeaderboardGroup = LeaderboardGroup(app_id) lg: LeaderboardGroup = LeaderboardGroup(app_id)
for proto in lg.leaderboards: for proto in lg.leaderboards:
print(f"fetching full leaderboard: {app_id} {proto.name}", file=sys.stderr) print(f"fetching full leaderboard: {app_id} {proto.name}", file=sys.stderr)
full: Leaderboard = proto.full() full: Leaderboard = proto.full(delay=args.request_delay)
with open(output_dir.joinpath(f"{full.app_id}_{full.name}.csv"), mode="w") as file: with open(output_dir.joinpath(f"{full.app_id}_{full.name}.csv"), mode="w") as file:
file.write(f"rank,steam_id,score,ugcid,details\n") file.write(f"rank,steam_id,score,ugcid,details\n")
for entry in full.entries: for entry in full.entries: