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
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} Season"
GAMES_PLAYED = 162
PULL_FIELDING = True
POST_BATTERS = True
POST_PITCHERS = True
POST_FIELDERS = True
POST_PLAYERS = True
PLAYER_DESCRIPTION = "2025"  # '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("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("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("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("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("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:]))