paper-dynasty-card-creation/scripts/custom_card_creation.py

import asyncio
import datetime

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

CARD_BASE_URL = f"{DB_URL}/v2/players"


async def main(args):
    """
    params:
        cardset_name: str - to be searched in pd database
        games_played: int - always 162
        pull_fielding: bool - always False
        post_batters: bool - whether or not to post batting cards, batting card ratings, and batter updates
        post_pitchers: bool - whether or not to post pitching cards, pitching card ratings, and pitching updates
        post_players: bool - whether or not to post player updates
        player_description: str - shows as cardset on card image and prefixes player name in discord
        is_liveseries: str - always False
    """
    arg_data = get_args(args)

    # cardset_name = input(f'What is the name of this Cardset? ')
    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["count"] == 0:
        print(f"I do not see a cardset named {cardset_name}")
        return
    cardset = c_query["cardsets"][0]
    del c_query
    input_path = f'data-input/{cardset["name"]} Cardset/'

    if "season" in arg_data:
        season = arg_data["season"]
    else:
        season = int(cardset["name"][:4])

    season_pct = 1
    print(f'Cardset ID: {cardset["id"]} / Season: {season}\nSeason %: {season_pct}\n')

    if "player_description" in arg_data:
        player_description = arg_data["player_description"]
    elif season_pct < 1:
        player_description = "Live"
    else:
        player_description = f"{season}"

    post_batters = (
        True
        if "post_batters" not in arg_data or arg_data["post_batters"].lower() == "true"
        else False
    )
    post_pitchers = (
        True
        if "post_pitchers" not in arg_data
        or arg_data["post_pitchers"].lower() == "true"
        else False
    )
    post_players = (
        True
        if "post_players" not in arg_data or arg_data["post_players"].lower() == "true"
        else False
    )
    pull_fielding = False
    is_liveseries = False
    ignore_limits = True
    is_custom = True

    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, is_custom
    # )

    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,
        pull_fielding,
        is_custom,
    )
    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":
                this_pitcher = pitching_stats.loc[df_data["bbref_id"]]
                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)

    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:]))