paper-dynasty-card-creation/live_series_update.py

import asyncio
import datetime
import batters.creation
import pitchers.creation
import pandas as pd
import sys

from creation_helpers import pd_players_df, pd_positions_df, get_args
from db_calls import db_get, db_patch, DB_URL
from exceptions import logger
CARD_BASE_URL = f'{DB_URL}/v2/players'

SEASON = 2025
CARDSET_NAME = f'{SEASON} Live'
GAMES_PLAYED = 100
PULL_FIELDING = True
POST_BATTERS = True
POST_PITCHERS = True
POST_FIELDERS = True
POST_PLAYERS = True
PLAYER_DESCRIPTION = 'Live'
IS_LIVESERIES = True
IGNORE_LIMITS = False


async def main(args):
    """
    params:
        cardset_name: str - to be searched in pd database
        games_played: int - from 1 - 162
        pull_fielding: bool - whether to pull fielding stats from bbref
        post_batters: bool - whether to post batting cards, batting card ratings, and batter updates
        post_pitchers: bool - whether to post pitching cards, pitching card ratings, and pitching updates
        post_fielders: bool - whether to post card positions
        post_players: bool - whether to post player updates
        p_desc_prefix: str - shows as cardset on card image and prefixes player name in discord
        is_liveseries: str - whether to look up players' current MLB club from MLB statsapi
        ignore_limits: bool - whether to ignore PA and TBF limits
    """
    # arg_data = get_args(args)

    # cardset_name = arg_data['cardset_name']
    print(f'Searching for cardset: {CARDSET_NAME}')
    c_query = await db_get('cardsets', params=[('name', CARDSET_NAME)])
    if c_query is None or c_query['count'] == 0:
        raise ValueError(f'Cardset {CARDSET_NAME} not found')

    cardset = c_query['cardsets'][0]
    del c_query
    input_path = f'data-input/{cardset["name"]} Cardset/'

    if GAMES_PLAYED < 1 or GAMES_PLAYED > 162:
        print(f'Game count has to be between 1 and 162.')
        return
    season_pct = GAMES_PLAYED / 162
    print(f'Cardset ID: {cardset["id"]} / Season: {SEASON}\nGame count: {GAMES_PLAYED} / Season %: {season_pct}\n')

    start_time = datetime.datetime.now()
    release_directory = f'{start_time.year}-{start_time.month}-{start_time.day}'

    data = await batters.creation.run_batters(
        cardset, input_path, POST_PLAYERS, CARD_BASE_URL, release_directory, PLAYER_DESCRIPTION, season_pct,
        POST_BATTERS, PULL_FIELDING, SEASON, IS_LIVESERIES, IGNORE_LIMITS
    )

    print(f'Batter updates are complete')
    start_time_two = datetime.datetime.now()
    run_time = start_time_two - start_time
    print(f'Total batting cards: {data["tot_batters"]}\nNew cardset batters: {data["new_batters"]}\n'
          f'Batter runtime: {round(run_time.total_seconds())} seconds\n')

    data = await pitchers.creation.run_pitchers(
        cardset, input_path, CARD_BASE_URL, SEASON, release_directory, PLAYER_DESCRIPTION, season_pct, POST_PLAYERS, POST_PITCHERS, IS_LIVESERIES, IGNORE_LIMITS
    )
    pitching_stats = data['pitching_stats']

    print(f'Pitcher updates are complete')
    start_time_three = datetime.datetime.now()
    p_run_time = datetime.datetime.now() - start_time_two
    print(f'Total pitching cards: {data["tot_pitchers"]}\nNew cardset pitchers: {data["new_pitchers"]}\n'
          f'Pitcher runtime: {round(p_run_time.total_seconds())} seconds\n')

    print(f'Running player position updates..')
    all_pos = await pd_positions_df(cardset['id'])

    player_updates = {}

    def set_all_positions(df_data):
        pos_series = all_pos.query(f'player_id == {df_data["player_id"]}')['position']
        pos_updates = []
        count = 1
        for this_pos in pos_series:
            if this_pos == 'P':
                try:
                    this_pitcher = pitching_stats.loc[df_data['bbref_id']]
                except KeyError as e:
                    logger.error(f'{e} / setting position to RP')
                    pos_updates.append((f'pos_{count}', 'RP'))
                    count += 1
                    break
                
                if this_pitcher['starter_rating'] > 3:
                    pos_updates.append((f'pos_{count}', 'SP'))
                    count += 1
                    if this_pitcher['relief_rating'] > 1 or not pd.isna(this_pitcher['closer_rating']):
                        pos_updates.append((f'pos_{count}', 'RP'))
                        count += 1
                else:
                    pos_updates.append((f'pos_{count}', 'RP'))
                    count += 1

                if not pd.isna(this_pitcher['closer_rating']):
                    pos_updates.append((f'pos_{count}', 'CP'))
                    count += 1
            else:
                pos_updates.append((f'pos_{count}', this_pos))
                count += 1

        if count == 1:
            pos_updates.append(('pos_1', 'DH'))
            count += 1

        while count <= 9:
            pos_updates.append((f'pos_{count}', 'False'))
            count += 1

        if len(pos_updates) > 0:
            if df_data.player_id not in player_updates.keys():
                player_updates[df_data.player_id] = pos_updates
            else:
                player_updates[df_data.player_id].extend(pos_updates)

    if 'promos' not in CARDSET_NAME.lower():
        all_players = await pd_players_df(cardset['id'])
        all_players.apply(set_all_positions, axis=1)

    print(f'Sending {len(player_updates)} player updates to PD database...')
    if POST_PLAYERS:
        for x in player_updates:
            await db_patch('players', object_id=x, params=player_updates[x])
    print(f'Player updates are complete\n')

    p_run_time = datetime.datetime.now() - start_time_three
    print(f'Player update runtime: {round(p_run_time.total_seconds())} seconds')
    t_run_time = datetime.datetime.now() - start_time
    print(f'Total runtime: {round(t_run_time.total_seconds())} seconds')


if __name__ == '__main__':
    asyncio.run(main(sys.argv[1:]))