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 = """

{player_log}

""" 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", "consistent_climbers_club_I": "連續爬升俱樂部 I", "consistent_climbers_club_II": "連續爬升俱樂部 II", "consistent_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"] 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.update( value=stage / 12 * 100, show_label=False, interactive=False, info=story[str(stage)], ) else: return gr.Slider.update( 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)