cal/major-domo-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
f8716c69cb
major-domo-legacy/db_calls.py
Cal Corum e54cf885ad Migrated to rotating log handler 
Added random_gif()
Moved back from exception-handler cog to local error handling
Updated !keepers to be season agnostic
Added new !sync param to update and clear local guild
Added error checking to !player command
2024-12-11 14:52:28 -06:00

1063 lines
38 KiB
Python
Raw Blame History

import requests
import logging
import aiohttp
import os
from bs4 import BeautifulSoup
import csv
AUTH_TOKEN = {'Authorization': f'Bearer {os.environ.get("API_TOKEN")}'}
# DB_URL = 'http://database/api'
DB_URL = 'https://sba.manticorum.com/api'
master_debug = True
logger = logging.getLogger('discord_app')
def param_char(other_params):
if other_params:
return '&'
else:
return '?'
def get_req_url(endpoint: str, api_ver: int = 3, object_id: int = None, params: list = None):
# Checking for hard-coded url
if '/api/' in endpoint:
return endpoint
req_url = f'{DB_URL}/v{api_ver}/{endpoint}{"/" if object_id is not None else ""}{object_id if object_id is not None else ""}'
if params is not None:
other_params = False
for x in params:
req_url += f'{param_char(other_params)}{x[0]}={x[1]}'
other_params = True
return req_url
def log_return_value(log_string: str):
if master_debug:
logger.info(f'return: {log_string[:1200]}{" [ S N I P P E D ]" if len(log_string) > 1200 else ""}\n')
else:
logger.debug(f'return: {log_string[:1200]}{" [ S N I P P E D ]" if len(log_string) > 1200 else ""}\n')
async def db_get(endpoint: str, api_ver: int = 3, object_id: int = None, params: list = None, none_okay: bool = True,
timeout: int = 3):
req_url = get_req_url(endpoint, api_ver=api_ver, object_id=object_id, params=params)
log_string = f'get:\n{endpoint} id: {object_id} params: {params}'
logger.info(log_string) if master_debug else logger.debug(log_string)
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as session:
async with session.get(req_url) as r:
if r.status == 200:
js = await r.json()
log_return_value(f'{js}')
return js
elif none_okay:
e = await r.text()
logger.error(e)
return None
else:
e = await r.text()
logger.error(e)
raise ValueError(f'DB: {e}')
async def db_patch(
endpoint: str, object_id: int, params: list, api_ver: int = 3, timeout: int = 3, payload: dict = None):
req_url = get_req_url(endpoint, api_ver=api_ver, object_id=object_id, params=params)
log_string = f'patch:\n{endpoint}/{object_id} {params}'
logger.info(log_string) if master_debug else logger.debug(log_string)
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as session:
async with session.patch(req_url, json=payload) as r:
if r.status == 200:
js = await r.json()
log_return_value(f'{js}')
return js
else:
e = await r.text()
logger.error(e)
raise ValueError(f'DB: {e}')
async def db_post(endpoint: str, api_ver: int = 3, payload: dict = None, timeout: int = 3):
req_url = get_req_url(endpoint, api_ver=api_ver)
log_string = f'post:\n{endpoint} payload: {payload}\ntype: {type(payload)}'
logger.info(log_string) if master_debug else logger.debug(log_string)
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as session:
async with session.post(req_url, json=payload) as r:
if r.status == 200:
js = await r.json()
log_return_value(f'{js}')
return js
else:
e = await r.text()
logger.error(e)
raise ValueError(f'DB: {e}')
async def db_put(endpoint: str, api_ver: int = 3, payload: dict = None, object_id: int = None, timeout: int = 3):
req_url = get_req_url(endpoint, api_ver=api_ver, object_id=object_id)
log_string = f'put:{req_url}\npayload: {payload}\nobject_id: {object_id}\ntype: {type(payload)}'
logger.info(log_string) if master_debug else logger.debug(log_string)
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as session:
async with session.put(req_url, json=payload) as r:
if r.status == 200:
js = await r.json()
log_return_value(f'{js}')
return js
else:
e = await r.text()
logger.error(e)
raise ValueError(f'DB: {e}')
# retries = 0
# while True:
# try:
# resp = requests.put(req_url, json=payload, headers=AUTH_TOKEN, timeout=timeout)
# break
# except requests.Timeout as e:
# logger.error(f'Post Timeout: {req_url} / retries: {retries} / timeout: {timeout}')
# if retries > 1:
# raise ConnectionError(f'DB: The internet was a bit too slow for me to grab the data I needed. Please '
# f'hang on a few extra seconds and try again.')
# timeout += [min(3, timeout), min(5, timeout)][retries]
# retries += 1
#
# if resp.status_code == 200:
# data = resp.json()
# log_string = f'{data}'
# if master_debug:
# logger.info(f'return: {log_string[:1200]}{" [ S N I P P E D ]" if len(log_string) > 1200 else ""}')
# else:
# logger.debug(f'return: {log_string[:1200]}{" [ S N I P P E D ]" if len(log_string) > 1200 else ""}')
# return data
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
async def db_delete(endpoint: str, object_id: int, api_ver: int = 3, timeout=3):
req_url = get_req_url(endpoint, api_ver=api_ver, object_id=object_id)
log_string = f'delete:\n{endpoint} {object_id}'
logger.info(log_string) if master_debug else logger.debug(log_string)
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as session:
async with session.delete(req_url) as r:
if r.status == 200:
js = await r.json()
log_return_value(f'{js}')
return js
else:
e = await r.text()
logger.error(e)
raise ValueError(f'DB: {e}')
async def get_team_by_abbrev(team_abbrev: str, season: int):
t_query = await db_get('teams', params=[('season', season), ('team_abbrev', team_abbrev)])
if t_query['count'] == 0:
return None
else:
return t_query['teams'][0]
async def get_team_by_owner(season, owner_id):
params = [('active_only', True), ('season', season), ('owner_id', owner_id)]
resp = await db_get('teams', params=params)
if resp['count'] > 0:
return resp['teams'][0]
else:
return None
async def get_player_by_name(season: int, player_name: str):
p_query = await db_get('players', params=[('name', player_name), ('season', season)])
if p_query['count'] == 0:
return None
player = p_query['players'][0]
if player['demotion_week'] is None:
player['demotion_week'] = 0
await put_player(player)
return player
async def put_player(this_player: dict):
this_player['team_id'] = this_player['team']['id']
return await db_put('players', object_id=this_player['id'], payload=this_player)
async def patch_draftpick(this_pick: dict):
logger.info(f'attempting to patch draftpick:\n{this_pick}')
this_pick['player_id'] = None if this_pick['player'] is None else this_pick['player']['id']
this_pick['origowner_id'] = None if this_pick['origowner'] is None else this_pick['origowner']['id']
this_pick['owner_id'] = None if this_pick['owner'] is None else this_pick['owner']['id']
return await db_patch('draftpicks', object_id=this_pick['id'], params=[], payload=this_pick)
async def get_player_photo(player_name):
req_url = f'https://www.thesportsdb.com/api/v1/json/1/searchplayers.php?p={player_name}'
try:
resp = requests.get(req_url, timeout=.5)
except Exception as e:
return None
if resp.status_code == 200 and resp.json()['player']:
if resp.json()['player'][0]['strSport'] == 'Baseball':
return resp.json()['player'][0]['strThumb']
return None
async def get_player_headshot(player_name):
req_url = f'https://www.baseball-reference.com/search/search.fcgi?search={player_name}'
try:
resp = requests.get(req_url, timeout=1).text
except Exception as e:
return None
soup = BeautifulSoup(resp, 'html.parser')
for item in soup.find_all('img'):
if 'headshot' in item['src']:
return item['src']
return await get_player_photo(player_name)
###
# TO BE DEPRECATED FUNCTIONS
###
# async def get_current(season=None):
# if season is not None:
# params = [('season', season)]
# else:
# params = []
# return await db_get('current', params=params)
#
#
# async def patch_current(
# week=None, freeze=None, season=None, transcount=None, bstatcount=None, pstatcount=None, bet_week=None,
# trade_deadline=None, pick_trade_start=None, pick_trade_end=None, injury_count=None):
# params = []
# if week is not None:
# params.append(('week', week))
# if freeze is not None:
# params.append(('freeze', freeze))
# if season is not None:
# params.append(('season', season))
# if transcount is not None:
# params.append(('transcount', transcount))
# if bstatcount is not None:
# params.append(('bstatcount', bstatcount))
# if pstatcount is not None:
# params.append(('pstatcount', pstatcount))
# if bet_week is not None:
# params.append(('bet_week', bet_week))
# if trade_deadline is not None:
# params.append(('trade_deadline', trade_deadline))
# if pick_trade_start is not None:
# params.append(('pick_trade_start', pick_trade_start))
# if pick_trade_end is not None:
# params.append(('pick_trade_end', pick_trade_end))
# if injury_count is not None:
# params.append(('injury_count', injury_count))
#
# resp = requests.patch(get_req_url('current', params=params), data=None, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_one_team(id_or_abbrev, season=None, is_pd=False, timeout=3):
# if season is not None:
# params = [('season', season)]
# else:
# params = None
#
# return await db_get('teams', object_id=id_or_abbrev, params=params)
#
#
# async def get_team_roster(team, current_or_next):
# resp = requests.get(f'{DB_URL}/v1/teams/{team["id"]}/roster/{current_or_next}', timeout=3)
# if resp.status_code == 200:
# return resp.json()
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_team_record(team, week_num):
# resp = requests.get(f'{DB_URL}/v1/teams/{team["id"]}/record/{week_num}', timeout=3)
# if resp.status_code == 200:
# return resp.json()
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def patch_team(
# team, manager=None, gmid=None, gmid2=None, mascot=None, stadium=None, thumbnail=None, color=None,
# dice_color=None):
# req_url = f'{DB_URL}/v1/teams/{team["id"]}'
# other_params = False
# if manager:
# req_url += f'{param_char(other_params)}manager={manager}'
# other_params = True
# if gmid:
# req_url += f'{param_char(other_params)}gmid={gmid}'
# other_params = True
# if gmid2:
# req_url += f'{param_char(other_params)}gmid2={gmid2}'
# other_params = True
# if mascot:
# req_url += f'{param_char(other_params)}mascot={mascot}'
# other_params = True
# if stadium:
# req_url += f'{param_char(other_params)}stadium={stadium}'
# other_params = True
# if thumbnail:
# req_url += f'{param_char(other_params)}thumbnail={thumbnail}'
# other_params = True
# if color:
# req_url += f'{param_char(other_params)}color={color}'
# other_params = True
# if dice_color:
# req_url += f'{param_char(other_params)}dice_color={dice_color}'
# other_params = True
#
# resp = requests.patch(req_url, data=None, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_schedule(
# season, team_abbrev1=None, team_abbrev2=None, away_abbrev=None, home_abbrev=None, week_start=None,
# week_end=None):
# req_url = f'{DB_URL}/v1/schedules?season={season}'
# if team_abbrev1:
# req_url += f'&team_abbrev1={team_abbrev1}'
# if team_abbrev2:
# req_url += f'&team_abbrev2={team_abbrev2}'
# if away_abbrev:
# req_url += f'&away_abbrev={away_abbrev}'
# if home_abbrev:
# req_url += f'&home_abbrev={home_abbrev}'
# if week_start:
# req_url += f'&week_start={week_start}'
# if week_end:
# req_url += f'&week_end={week_end}'
#
# return await db_get(req_url)
#
#
# async def get_one_schedule(
# season, team_abbrev1=None, team_abbrev2=None, away_abbrev=None, home_abbrev=None, week=None):
# req_url = f'{DB_URL}/v1/schedules?season={season}'
# if team_abbrev1:
# req_url += f'&team_abbrev1={team_abbrev1}'
# if team_abbrev2:
# req_url += f'&team_abbrev2={team_abbrev2}'
# if away_abbrev:
# req_url += f'&away_abbrev={away_abbrev}'
# if home_abbrev:
# req_url += f'&home_abbrev={home_abbrev}'
# if week:
# req_url += f'&week_start={week}'
# req_url += f'&week_end={week}'
#
# resp = await db_get(req_url)
# if len(resp) > 2:
# raise ValueError(f'One schedule requested, but {len(resp)} were returned')
# elif len(resp) == 0:
# raise ValueError(f'No schedules found')
# else:
# key = [*resp]
# return_val = resp[key[0]]
# return return_val
#
#
# async def get_players(season, team_abbrev=None, sort=None, injured=None):
# req_url = f'{DB_URL}/v2/players?season={season}'
# if team_abbrev:
# req_url += f'&team_abbrev={team_abbrev}'
# if sort:
# req_url += f'&sort={sort}'
# if injured:
# req_url += f'&injured={injured}'
#
# return await db_get(req_url)
#
#
# async def get_one_player(id_or_name, season=None):
# req_url = f'{DB_URL}/v1/players/{id_or_name}'
# if season is not None:
# req_url += f'?season={season}'
#
# return await db_get(req_url)
#
#
# # async def patch_player(
# # pid, name=None, wara=None, image=None, image2=None, team_id=None, season=None, pitcher_injury=None, pos_1=None,
# # pos_2=None, pos_3=None, pos_4=None, pos_5=None, pos_6=None, pos_7=None, pos_8=None, last_game=None,
# # last_game2=None, il_return=None, demotion_week=None, headshot=None, vanity_card=None, injury_rating=None):
# # """
# # For values being patched to "None", set to False
# #
# # :param pid:
# # :param name:
# # :param wara:
# # :param image:
# # :param image2:
# # :param team_id:
# # :param season:
# # :param pitcher_injury:
# # :param pos_1:
# # :param pos_2:
# # :param pos_3:
# # :param pos_4:
# # :param pos_5:
# # :param pos_6:
# # :param pos_7:
# # :param pos_8:
# # :param last_game:
# # :param last_game2:
# # :param il_return:
# # :return:
# # """
# # req_url = f'{DB_URL}/v2/players/{pid}'
# # other_params = False
# # if name is not None:
# # req_url += f'{param_char(other_params)}name={name}'
# # other_params = True
# # if wara is not None:
# # req_url += f'{param_char(other_params)}wara={wara}'
# # other_params = True
# # if image is not None:
# # req_url += f'{param_char(other_params)}image={image}'
# # other_params = True
# # if image2 is not None:
# # if not image2:
# # req_url += f'{param_char(other_params)}image2=False'
# # else:
# # req_url += f'{param_char(other_params)}image2={image2}'
# # other_params = True
# # if team_id is not None:
# # req_url += f'{param_char(other_params)}team_id={team_id}'
# # other_params = True
# # if season is not None:
# # req_url += f'{param_char(other_params)}season={season}'
# # other_params = True
# # if pitcher_injury is not None:
# # if not pitcher_injury:
# # req_url += f'{param_char(other_params)}pitcher_injury=False'
# # else:
# # req_url += f'{param_char(other_params)}pitcher_injury={pitcher_injury}'
# # other_params = True
# # if pos_1 is not None:
# # req_url += f'{param_char(other_params)}pos_1={pos_1}'
# # other_params = True
# # if pos_2 is not None:
# # if not pos_2:
# # req_url += f'{param_char(other_params)}pos_2=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_2={pos_2}'
# # other_params = True
# # if pos_3 is not None:
# # if not pos_3:
# # req_url += f'{param_char(other_params)}pos_3=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_3={pos_3}'
# # other_params = True
# # if pos_4 is not None:
# # if not pos_4:
# # req_url += f'{param_char(other_params)}pos_4=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_4={pos_4}'
# # other_params = True
# # if pos_5 is not None:
# # if not pos_5:
# # req_url += f'{param_char(other_params)}pos_5=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_5={pos_5}'
# # other_params = True
# # if pos_6 is not None:
# # if not pos_6:
# # req_url += f'{param_char(other_params)}pos_6=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_6={pos_6}'
# # other_params = True
# # if pos_7 is not None:
# # if not pos_7:
# # req_url += f'{param_char(other_params)}pos_7=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_7={pos_7}'
# # other_params = True
# # if pos_8 is not None:
# # if not pos_8:
# # req_url += f'{param_char(other_params)}pos_8=False'
# # else:
# # req_url += f'{param_char(other_params)}pos_8={pos_8}'
# # other_params = True
# # if last_game is not None:
# # if not last_game:
# # req_url += f'{param_char(other_params)}last_game=False'
# # else:
# # req_url += f'{param_char(other_params)}last_game={last_game}'
# # other_params = True
# # if last_game2 is not None:
# # req_url += f'{param_char(other_params)}last_game2={last_game2}'
# # other_params = True
# # if il_return is not None:
# # if not il_return:
# # req_url += f'{param_char(other_params)}il_return=false'
# # else:
# # req_url += f'{param_char(other_params)}il_return={il_return}'
# # other_params = True
# # if demotion_week is not None:
# # req_url += f'{param_char(other_params)}demotion_week={demotion_week}'
# # other_params = True
# # if headshot is not None:
# # if not headshot:
# # req_url += f'{param_char(other_params)}headshot=false'
# # else:
# # req_url += f'{param_char(other_params)}headshot={headshot}'
# # other_params = True
# # if vanity_card is not None:
# # if not vanity_card:
# # req_url += f'{param_char(other_params)}vanity_card=false'
# # else:
# # req_url += f'{param_char(other_params)}vanity_card={vanity_card}'
# # other_params = True
# # if injury_rating is not None:
# # req_url += f'{param_char(other_params)}injury_rating={injury_rating}'
# # other_params = True
# #
# # resp = requests.patch(req_url, headers=AUTH_TOKEN, timeout=3)
# # if resp.status_code == 200:
# # return True
# # else:
# # logger.warning(resp.text)
# # raise ValueError(f'DB: {resp.text}')
#
#
# async def get_standings(season=None, team_abbrev=None, league_abbrev=None, division_abbrev=None):
# req_url = f'{DB_URL}/v1/standings'
# other_params = False
# if season:
# req_url += f'{param_char(other_params)}season={season}'
# other_params = True
# if team_abbrev:
# req_url += f'{param_char(other_params)}team_abbrev={team_abbrev}'
# other_params = True
# if league_abbrev:
# req_url += f'{param_char(other_params)}league_abbrev={league_abbrev}'
# other_params = True
# if division_abbrev:
# req_url += f'{param_char(other_params)}division_abbrev={division_abbrev}'
# other_params = True
#
# return await db_get(req_url)
#
#
# async def get_results(season, team_abbrev=None, week=None, away_abbrev=None, home_abbrev=None):
# req_url = f'{DB_URL}/v1/results?season={season}'
# if team_abbrev:
# req_url += f'&team_abbrev={team_abbrev}'
# if week:
# req_url += f'&week={week}'
# if away_abbrev:
# req_url += f'&away_abbrev={away_abbrev}'
# if home_abbrev:
# req_url += f'&home_abbrev={home_abbrev}'
#
# return await db_get(req_url)
#
#
# async def get_transactions(
# season, team_abbrev=None, week_start=None, week_end=None, cancelled=None, frozen=None,
# player_name=None, player_id=None, move_id=None, timeout=3):
# req_url = f'{DB_URL}/v1/transactions?season={season}'
# if team_abbrev:
# req_url += f'&team_abbrev={team_abbrev}'
# if week_start:
# req_url += f'&week_start={week_start}'
# if week_end:
# req_url += f'&week_end={week_end}'
# if cancelled:
# req_url += f'&cancelled={cancelled}'
# if frozen:
# req_url += f'&frozen={frozen}'
# if player_name:
# req_url += f'&player_name={player_name}'
# if player_id:
# req_url += f'&player_id={player_id}'
# if move_id:
# req_url += f'&move_id={move_id}'
#
# return await db_get(req_url)
#
#
# async def post_transactions(moves: list):
# req_url = f'{DB_URL}/v1/transactions'
# data = {
# 'count': len(moves),
# 'moves': moves
# }
# payload = json.dumps(data)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def patch_transaction(move_id, frozen=None, cancelled=None):
# req_url = f'{DB_URL}/v1/transactions/{move_id}'
# other_params = False
# if frozen is not None:
# req_url += f'{param_char(other_params)}frozen={frozen}'
# other_params = True
# if cancelled is not None:
# req_url += f'{param_char(other_params)}cancelled={cancelled}'
# other_params = True
#
# resp = requests.patch(req_url, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_one_battingseason(player_id):
# req_url = f'{DB_URL}/v1/battingseasons/{player_id}'
#
# return await db_get(req_url)
#
#
# async def get_one_pitchingseason(player_id):
# req_url = f'{DB_URL}/v1/pitchingseasons/{player_id}'
#
# return await db_get(req_url)
#
#
# async def get_one_fieldingseason(player_id):
# req_url = f'{DB_URL}/v1/fieldingseasons/{player_id}'
#
# return await db_get(req_url)
#
#
# async def get_battingcareer(name_or_id):
# req_url = f'{DB_URL}/v1/battingcareer/{name_or_id}'
#
# return await db_get(req_url)
#
#
# async def get_pitchingcareer(name_or_id):
# req_url = f'{DB_URL}/v1/pitchingcareer/{name_or_id}'
#
# return await db_get(req_url)
#
#
# async def get_battingstat(
# season, s_type='regular', team_abbrev=None, player_name=None, player_id=None, week_start=None,
# week_end=None, game_num=None, position=None, timeout=10):
# req_url = f'{DB_URL}/v1/battingstats/s{season}/{s_type}'
# other_params = False
# if team_abbrev:
# req_url += f'{param_char(other_params)}team_abbrev={team_abbrev}'
# other_params = True
# if player_name:
# req_url += f'{param_char(other_params)}player_name={player_name}'
# other_params = True
# if player_id:
# req_url += f'{param_char(other_params)}player_id={player_id}'
# other_params = True
# if week_start:
# req_url += f'{param_char(other_params)}week_start={week_start}'
# other_params = True
# if week_end:
# req_url += f'{param_char(other_params)}week_end={week_end}'
# other_params = True
# if game_num:
# req_url += f'{param_char(other_params)}game_num={game_num}'
# other_params = True
# if position:
# req_url += f'{param_char(other_params)}position={position}'
# other_params = True
#
# return await db_get(req_url, timeout=timeout)
#
#
# async def post_battingstats(stats: list):
# req_url = f'{DB_URL}/v1/battingstats'
# data = {
# 'count': len(stats),
# 'stats': stats
# }
# payload = json.dumps(data)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def delete_battingstats(game_obj: dict):
# req_url = f'{DB_URL}/v1/battingstats'
# payload = json.dumps(game_obj)
#
# resp = requests.delete(req_url, data=payload, headers=AUTH_TOKEN)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def recalc_batting_seasons(season, team_id):
# req_url = f'{DB_URL}/v1/battingseasons/recalculate?season={season}&team_id={team_id}'
#
# resp = requests.post(req_url, headers=AUTH_TOKEN, timeout=45)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def recalc_fielding_seasons(season, team_id):
# req_url = f'{DB_URL}/v1/fieldingseasons/recalculate?season={season}&team_id={team_id}'
#
# resp = requests.post(req_url, headers=AUTH_TOKEN, timeout=45)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_pitchingstat(
# season, s_type='regular', team_abbrev=None, player_name=None, player_id=None, week_start=None,
# week_end=None, game_num=None, timeout=10):
# req_url = f'{DB_URL}/v1/pitchingstats/s{season}/{s_type}'
# other_params = False
# if team_abbrev:
# req_url += f'{param_char(other_params)}team_abbrev={team_abbrev}'
# other_params = True
# if player_name:
# req_url += f'{param_char(other_params)}player_name={player_name}'
# other_params = True
# if player_id:
# req_url += f'{param_char(other_params)}player_id={player_id}'
# other_params = True
# if week_start:
# req_url += f'{param_char(other_params)}week_start={week_start}'
# other_params = True
# if game_num:
# req_url += f'{param_char(other_params)}game_num={game_num}'
# other_params = True
# if week_end:
# req_url += f'{param_char(other_params)}week_end={week_end}'
# other_params = True
#
# return await db_get(req_url, timeout=timeout)
#
#
# async def post_pitchingstats(stats: list):
# req_url = f'{DB_URL}/v1/pitchingstats'
# data = {
# 'count': len(stats),
# 'stats': stats
# }
# payload = json.dumps(data)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def delete_pitchingstats(game_obj: dict):
# req_url = f'{DB_URL}/v1/pitchingstats'
# payload = json.dumps(game_obj)
#
# resp = requests.delete(req_url, data=payload, headers=AUTH_TOKEN)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def recalc_pitching_seasons(season, team_id):
# req_url = f'{DB_URL}/v1/pitchingseasons/recalculate?season={season}&team_id={team_id}'
#
# resp = requests.post(req_url, headers=AUTH_TOKEN, timeout=45)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_draftdata():
# return await db_get('draftdata')
#
#
# async def patch_draftdata(
# currentpick=None, timer=None, pick_deadline=None, result_channel=None, ping_channel=None, pick_minutes=None):
# req_url = f'{DB_URL}/v1/draftdata'
# other_params = False
# if currentpick is not None:
# req_url += f'{param_char(other_params)}currentpick={currentpick}'
# other_params = True
# if timer is not None:
# req_url += f'{param_char(other_params)}timer={timer}'
# other_params = True
# if pick_deadline is not None:
# req_url += f'{param_char(other_params)}pick_deadline={pick_deadline}'
# other_params = True
# if result_channel is not None:
# req_url += f'{param_char(other_params)}result_channel={result_channel}'
# other_params = True
# if ping_channel is not None:
# req_url += f'{param_char(other_params)}ping_channel={ping_channel}'
# other_params = True
# if pick_minutes is not None:
# req_url += f'{param_char(other_params)}pick_minutes={pick_minutes}'
# other_params = True
#
# resp = requests.patch(req_url, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_draftpicks(
# season, owner_team=None, orig_owner_team=None, round_start=None, round_end=None, team_season=None,
# overall_start=None, overall_end=None):
# req_url = f'{DB_URL}/v1/draftpicks?season={season}'
# if owner_team:
# req_url += f'&owner_team_abbrev={owner_team["abbrev"]}'
# if orig_owner_team:
# req_url += f'&orig_team_abbrev={orig_owner_team["abbrev"]}'
# if round_start:
# req_url += f'&pick_round_start={round_start}'
# if round_end:
# req_url += f'&pick_round_end={round_end}'
# if team_season:
# req_url += f'&team_season={team_season}'
# if overall_start:
# req_url += f'&overall_start={overall_start}'
# if overall_end:
# req_url += f'&overall_end={overall_end}'
#
# return await db_get(req_url)
#
#
# async def get_one_draftpick_search(season, orig_owner_abbrev, round_num: int, team_season: int = None):
# req_url = f'{DB_URL}/v1/draftpicks?season={season}'
# req_url += f'&orig_team_abbrev={orig_owner_abbrev}'
# req_url += f'&pick_round_start={round_num}'
# req_url += f'&pick_round_end={round_num}'
# if team_season is not None:
# req_url += f'&team_season={team_season}'
#
# picks = await db_get(req_url)
# if len(picks) == 0:
# return None
# elif len(picks) == 1:
# for x in picks:
# return picks[x]
#
#
# async def get_one_draftpick_byid(pick_id: int):
# req_url = f'{DB_URL}/v1/draftpicks/{pick_id}'
#
# return await db_get(req_url)
#
#
# async def get_one_draftpick_byoverall(season: int, overall: int):
# req_url = f'{DB_URL}/v1/draftpicks?season={season}&overall={overall}'
#
# resp = await db_get(req_url)
# if len(resp) != 1:
# return None
# else:
# key = [*resp]
# return resp[key[0]]
#
#
# async def patch_draftpick(
# pick_id: int, overall: int = None, round: int = None, orig_owner_id: int = None, owner_id: int = None,
# season: int = None, player_id: int = None):
# req_url = f'{DB_URL}/v1/draftpicks/{pick_id}'
# other_params = False
# if overall is not None:
# req_url += f'{param_char(other_params)}overall={overall}'
# other_params = True
# if round is not None:
# req_url += f'{param_char(other_params)}round={round}'
# other_params = True
# if orig_owner_id is not None:
# req_url += f'{param_char(other_params)}orig_owner_id={orig_owner_id}'
# other_params = True
# if owner_id is not None:
# req_url += f'{param_char(other_params)}owner_id={owner_id}'
# other_params = True
# if season is not None:
# req_url += f'{param_char(other_params)}season={season}'
# other_params = True
# if player_id is not None:
# req_url += f'{param_char(other_params)}player_id={player_id}'
# other_params = True
#
# resp = requests.patch(req_url, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_awards(player_id=None, player_name=None, season=None):
# req_url = f'{DB_URL}/v1/awards'
# other_params = False
# if player_id:
# req_url += f'{param_char(other_params)}player_id={player_id}'
# other_params = True
# if player_name:
# req_url += f'{param_char(other_params)}player_name={player_name}'
# other_params = True
# if season:
# req_url += f'{param_char(other_params)}season={season}'
# other_params = True
#
# resp = requests.get(req_url, timeout=3)
# if resp.status_code == 200:
# return resp.json()
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def post_award(award: dict):
# req_url = f'{DB_URL}/v1/awards'
# award_model = {
# 'name': award['name'],
# 'season': award['season'],
# 'timing': award['timing'],
# }
# if award['manager1']:
# award_model['manager1_id'] = award['manager1']['id']
# if award['manager2']:
# award_model['manager2_id'] = award['manager2']['id']
# if award['player']:
# award_model['player_id'] = award['player']['id']
# if award['team']:
# award_model['team_id'] = award['team']['id']
# if award['image']:
# award_model['image'] = award['image']
#
# payload_prep = {'count': 1, 'awards': [award_model]}
# payload = json.dumps(payload_prep)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_one_manager(manager_name):
# req_url = f'{DB_URL}/v1/managers/{manager_name}'
#
# return await db_get(req_url)
#
#
# async def post_dice(dice_bag: list):
# req_url = f'{DB_URL}/v1/dice'
# dice_model = {
# 'count': len(dice_bag),
# 'rolls': dice_bag
# }
#
# payload = json.dumps(dice_model)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_player_photo(player_name):
# req_url = f'https://www.thesportsdb.com/api/v1/json/1/searchplayers.php?p={player_name}'
#
# try:
# resp = requests.get(req_url, timeout=.5)
# except Exception as e:
# return None
# if resp.status_code == 200 and resp.json()['player']:
# if resp.json()['player'][0]['strSport'] == 'Baseball':
# return resp.json()['player'][0]['strThumb']
# return None
#
#
# async def get_player_headshot(player_name):
# req_url = f'https://www.baseball-reference.com/search/search.fcgi?search={player_name}'
#
# try:
# resp = requests.get(req_url, timeout=1).text
# except Exception as e:
# return None
# soup = BeautifulSoup(resp, 'html.parser')
# for item in soup.find_all('img'):
# if 'headshot' in item['src']:
# return item['src']
# return await get_player_photo(player_name)
#
#
# async def post_draft_list(this_list: list):
# req_url = f'{DB_URL}/v1/draft-list'
# list_model = {
# 'count': len(this_list),
# 'draft_list': this_list
# }
#
# payload = json.dumps(list_model)
#
# resp = requests.post(req_url, payload, headers=AUTH_TOKEN, timeout=3)
# if resp.status_code == 200:
# return True
# else:
# logger.warning(resp.text)
# raise ValueError(f'DB: {resp.text}')
#
#
# async def get_draft_list(team):
# req_url = f'{DB_URL}/v1/draft-list/{team["id"]}'
#
# return await db_get(req_url)
Reference in New Issue View Git Blame Copy Permalink