Coverage for src/aggregator/artist_aggregate.py: 100%
36 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-05-14 23:08 +0000
« prev ^ index » next coverage.py v7.4.4, created at 2024-05-14 23:08 +0000
1"""Module for additional aggregator to collect related artist data from Spotify."""
2import json
4import requests
5from loguru import logger
8def get_related_artists(artists_ids, headers, stage=0):
9 """Function for finding similar artists by their artists_ids,
10 query headers and number of stages are used to access the service"""
11 artists_ids = list(set(artists_ids))
12 total_artists = len(artists_ids)
14 artists_ids_list = []
15 artists_ids_list.extend(artists_ids)
16 process_num = 0
17 for artist_id in artists_ids:
18 process_num += 1
19 logger.debug(f'Collect {process_num} of {total_artists}')
20 response = requests.get(
21 f'https://api.spotify.com/v1/artists/{artist_id}/related-artists',
22 headers=headers,
23 timeout=10
24 )
26 for artist in response.json().get('artists'):
27 artists_ids_list.append(artist.get('id'))
29 logger.debug(f'==============STAGE={stage}=============')
30 logger.info(f' Total related artists: {len(artists_ids_list)}')
31 logger.info(f'Total unique related artists: {len(set(artists_ids_list))}')
32 logger.debug('==================================')
34 artists_ids_list = list(set(artists_ids_list))
36 return artists_ids_list
39def artist_aggregate_main(source_file_path, id_file_path, headers):
40 """Main function for data aggregation."""
41 logger.info('Starting artist aggregation')
42 with open(
43 source_file_path,
44 mode='r',
45 encoding='utf-8'
46 ) as file:
47 artists_data = json.load(file)
49 followed_artists_ids = []
50 for artist in artists_data.get('artists'):
51 followed_artists_ids.append(artist.get('uri').split(':')[2])
53 logger.info(f"Artist aggregation stage one")
54 artists_ids_list_stage_one = get_related_artists(
55 followed_artists_ids,
56 headers=headers,
57 stage=1
58 )
60 logger.info(f"Artist aggregation stage two")
61 artists_ids_list = get_related_artists(
62 artists_ids_list_stage_one,
63 headers=headers,
64 stage=2
65 )
67 json_string = json.dumps(
68 artists_ids_list,
69 indent=4,
70 ensure_ascii=False
71 )
73 with open(
74 id_file_path,
75 mode='w',
76 encoding='utf-8'
77 ) as file:
78 file.write(json_string)
79 logger.info(f"Artist aggregation complete")
81# if __name__ == '__main__':
82# artist_aggregate_main(
83# source_file_path='src/aggregator/resources/spotify-followed-artists.json',
84# id_file_path='src/aggregator/resources/artists-ids-list.json',
85# headers={'Authorization': f'{token_type} {access_token}'}
86# )