Coverage for src / utils / spl_util.py: 0%

1import logging 

2 

3import pandas as pd 

4from dateutil import parser 

5from dateutil.parser import isoparse 

6from pandas import json_normalize 

7 

8from src.api import spl 

9from src.utils import progress_util, store_util 

10 

11 

12def get_unclaimed_sps_balance_history_for_token(username, start_date=None): 

13 limit = 1000 

14 offset = None 

15 

16 msg_prefix = 'SPS UNCLAIMED (' + str(username) + ') ' 

17 token_params = store_util.get_token_dict(username) 

18 

19 total_items = 0 

20 complete_result = [] 

21 while True: 

22 

23 data = spl.get_unclaimed_sps_balance_history_for_token_impl( 

24 offset=offset, 

25 limit=limit, 

26 token_params=token_params 

27 ) 

28 

29 if data: 

30 print('sps_unclaimed: ' + str(data[-1]['created_date'])) 

31 complete_result += data 

32 offset = data[-1]["id"] 

33 total_items += len(data) 

34 progress_util.update_season_msg( 

35 msg_prefix + 

36 ' get unclaimed sps balance history total found items: ' + 

37 str(total_items)) 

38 

39 created_date = parser.parse(complete_result[-1]['created_date']) 

40 if start_date and created_date < start_date: 

41 progress_util.update_season_msg( 

42 msg_prefix + 

43 ' last pull contains all season information data from ' + 

44 str(start_date) + ' till NOW') 

45 break 

46 else: 

47 progress_util.update_season_msg( 

48 msg_prefix + 

49 ': last pull contains no data assume all data is collected ' + 

50 str(start_date) + ' till NOW') 

51 break 

52 progress_util.update_season_msg(msg_prefix + ': all data pulled') 

53 return complete_result 

54 

55 

56def get_balance_history_for_token(username, token='DEC', start_date=None): 

57 limit = 1000 

58 

59 token_params = store_util.get_token_dict(username) 

60 

61 msg_prefix = str(token) + ' (' + str(username) + ') ' 

62 complete_result = [] 

63 from_date = None # from is none the spl api will make it to current date 

64 last_update_date = None # Not needed for the first call 

65 

66 # Background information from Investygator 

67 # The reason there's 2 dates is that the "from" (or, created_date in thedata) is essentially the block date, 

68 # and the last_update_date is when it was actually written to the DB. 

69 # 

70 # Since multiple balance updates can happen in the same transaction, sometimes they will have the same "from" date, 

71 # which can lead to items getting cut off with a limit. 

72 # 

73 # The last_update_date should always be distinct, and ensures that you don't get skipped results. 

74 # However, created_date is the only one indexed, so it's needed for performance reasons. 

75 while True: 

76 data = spl.get_balance_history_for_token_impl_v2( 

77 token=token, 

78 from_date=from_date, 

79 last_update_date=last_update_date, 

80 limit=limit, 

81 token_params=token_params 

82 ) 

83 

84 if data: 

85 update_message(data, msg_prefix) 

86 

87 complete_result += data 

88 # Update the parameters for the next request 

89 from_date = data[-1]["created_date"] 

90 last_update_date = data[-1]["last_update_date"] 

91 

92 if start_date and parser.parse(from_date) < start_date: 

93 progress_util.update_season_msg( 

94 msg_prefix + 

95 ': last pull contains all season information data from ' + 

96 str(start_date) + ' till NOW') 

97 break 

98 

99 else: 

100 progress_util.update_season_msg( 

101 msg_prefix + 

102 ': last pull contains no data assume all data is collected ' + 

103 str(start_date) + ' till NOW') 

104 break 

105 

106 return complete_result 

107 

108 

109def update_message(data, msg_prefix): 

110 date_str = data[-1]['created_date'] 

111 date_obj = isoparse(date_str) 

112 formatted_date = date_obj.strftime('%Y-%m-%d %H:%M:%S') 

113 

114 progress_util.update_season_msg( 

115 msg_prefix + 'get balance history: ' + 

116 str(formatted_date) 

117 ) 

118 

119 

120def get_battle_history_df(account): 

121 return spl.get_battle_history_df(account, store_util.get_token_dict(account)) 

122 

123 

124def is_season_reward_claimed(account, season_id): 

125 df = spl.get_player_history_season_rewards_df(store_util.get_token_dict(account)) 

126 if df.empty: 

127 # in this case there are not season rewards found at all assume inactive account or rental account 

128 # proceed processing balances 

129 logging.info('No season rewards found at all for account: ' + str(account)) 

130 logging.info('Assume inactive account or rental account continue processing season for : ' + str(account)) 

131 return True 

132 

133 if df.loc[df.season == season_id].empty: 

134 logging.info('Season results not claimed yet for account: ' + str(account)) 

135 logging.info('Stop season processing for: ' + str(account)) 

136 return False 

137 

138 logging.info('Continue season results are claimed for account: ' + str(account)) 

139 return True 

140 

141 

142def get_ability_list(): 

143 cards = spl.get_card_details() 

144 abilities_df = json_normalize(cards['stats']).abilities.dropna() 

145 flattened_abilities = [ability for sublist in abilities_df for ability in sublist if sublist] 

146 

147 unique_abilities = {ability for sublist in flattened_abilities for ability in 

148 (sublist if isinstance(sublist, list) else [sublist])} 

149 

150 series = pd.Series(list(unique_abilities)) 

151 return series.sort_values()