paper-dynasty-card-creation/refresh_cards.py

import asyncio
import datetime
import sys
import urllib.parse

from db_calls import db_get, url_get, db_post, db_patch
from exceptions import logger

CARD_BASE_URL = "https://pd.manticorum.com/api/v2/players"
CARDSET_ID = 20
SKIP_ARMS = False
SKIP_BATS = False
START_ID = None
TEST_COUNT_MAX = None
HTML_CARDS = False


async def main(args):
    p_query = await db_get(
        "players",
        params=[("inc_dex", False), ("cardset_id", CARDSET_ID), ("short_output", True)],
    )
    if p_query["count"] == 0:
        raise ValueError("No players returned from Paper Dynasty API")
    all_players = p_query["players"]
    del p_query

    errors = []
    successes = []
    cxn_error = False
    count = -1
    start_time = datetime.datetime.now()

    for x in all_players:
        if "pitching" in x["image"] and SKIP_ARMS:
            pass
        elif "batting" in x["image"] and SKIP_BATS:
            pass
        elif START_ID is not None and START_ID > x["player_id"]:
            pass
        # elif 'sombaseball' in x['image']:
        #     errors.append((x, f'Bad card url: {x["image"]}'))
        elif "sombaseball" in x["image"]:
            print(
                f'Found old image URL for {x["p_name"]} - setting card type by pos_1: {x["pos_1"]}'
            )
            release_dir = f"{start_time.year}-{start_time.month}-{start_time.day}"
            if x["pos_1"] in ["SP", "RP", "CP", "P"]:
                image_url = (
                    f'{CARD_BASE_URL}/{x["player_id"]}/pitchingcard'
                    f'{urllib.parse.quote("?d=")}{release_dir}'
                )
            else:
                image_url = (
                    f'{CARD_BASE_URL}/{x["player_id"]}/battingcard'
                    f'{urllib.parse.quote("?d=")}{release_dir}'
                )
            this_player = await db_patch(
                "players", object_id=x["player_id"], params=[("image", image_url)]
            )
        else:
            count += 1
            if count % 20 == 0:
                print(f'Card #{count + 1} being updated is {x["p_name"]}...')
            elif TEST_COUNT_MAX is not None and TEST_COUNT_MAX < count:
                print("Done setting URLs")
                break

            try:
                resp = await db_post(f'players/{x["player_id"]}/image-reset')

            except ConnectionError as e:
                if cxn_error:
                    raise e

                cxn_error = True
                errors.append((x, e))

            except ValueError as e:
                errors.append((x, e))

    count = -1
    for x in all_players:
        if HTML_CARDS:
            card_url = f'{x["image"]}&html=true'
            timeout = 2
        else:
            card_url = x["image"]
            timeout = 6

        if TEST_COUNT_MAX is not None and TEST_COUNT_MAX < count:
            print("Done pulling images")
            break

        try:
            logger.info("calling the card url")
            resp = await url_get(card_url, timeout=timeout)

        except ConnectionError as e:
            if cxn_error:
                raise e

            cxn_error = True
            errors.append((x, e))

        except ValueError as e:
            errors.append((x, e))

        else:
            if x["image2"] is not None:
                if HTML_CARDS:
                    card_url = f'{x["image2"]}&html=true'
                else:
                    card_url = x["image2"]

                if "sombaseball" in x["image2"]:
                    errors.append((x, f'Bad card url: {x["image"]}'))
                else:
                    try:
                        resp = await url_get(card_url, timeout=6)

                    except ConnectionError as e:
                        if cxn_error:
                            raise e

                        cxn_error = True
                        errors.append((x, e))

                    except ValueError as e:
                        errors.append((x, e))

                    else:
                        successes.append(x)
            else:
                successes.append(x)

        count += 1

    if len(errors) > 0:
        logger.error("All Errors:")
        for x in errors:
            logger.error(f'ID {x[0]["player_id"]} {x[0]["p_name"]} - Error: {x[1]}')

    if len(successes) > 0:
        logger.debug("All Successes:")
        for x in successes:
            logger.info(f'ID {x["player_id"]} {x["p_name"]}')

    p_run_time = datetime.datetime.now() - start_time
    print(f"\nAll done!\nErrors: {len(errors)}\nSuccesses: {len(successes)}")
    print(f"Total runtime: {p_run_time.total_seconds()} seconds")


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