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.static.static_values_enum import Edition 

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

14 

15 

16def get_season_played(): 

17 season_played = store_util.get_seasons_played_list() 

18 return season_played 

19 

20 

21def first_played_season(): 

22 season_played = store_util.get_seasons_played_list() 

23 first_season = '' 

24 if len(season_played) > 0: 

25 first_season = season_played[-1] 

26 return first_season 

27 

28 

29def get_season_end_date(season_id): 

30 if season_id != '': 

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

32 from_date = parser.parse(season_end_date) 

33 return from_date 

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

35 

36 

37def generate_season_hive_blog(season_id, users): 

38 progress_util.set_season_title('Generate hive blog') 

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

40 

41 season_info_store = { 

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

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

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

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

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

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

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

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

50 'season'), 

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

52 'season') 

53 } 

54 

55 start_date, end_date = season_balances_info.get_start_end_time_season(season_id) 

56 tournaments_info_dict = {} 

57 purchases_dict = {} 

58 sold_dict = {} 

59 last_season_rewards_dict = {} 

60 last_season_league_rewards_dict = {} 

61 for account_name in users: 

62 # get tournament information 

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

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

65 start_date, 

66 end_date) 

67 

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

69 

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

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

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

73 

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

75 account_name, 

76 transactions) 

77 

78 # get last season rewards 

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

80 last_season_rewards_dict[account_name] = get_last_season_reward_draws(transactions) 

81 last_season_league_rewards_dict[account_name] = get_last_season_league_rewards(transactions) 

82 

83 # print single post for each account 

84 report = hive_blog.write_blog_post(users, 

85 season_info_store, 

86 last_season_rewards_dict, 

87 last_season_league_rewards_dict, 

88 tournaments_info_dict, 

89 purchases_dict, 

90 sold_dict, 

91 season_id) 

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

93 progress_util.update_season_msg('Done') 

94 return report 

95 

96 

97def get_last_season_reward_draws(transactions): 

98 df = pd.DataFrame() 

99 for transaction in transactions: 

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

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

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

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

104 if 'result' in data: 

105 reward_result = data['result'] 

106 if reward_result['success']: 

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

108 temp_df['sub_type'] = trx['sub_type'] 

109 df = extract_card_info(df, temp_df) 

110 else: 

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

112 

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

114 df.reset_index(drop=True) 

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

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

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

118 df.loc[not_na_index, 'edition_name'] = df.loc[not_na_index, 'edition'].apply(lambda x: Edition(x).name) 

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

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

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

122 return df 

123 

124 

125def get_last_season_league_rewards(transactions): 

126 df = pd.DataFrame() 

127 for transaction in transactions: 

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

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

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

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

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

133 league_format = data['format'] 

134 tier = data['tier'] 

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

136 for sub_type in types: 

137 rewards = trx['rewards'][sub_type] 

138 if rewards: 

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

140 temp_df['format'] = league_format 

141 temp_df['tier'] = tier 

142 temp_df['sub_type'] = sub_type 

143 df = extract_card_info(df, temp_df) 

144 

145 else: 

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

147 

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

149 df.reset_index(drop=True) 

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

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

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

153 df.loc[not_na_index, 'edition_name'] = df.loc[not_na_index, 'edition'].apply(lambda x: Edition(x).name) 

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

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

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

157 return df 

158 

159 

160def extract_card_info(df, input_df): 

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

162 # Expand 'card' column into multiple columns 

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

164 # Concatenate temp_df and card_df along columns axis 

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

166 df = pd.concat([df, input_df]) 

167 return df