feat: add load and save player data process

app.py

@@ -10,6 +10,7 @@ from utils.utils import (
     get_player_adventure_logs_html,
     get_player_achievements,
     get_current_story,
+    save_latest_player_data,
 )
 
 seafoam = Seafoam()
@@ -96,6 +97,7 @@ with gr.Blocks(theme=seafoam, css=get_content("css/style.css")) as demo:
         player_info_query_btn = gr.Button(
             "Query", elem_id="trigger_button", visible=False
         )
+        pull_newest_player_info_btn = gr.Button("", visible=False)
 
     # actions when player login
     player_info_query_btn.click(get_player_info, player_backend_id, player_info).then(
@@ -112,5 +114,7 @@ with gr.Blocks(theme=seafoam, css=get_content("css/style.css")) as demo:
         player_info_query_btn.click(get_current_story, None, adventure)
     )
 
+    pull_newest_player_info_btn.click(save_latest_player_data, None, None)
+
 if __name__ == "__main__":
     demo.launch()

utils/utils.py

@@ -1,12 +1,28 @@
+import os
 import json
-from datetime import datetime
+import logging
+from datetime import datetime, timedelta, date
 from typing import List
+from dataclasses import asdict
 
 import gradio as gr
+from google.oauth2.service_account import Credentials
+from google.cloud import bigquery
+
+from mes_player_model import Player
 
 MEDIA_PATH = "medias/"
 MEDIA_FILE_TYPE = ".png"
 
+SCOPES = ["https://www.googleapis.com/auth/bigquery"]
+SERVICE_ACCOUNT_INFO = os.getenv("GBQ_TOKEN")
+service_account_info_dict = json.loads(SERVICE_ACCOUNT_INFO)
+
+creds = Credentials.from_service_account_info(
+        service_account_info_dict, scopes=SCOPES
+    )
+client = bigquery.Client(credentials=creds, project=service_account_info_dict['project_id'])
+
 
 def get_content(file_name: str) -> str:
     with open(file_name, "r", encoding="utf-8") as file:
@@ -127,3 +143,38 @@ def get_current_story():
             interactive=False,
             info="狐貍貓與光束守護者的旅程將於 2023/12/04 開始！敬請期待！",
         )
+
+
+def query_bq_table(client, sql):
+    try:
+        query_job = client.query(sql)
+        query_job.result()
+        return query_job.to_dataframe()
+    except Exception as e:
+        logging.error(f"Query Failed: {e}")
+        raise
+
+def load_player_statuses(client, date_str):
+    table_name = f"mes_report_{date_str}"
+    sql = f"SELECT * FROM `data_mart.{table_name}`"
+    return {
+        row["player_backend_user_id"]: Player.from_dict(row)
+        for _, row in query_bq_table(client, sql).iterrows()
+    }
+
+def get_date_strs(delta_days=0):
+    target_date = datetime.now().date() - timedelta(days=delta_days)
+    return target_date.strftime("%Y%m%d")
+
+def save_latest_player_data():
+    latest_player_data = load_player_statuses(client, '20231113')
+    latest_player_data_as_dict = {key: asdict(value) for key, value in latest_player_data.items()}
+
+    def date_serializer(obj):
+        if isinstance(obj, date):
+            return obj.isoformat()
+        raise TypeError("Type not serializable")
+    
+    with open('latest_player_data.json', 'w') as fp:
+        print("Saving latest player data...")
+        json.dump(latest_player_data_as_dict, fp, default=date_serializer)