agresdominik/predictify
1
0
Fork 0
mirror of https://github.com/agresdominik/predictify.git synced 2026-04-21 17:55:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactored to own logger

Browse Source
This commit is contained in:
agres
2025-03-23 23:19:12 +01:00
parent 87f84b250e
commit db983cfc8f
1 changed files with 28 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gdpr_export.py
+28 -25
View File
@@ -1,15 +1,16 @@
import json
import logging as log
import os
from auth import simple_authenticate
from database_handler import Database, Table
from logger import LoggerWrapper
from spotify_api import get_multiple_field_information
# Define the absolute folder path to the folder containing the gdrp retrieved data
folder_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'data', 'gdpr_data')
# Define the db
db = Database()
log = LoggerWrapper()
def _read_gdrp_data() -> list:
@@ -20,32 +21,34 @@ def _read_gdrp_data() -> list:
:return: all_songs_played: A dict with an items field containing all songs played for the user
"""
all_songs_played = []
try:
for filename in os.listdir(folder_path):
for filename in os.listdir(folder_path):
if filename.endswith('.json'):
file_path = os.path.join(folder_path, filename)
if filename.endswith('.json'):
file_path = os.path.join(folder_path, filename)
with open(file_path, 'r') as file:
data = json.load(file)
with open(file_path, 'r') as file:
data = json.load(file)
for entry in data:
# This removes all podcasts from the list
if entry['spotify_track_uri'] is None:
continue
try:
track = {
'timestamp': entry['ts'],
'id': _extract_id(entry['spotify_track_uri']),
'track_name': entry['master_metadata_track_name'],
'artist_name': entry['master_metadata_album_artist_name'],
'album_name': entry['master_metadata_album_album_name'],
'conn_country': entry['conn_country'],
'ms_played': entry['ms_played']
}
all_songs_played.append(track)
except Exception as e:
print(f'Missing field: {e}')
for entry in data:
# This removes all podcasts from the list
if entry['spotify_track_uri'] is None:
continue
try:
track = {
'timestamp': entry['ts'],
'id': _extract_id(entry['spotify_track_uri']),
'track_name': entry['master_metadata_track_name'],
'artist_name': entry['master_metadata_album_artist_name'],
'album_name': entry['master_metadata_album_album_name'],
'conn_country': entry['conn_country'],
'ms_played': entry['ms_played']
}
all_songs_played.append(track)
except Exception as e:
log.warning(f'Missing field from gdpr data: {e}')
except Exception as e:
log.error(f'Failed to read gdpr data: {e}')
all_songs_played = sorted(all_songs_played, key=lambda x: x['timestamp'])
return all_songs_played
@@ -139,7 +142,7 @@ def _insert_data_into_db(all_songs_played: list):
log.error(f'Failed adding {entry} to database, error {e}')
def export_gdpr_data(n_limit: int = 100):
def export_gdpr_data(n_limit: int = 100) -> None:
all_songs_played = _read_gdrp_data()
all_songs_played = all_songs_played[-n_limit:]
all_songs_catalogued = _populate_ids(all_songs_played)