Pynamodb study

database.py

from pynamodb.attributes import UnicodeAttribute, NumberAttribute, BooleanAttribute, UTCDateTimeAttribute, JSONAttribute
from pynamodb.models import Model
from time import sleep

Access_Key = "AKIASDKHWTHT5FZB5TMC"
Secret_Key = "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"
Region = "ap-northeast-2"


class BaseModel(Model):
    def commit(self):
        while True:
            try:
                self.save()
                break
            except Exception as e:
                sleep(1.1)
                print(e)

    class BaseMeta:
        region = Region
        aws_access_key_id = Access_Key
        aws_secret_access_key = Secret_Key
        table_name = None


class Player(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_Player"

    puuid = UnicodeAttribute(hash_key=True)
    game_name = UnicodeAttribute(null=False)
    tag_line = UnicodeAttribute(null=True)
    is_master = BooleanAttribute(null=False, default=False)
    last_matched_at = UTCDateTimeAttribute(null=True)
    last_matched_game_id = UnicodeAttribute(null=True)
    last_checked_at = UTCDateTimeAttribute(null=True)
    last_used_deck_code = UnicodeAttribute(null=True)
    created_at = UTCDateTimeAttribute(null=False)
    updated_at = UTCDateTimeAttribute(null=False)

    def __repr__(self):
        return f"Player(id={self.id}, game_name={self.game_name}, tag_line={self.tag_line}, puuid={self.puuid}, is_master={self.is_master})"

    def __str__(self):
        return self.__repr__()


class GameVersion(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_GameVersion"

    game_version = UnicodeAttribute(hash_key=True)
    total_match_count = NumberAttribute(null=False, default=0)
    created_at = UTCDateTimeAttribute(null=False)
    updated_at = UTCDateTimeAttribute(null=False)

    def __repr__(self):
        return f"GameVersion(game_version={self.game_version}, total_match_count={self.total_match_count})"

    def __str__(self):
        return self.__repr__()


class SingleMetaDeckAnalyze(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_SingleMetaDeckAnalyze"

    id = UnicodeAttribute(hash_key=True)
    game_version = UnicodeAttribute(null=False)
    deck_code = UnicodeAttribute(null=False)
    win_count = NumberAttribute(null=False, default=0)
    lose_count = NumberAttribute(null=False, default=0)
    first_start_win_count = NumberAttribute(null=False, default=0)
    first_start_lose_count = NumberAttribute(null=False, default=0)
    turns = JSONAttribute(null=False, default={
        str(turn_count): {
            "win": 0,
            "lose": 0
        } for turn_count in range(1, 51)
    })

    def __repr__(self):
        return f"SingleMetaDeckAnalyze(id={self.id}, game_version={self.game_version}, deck_code={self.deck_code}, win_count={self.win_count}, lose_count={self.lose_count}, first_start_win_count={self.first_start_win_count}, first_start_lose_count={self.first_start_lose_count})"

    def __str__(self):
        return self.__repr__()


class SingleMetaDeckCodeAnalyze(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_SingleMetaDeckCodeAnalyze"

    id = UnicodeAttribute(hash_key=True)
    single_meta_deck_analyze_id = UnicodeAttribute(null=False)
    deck_code = UnicodeAttribute(null=False)
    win_count = NumberAttribute(null=False, default=0)
    lose_count = NumberAttribute(null=False, default=0)

    def __repr__(self):
        return f"SingleMetaDeckCodeAnalyze(id={self.id}, single_meta_deck_analyze_id={self.single_meta_deck_analyze_id}, deck_code={self.deck_code}, win_count={self.win_count}, lose_count={self.lose_count})"

    def __str__(self):
        return self.__repr__()


class DoubleMetaDeckAnalyze(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_DoubleMetaDeckAnalyze"

    id = UnicodeAttribute(hash_key=True)
    my_deck_id = UnicodeAttribute(null=False, range_key=True)
    opponent_deck_id = UnicodeAttribute(null=False)
    win_count = NumberAttribute(null=False, default=0)
    lose_count = NumberAttribute(null=False, default=0)
    first_start_win_count = NumberAttribute(null=False, default=0)
    first_start_lose_count = NumberAttribute(null=False, default=0)

    def __repr__(self):
        return f"DoubleMetaDeckAnalyze(id={self.id}, my_deck_id={self.my_deck_id}, opponent_deck_id={self.opponent_deck_id}, win_count={self.win_count}, lose_count={self.lose_count}, first_start_win_count={self.first_start_win_count}, first_start_lose_count={self.first_start_lose_count})"

    def __str__(self):
        return self.__repr__()


class DoubleMetaDeckTurnAnalyze(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_DoubleMetaDeckTurnAnalyze"

    id = UnicodeAttribute(hash_key=True)
    double_meta_deck_analyze_id = UnicodeAttribute(null=False, range_key=True)
    turn_count = NumberAttribute(null=False)
    win_count = NumberAttribute(null=False, default=0)
    lose_count = NumberAttribute(null=False, default=0)

    def __repr__(self):
        return f"DoubleMetaDeckTurnAnalyze(id={self.id}, double_meta_deck_analyze_id={self.double_meta_deck_analyze_id}, turn_count={self.turn_count}, win_count={self.win_count}, lose_count={self.lose_count})"

    def __str__(self):
        return self.__repr__()


class Card(BaseModel):
    class Meta(BaseModel.BaseMeta):
        table_name = "LOR_Card"

    id = UnicodeAttribute(hash_key=True)
    name = UnicodeAttribute(null=False)
    region = UnicodeAttribute(null=False)
    type = UnicodeAttribute(null=False)
    set = UnicodeAttribute(null=False)
    is_champion = BooleanAttribute(null=False, default=False)

    def __repr__(self):
        return f"Card(id={self.id}, name={self.name}, region={self.region}, type={self.type}, set={self.set}, is_champion={self.is_champion})"

    def __str__(self):
        return self.__repr__()


if __name__ == "__main__":
    for table in [SingleMetaDeckAnalyze, SingleMetaDeckCodeAnalyze, DoubleMetaDeckAnalyze, DoubleMetaDeckTurnAnalyze, Card, GameVersion]:
        if not table.exists():
            table.create_table(
                read_capacity_units=1,
                write_capacity_units=1,
                wait=False
            )

use.py

import os
from database import Card,  GameVersion, SingleMetaDeckAnalyze, SingleMetaDeckCodeAnalyze, DoubleMetaDeckAnalyze, DoubleMetaDeckTurnAnalyze
from datetime import datetime
from typing import Dict, Union, List, Optional
from collections import defaultdict
from csv import DictReader
from lor_deckcodes import LoRDeck, CardCodeAndCount

DISCORD_WEBHOOKS_ANALYZE_MATCH_DATA_LOG = os.environ.get(
    "DISCORD_WEBHOOKS_ANALYZE_MATCH_DATA_LOG")


def analyze_match_data(s3_bucket: str, s3_path: str) -> int:
    total_match_count = 0

    with open("test.csv", "r", encoding="utf-8") as f:
        lines = f.read().splitlines()

    total_match_count = len(lines) - 1

    game_version_analyze = defaultdict(lambda: {
        "total_match_count": 0,
    })

    single_meta_deck_analyze = defaultdict(
        lambda:
        {
            "win": 0,
            "lose": 0,
            "turn": defaultdict(
                lambda:
                {
                    "win": 0,
                    "lose": 0,
                }
            ),
            "single_meta_deck_code_analyze": defaultdict(
                lambda:
                {
                    "win": 0,
                    "lose": 0,
                }
            ),
            "first_start_win_count": 0,
            "first_start_lose_count": 0,
        }
    )

    double_meta_deck_analyze = defaultdict(
        lambda:
        {
            "win": 0,
            "lose": 0,
            "turn": defaultdict(
                lambda:
                {
                    "win": 0,
                    "lose": 0,
                }
            ),
            "first_start_win_count": 0,
            "first_start_lose_count": 0,
        }
    )

    db_card_champions: list[Card] = list(Card.scan(Card.is_champion == True))

    all_card_champion_ids = set([card.id for card in db_card_champions])

    for row in DictReader(lines):
        print(f"finding : {row['match_id']}")
        game_version_analyze[row["game_version"]]["total_match_count"] += 1

        winner_deck = LoRDeck.from_deckcode(
            row["win_user_deck_code"])
        winner_new_deck_code = LoRDeck(
            [
                CardCodeAndCount(card, 1) for card in sorted([card.card_code for card in winner_deck.cards]) if card in all_card_champion_ids
            ]
        ).encode()

        loser_deck = LoRDeck.from_deckcode(
            row["loss_user_deck_code"])
        loser_new_deck_code = LoRDeck(
            [
                CardCodeAndCount(card, 1) for card in sorted([card.card_code for card in loser_deck.cards]) if card in all_card_champion_ids
            ]
        ).encode()

        single_meta_deck_analyze[(row["game_version"], winner_new_deck_code)
                                 ]["win"] += 1
        single_meta_deck_analyze[(row["game_version"], winner_new_deck_code)
                                 ]["turn"][row["total_turn_count"]]["win"] += 1
        single_meta_deck_analyze[(row["game_version"], winner_new_deck_code)
                                 ]["single_meta_deck_code_analyze"][row["win_user_deck_code"]]["win"] += 1

        is_winner_first_start = (int(row["total_turn_count"]) %
                                 2 == int(row["win_user_order_of_play"]))
        if is_winner_first_start:
            single_meta_deck_analyze[(row["game_version"], winner_new_deck_code)
                                     ]["first_start_win_count"] += 1

        single_meta_deck_analyze[(row["game_version"], loser_new_deck_code)
                                 ]["lose"] += 1
        single_meta_deck_analyze[(row["game_version"], loser_new_deck_code)
                                 ]["turn"][row["total_turn_count"]]["lose"] += 1
        single_meta_deck_analyze[(row["game_version"], winner_new_deck_code)
                                 ]["single_meta_deck_code_analyze"][row["loss_user_deck_code"]]["lose"] += 1

        is_loser_first_start = (int(row["total_turn_count"]) %
                                2 == int(row["loss_user_order_of_play"]))
        if is_loser_first_start:
            single_meta_deck_analyze[(row["game_version"], loser_new_deck_code)
                                     ]["first_start_lose_count"] += 1

        double_meta_deck_analyze[(
            row["game_version"], winner_new_deck_code, loser_new_deck_code)]["win"] += 1
        double_meta_deck_analyze[(
            row["game_version"], loser_new_deck_code, winner_new_deck_code)]["lose"] += 1

        if is_winner_first_start:
            double_meta_deck_analyze[(
                row["game_version"], winner_new_deck_code, loser_new_deck_code)]["first_start_win_count"] += 1
        if is_loser_first_start:
            double_meta_deck_analyze[(
                row["game_version"], loser_new_deck_code, winner_new_deck_code)]["first_start_lose_count"] += 1

        double_meta_deck_analyze[(
            row["game_version"], winner_new_deck_code, loser_new_deck_code)]["turn"][row["total_turn_count"]]["win"] += 1
        double_meta_deck_analyze[(
            row["game_version"], loser_new_deck_code, winner_new_deck_code)]["turn"][row["total_turn_count"]]["lose"] += 1

    total = len(game_version_analyze.keys())
    now = 1
    for key, value in game_version_analyze.items():
        print(now, total)
        now += 1

        try:
            db_game_version = GameVersion.get(key)
        except GameVersion.DoesNotExist:
            db_game_version = GameVersion(
                game_version=key,
            )
            db_game_version.created_at = datetime.utcnow()
            db_game_version.updated_at = datetime.utcnow()

        db_game_version.total_match_count += value["total_match_count"]
        db_game_version.commit()

    total = len(single_meta_deck_analyze.keys())
    now = 1
    for key, value in single_meta_deck_analyze.items():
        game_version, deck_code = key
        print(now, total)
        now += 1
        try:
            db_single_meta_deck_analyze = SingleMetaDeckAnalyze.get(
                f"{game_version}#{deck_code}"
            )
        except SingleMetaDeckAnalyze.DoesNotExist:
            db_single_meta_deck_analyze = SingleMetaDeckAnalyze(
                f"{game_version}#{deck_code}",
                game_version=key[0],
                deck_code=key[1],
            )

        db_single_meta_deck_analyze.win_count += value["win"]
        db_single_meta_deck_analyze.lose_count += value["lose"]
        db_single_meta_deck_analyze.first_start_win_count += value["first_start_win_count"]
        db_single_meta_deck_analyze.first_start_lose_count += value["first_start_lose_count"]

        for turn_key, turn_value in value["turn"].items():
            if db_single_meta_deck_analyze.turns.get(turn_key) is None:
                db_single_meta_deck_analyze.turns[turn_key] = {
                    "win": turn_value["win"],
                    "lose": turn_value["lose"],
                }
            else:
                db_single_meta_deck_analyze.turns[turn_key]["win"] += turn_value["win"]
                db_single_meta_deck_analyze.turns[turn_key]["lose"] += turn_value["lose"]

        db_single_meta_deck_analyze.commit()

        for single_meta_deck_code_analyze_key, single_meta_deck_code_analyze_value in value["single_meta_deck_code_analyze"].items():
            try:
                db_single_meta_deck_code_analyze = SingleMetaDeckCodeAnalyze.get(
                    f"{db_single_meta_deck_analyze.id}#{single_meta_deck_code_analyze_key}",
                    db_single_meta_deck_analyze.id
                )
            except SingleMetaDeckCodeAnalyze.DoesNotExist:
                db_single_meta_deck_code_analyze = SingleMetaDeckCodeAnalyze(
                    f"{db_single_meta_deck_analyze.id}#{single_meta_deck_code_analyze_key}",
                    single_meta_deck_analyze_id=db_single_meta_deck_analyze.id,
                    deck_code=single_meta_deck_code_analyze_key,
                )

            db_single_meta_deck_code_analyze.win_count += single_meta_deck_code_analyze_value["win"]
            db_single_meta_deck_code_analyze.lose_count += single_meta_deck_code_analyze_value["lose"]
            db_single_meta_deck_code_analyze.commit()

    total = len(double_meta_deck_analyze.keys())
    now = 1
    for key, value in double_meta_deck_analyze.items():
        game_version, my_deck_code, opponent_deck_code = key
        print(now, total)
        now += 1

        try:
            db_double_meta_deck_analyze = DoubleMetaDeckAnalyze.get(
                f"{game_version}#{my_deck_code}#{opponent_deck_code}",
                f"{game_version}#{my_deck_code}",
            )
        except DoubleMetaDeckAnalyze.DoesNotExist:
            db_double_meta_deck_analyze = DoubleMetaDeckAnalyze(
                f"{game_version}#{my_deck_code}#{opponent_deck_code}",
                my_deck_id=f"{game_version}#{my_deck_code}",
                opponent_deck_id=f"{game_version}#{opponent_deck_code}",
            )

        db_double_meta_deck_analyze.win_count += value["win"]
        db_double_meta_deck_analyze.lose_count += value["lose"]
        db_double_meta_deck_analyze.first_start_win_count += value["first_start_win_count"]
        db_double_meta_deck_analyze.first_start_lose_count += value["first_start_lose_count"]
        db_double_meta_deck_analyze.commit()

        for turn_key, turn_value in value["turn"].items():
            try:
                db_double_meta_deck_turn_analyze = DoubleMetaDeckTurnAnalyze.get(
                    f"{db_double_meta_deck_analyze.id}#{turn_key}",
                    db_double_meta_deck_analyze.id,
                )
            except DoubleMetaDeckTurnAnalyze.DoesNotExist:
                db_double_meta_deck_turn_analyze = DoubleMetaDeckTurnAnalyze(
                    f"{db_double_meta_deck_analyze.id}#{turn_key}",
                    double_meta_deck_analyze_id=db_double_meta_deck_analyze.id,
                    turn_count=int(turn_key),
                )

            db_double_meta_deck_turn_analyze.win_count += turn_value["win"]
            db_double_meta_deck_turn_analyze.lose_count += turn_value["lose"]
            db_double_meta_deck_turn_analyze.commit()

    return total_match_count


def lambda_handler(event, context):
    log = {
        "start": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "total_match_count": 0,
        "target_files": [],
        "end": None,
    }

    total_match_count = 0

    total_match_count += analyze_match_data(None, None)

    log["total_match_count"] = total_match_count
    log["end"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    success_color = 0x2ECC71

    return log


print(lambda_handler(None, None))