587 lines
26 KiB
Python
587 lines
26 KiB
Python
import datetime
|
|
import logging
|
|
import math
|
|
|
|
import pydantic
|
|
from sqlalchemy import func
|
|
from api_calls import db_get, db_post
|
|
from in_game.gameplay_models import CACHE_LIMIT, Card, CardBase, Lineup, Player, PlayerBase, Session, Team, TeamBase, select, or_, Game, Play
|
|
from exceptions import DatabaseError, log_exception, PlayNotFoundException
|
|
|
|
|
|
logger = logging.getLogger('discord_app')
|
|
|
|
|
|
class DecisionModel(pydantic.BaseModel):
|
|
game_id: int
|
|
season: int
|
|
week: int
|
|
pitcher_id: int
|
|
pitcher_team_id: int
|
|
win: int = 0
|
|
loss: int = 0
|
|
hold: int = 0
|
|
is_save: int = 0
|
|
is_start: bool = False
|
|
b_save: int = 0
|
|
irunners: int = 0
|
|
irunners_scored: int = 0
|
|
rest_ip: float = 0
|
|
rest_required: int = 0
|
|
game_finished: int = 0
|
|
|
|
|
|
def get_games_by_channel(session: Session, channel_id: int) -> list[Game]:
|
|
logger.info(f'Getting games in channel {channel_id}')
|
|
return session.exec(select(Game).where(Game.channel_id == channel_id, Game.active)).all()
|
|
|
|
|
|
def get_channel_game_or_none(session: Session, channel_id: int) -> Game | None:
|
|
logger.info(f'Getting one game from channel {channel_id}')
|
|
all_games = get_games_by_channel(session, channel_id)
|
|
if len(all_games) > 1:
|
|
err = 'Too many games found in get_channel_game_or_none'
|
|
logger.error(f'cogs.gameplay - get_channel_game_or_none - channel_id: {channel_id} / {err}')
|
|
raise Exception(err)
|
|
elif len(all_games) == 0:
|
|
return None
|
|
return all_games[0]
|
|
|
|
|
|
def get_active_games_by_team(session: Session, team: Team) -> list[Game]:
|
|
logger.info(f'Getting game for team {team.lname}')
|
|
return session.exec(select(Game).where(Game.active, or_(Game.away_team_id == team.id, Game.home_team_id == team.id))).all()
|
|
|
|
|
|
async def get_team_or_none(
|
|
session: Session, team_id: int | None = None, gm_id: int | None = None, team_abbrev: str | None = None, skip_cache: bool = False) -> Team | None:
|
|
logger.info(f'Getting team or none / team_id: {team_id} / gm_id: {gm_id} / team_abbrev: {team_abbrev} / skip_cache: {skip_cache}')
|
|
if team_id is None and gm_id is None and team_abbrev is None:
|
|
err = 'One of "team_id", "gm_id", or "team_abbrev" must be included in search'
|
|
logger.error(f'gameplay_models - get_team - {err}')
|
|
raise TypeError(err)
|
|
|
|
if not skip_cache:
|
|
if team_id is not None:
|
|
this_team = session.get(Team, team_id)
|
|
else:
|
|
if gm_id is not None:
|
|
statement = select(Team).where(Team.gmid == gm_id)
|
|
else:
|
|
statement = select(Team).where(func.lower(Team.abbrev) == team_abbrev.lower())
|
|
this_team = session.exec(statement).one_or_none()
|
|
|
|
if this_team is not None:
|
|
logger.debug(f'we found a team: {this_team} / created: {this_team.created}')
|
|
tdelta = datetime.datetime.now() - this_team.created
|
|
logger.debug(f'tdelta: {tdelta}')
|
|
if tdelta.total_seconds() < CACHE_LIMIT:
|
|
return this_team
|
|
else:
|
|
session.delete(this_team)
|
|
session.commit()
|
|
|
|
def cache_team(json_data: dict) -> Team:
|
|
# logger.info(f'gameplay_models - get_team - cache_team - writing a team to cache: {json_data}')
|
|
valid_team = TeamBase.model_validate(json_data, from_attributes=True)
|
|
# logger.info(f'gameplay_models - get_team - cache_team - valid_team: {valid_team}')
|
|
db_team = Team.model_validate(valid_team)
|
|
# logger.info(f'gameplay_models - get_team - cache_team - db_team: {db_team}')
|
|
session.add(db_team)
|
|
session.commit()
|
|
session.refresh(db_team)
|
|
return db_team
|
|
|
|
if team_id is not None:
|
|
t_query = await db_get('teams', object_id=team_id, params=[('inc_packs', False)])
|
|
if t_query is not None:
|
|
return cache_team(t_query)
|
|
|
|
elif gm_id is not None:
|
|
t_query = await db_get('teams', params=[('gm_id', gm_id)])
|
|
if t_query['count'] != 0:
|
|
for team in [x for x in t_query['teams'] if 'gauntlet' not in x['abbrev'].lower()]:
|
|
return cache_team(team)
|
|
|
|
elif team_abbrev is not None:
|
|
t_query = await db_get('teams', params=[('abbrev', team_abbrev)])
|
|
if t_query['count'] != 0:
|
|
for team in [x for x in t_query['teams'] if 'gauntlet' not in x['abbrev'].lower()]:
|
|
return cache_team(team)
|
|
|
|
return None
|
|
|
|
|
|
async def get_player_or_none(session: Session, player_id: int, skip_cache: bool = False) -> Player | None:
|
|
logger.info(f'gameplay_models - get_player_or_none - player_id: {player_id}')
|
|
if not skip_cache:
|
|
this_player = session.get(Player, player_id)
|
|
|
|
if this_player is not None:
|
|
logger.info(f'we found a cached player: {this_player} / created: {this_player.created}')
|
|
tdelta = datetime.datetime.now() - this_player.created
|
|
logger.debug(f'tdelta: {tdelta}')
|
|
if tdelta.total_seconds() < CACHE_LIMIT:
|
|
return this_player
|
|
else:
|
|
session.delete(this_player)
|
|
session.commit()
|
|
|
|
def cache_player(json_data: dict) -> Player:
|
|
logger.info(f'gameplay_models - get_player_or_none - cache_player - caching player data: {json_data}')
|
|
valid_player = PlayerBase.model_validate(json_data, from_attributes=True)
|
|
db_player = Player.model_validate(valid_player)
|
|
session.add(db_player)
|
|
session.commit()
|
|
session.refresh(db_player)
|
|
return db_player
|
|
|
|
p_query = await db_get('players', object_id=player_id, params=[('inc_dex', False)])
|
|
if p_query is not None:
|
|
if 'id' not in p_query:
|
|
p_query['id'] = p_query['player_id']
|
|
if 'name' not in p_query:
|
|
p_query['name'] = p_query['p_name']
|
|
return cache_player(p_query)
|
|
|
|
return None
|
|
|
|
|
|
def get_player_id_from_dict(json_data: dict) -> int:
|
|
logger.info(f'Getting player from dict {json_data}')
|
|
if 'player_id' in json_data:
|
|
return json_data['player_id']
|
|
elif 'id' in json_data:
|
|
return json_data['id']
|
|
log_exception(KeyError, 'Player ID could not be extracted from json data')
|
|
|
|
|
|
def get_player_name_from_dict(json_data: dict) -> str:
|
|
logger.info(f'Getting player from dict {json_data}')
|
|
if 'name' in json_data:
|
|
return json_data['name']
|
|
elif 'p_name' in json_data:
|
|
return json_data['p_name']
|
|
log_exception(KeyError, 'Player name could not be extracted from json data')
|
|
|
|
|
|
async def get_or_create_ai_card(session: Session, player: Player, team: Team, skip_cache: bool = False, dev_mode: bool = False) -> Card:
|
|
logger.info(f'Getting or creating card for {player.name_with_desc} on the {team.sname}')
|
|
if not team.is_ai:
|
|
err = f'Cannot create AI cards for human teams'
|
|
logger.error(f'gameplay_models - get_or_create_ai_card: {err}')
|
|
raise TypeError(err)
|
|
|
|
logger.info(f'gameplay_models - get_or_create_ai_card - player.id: {player.id} / team.id: {team.id}')
|
|
if not skip_cache:
|
|
c_query = session.exec(select(Card).where(Card.player == player, Card.team == team)).all()
|
|
|
|
if len(c_query) > 0:
|
|
this_card = c_query[0]
|
|
logger.info(f'we found a cached card: {this_card} / created: {this_card.created}')
|
|
tdelta = datetime.datetime.now() - this_card.created
|
|
logger.debug(f'tdelta: {tdelta}')
|
|
if tdelta.total_seconds() < CACHE_LIMIT:
|
|
return this_card
|
|
else:
|
|
session.delete(this_card)
|
|
session.commit()
|
|
|
|
async def pull_card(p: Player, t: Team):
|
|
c_query = await db_get('cards', params=[('team_id', t.id), ('player_id', p.id)])
|
|
if c_query['count'] > 0:
|
|
json_data = c_query['cards'][0]
|
|
logger.info(f'gameplay_models - get_or_create_ai_card - pull_card - caching json_data: {json_data}')
|
|
json_data['team_id'] = json_data['team']['id']
|
|
json_data['player_id'] = get_player_id_from_dict(json_data['player'])
|
|
valid_card = CardBase.model_validate(c_query['cards'][0], from_attributes=True)
|
|
db_card = Card.model_validate(valid_card)
|
|
session.add(db_card)
|
|
session.commit()
|
|
session.refresh(db_card)
|
|
return db_card
|
|
else:
|
|
return None
|
|
|
|
this_card = await pull_card(player, team)
|
|
if this_card is not None:
|
|
return this_card
|
|
|
|
logger.info(f'gameplay_models - get_or_create_ai_card: creating {player.description} {player.name} card for {team.abbrev}')
|
|
|
|
if dev_mode:
|
|
this_card = Card(player=player, team=team)
|
|
session.add(this_card)
|
|
session.commit()
|
|
session.refresh(this_card)
|
|
return this_card
|
|
|
|
await db_post(
|
|
'cards',
|
|
payload={'cards': [
|
|
{'player_id': player.id, 'team_id': team.id, 'pack_id': 1}
|
|
]}
|
|
)
|
|
|
|
this_card = await pull_card(player, team)
|
|
if this_card is not None:
|
|
return this_card
|
|
|
|
err = f'Could not create {player.name} card for {team.abbrev}'
|
|
logger.error(f'gameplay_models - get_or_create_ai_card - {err}')
|
|
raise LookupError(err)
|
|
|
|
|
|
async def get_card_or_none(session: Session, card_id: int, skip_cache: bool = False) -> Card | None:
|
|
logger.info(f'Getting card {card_id}')
|
|
if not skip_cache:
|
|
this_card = session.get(Card, card_id)
|
|
|
|
if this_card is not None:
|
|
logger.info(f'we found a cached card: {this_card} / created: {this_card.created}')
|
|
tdelta = datetime.datetime.now() - this_card.created
|
|
logger.debug(f'tdelta: {tdelta}')
|
|
if tdelta.total_seconds() < CACHE_LIMIT:
|
|
return this_card
|
|
else:
|
|
session.delete(this_card)
|
|
session.commit()
|
|
|
|
def cache_card(json_data: dict) -> Card:
|
|
valid_card = CardBase.model_validate(json_data, from_attributes=True)
|
|
db_card = Card.model_validate(valid_card)
|
|
session.add(db_card)
|
|
session.commit()
|
|
session.refresh(db_card)
|
|
return db_card
|
|
|
|
c_query = await db_get('cards', object_id=card_id)
|
|
if c_query is not None:
|
|
c_query['team_id'] = c_query['team']['id']
|
|
c_query['player_id'] = get_player_id_from_dict(c_query['player'])
|
|
|
|
this_player = await get_player_or_none(session, player_id=c_query['player_id'])
|
|
this_team = await get_team_or_none(session, team_id=c_query['team_id'])
|
|
|
|
if this_player is None:
|
|
raise LookupError(f'Player ID {c_query["player_id"]} not found during card check')
|
|
if this_team is None:
|
|
raise LookupError(f'Team ID {c_query["team_id"]} not found during card check')
|
|
|
|
return cache_card(c_query)
|
|
|
|
return None
|
|
|
|
|
|
def get_game_lineups(session: Session, this_game: Game, specific_team: Team = None, is_active: bool = None) -> list[Lineup]:
|
|
logger.info(f'Getting lineups for game {this_game.id} / specific_team: {specific_team} / is_active: {is_active}')
|
|
st = select(Lineup).where(Lineup.game == this_game)
|
|
|
|
if specific_team is not None:
|
|
st = st.where(Lineup.team == specific_team)
|
|
if is_active is not None:
|
|
st = st.where(Lineup.active == is_active)
|
|
|
|
return session.exec(st).all()
|
|
|
|
|
|
def get_players_last_pa(session: Session, lineup_member: Lineup, none_okay: bool = False):
|
|
logger.info(f'Getting last AB for {lineup_member.player.name_with_desc} on the {lineup_member.team.lname}')
|
|
last_pa = session.exec(select(Play).where(Play.game == lineup_member.game, Play.batter == lineup_member).order_by(Play.id.desc()).limit(1)).all()
|
|
if len(last_pa) == 1:
|
|
return last_pa[0]
|
|
else:
|
|
if none_okay:
|
|
return None
|
|
else:
|
|
log_exception(PlayNotFoundException, f'No play found for {lineup_member.player.name_with_desc}\'s last AB')
|
|
|
|
|
|
def get_one_lineup(session: Session, this_game: Game, this_team: Team, active: bool = True, position: str = None, batting_order: int = None) -> Lineup:
|
|
logger.info(f'Getting one lineup / this_game: {this_game.id} / this_team: {this_team.lname} / active: {active}, position: {position}, batting_order: {batting_order}')
|
|
if position is None and batting_order is None:
|
|
raise KeyError('Position or batting order must be provided for get_one_lineup')
|
|
|
|
st = select(Lineup).where(Lineup.game == this_game, Lineup.team == this_team, Lineup.active == active)
|
|
if position is not None:
|
|
st = st.where(Lineup.position == position)
|
|
else:
|
|
st = st.where(Lineup.batting_order == batting_order)
|
|
|
|
return session.exec(st).one()
|
|
|
|
|
|
def get_last_team_play(session: Session, this_game: Game, this_team: Team, none_okay: bool = False):
|
|
logger.info(f'Getting last play for the {this_team.lname} in game {this_game.id}')
|
|
last_play = session.exec(select(Play).join(Lineup, onclause=Lineup.id == Play.batter_id).where(Play.game == this_game, Lineup.team == this_team).order_by(Play.id.desc()).limit(1)).all()
|
|
|
|
if len(last_play) == 1:
|
|
return last_play[0]
|
|
else:
|
|
if none_okay:
|
|
return None
|
|
else:
|
|
log_exception(PlayNotFoundException, f'No last play found for the {this_team.sname}')
|
|
|
|
|
|
def get_sorted_lineups(session: Session, this_game: Game, this_team: Team) -> list[Lineup]:
|
|
logger.info(f'Getting sorted lineups for the {this_team.lname} in game {this_game.id}')
|
|
custom_order = {'P': 1, 'C': 2, '1B': 3, '2B': 4, '3B': 5, 'SS': 6, 'LF': 7, 'CF': 8, 'RF': 9}
|
|
|
|
all_lineups = session.exec(select(Lineup).where(Lineup.game == this_game, Lineup.active == True, Lineup.team == this_team)).all()
|
|
|
|
sorted_lineups = sorted(all_lineups, key=lambda x: custom_order.get(x.position, float('inf')))
|
|
return sorted_lineups
|
|
|
|
|
|
def get_db_ready_plays(session: Session, this_game: Game, db_game_id: int):
|
|
logger.info(f'Getting db ready plays for game {this_game.id}')
|
|
all_plays = session.exec(select(Play).where(Play.game == this_game)).all()
|
|
|
|
obc_list = ['000', '001', '010', '100', '011', '101', '110', '111']
|
|
|
|
return_plays = []
|
|
for play in all_plays:
|
|
dump = play.model_dump()
|
|
dump['game_id'] = db_game_id
|
|
dump['on_base_code'] = obc_list[play.on_base_code]
|
|
dump['batter_id'] = play.batter.player.id
|
|
dump['pitcher_id'] = play.pitcher.player.id
|
|
dump['catcher_id'] = play.catcher.player.id
|
|
if 'runner_id' in dump and dump['runner_id'] is not None:
|
|
dump['runner_id'] = play.runner.player.id
|
|
if 'defender_id' in dump and dump['defender_id'] is not None:
|
|
dump['defender_id'] = play.defender.player.id
|
|
if 'on_first_id' in dump and dump['on_first_id'] is not None:
|
|
dump['on_first_id'] = play.on_first.player.id
|
|
if 'on_second_id' in dump and dump['on_second_id'] is not None:
|
|
dump['on_second_id'] = play.on_second.player.id
|
|
if 'on_third_id' in dump and dump['on_third_id'] is not None:
|
|
dump['on_third_id'] = play.on_third.player.id
|
|
return_plays.append(dump)
|
|
|
|
return {'plays': return_plays}
|
|
|
|
|
|
def get_db_ready_decisions(session: Session, this_game: Game, db_game_id: int) -> list[DecisionModel]:
|
|
logger.info(f'Game {this_game.id} | Getting db ready decisions for game {this_game.id}')
|
|
winner = None
|
|
loser = None
|
|
save = None
|
|
away_starter = None
|
|
home_starter = None
|
|
away_finisher = None
|
|
home_finisher = None
|
|
b_save = []
|
|
holds = []
|
|
|
|
away_pitcher = None
|
|
home_pitcher = None
|
|
|
|
decisions = {
|
|
# { <player_id>: DecisionModel }
|
|
}
|
|
|
|
final_inning = session.exec(select(func.max(Play.inning_num)).where(Play.game == this_game)).one()
|
|
|
|
# Get starting pitchers and update this as a pointer for the play crawl
|
|
for play in session.exec(select(Play).where(Play.game == this_game)).all():
|
|
logger.info(f'Game {this_game.id} | Crawling play #{play.play_num}')
|
|
runs_scored = 0
|
|
if play.inning_half == 'top':
|
|
if home_starter is None:
|
|
logger.info(f'Game {this_game.id} | Setting home starter to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
home_starter = play.pitcher
|
|
|
|
if home_finisher is None:
|
|
logger.info(f'Game {this_game.id} | Setting home finisher to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
home_finisher = play.pitcher
|
|
|
|
if home_pitcher != play.pitcher:
|
|
logger.info(f'Game {this_game.id} | Setting home pitcher to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
home_pitcher = play.pitcher
|
|
|
|
if save == play.pitcher:
|
|
if play.home_score > play.away_score:
|
|
if play.pitcher not in holds:
|
|
logger.info(f'Game {this_game.id} | Appending {play.pitcher.player.name_with_desc} to holds on play #{play.play_num}')
|
|
holds.append(play.pitcher)
|
|
else:
|
|
if play.pitcher not in b_save:
|
|
logger.info(f'Game {this_game.id} | Appending {play.pitcher.player.name_with_desc} to blown saves on play #{play.play_num}')
|
|
b_save.append(play.pitcher)
|
|
|
|
elif play.home_score > play.away_score and play.home_score - play.away_score <= 3 and home_pitcher != home_starter and play.inning_num + 2 >= final_inning:
|
|
logger.info(f'Game {this_game.id} | Setting {play.pitcher.player.name_with_desc} to save on play #{play.play_num}')
|
|
save = home_pitcher
|
|
|
|
elif play.inning_half == 'bot':
|
|
if away_starter is None:
|
|
logger.info(f'Game {this_game.id} | Setting away starter to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
away_starter = play.pitcher
|
|
|
|
if away_finisher is None:
|
|
logger.info(f'Game {this_game.id} | Setting away finisher to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
away_finisher = play.pitcher
|
|
|
|
if away_pitcher != play.pitcher:
|
|
logger.info(f'Game {this_game.id} | Setting away pitcher to {play.pitcher.player.name_with_desc} on play #{play.play_num}')
|
|
away_pitcher = play.pitcher
|
|
|
|
if save == play.pitcher:
|
|
if play.away_score > play.home_score:
|
|
if play.pitcher not in holds:
|
|
logger.info(f'Game {this_game.id} | Appending {play.pitcher.player.name_with_desc} to holds on play #{play.play_num}')
|
|
holds.append(play.pitcher)
|
|
else:
|
|
if play.pitcher not in b_save:
|
|
logger.info(f'Game {this_game.id} | Appending {play.pitcher.player.name_with_desc} to blown saves on play #{play.play_num}')
|
|
b_save.append(play.pitcher)
|
|
|
|
if play.is_go_ahead:
|
|
run_diff = play.home_score - play.away_score
|
|
for x in [play.on_first_final, play.on_second_final, play.on_third_final, play.batter_final]:
|
|
runs_scored += 1 if x == 4 else 0
|
|
|
|
if play.inning_half == 'top':
|
|
run_diff -= runs_scored
|
|
else:
|
|
run_diff += runs_scored
|
|
|
|
logger.info(f'run_diff for go-ahead: {run_diff}')
|
|
logger.info(f'go-ahead play: {play}')
|
|
count = 1
|
|
|
|
for runner, dest in [(play.on_third, play.on_third_final), (play.on_second, play.on_second_final), (play.on_first, play.on_first_final), (play.batter, play.batter_final)]:
|
|
logger.info(f'Game {this_game.id} | Looking for go-ahead runner / runner, dest: {runner}, {dest} / count: {count}')
|
|
if dest == 4 and count == abs(run_diff):
|
|
winning_play = get_players_last_pa(session, runner)
|
|
|
|
loser = winning_play.pitcher
|
|
logger.info(f'Game {this_game.id} | Setting loser to {loser} on play #{play.play_num}')
|
|
|
|
if save == loser:
|
|
logger.info(f'Game {this_game.id} | Appending {loser} to blown saves on play #{play.play_num}')
|
|
b_save.append(loser)
|
|
|
|
winner = home_pitcher if play.inning_half == 'bot' else away_pitcher
|
|
logger.info(f'Game {this_game.id} | Setting winner to {winner} on play #{play.play_num}')
|
|
break
|
|
|
|
count += 1
|
|
|
|
if winner is None:
|
|
logger.info(f'Game {this_game.id} | Setting winner to {winner} by default on play #{play.play_num}')
|
|
winner = home_pitcher if play.inning_half == 'bot' else away_pitcher
|
|
|
|
if loser is None:
|
|
logger.info(f'Game {this_game.id} | Setting loser to {loser} by default on play #{play.play_num}')
|
|
loser = play.pitcher
|
|
|
|
if play.is_tied and runs_scored == 0:
|
|
logger.info(f'Game {this_game.id} | Clearing winner and loser on play #{play.play_num}')
|
|
winner, loser = None, None
|
|
|
|
if save is not None:
|
|
logger.info(f'Game {this_game.id} | Appending current save pitcher {save.player.name_with_desc} to blown saves and clearing save on play #{play.play_num}')
|
|
b_save.append(save)
|
|
save = None
|
|
|
|
if play.pitcher_id not in decisions:
|
|
logger.info(f'Game {this_game.id} | Adding {play.pitcher.player.name} to decisions dict on play #{play.play_num}')
|
|
decisions[play.pitcher.player_id] = DecisionModel(
|
|
game_id=db_game_id,
|
|
season=this_game.season,
|
|
week=0 if this_game.week is None else this_game.week,
|
|
pitcher_id=play.pitcher.player_id,
|
|
pitcher_team_id=play.pitcher.team_id
|
|
)
|
|
|
|
logger.info(f'winner: {winner} / loser: {loser}')
|
|
decisions[winner.player_id].win = 1
|
|
decisions[loser.player_id].loss = 1
|
|
decisions[away_starter.player_id].is_start = True
|
|
decisions[home_starter.player_id].is_start = True
|
|
decisions[away_finisher.player_id].game_finished = 1
|
|
decisions[home_finisher.player_id].game_finished = 1
|
|
|
|
for lineup in holds:
|
|
decisions[lineup.player_id].hold = 1
|
|
|
|
if save is not None:
|
|
decisions[save.player_id].is_save = 1
|
|
decisions[save.player_id].hold = 0
|
|
|
|
for lineup in b_save:
|
|
decisions[lineup.player_id].b_save = 1
|
|
decisions[lineup.player_id].save = 0
|
|
|
|
return [x.model_dump() for x in decisions.values()]
|
|
|
|
|
|
async def post_game_rewards(session: Session, winning_team: Team, losing_team: Team, this_game: Game):
|
|
wr_query = await db_get(
|
|
'gamerewards', params=[('name', f'{"Short" if this_game.short_game else "Full"} Game Win')])
|
|
lr_query = await db_get(
|
|
'gamerewards', params=[('name', f'{"Short" if this_game.short_game else "Full"} Game Loss')])
|
|
if not wr_query['count'] or not lr_query['count']:
|
|
raise DatabaseError(f'Game rewards were not found. Leaving this game active.')
|
|
|
|
win_reward = wr_query['gamerewards'][0]
|
|
loss_reward = lr_query['gamerewards'][0]
|
|
win_string = f'1x {win_reward["pack_type"]["name"]} Pack\n'
|
|
|
|
# Post Campaign Team Choice packs
|
|
if this_game.ai_team is not None and losing_team.is_ai and 'gauntlet' not in this_game.game_type and not this_game.short_game:
|
|
g_query = await db_get('games', params=[('team1_id', winning_team.id), ('season', this_game.season), ('forfeit', False)])
|
|
|
|
win_points = 0
|
|
for x in g_query['games']:
|
|
if (x['away_score'] > x['home_score'] and x['away_team']['id'] == winning_team.id) or (x['home_score'] > x['away_score'] and x['home_team']['id'] == winning_team.id):
|
|
if x['game_type'] == 'minor-league':
|
|
win_points += 1
|
|
|
|
elif x['game_type'] in ['major-league', 'flashback']:
|
|
win_points += 2
|
|
|
|
elif x['game_type'] == 'hall-of-fame':
|
|
win_points += 3
|
|
|
|
if this_game.game_type == 'minor-league':
|
|
this_game_points = 1
|
|
elif this_game.game_type in ['major-league', 'flashback']:
|
|
this_game_points = 2
|
|
else:
|
|
this_game_points = 3
|
|
|
|
pre_game_points = win_points - this_game_points
|
|
if math.floor(win_points / 6) > math.floor(pre_game_points / 6):
|
|
await db_post('packs/one', payload={
|
|
'team_id': winning_team.id,
|
|
'pack_type_id': 8,
|
|
'pack_team_id': losing_team.id
|
|
})
|
|
win_string += f'1x {losing_team.abbrev} Team Choice pack\n'
|
|
|
|
win_string += f'{win_reward["money"]}₼\n'
|
|
loss_string = f'{loss_reward["money"]}₼\n'
|
|
|
|
if 'gauntlet' in this_game.game_type:
|
|
winning_abbrev = winning_team.abbrev.lower()
|
|
if 'gauntlet' in winning_abbrev:
|
|
winning_team = await get_team_or_none(session, team_abbrev=winning_abbrev.split('-')[1])
|
|
if winning_team is None:
|
|
raise DatabaseError(f'Main team not found for {winning_abbrev}')
|
|
|
|
losing_abbrev = losing_team.abbrev.lower()
|
|
if 'gauntlet' in losing_abbrev:
|
|
losing_team = await get_team_or_none(session, team_abbrev=losing_abbrev.split('-')[1])
|
|
if losing_team is None:
|
|
raise DatabaseError(f'Main team not found for {losing_abbrev}')
|
|
|
|
await db_post('packs/one', payload={'team_id': winning_team.id, 'pack_type_id': win_reward['pack_type']['id']})
|
|
await db_post(f'teams/{winning_team.id}/money/{win_reward["money"]}')
|
|
await db_post(f'teams/{losing_team.id}/money/{loss_reward["money"]}')
|
|
|
|
return win_string, loss_string
|