1
0
personal-data/personal_data/fetchers/steam_community.py

105 lines
3.2 KiB
Python
Raw Permalink Normal View History

2024-09-01 14:55:45 +00:00
import dataclasses
import datetime
import logging
import re
from collections.abc import Iterator
2024-10-03 21:24:12 +00:00
from typing import Any
2024-09-01 14:55:45 +00:00
2024-10-03 21:24:12 +00:00
import bs4
from .. import html_util, parse_util, secrets
2024-09-01 14:55:45 +00:00
from ..data import DeduplicateMode, Scraper
logger = logging.getLogger(__name__)
URL_SITE_ROOT = 'https://steamcommunity.com/'
2024-10-03 21:24:12 +00:00
URL_GAME_ACHIVEMENTS = URL_SITE_ROOT + 'id/{username}/stats/appid/{appid}'
URL_USER_RECENT_ACTIVITY = URL_SITE_ROOT + 'id/{username}'
2024-09-01 14:55:45 +00:00
FORMAT_DATE_HEADER = '%d/%m/%YYYY'
@dataclasses.dataclass(frozen=True)
class SteamAchievement(Scraper):
2024-10-25 19:29:44 +00:00
"""Downloads all trophies for the given user."""
2024-10-25 19:47:44 +00:00
dataset_name = 'games_played'
2024-09-01 14:55:45 +00:00
deduplicate_mode = DeduplicateMode.BY_ALL_COLUMNS
def scrape(self) -> Iterator[dict[str, Any]]:
2024-10-25 19:47:44 +00:00
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:
2024-10-25 19:29:44 +00:00
yield from self._scrape_app_achievements(username, appid)
2024-09-01 14:55:45 +00:00
2024-10-25 19:29:44 +00:00
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
2024-10-25 19:30:12 +00:00
def _scrape_app_achievements(
2024-10-25 19:52:20 +00:00
self,
username: str,
appid: int,
2024-10-25 19:30:12 +00:00
) -> Iterator[dict[str, Any]]:
2024-09-01 14:55:45 +00:00
url = URL_GAME_ACHIVEMENTS.format(
username=username,
2024-10-03 21:24:12 +00:00
appid=appid,
2024-09-01 14:55:45 +00:00
)
response = self.session.get(url)
response.raise_for_status()
# Parse data
soup = bs4.BeautifulSoup(response.content, 'lxml')
2024-10-03 21:24:12 +00:00
game_name: str = re.match(
2024-10-10 22:54:01 +00:00
r'Steam Community :: (.+) :: .*',
soup.head.title.get_text(),
2024-10-03 21:24:12 +00:00
).group(1)
2024-09-01 14:55:45 +00:00
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()
2024-10-25 19:47:44 +00:00
trophy_icon: str = entry.select_one('img')['src']
2024-09-01 14:55:45 +00:00
time_acquired_html: str = entry.select_one('.achieveUnlockTime')
if time_acquired_html is None:
continue
2024-10-03 21:24:12 +00:00
time_acquired_text: str = (
time_acquired_html.get_text().strip().removeprefix('Unlocked ')
)
2024-09-01 14:55:45 +00:00
time_acquired: datetime.datetime = parse_util.parse_time(time_acquired_text)
yield {
'game.name': game_name,
2024-10-25 19:47:44 +00:00
#'game.release_date': None,
2024-09-01 14:55:45 +00:00
'me.last_played_time': time_acquired,
# Trophy Data
'trophy.name': trophy_name,
'trophy.desc': trophy_desc,
'trophy.icon': trophy_icon,
2024-10-25 19:29:44 +00:00
# Ids
'steam.appid': appid,
2024-09-01 14:55:45 +00:00
}
del entry, time_acquired