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

1import datetime 

2import json 

3import logging 

4 

5import pandas as pd 

6from dateutil import parser 

7from dateutil.parser import isoparse 

8 

9from src import season_balances_info, market_info 

10from src.api import hive 

11from src.configuration import store, config 

12from src.utils import store_util, progress_util, hive_blog, tournaments_info 

13 

14 

15def get_season_played(): 

16 season_played = store_util.get_seasons_played_list() 

17 return season_played 

18 

19 

20def first_played_season(): 

21 season_played = store_util.get_seasons_played_list() 

22 first_season = '' 

23 if len(season_played) > 0: 

24 first_season = season_played[-1] 

25 return first_season 

26 

27 

28def get_season_end_date(season_id): 

29 if season_id != '': 

30 season_end_date = store.season_end_dates.loc[(store.season_end_dates.id == int(season_id) - 1)].end_date.iloc[0] 

31 from_date = parser.parse(season_end_date) 

32 return from_date 

33 return datetime.date(2018, 5, 26) # start of splinterlands 

34 

35 

36def generate_season_hive_blog(season_id, users): 

37 progress_util.set_season_title('Generate hive blog') 

38 progress_util.update_season_msg('Start collecting last season data') 

39 

40 season_info_store = { 

41 'sps': store_util.get_season_values(store.season_sps, season_id, users), 

42 'dec': store_util.get_season_values(store.season_dec, season_id, users), 

43 'merits': store_util.get_season_values(store.season_merits, season_id, users), 

44 'credits': store_util.get_season_values(store.season_credits, season_id, users), 

45 'vouchers': store_util.get_season_values(store.season_vouchers, season_id, users), 

46 'glint': store_util.get_season_values(store.season_glint, season_id, users), 

47 'unclaimed_sps': store_util.get_season_values(store.season_unclaimed_sps, season_id, users), 

48 'modern_battle': store_util.get_season_values(store.season_modern_battle_info, season_id, users, 

49 'season'), 

50 'wild_battle': store_util.get_season_values(store.season_wild_battle_info, season_id, users, 

51 'season') 

52 } 

53 

54 start_date, end_date = season_balances_info.get_start_end_time_season(season_id) 

55 tournaments_info_dict = {} 

56 purchases_dict = {} 

57 sold_dict = {} 

58 last_season_rewards_dict = {} 

59 last_season_league_rewards_dict = {} 

60 for account_name in users: 

61 # get tournament information 

62 progress_util.update_season_msg('Collecting tournament information for: ' + str(account_name)) 

63 tournaments_info_dict[account_name] = tournaments_info.get_tournaments_info(account_name, 

64 start_date, 

65 end_date) 

66 

67 progress_util.update_season_msg('Collecting bought and sold cards for: ' + str(account_name)) 

68 

69 from_date = isoparse(start_date).replace(tzinfo=None) 

70 till_date = isoparse(end_date).replace(tzinfo=None) 

71 transactions = hive.get_spl_transactions(account_name, from_date, till_date) 

72 

73 purchases_dict[account_name], sold_dict[account_name] = market_info.get_purchased_sold_cards( 

74 account_name, 

75 transactions) 

76 

77 # get last season rewards 

78 progress_util.update_season_msg('Collecting last season reward draws for: ' + str(account_name)) 

79 last_season_rewards_dict[account_name] = get_last_season_reward_draws(transactions) 

80 last_season_league_rewards_dict[account_name] = get_last_season_league_rewards(transactions) 

81 

82 # print single post for each account 

83 report = hive_blog.write_blog_post(users, 

84 season_info_store, 

85 last_season_rewards_dict, 

86 last_season_league_rewards_dict, 

87 tournaments_info_dict, 

88 purchases_dict, 

89 sold_dict, 

90 season_id) 

91 progress_util.set_season_title('Generate hive blog finished ') 

92 progress_util.update_season_msg('Done') 

93 return report 

94 

95 

96def get_last_season_reward_draws(transactions): 

