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

1"""Module for additional aggregator to collect related artist data from Spotify.""" 

2import json 

3 

4import requests 

5from loguru import logger 

6 

7 

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) 

13 

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 ) 

25 

26 for artist in response.json().get('artists'): 

27 artists_ids_list.append(artist.get('id')) 

28 

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('==================================') 

33 

34 artists_ids_list = list(set(artists_ids_list)) 

35 

36 return artists_ids_list 

37 

38 

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) 

48 

49 followed_artists_ids = [] 

50 for artist in artists_data.get('artists'): 

51 followed_artists_ids.append(artist.get('uri').split(':')[2]) 

52 

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 ) 

59 

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 ) 

66 

67 json_string = json.dumps( 

68 artists_ids_list, 

69 indent=4, 

70 ensure_ascii=False 

71 ) 

72 

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") 

80 

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# )