From 87189f1bd83c9aa41857010642e0c8fcc1575fcc Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Sun, 25 Feb 2024 01:38:44 +0100 Subject: [PATCH] PSN Profiles --- README.md | 11 +++ personal_data/__main__.py | 26 ++++-- personal_data/fetchers/psnprofiles.py | 113 ++++++++++++++++++++++++++ personal_data/html_util.py | 73 +++++++++++++++++ personal_data/parse_util.py | 25 ++++++ 5 files changed, 242 insertions(+), 6 deletions(-) create mode 100644 README.md create mode 100644 personal_data/fetchers/psnprofiles.py create mode 100644 personal_data/html_util.py create mode 100644 personal_data/parse_util.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..055cbc4 --- /dev/null +++ b/README.md @@ -0,0 +1,11 @@ + +# Personal Data Fetcher Systems + +This is a collection of small fetchers for personal data spread around the internet. + +## Ideas for more fetchers + +- [ ] Save data for Theatherythm: Most played songs, and liked songs. +- [ ] YouTube (Music): Liked videos with title and URL. +- [ ] Playstation: Achivement dates, hours played and last played dates + diff --git a/personal_data/__main__.py b/personal_data/__main__.py index f76de5a..538357a 100644 --- a/personal_data/__main__.py +++ b/personal_data/__main__.py @@ -7,19 +7,22 @@ import io import browsercookie from frozendict import frozendict import logging +import cfscrape logger = logging.getLogger(__name__) import personal_data.fetchers.playstation import personal_data.fetchers.crunchyroll +import personal_data.fetchers.psnprofiles def determine_scrapers(): scrapers = [] #scrapers += personal_data.fetchers.playstation.SCRAPERS - scrapers += personal_data.fetchers.crunchyroll.SCRAPERS + #scrapers += personal_data.fetchers.crunchyroll.SCRAPERS + scrapers += personal_data.fetchers.psnprofiles.SCRAPERS return scrapers -def extend_csv_file(filename, new_dicts , deduplicate = False): +def extend_csv_file(filename, new_dicts, deduplicate = False): dicts = [] try: with open(filename, 'r') as csvfile: @@ -31,10 +34,15 @@ def extend_csv_file(filename, new_dicts , deduplicate = False): logger.info('Creating file: %s', filename) pass + original_num_dicts = len(dicts) dicts += [frozendict(d) for d in new_dicts] del new_dicts - fieldnames = list(dicts[0].keys()) + fieldnames = [] + for d in dicts: + for k in d.keys(): + if k not in fieldnames: + fieldnames.append(k) if deduplicate: dicts = sorted(set(dicts), key = lambda d: d[fieldnames[0]]) @@ -50,10 +58,11 @@ def extend_csv_file(filename, new_dicts , deduplicate = False): with open(filename, 'w') as csvfile: csvfile.write(output_csv) del csvfile + logger.warning('Extended CSV "%s" from %d to %d lines', filename, original_num_dicts, len(dicts)) STANDARD_HEADERS = { 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:122.0) Gecko/20100101 Firefox/122.0', - "Accept": "application/json, text/plain, */*", + #"Accept": "application/json, text/plain, */*", 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate, br', } @@ -62,15 +71,20 @@ def main(): cookiejar = browsercookie.firefox() logger.warning('Got cookiejar from firefox: %s cookies', len(cookiejar)) - session = requests_cache.CachedSession('web_cache', cookies = cookiejar) + #session = requests_cache.CachedSession('web_cache', cookies = cookiejar) + session = cfscrape.create_scraper() for cookie in cookiejar: session.cookies.set_cookie(cookie) for scraper in determine_scrapers(): logger.warning('Running scraper: %s', scraper.dataset_name) - result_rows = list(scraper.scraper(session)) + result_rows = list() + for result in scraper.scraper(session): + result_rows.append(result) + del result extend_csv_file('output/'+scraper.dataset_name, result_rows, deduplicate = scraper.deduplicate) + logger.warning('Scraper done: %s', scraper.dataset_name) if __name__ == '__main__': main() diff --git a/personal_data/fetchers/psnprofiles.py b/personal_data/fetchers/psnprofiles.py new file mode 100644 index 0000000..4b4d711 --- /dev/null +++ b/personal_data/fetchers/psnprofiles.py @@ -0,0 +1,113 @@ +import secrets +import functools +import re +import logging +import bs4 +import datetime + +from personal_data.data import Scraper +import personal_data.html_util +import personal_data.parse_util + +logger = logging.getLogger(__name__) + +URL_PROFILE = 'https://psnprofiles.com/{psn_id}' + +FORMAT_DATE_HEADER = '%a, %d %b %Y %H:%M:%S GMT' +FORMAT_DAY_MONTH_YEAR = '%d %B %Y' + +def game_psnprofiles_id_from_url(relative_url: str) -> int: + m = re.match(r'/(?:trophy|trophies)/(\d+)\-(?:[\w-]+)(/[\w-]*)?', relative_url) + result = m.group(1) + return int(result) + +assert game_psnprofiles_id_from_url('/trophies/21045-theatrhythm-final-bar-line') +assert game_psnprofiles_id_from_url('/trophies/21045-theatrhythm-final-bar-line/') +assert game_psnprofiles_id_from_url('/trophies/21045-theatrhythm-final-bar-line/Jmaanes') +assert game_psnprofiles_id_from_url('/trophy/21045-theatrhythm-final-bar-line/19-seasoned-hunter') + +def scrape_personal_page(session): + # Request to get watch history + logger.info('Getting Watchlist') + url = URL_PROFILE.format(psn_id = secrets.PLAYSTATION_PSN_ID) + response = session.get(url) + response.raise_for_status() + + NOW = datetime.datetime.strptime(response.headers['Date'], FORMAT_DATE_HEADER) + + # Parse data + soup = bs4.BeautifulSoup(response.content, 'lxml') + soup = personal_data.html_util.normalize_soup_slightly(soup, classes = False) + + # Recent trophies. + soup_recent_tropies = soup.select('ul#recent-trophies > li') + assert len(soup_recent_tropies) > 0, url + for row in soup_recent_tropies: + cells = row.select_one('.info .box td').find_all('div') + + trophy_name = cells[0].get_text().strip() + trophy_desc = cells[1].get_text().strip() + game_name = cells[2].a.extract().get_text().strip() + + psnprofiles_id = game_psnprofiles_id_from_url(cells[0].find('a')['href']) + trophy_icon = row.find(class_='icon').find('img')['src'] + + gotten_at = cells[2].get_text().strip().removesuffix(' in').removesuffix(' ago') + gotten_at = personal_data.parse_util.parse_duration(gotten_at) + time_acquired = NOW - gotten_at + + 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, + 'psnprofiles.game_id': psnprofiles_id, + } + + del row, cells, time_acquired + + # Games table + table_rows = soup.find(id = 'gamesTable').find_all('tr') + assert len(table_rows) > 0, url + + for row in table_rows: + cells = row.find_all('td') + + # Check for pagination + if re.match(r'show \d+ more games', cells[0].get_text().strip(), re.IGNORECASE): + break + + game_name = cells[1].find(class_ = 'title').get_text() + psnprofiles_id = game_psnprofiles_id_from_url(cells[0].find('a')['href']) + game_icon = cells[0].find('img')['src'] + + game_name = row.select_one('.title').get_text() + game_platform = row.select_one('.platform').get_text() + + small_infos = cells[1].find_all('div') + if len(small_infos) > 2: + time_played_div = small_infos[2] + time_played_div.sup.extract() + time_played = datetime.datetime.strptime(time_played_div.get_text().strip(), FORMAT_DAY_MONTH_YEAR).date() + else: + time_played = None + + d = { + # Important fields + 'game.name': game_name, + + # Secondary fields + 'game.platform': game_platform, + 'game.icon': game_icon, + 'psnprofiles.game_id': psnprofiles_id, + } + if time_played: + d['me.last_played_time'] = time_played + yield d + +SCRAPERS = [ + Scraper(scrape_personal_page, 'games_played_playstation', deduplicate = True) +] diff --git a/personal_data/html_util.py b/personal_data/html_util.py new file mode 100644 index 0000000..b970f13 --- /dev/null +++ b/personal_data/html_util.py @@ -0,0 +1,73 @@ +import re +import bs4 + +HTML_TAGS_MOSTLY_CONTENTLESS: set[str] = {'style', 'svg', 'link', 'br', 'math', + 'canvas'} + +HTML_TAGS_WITH_LITTLE_CONTENT: set[str] = { 'head', 'script', 'meta' } | HTML_TAGS_MOSTLY_CONTENTLESS + +def normalize_text(text: str) -> str: + text = text.replace('\t', ' ') + text = text.replace('\r', '') + text = re.sub(r'\s*\n\s*\n\s*', '\n\n', text) + text = re.sub(r' +', ' ', text) + text = re.sub(r'^\s+', '', text) + text = re.sub(r'\s+$', '', text) + return text.encode('utf-8') + +def normalize_soup_bs4(soup: bs4.BeautifulSoup) -> bytes: + for comment in soup(text=lambda text: isinstance(text, bs4.Comment)): + comment.extract() + del comment + for element_name in HTML_TAGS_WITH_LITTLE_CONTENT: + for script_elements in soup(element_name): + script_elements.decompose() + del element_name + soup.smooth() + return soup + +def normalize_soup_lxml(soup) -> bytes: + for element_name in HTML_TAGS_WITH_LITTLE_CONTENT: + for script_elements in soup.cssselect(element_name): + script_elements.drop_tree() + del script_elements + del element_name + return soup + +def normalize_soup(soup) -> bytes: + text = None + if isinstance(soup, bs4.BeautifulSoup): + text = normalize_soup_bs4(soup).get_text() + else: + text = normalize_soup_lxml(soup).text_content() + return normalize_text(text) + +def normalize_soup_slightly(soup, classes = True, scripts = True, comments = True): + # Little if any content + for tag in HTML_TAGS_MOSTLY_CONTENTLESS: + for e in soup.select(tag): + e.decompose() + + if classes: + for e in soup.find_all(class_=True): + del e['class'] + for e in soup.find_all('script', src=True): + e.decompose() + for e in soup.find_all(style=True): + del e['style'] + for e in soup.select('a'): + del e['height'], e['target'], e['rel'], e['onclick'] + + for e in soup.select('a[href=""]'): + del e['href'] + + if scripts: + for e in soup.find_all('script'): + e.decompose() + + if comments: + for c in soup.find_all(string=lambda text: isinstance(text, bs4.Comment)): + c.extract() + + soup.smooth() + return soup diff --git a/personal_data/parse_util.py b/personal_data/parse_util.py new file mode 100644 index 0000000..2b3ccb5 --- /dev/null +++ b/personal_data/parse_util.py @@ -0,0 +1,25 @@ + +import datetime + +DATETIME_UNITS = { + 'second': datetime.timedelta(seconds = 1), + 'seconds': datetime.timedelta(seconds = 1), + 'minute': datetime.timedelta(minutes = 1), + 'minutes': datetime.timedelta(minutes = 1), + 'hour': datetime.timedelta(hours = 1), + 'hours': datetime.timedelta(hours = 1), + 'day': datetime.timedelta(days = 1), + 'days': datetime.timedelta(days = 1), + 'week': datetime.timedelta(days = 7), + 'weeks': datetime.timedelta(days = 7), + 'month': datetime.timedelta(days = 30), + 'months': datetime.timedelta(days = 30), + 'year': datetime.timedelta(days = 365), + 'years': datetime.timedelta(days = 365), +} + +def parse_duration(text: str) -> datetime.timedelta: + (num, unit) = text.split(' ') + num = int(num) + unit = DATETIME_UNITS[unit] + return unit * num