1
0

Implemented partial Steam support

This commit is contained in:
Jon Michael Aanes 2024-09-01 16:55:45 +02:00
parent c6fd5bc5cc
commit 263b0b2843
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
5 changed files with 91 additions and 6 deletions

View File

@ -0,0 +1,70 @@
import dataclasses
import datetime
import logging
import re
import bs4
from typing import Any
from collections.abc import Iterator
from ..data import DeduplicateMode, Scraper
from .. import secrets, parse_util, html_util
logger = logging.getLogger(__name__)
URL_SITE_ROOT = 'https://steamcommunity.com/'
URL_GAME_ACHIVEMENTS = URL_SITE_ROOT+'id/{username}/stats/appid/{appid}'
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]]:
yield from self.scrape_app(105600)
def scrape_app(self, appid: int) -> Iterator[dict[str, Any]]:
url = URL_GAME_ACHIVEMENTS.format(
username=secrets.STEAM_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 :: (.+) :: Jmaa', 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

View File

@ -36,14 +36,26 @@ def parse_response_datetime(response) -> datetime.datetime:
).replace(tzinfo=datetime.UTC)
LOCAL_TIMEZONE = datetime.datetime.now(datetime.timezone.utc).astimezone().tzinfo
LOCAL_TIMEZONE = datetime.datetime.now(datetime.UTC).astimezone().tzinfo
def try_parse(text:str, fmt:str) -> datetime.datetime | None:
try:
time = datetime.datetime.strptime(text, fmt)
if time.tzinfo is None:
time = time.replace(tzinfo=LOCAL_TIMEZONE)
except:
time = None
return time
def parse_time(text: str) -> datetime.datetime:
text = text.replace('\n', ' ')
text = text.strip()
time = datetime.datetime.strptime(text, '%d %b %Y %I:%M:%S %p')
time = time.replace(tzinfo=LOCAL_TIMEZONE)
time = try_parse(text, '%d %b %Y %I:%M:%S %p')
time = time or try_parse(text, '%d %b, %Y @ %I:%M%p')
assert time.tzinfo is not None, time
return time

View File

@ -15,6 +15,9 @@ PLAYSTATION_PSN_ID = load_secret('PLAYSTATION_PSN_ID')
# Partisia Blockchain
PBC_ACCOUNT_ADDRESS = load_secret('PBC_ACCOUNT_ADDRESS')
# Steam
STEAM_USERNAME = load_secret('STEAM_USERNAME')
# Kucoin
KUCOIN_KEY = load_secret('KUCOIN_KEY')
KUCOIN_SECRET = load_secret('KUCOIN_SECRET')

View File

@ -60,7 +60,7 @@ LIST_2 = [
27,
31,
134506,
tzinfo=datetime.timezone.utc,
tzinfo=datetime.UTC,
),
'weight.kg': Decimal('73.6'),
},
@ -75,7 +75,7 @@ LIST_2 = [
36,
9,
590355,
tzinfo=datetime.timezone.utc,
tzinfo=datetime.UTC,
),
'weight.kg': Decimal('74.7'),
},

View File

@ -8,7 +8,7 @@ from personal_data.main import to_value
PARSE_MAPPINGS = [
(
'2024-04-28 21:35:40+00:00',
datetime.datetime(2024, 4, 28, 21, 35, 40, tzinfo=datetime.timezone.utc),
datetime.datetime(2024, 4, 28, 21, 35, 40, tzinfo=datetime.UTC),
),
(
'0003791e9f5f3691b8bbbe0d12a7ae9c3f2e89db38',