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): dataset_name = 'games_played_TODO' deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS def scrape(self) -> Iterator[dict[str, Any]]: username = secrets.STEAM_USERNAME for appid in self.determine_appids_from_recent_activity(username): yield from self.scrape_app(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(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() NOW = parse_util.parse_response_datetime(response) # 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, 'me.last_played_time': time_acquired, # Trophy Data 'trophy.name': trophy_name, 'trophy.desc': trophy_desc, 'trophy.icon': trophy_icon, } del entry, time_acquired