personal-data/personal_data/fetchers/steam_community.py

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