105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
import dataclasses
|
|
import datetime
|
|
import logging
|
|
import re
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import bs4
|
|
|
|
from .. import html_util, parse_util, secrets
|
|
from ..data import DeduplicateMode, Scraper
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
URL_SITE_ROOT = 'https://steamcommunity.com/'
|
|
|
|
URL_GAME_ACHIVEMENTS = URL_SITE_ROOT + 'id/{username}/stats/appid/{appid}'
|
|
URL_USER_RECENT_ACTIVITY = URL_SITE_ROOT + 'id/{username}'
|
|
|
|
FORMAT_DATE_HEADER = '%d/%m/%YYYY'
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class SteamAchievement(Scraper):
|
|
"""Downloads all trophies for the given user."""
|
|
|
|
dataset_name = 'games_played'
|
|
deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS
|
|
|
|
def scrape(self) -> Iterator[dict[str, Any]]:
|
|
username: str = secrets.steam_username()
|
|
appids = list(self._determine_appids_from_recent_activity(username))
|
|
logger.info('Found %d Steam Apps', len(appids))
|
|
for appid in appids:
|
|
yield from self._scrape_app_achievements(username, appid)
|
|
|
|
def _determine_appids_from_recent_activity(self, username: str) -> Iterator[int]:
|
|
url = URL_USER_RECENT_ACTIVITY.format(
|
|
username=username,
|
|
)
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
|
|
soup = html_util.normalize_soup_slightly(
|
|
bs4.BeautifulSoup(response.content, 'lxml'),
|
|
classes=False,
|
|
)
|
|
|
|
for entry_a in soup.select('.recent_games .recent_game .game_info_cap a'):
|
|
href = entry_a['href']
|
|
appid = int(href.split('/')[-1])
|
|
yield appid
|
|
|
|
def _scrape_app_achievements(
|
|
self,
|
|
username: str,
|
|
appid: int,
|
|
) -> Iterator[dict[str, Any]]:
|
|
url = URL_GAME_ACHIVEMENTS.format(
|
|
username=username,
|
|
appid=appid,
|
|
)
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
|
|
# Parse data
|
|
soup = bs4.BeautifulSoup(response.content, 'lxml')
|
|
|
|
game_name: str = re.match(
|
|
r'Steam Community :: (.+) :: .*',
|
|
soup.head.title.get_text(),
|
|
).group(1)
|
|
|
|
soup = html_util.normalize_soup_slightly(
|
|
soup,
|
|
classes=False,
|
|
)
|
|
|
|
for entry in soup.select('.achieveRow'):
|
|
trophy_name: str = entry.select_one('h3').get_text()
|
|
trophy_desc: str = entry.select_one('h5').get_text()
|
|
trophy_icon: str = entry.select_one('img')['src']
|
|
|
|
time_acquired_html: str = entry.select_one('.achieveUnlockTime')
|
|
if time_acquired_html is None:
|
|
continue
|
|
time_acquired_text: str = (
|
|
time_acquired_html.get_text().strip().removeprefix('Unlocked ')
|
|
)
|
|
time_acquired: datetime.datetime = parse_util.parse_time(time_acquired_text)
|
|
|
|
yield {
|
|
'game.name': game_name,
|
|
#'game.release_date': None,
|
|
'me.last_played_time': time_acquired,
|
|
# Trophy Data
|
|
'trophy.name': trophy_name,
|
|
'trophy.desc': trophy_desc,
|
|
'trophy.icon': trophy_icon,
|
|
# Ids
|
|
'steam.appid': appid,
|
|
}
|
|
|
|
del entry, time_acquired
|