Coverage for src / battle_store.py: 0%
116 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-01 10:28 +0000
1import json
2import logging
4import pandas as pd
6from src.configuration import config, store
7from src.static.static_values_enum import MatchType, Format
8from src.utils import store_util, progress_util, spl_util
11def get_uid_array(team):
12 uid_array = [team['summoner']['uid']]
13 for monster in team['monsters']:
14 uid_array.append(monster['uid'])
15 return uid_array
18def add_battle_store_big_my(account, team, battle, match_type):
19 match_format = get_battle_format(battle['format'])
20 created_date = battle['created_date']
21 mana_cap = battle['mana_cap']
22 rulesets = battle['ruleset']
23 inactive = battle['inactive']
24 battle_id = battle['battle_queue_id_1']
25 winner = battle['winner']
26 opponent = battle['player_2'] if battle['player_1'] == account else battle['player_1']
27 result = "win" if winner == account else "loss"
29 cards = list(team['monsters'])
30 cards.append(team['summoner'])
31 ruleset_split = ['None', 'None', 'None']
32 for idx, ruleset in enumerate(rulesets.split('|')):
33 ruleset_split[idx] = ruleset
35 for card in cards:
36 # add new row
37 card_id = card['card_detail_id']
38 card_name = config.card_details_df.loc[card_id]['name']
39 df = pd.DataFrame({'card_detail_id': card_id,
40 'card_name': card_name,
41 'card_type': config.card_details_df.loc[card_id]['type'],
42 'rarity': config.card_details_df.loc[card_id]['rarity'],
43 'color': config.card_details_df.loc[card_id]['color'],
44 'secondary_color': config.card_details_df.loc[card_id]['secondary_color'],
45 'xp': card['xp'] if 'xp' in card else None,
46 'gold': card['gold'],
47 'level': card['level'],
48 'edition': card['edition'],
49 'account': account,
50 'opponent': opponent,
51 'created_date': created_date,
52 'match_type': match_type,
53 'format': match_format,
54 'mana_cap': mana_cap,
55 'ruleset1': ruleset_split[0],
56 'ruleset2': ruleset_split[1],
57 'ruleset3': ruleset_split[2],
58 'inactive': inactive,
59 'battle_id': battle_id,
60 'winner': winner,
61 'result': result,
62 }, index=[0])
63 store.battle_big = pd.concat([store.battle_big, df], ignore_index=True)
66def log_battle_note(count):
67 limit = 50
68 if count == limit:
69 logging.info(str(limit) + ' or more battles to process consider running this program more often')
70 logging.info('SPL API limits ' + str(limit) + ' battle history')
71 else:
72 logging.info(str(count) + ' battles to process')
75def add_rating_log(account, battle):
76 if battle['player_1'] == account:
77 final_rating = battle['player_1_rating_final']
78 else:
79 final_rating = battle['player_2_rating_final']
81 match_format = get_battle_format(battle['format'])
83 df = pd.DataFrame({'created_date': battle['created_date'],
84 'account': account,
85 'rating': final_rating,
86 'format': match_format,
87 }, index=[0])
88 store.rating = pd.concat([store.rating, df], ignore_index=True)
91def get_battle_format(battle_format):
92 # format = null when wild
93 if battle_format:
94 return battle_format
95 else:
96 return Format.wild.value
99def add_losing_battle_team(account, team, battle):
100 match_type = battle['match_type']
101 match_format = get_battle_format(battle['format'])
102 created_date = battle['created_date']
103 mana_cap = battle['mana_cap']
104 rulesets = battle['ruleset']
105 inactive = battle['inactive']
106 battle_id = battle['battle_queue_id_1']
107 opponent = battle['player_2'] if battle['player_1'] == account else battle['player_1']
109 cards = list(team['monsters'])
110 cards.append(team['summoner'])
111 ruleset_split = ['None', 'None', 'None']
112 for idx, ruleset in enumerate(rulesets.split('|')):
113 ruleset_split[idx] = ruleset
115 for card in cards:
116 # add new row
117 card_id = card['card_detail_id']
118 card_name = config.card_details_df.loc[card_id]['name']
119 df = pd.DataFrame({'card_detail_id': card_id,
120 'card_name': card_name,
121 'card_type': config.card_details_df.loc[card_id]['type'],
122 'rarity': config.card_details_df.loc[card_id]['rarity'],
123 'color': config.card_details_df.loc[card_id]['color'],
124 'secondary_color': config.card_details_df.loc[card_id]['secondary_color'],
125 'xp': card['xp'] if 'xp' in card else None,
126 'gold': card['gold'],
127 'level': card['level'],
128 'edition': card['edition'],
129 'account': account,
130 'created_date': created_date,
131 'match_type': match_type,
132 'format': match_format,
133 'mana_cap': mana_cap,
134 'ruleset1': ruleset_split[0],
135 'ruleset2': ruleset_split[1],
136 'ruleset3': ruleset_split[2],
137 'inactive': inactive,
138 'battle_id': battle_id,
139 'opponent': opponent,
140 }, index=[0])
141 store.losing_big = pd.concat([store.losing_big, df], ignore_index=True)
144def get_battles_to_process(account):
145 battle_history = spl_util.get_battle_history_df(account)
146 if battle_history is not None and \
147 not battle_history.empty and \
148 not store.last_processed.empty and \
149 not store.last_processed.loc[(store.last_processed.account == account)].empty:
150 # filter out already processed
151 last_processed_date = \
152 store.last_processed.loc[(store.last_processed.account == account)].last_processed.values[0]
153 battle_history = battle_history.loc[(battle_history['created_date'] > last_processed_date)]
154 return battle_history
157def process_battle(account):
158 battle_history = get_battles_to_process(account)
159 if battle_history is not None:
160 log_battle_note(len(battle_history.index))
161 if not battle_history.empty:
162 for index, battle in battle_history.iterrows():
163 match_type = battle['match_type']
165 battle_details = battle.details
166 if not is_surrender(battle_details) and not is_training(battle):
167 winner_name = battle_details['winner']
169 if battle_details['team1']['player'] == account:
170 my_team = battle_details['team1']
171 opponent_team = battle_details['team2']
172 else:
173 my_team = battle_details['team2']
174 opponent_team = battle_details['team1']
176 if 'is_brawl' in battle_details and battle_details['is_brawl']:
177 match_type = MatchType.BRAWL.value
179 add_battle_store_big_my(account,
180 my_team,
181 battle, match_type)
183 # If a ranked match also log the rating
184 if match_type and match_type == MatchType.RANKED.value:
185 add_rating_log(account, battle)
187 # If the battle is lost store losing battle
188 if winner_name != account:
189 add_losing_battle_team(account,
190 opponent_team,
191 battle)
192 else:
193 logging.debug("Surrender match skip")
195 last_processed_date = \
196 battle_history.sort_values(by='created_date', ascending=False)['created_date'].iloc[0]
197 update_last_processed_df(account, last_processed_date)
198 else:
199 logging.debug('No battles to process.')
200 else:
201 logging.debug('None battles to process.')
204def update_last_processed_df(account, last_processed_date):
205 if store.last_processed.empty or store.last_processed.loc[(store.last_processed.account == account)].empty:
206 # create
207 new = pd.DataFrame({'account': [account],
208 'last_processed': [last_processed_date]},
209 index=[0])
210 store.last_processed = pd.concat([store.last_processed, new], ignore_index=True)
211 else:
212 store.last_processed.loc[
213 (store.last_processed.account == account),
214 'last_processed'] = last_processed_date
217def is_surrender(battle_details):
218 return 'type' in battle_details and battle_details['type'] == 'Surrender'
221def is_training(battle):
222 return (battle['match_type'] == MatchType.CHALLENGE.value
223 and json.loads(battle['settings'])['is_training'] == 'true')
226def process_battles():
227 progress_util.update_daily_msg("Start processing battles")
228 for account in store_util.get_account_names():
229 if store_util.get_token_dict(account):
230 progress_util.update_daily_msg("...processing: " + account)
231 process_battle(account)
232 else:
233 logging.info('Skip battle process... not token found. Check config page.')