File size: 7,625 Bytes
4e6be8d
 
 
 
d4ccbc6
 
 
 
4e6be8d
d4ccbc6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
67b558d
d4ccbc6
 
 
4e6be8d
 
 
 
 
 
 
 
115ff47
4e6be8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
67b558d
 
 
4e6be8d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d4ccbc6
 
26ab58a
d4ccbc6
eca6f3a
4e6be8d
 
 
 
7b96f10
4e6be8d
 
 
7b96f10
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
import json
import logging
import hashlib
from dataclasses import dataclass, field, InitVar, asdict
from typing import List, Dict, Union
from datetime import datetime, date

import pandas as pd
from google.cloud import bigquery


@dataclass
class Player:
    player_backend_user_id: str
    player_nickname: str = field(default=None)
    player_group: int = field(default=None)
    partners: List[str] = field(default_factory=list)
    badges: List[str] = field(default_factory=list)
    rewards_status: Union[str, Dict[str, Dict[str, Union[bool, str]]]] = field(
        default_factory=dict
    )
    adventure_logs: List[str] = field(default_factory=list)
    total_active_days: int = field(default=0)
    total_gained_scores: int = field(default=0)
    total_finished_contents: int = field(default=0)
    finished_videos_before_campaign: List[str] = field(default_factory=list)
    created_at_date: datetime.date = field(default_factory=date.today)
    updated_at_date: datetime.date = field(default_factory=date.today)

    available_achievements: InitVar[list] = field(default=None)
    init: InitVar[bool] = field(default=False)

    def __post_init__(self, available_achievements: List, init: bool):
        # If is_init is True, it means that the player is newly created
        # and the rewards_status is not yet initialized
        if init:
            self.assign_group()
            self.add_adventure_log("你並未參與這次與狐貍貓的冒險,請在下次活動中參與冒險吧!")
            if available_achievements is not None:
                self.rewards_status.update(
                    {
                        achievement_key: {
                            "is_completed": False,
                            "is_issued": False,
                        }
                        for achievement_key in available_achievements
                    }
                )
                self.serialize_rewards_status()
        else:
            self.badges = self.badges.tolist()
            self.partners = self.partners.tolist()
            self.adventure_logs = self.adventure_logs.tolist()
            self.finished_videos_before_campaign = (
                self.finished_videos_before_campaign.tolist()
            )
            self.assign_weekly_partner()

    # Serialization and Deserialization is for content type of rewards_status
    # rewards_status have to be serialized before inserting into BigQuery
    def serialize_rewards_status(self):
        self.rewards_status = json.dumps(self.rewards_status)

    def deserialize_rewards_status(self):
        self.rewards_status = json.loads(self.rewards_status)

    def add_partner(self, partner: str):
        if partner and partner not in self.partners:
            self.partners.append(partner)
            logging.info(
                f"Player {self.player_backend_user_id} has a new partner: {partner}"
            )

    def add_badge(self, badge: str):
        if badge and badge not in self.badges:
            self.badges.append(badge)
            logging.info(
                f"Player {self.player_backend_user_id} earned a new badge: {badge}"
            )

    def add_adventure_log(self, log: str):
        self.adventure_logs.append(log)
        logging.info(
            f"Player {self.player_backend_user_id} has a new adventure log: {log}"
        )

    def update_total_gained_scores(self, gained_scores: int):
        self.total_gained_scores = gained_scores

    def update_total_finished_contents(self, finished_contents_count: int):
        self.total_finished_contents = finished_contents_count

    def update_total_active_days(self, active_days: int):
        self.total_active_days = active_days

    def update_rewards_status(
        self, key: str, value: bool, target=["is_completed", "is_issued"]
    ):
        self.deserialize_rewards_status()
        self.rewards_status[key][target] = value
        self.serialize_rewards_status()

    def hash_user_id(self):
        hashed = hashlib.sha256(self.player_backend_user_id.encode()).hexdigest()
        hash_int = int(hashed, 16)
        return hash_int % 40

    def assign_group(self):
        self.player_group = self.hash_user_id()
        partner_group = self.player_group // 10
        group_base_partners = {
            0: "phoenix_1",
            1: "pegasus_1",
            2: "dragon_1",
            3: "griffin_1",
        }
        self.add_partner(group_base_partners[partner_group])

    def assign_weekly_partner(self):
        event_start_date = datetime(2023, 12, 4).date()
        current_date = datetime.now().date()
        weeks_elapsed = (current_date - event_start_date).days // 7
        if weeks_elapsed not in range(0, 4):
            return

        group_base_partners = {
            0: "phoenix",
            1: "pegasus",
            2: "dragon",
            3: "griffin",
        }

        base_partner = group_base_partners[self.player_group // 10]
        stage_partner = (
            f"{base_partner}_{weeks_elapsed + 1}"  # +1 to start from stage 1
        )

        if stage_partner not in self.partners:
            self.add_partner(stage_partner)

    def display_player_info(self):
        logging.info(f"Player Backend User ID: {self.player_backend_user_id}")
        logging.info(f"Player Group: {self.player_group}")
        logging.info(f"Partners: {self.partners}")
        logging.info(f"Badges: {self.badges}")
        logging.info(f"Adventure Logs: {self.adventure_logs}")
        logging.info(f"Rewards Status: {self.rewards_status}")
        logging.info(f"Total Gained Scores: {self.total_gained_scores}")
        logging.info(f"Total Finished Contents: {self.total_finished_contents}")
        logging.info(f"Total Active Days: {self.total_active_days}")
        logging.info(f"Created At: {self.created_at_date}")
        logging.info(f"Updated At: {self.updated_at_date}")

    def get_incomplete_rewards(self):
        self.deserialize_rewards_status()
        incomplete_rewards = list(
            {
                k: v for k, v in self.rewards_status.items() if not v["is_completed"]
            }.keys()
        )
        self.serialize_rewards_status()
        return incomplete_rewards

    @staticmethod
    def get_big_query_schema():
        return [
            bigquery.SchemaField("player_backend_user_id", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("player_group", "INTEGER", mode="NULLABLE"),
            bigquery.SchemaField("partners", "STRING", mode="REPEATED"),
            bigquery.SchemaField("badges", "STRING", mode="REPEATED"),
            bigquery.SchemaField("rewards_status", "STRING", mode="REQUIRED"),
            bigquery.SchemaField("adventure_logs", "STRING", mode="REPEATED"),
            bigquery.SchemaField("total_active_days", "INTEGER", mode="REQUIRED"),
            bigquery.SchemaField("total_gained_scores", "INTEGER", mode="REQUIRED"),
            bigquery.SchemaField("total_finished_contents", "INTEGER", mode="REQUIRED"),
            bigquery.SchemaField("created_at_date", "DATE", mode="REQUIRED"),
            bigquery.SchemaField("updated_at_date", "DATE", mode="REQUIRED"),
        ]

    @staticmethod
    def from_dict(series: pd.Series) -> "Player":
        data = series.copy()
        return Player(**data)

    def to_dict(self) -> Dict:
        data = asdict(self)
        # Convert datetime.date objects to string
        for date_field in ["created_at_date", "updated_at_date"]:
            if data.get(date_field):
                data[date_field] = data[date_field].strftime("%Y-%m-%d")

        return data