1
Fork 0
mirror of https://github.com/Steffo99/steamleaderboards.git synced 2024-10-16 06:17:29 +00:00

Add placement-based limiting

As mentioned in #1.
This commit is contained in:
Steffo 2024-05-26 10:32:21 +02:00
parent 446b715daa
commit e23210abaf
Signed by: steffo
GPG key ID: 5ADA3868646C3FC0
3 changed files with 19 additions and 9 deletions

2
.vscode/launch.json vendored
View file

@ -6,7 +6,7 @@
"type": "debugpy", "type": "debugpy",
"request": "launch", "request": "launch",
"module": "steamleaderboards", "module": "steamleaderboards",
"args": ["-o", "./data", "383980"] "args": ["-o", "./data", "383980", "--limit", "10", "--request-delay", "1.5"]
} }
] ]
} }

View file

@ -59,7 +59,7 @@ class ProtoLeaderboard:
class Leaderboard: class Leaderboard:
# noinspection PyMissingConstructor # noinspection PyMissingConstructor
def __init__(self, app_id=None, lbid=None, *, protoleaderboard=None, limit=5000, delay=None): def __init__(self, app_id=None, lbid=None, *, protoleaderboard=None, limit=None, delay=None):
if protoleaderboard: if protoleaderboard:
self.url = protoleaderboard.url self.url = protoleaderboard.url
self.lbid = protoleaderboard.lbid self.lbid = protoleaderboard.lbid
@ -80,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 limit is None:
limit = 5000
if delay is None: if delay is None:
delay = 0.5 delay = 0.5
next_request_url = self.url next_request_url = self.url
@ -89,12 +91,16 @@ class Leaderboard:
_bs = BeautifulSoup(xml.content, features="lxml-xml") _bs = BeautifulSoup(xml.content, features="lxml-xml")
for entry in _bs.find_all("entry"): for entry in _bs.find_all("entry"):
self.entries.append(Entry(entry)) self.entries.append(Entry(entry))
try: if len(self.entries) >= limit:
next_request_url = _bs.find_all("nextRequestURL")[0].text next_request_url = None
except IndexError: break
next_request_url = None
else: else:
time.sleep(delay) try:
next_request_url = _bs.find_all("nextRequestURL")[0].text
except IndexError:
next_request_url = None
else:
time.sleep(delay)
def __repr__(self): def __repr__(self):
if self.name: if self.name:

View file

@ -2,6 +2,7 @@ import argparse
import importlib.metadata import importlib.metadata
import pathlib import pathlib
import sys import sys
import time
from . import LeaderboardGroup, ProtoLeaderboard, Leaderboard, Entry from . import LeaderboardGroup, ProtoLeaderboard, Leaderboard, Entry
@ -11,7 +12,8 @@ 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("-d", "--request-delay", dest="request_delay", help="How long to wait between two requests to the scoreboard API.", type=float, default=0.5)
parser.add_argument("-l", "--limit", dest="limit", help="How many entries to retrieve for each scoreboard.", type=int, default=5000)
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():
@ -32,11 +34,13 @@ 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(delay=args.request_delay) full: Leaderboard = proto.full(limit=args.limit, 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:
file.write(f"{entry.rank!r},{entry.steam_id!r},{entry.score!r},{entry.ugcid!r},{entry.details!r}\n") file.write(f"{entry.rank!r},{entry.steam_id!r},{entry.score!r},{entry.ugcid!r},{entry.details!r}\n")
print(f"done, resting for {args.request_delay} seconds", file=sys.stderr)
time.sleep(args.request_delay)
if len(lg.leaderboards) == 0: if len(lg.leaderboards) == 0:
print(f"game has no leaderboards", file=sys.stderr) print(f"game has no leaderboards", file=sys.stderr)