97 df = pd.DataFrame() 

98 for transaction in transactions: 

99 if transaction['operation'] == 'sm_purchase': 

100 trx = json.loads(transaction['trx_info']['result']) 

101 if trx['type'] in ['reward_draw', 'reward_merits']: 

102 data = json.loads(trx['data']) 

103 if 'result' in data: 

104 reward_result = data['result'] 

105 if reward_result['success']: 

106 temp_df = pd.DataFrame(reward_result['rewards']) 

107 temp_df['sub_type'] = trx['sub_type'] 

108 df = extract_card_info(df, temp_df) 

109 else: 

110 logging.info('Get last season reward draws, skipping transaction type: ' + trx['type']) 

111 

112 if not df.empty and 'card_detail_id' in df.columns: 

113 df.reset_index(drop=True) 

114 df.index = range(len(df)) 

115 not_na_index = df['card_detail_id'].notna() 

116 if not_na_index.any(): # Check if there are non-NaN values in 'card_detail_id' 

117 df.loc[not_na_index, 'bcx'] = df.loc[not_na_index, 'quantity'] 

118 df.loc[not_na_index, 'card_name'] = df.loc[not_na_index, 'card_detail_id'].apply( 

119 lambda x: config.card_details_df.loc[x, 'name']) 

120 return df 

121 

122 

123def get_last_season_league_rewards(transactions): 

124 df = pd.DataFrame() 

125 for transaction in transactions: 

126 if (transaction['operation'] == 'sm_claim_reward' 

127 and transaction['trx_info']['success'] is True): 

128 data = json.loads(transaction['trx_info']['data']) 

129 trx = json.loads(transaction['trx_info']['result']) 

130 if data['type'] in ['league']: 

131 league_format = data['format'] 

132 tier = data['tier'] 

133 types = ['minor', 'major', 'ultimate'] 

134 for sub_type in types: 

135 rewards = trx['rewards'][sub_type] 

136 if rewards: 

137 temp_df = pd.DataFrame(rewards['result']['rewards']) 

138 temp_df['format'] = league_format 

139 temp_df['tier'] = tier 

140 temp_df['sub_type'] = sub_type 

141 df = extract_card_info(df, temp_df) 

142 

143 else: 

144 logging.info('Get last season league chest rewards, skipping transaction type: ' + data['type']) 

145 

146 if not df.empty and 'card_detail_id' in df.columns: 

147 df.reset_index(drop=True) 

148 df.index = range(len(df)) 

149 not_na_index = df['card_detail_id'].notna() 

150 if not_na_index.any(): # Check if there are non-NaN values in 'card_detail_id' 

151 df.loc[not_na_index, 'bcx'] = df.loc[not_na_index, 'quantity'] 

152 df.loc[not_na_index, 'card_name'] = df.loc[not_na_index, 'card_detail_id'].apply( 

153 lambda x: config.card_details_df.loc[x, 'name']) 

154 return df 

155 

156 

157def rename_duplicate_card_detail_id(df): 

158 cols = list(df.columns) 

159 idxs = [i for i, c in enumerate(cols) if c == "card_detail_id"] 

160 if len(idxs) <= 1: 

161 return df 

162 

163 # Example rule: if 'skin' exists, the FIRST occurrence is skin-related 

164 if 'skin' in df.columns: 

165 cols[idxs[0]] = "card_detail_id_skin" 

166 

167 df.columns = cols 

168 return df 

169 

170 

171def extract_card_info(df, input_df): 

172 if 'card' in input_df.columns.tolist(): 

173 # Expand 'card' column into multiple columns 

174 card_df = pd.json_normalize(input_df['card']) 

175 # Concatenate temp_df and card_df along columns axis 

176 input_df = pd.concat([input_df.drop(columns=['card']), card_df], axis=1) 

177 

178 input_df = rename_duplicate_card_detail_id(input_df) 

179 df = pd.concat([df, input_df], ignore_index=True) 

180 return df