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 SteamAchievementScraper(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