utils/utils.py

import os
import json
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 utils.mes_player_model import Player

AVATAR_PATH = "avatar/"
AVATAR_FILE_TYPE = ".png"
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:
        content = file.read()
    return content


def get_player_partners(player_info: gr.State) -> List[str]:
    return [
        f"{MEDIA_PATH}{partner}{MEDIA_FILE_TYPE}" for partner in player_info["partners"]
    ]


def get_player_badges(player_info: gr.State) -> List[str]:
    return [f"{MEDIA_PATH}{badge}{MEDIA_FILE_TYPE}" for badge in player_info["badges"]]


def get_player_avatar(player_info: gr.State) -> str:
    return f"{AVATAR_PATH}avatar_{player_info['player_group'] + 1}{AVATAR_FILE_TYPE}"


def get_player_adventure_logs(player_info: gr.State) -> List[str]:
    log_template = """<div class="adventure"><p>{player_log}</p></div>"""
    return [
        log_template.format(player_log=player_log)
        for player_log in player_info["adventure_logs"]
    ]


def get_player_adventure_logs_html(player_info: gr.State) -> str:
    adventure_logs = "".join(get_player_adventure_logs(player_info))
    template_content = get_content("htmls/adventure_template.html")
    return template_content.replace("{logs}", adventure_logs)


def get_player_achievements(player_info: gr.State) -> List[str]:
    achivement_name_map = {
        "participation_star": "參賽之星",
        "star_score_settler": "星際積分領航者",
        "interstellar_traveler_I": "星際旅行者 I",
        "interstellar_traveler_II": "星際旅行者 II",
        "interstellar_traveler_III": "星際旅行者 III",
        "interstellar_traveler_IV": "星際旅行者 IV",
        "climbers_club_I": "爬升俱樂部 I",
        "climbers_club_II": "爬升俱樂部 II",
        "climbers_club_III": "爬升俱樂部 III",
        "star_cluster_detector": "星團探測官",
        "starry_vigilante": "群星瞭望者",
        "planetary_decoder": "行星解碼",
        "galactic_librarian": "星系圖書館員",
        "energy_enthusiast_I": "能量狂熱者 I",
        "energy_enthusiast_II": "能量狂熱者 II",
        "energy_enthusiast_III": "能量狂熱者 III",
        "energy_enthusiast_IV": "能量狂熱者 IV",
        "knowledge_planet_explorer_I": "知識星球探險家 I",
        "knowledge_planet_explorer_II": "知識星球探險家 II",
        "scientific_expedition_explorer_I": "科學探險探險家 I",
        "scientific_expedition_explorer_II": "科學探險探險家 II",
        "cultural_celebration_explorer_I": "文化慶典探險家 I",
        "cultural_celebration_explorer_II": "文化慶典探險家 II",
        "youth_literature_explorer_I": "青春文學探險家 I",
        "youth_literature_explorer_II": "青春文學探險家 II",
        "path_to_wealth_explorer_I": "財富之路探險家 I",
        "path_to_wealth_explorer_II": "財富之路探險家 II",
        "cultivation_universe_explorer_I": "素養宇宙探險家 I",
        "cultivation_universe_explorer_II": "素養宇宙探險家 II",
        "electronic_and_information_college_explorer_I": "電資學院探險家 I",
        "electronic_and_information_college_explorer_II": "電資學院探險家 II",
        "star_warrior": "星空艦長",
    }
    if not isinstance(player_info["rewards_status"], dict):
        rewards_status = json.loads(player_info["rewards_status"])
    else:
        rewards_status = player_info["rewards_status"]
    if "routine_checker" in rewards_status:
        del rewards_status["routine_checker"]
    return [
        (
            achivement_name_map[achievement_key],
            "完成" if achievement_value["is_completed"] else "未完成",
        )
        for achievement_key, achievement_value in rewards_status.items()
    ]


def get_current_story():
    with open("story.json", "r", encoding="utf-8") as file:
        story = json.load(file)

    storyline_date = {
        (datetime(2023, 12, 4).date(), datetime(2023, 12, 5).date()): 1,
        (datetime(2023, 12, 6).date(), datetime(2023, 12, 7).date()): 2,
        (datetime(2023, 12, 8).date(), datetime(2023, 12, 9).date()): 3,
        (datetime(2023, 12, 10).date(), datetime(2023, 12, 11).date()): 4,
        (datetime(2023, 12, 12).date(), datetime(2023, 12, 13).date()): 5,
        (datetime(2023, 12, 14).date(), datetime(2023, 12, 15).date()): 6,
        (datetime(2023, 12, 16).date(), datetime(2023, 12, 17).date()): 7,
        (datetime(2023, 12, 18).date(), datetime(2023, 12, 19).date()): 8,
        (datetime(2023, 12, 20).date(), datetime(2023, 12, 22).date()): 9,
        (datetime(2023, 12, 23).date(), datetime(2023, 12, 25).date()): 10,
        (datetime(2023, 12, 26).date(), datetime(2023, 12, 27).date()): 11,
        (datetime(2023, 12, 28).date(), datetime(2023, 12, 29).date()): 12,
    }

    def get_stage(storyline_date):
        current_date = datetime.now().date()
        for (start_date, end_date), stage in storyline_date.items():
            if start_date <= current_date <= end_date:
                return stage
        return None

    stage = get_stage(storyline_date)

    if stage:
        return gr.Slider(
            value=stage / 12 * 100,
            show_label=False,
            interactive=False,
            info=story[str(stage)],
        )
    else:
        return gr.Slider(
            value=0,
            show_label=False,
            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)
        return "finished"


def render_player_data(player_info: gr.State):
    player_avatar = get_player_avatar(player_info)
    player_partners = get_player_partners(player_info)
    player_badges = get_player_badges(player_info)
    player_adventure_logs = get_player_adventure_logs_html(player_info)
    player_achievements = get_player_achievements(player_info)
    current_story = get_current_story()

    return (
        player_avatar,
        player_partners,
        player_badges,
        player_adventure_logs,
        player_achievements,
        current_story,
    )