app.py

import json
import logging

import gradio as gr

from theme import Seafoam
from utils.mes_player_model import Player
from utils.mes_achievements import Achievement
from utils.mes_player_activity_model import PlayerActivity
from utils.utils import (
    get_content,
    render_player_data,
    save_latest_player_data,
    render_finished,
)
from utils.completion_reward import CompletionReward
from utils.completion_reward_utils import (
    get_llm_response,
    set_player_name,
    set_player_selected_character,
    create_certificate,
    complete_reward,
    check_is_in_completion_reward,
    get_llm_response_once,
)

seafoam = Seafoam()


def get_player_info(player_backend_user_id):
    try:
        with open("latest_player_data.json", "r", encoding="utf-8") as file:
            player_info = json.load(file)
    except FileNotFoundError:
        logging.error("No player data found. Loading new player data.")
        save_latest_player_data()
        with open("latest_player_data.json", "r", encoding="utf-8") as file:
            player_info = json.load(file)
        with open("latest_player_data.json", "r") as f:
            datas = json.load(f)

        processed_datas = {}

        for k, v in datas.items():
            log = v["adventure_logs"]
            filtered_log = [record for record in log if not record.startswith("恭喜")]
            processed_log = [record.rsplit(",", 1)[0] for record in filtered_log]
            concat_log = " ".join(processed_log)
            processed_datas[k] = concat_log

        temp_log = """你開始了星空探險隊的冒險！ 你在離家 2 天後，找到了迷失已久的星際繪圖師，並在他的引導下進入了下一個未知的領域。 在完成了 41 關學習內容後，你獲得了遙遠星球上的神秘遺產。 在某天，你在冰晶極地找到了一顆具有神秘力量的寒冰寶石。 你在奇特的花園中找到了一種新的能源。 在雲端星球中，你遇見了一位雲之精靈，並從她那裡獲得了一個神祕的能量水球。 132 天之後，你們獲得了一個神秘的地圖，這將引導你們找到隱藏的神秘能量源。\n你們在一個古老的地下洞穴中找到了一個巨大的藍色能量晶體，這將極大地援助你們的旅程。 你在混亂的黑洞中找到了一片完好無損的石碑，上面記載著光束寶石的由來和使用方式。 你用寶石的力量召喚傳說中的先知，獲得了狐狸貓族的力量, 向黑洞發動了終極一擊。 你在繼續活動了 7 天後，你的行為感動了星際，你獲得了一個可以轉換成任何形象的能量護盾。 你在一場巨大的黑洞爆炸後，成功找回了去往家鄉的方向。 連日交戰後達成了學習 209 個內容的紀錄，獲得了星系長老的讚賞，並贈與你們神秘的光護符。 你成功與神秘的月亮寶石建立連結，從而獲得月亮之力。 你在遙遠的星海之上發現了一顆寂靜的星體，星體的中心藏有一個帶著神秘能量的寶藏。 你在神祕的狐狸貓族祖先墳場找到了一塊古老的碑文，裡面記載了如何使用和提煉能量的方法。 在經歷了若干日月後，你終於在 229 號洞穴找到了第一塊結晶。 你們在探索神秘洞穴的過程中，發現了一個神秘寶箱，裡面有 229 個神秘道具。 你們在冬季島上找到了冰晶葉，並獲得了 678 點的能量。 你剛剛結束了 14 天的劍橋學堂，你們獲得了 726 點的能量供給。 你剛剛結束了 14 天的劍橋學堂，你們獲得了 726 點的能量供給。 你在星空探險隊的冒險期間和狐貍貓一起度過了聖誕佳節 🎄 你從超空間黑洞脫離時發現了一串神秘的密碼，這個密碼包含著黑洞能量的謎題。 你們在一個神秘的星球上探險，在那裡你們發現了新的能量源。 你們在一片未知的領域中尋找到了狐狸貓族的古老遺蹟，並獲得了 780 的能量點數。 你收集了 836 個痕跡之星，並用風暴龍的力量打開了神秘之門。 你收集了 941 個痕跡之星，並用風暴龍的力量打開了神秘之門。 你結束了 2023 與狐貍貓的星空探險！"""
        processed_datas["[email protected]"] = temp_log
        processed_datas[
            "http://googleid.junyiacademy.org/115084976189396533674"
        ] = temp_log
        processed_datas[
            "http://googleid.junyiacademy.org/106428943548495187296"
        ] = temp_log

        with open("processed_adventure_logs.json", "w") as f:
            json.dump(processed_datas, f)

    if player_backend_user_id in player_info:
        return player_info[player_backend_user_id]
    else:
        logging.info(
            f"No data found for player ID {player_backend_user_id}. Initializing new player data."
        )

        new_player = Player(
            player_backend_user_id=player_backend_user_id,
            init=True,
            available_achievements=Achievement.get_available_achievements(),
        )

        return new_player.to_dict()


def create_new_player_activity():
    return PlayerActivity()


def get_player_logs(player_backend_user_id):
    with open("processed_adventure_logs.json") as file:
        player_logs = json.load(file)
    try:
        player_logs = player_logs[player_backend_user_id]
        return player_logs
    except KeyError:
        logging.error(
            f"Player {player_backend_user_id} not found in processed_player_adventure_logs.json"
        )


def init_reward():
    return CompletionReward()


# start of gradio interface
with gr.Blocks(theme=seafoam, css=get_content("css/style.css")) as demo:
    player_info = gr.State()
    player_logs = gr.State()
    completion_reward = gr.State(init_reward)
    player_activity_tracker = gr.State(create_new_player_activity)

    with gr.Tab("個人化戰報", elem_id="personal_report"):
        with gr.Row():
            with gr.Column(
                scale=1,
                elem_classes="gallery_container",
            ):
                pet_description = gr.Markdown("# 夥伴", elem_id="pet_avatar_description")
                pet_gallery = gr.Gallery(
                    [],
                    label="夥伴",
                    preview=False,
                    elem_id="pet_gallery",
                    columns=30,
                    height=200,
                )

                badge_description = gr.Markdown(
                    "# 徽章", elem_id="badge_avatar_description"
                )
                badge_gallery = gr.Gallery(
                    [],
                    label="徽章",
                    preview=False,
                    elem_id="badge_gallery",
                    columns=30,
                    height=200,
                )

            with gr.Column(scale=1, elem_id="player_avatar_container"):
                avatar = gr.Image(
                    "avatar/blank_avatar.png",
                    elem_id="player_avatar",
                )
                avatar_description = gr.Markdown(
                    "# 光束守護者", elem_id="player_avatar_description"
                )

            with gr.Column(scale=1):
                adventure_description = gr.Markdown(
                    "# 冒險階段", elem_id="adventure_description"
                )
                adventure = gr.Slider(
                    value=0,
                    show_label=False,
                    interactive=False,
                    elem_id="adventure_slider",
                    info="",
                )
                achievements_description = gr.Markdown(
                    "# 達成成就", elem_id="achievements_description"
                )
                achievements = gr.HighlightedText(
                    value=[],
                    elem_classes="achievements",
                    color_map={
                        "完成": "green",
                        "未完成": "red",
                    },
                )

        with gr.Row():
            html = (
                "<div style='max-width:100%; max-height:360px; overflow:auto'>"
                + get_content("htmls/adventure_blank.html")
            )
            adventure_log = gr.HTML(
                html, label="Adventure Log", elem_id="adventure_log"
            )

        # handling player info
        with gr.Row():
            player_backend_id = gr.Textbox(
                "", elem_id="player_backend_id", visible=False
            )
            player_info_query_btn = gr.Button(
                "Query", elem_id="trigger_button", visible=False
            )
            pull_newest_player_data = gr.Textbox("", visible=False)
            update_status = gr.Textbox("", visible=False)

    with gr.Tab("完賽獎勵", elem_id="completion_reward"):
        with gr.Row():
            start_make_reward = gr.Button(
                "開始製作完賽獎勵！", visible=False, elem_id="start_make_reward"
            )

        with gr.Row():
            not_participate = gr.Markdown(
                "# 同學並未參與 2023 的星空探險隊唷！", visible=False, elem_id="not_participate"
            )

        with gr.Row():
            not_start = gr.Markdown(
                "# 完賽獎勵還沒有開放申請，但是就快了，請敬請期待！", visible=False, elem_id="not_start"
            )

        with gr.Row():
            already_issued = gr.Image(visible=False)

        with gr.Row():
            player_name_title = gr.Markdown(
                "# 選擇玩家暱稱", visible=False, elem_id="player_name_title"
            )

        with gr.Row():
            player_name = gr.Textbox(
                label="玩家暱稱",
                info="請輸入想要用在完賽獎勵上的玩家暱稱，上限為 10 字，獎勵發送後無法更改也無法補發，還請同學們謹慎填寫。",
                interactive=True,
                elem_id="player_name",
                visible=False,
                placeholder="請在這裡輸入暱稱",
            )

        with gr.Row():
            player_name_too_long = gr.Markdown(
                "# 暱稱過長，請重新輸入", visible=False, elem_id="player_name_too_long"
            )

        with gr.Row():
            confirm_player_name = gr.Button(
                "確認暱稱", elem_id="confirm_player_name", visible=False
            )
            cancel_player_name = gr.Button(
                "取消", elem_id="cancel_player_name", visible=False
            )

        with gr.Row():
            player_name_next_step = gr.Button(
                "下一步", visible=False, elem_id="player_name_next_step"
            )

        with gr.Row():
            story_title = gr.Markdown("# 選擇冒險故事", visible=False, elem_id="story_title")

        with gr.Row():
            story_description = gr.Markdown(
                "有五位星際夥伴來為你撰寫專屬於你的冒險故事，每位星際夥伴都會得到同學冒險週記的內容，並寫成一段故事，請同學選擇自己最喜歡的一段故事，這段故事將會被用來印在獎狀上<br><br>小叮嚀：請同學選擇喜歡的故事內容，而不是星際夥伴唷！星際夥伴可是不會和同學們一起冒險的！最後，撰寫故事的過程大概需要幾分鐘的時間，請同學耐心等待",
                visible=False,
                elem_id="story_description",
            )

        with gr.Row():
            openai_description = gr.Markdown(
                "# 露米娜", elem_id="openai_description", visible=False
            )
            aws_description = gr.Markdown(
                "# 索拉拉", elem_id="aws_description", visible=False
            )
            google_description = gr.Markdown(
                "# 薇丹特", elem_id="google_description", visible=False
            )
            mtk_description = gr.Markdown(
                "# 蔚藍", elem_id="mtk_description", visible=False
            )

        with gr.Row():
            openai_img = gr.Image(
                "medias/lumina.png",
                visible=False,
                elem_id="openai_img",
                interactive=False,
                show_download_button=False,
            )
            aws_img = gr.Image(
                "medias/solara.png",
                visible=False,
                elem_id="aws_img",
                interactive=False,
                show_download_button=False,
            )
            google_img = gr.Image(
                "medias/verdant.png",
                visible=False,
                elem_id="google_img",
                interactive=False,
                show_download_button=False,
            )
            mtk_img = gr.Image(
                "medias/azure.png",
                visible=False,
                elem_id="mtk_img",
                interactive=False,
                show_download_button=False,
            )

        with gr.Row():
            start_generate_story = gr.Button(
                "開始寫作故事", visible=False, elem_id="start_generate_story"
            )
            weaving = gr.Button(
                "星際夥伴努力撰寫故事中...",
                visible=False,
                elem_id="weaving",
                size="lg",
                interactive=False,
            )

        with gr.Row():
            bot1 = gr.Chatbot(visible=False, elem_id="bot1")
            bot2 = gr.Chatbot(visible=False, elem_id="bot2")
            bot3 = gr.Chatbot(visible=False, elem_id="bot3")
            bot4 = gr.Chatbot(visible=False, elem_id="bot4")

        with gr.Row():
            select_story = gr.Radio(
                ["露米娜", "索拉拉", "薇丹特", "蔚藍"],
                interactive=True,
                label="選擇故事",
                visible=False,
                elem_id="select_story",
            )

        with gr.Row():
            confirm_story = gr.Button("確認故事", visible=False, elem_id="confirm_story")
            cancel_story = gr.Button("取消", visible=False, elem_id="cancel_story")

        with gr.Row():
            start_generate_certificate = gr.Button(
                "開始製作完賽獎勵！", visible=False, elem_id="start_generate_certificate"
            )
            processing = gr.Button(
                "製作中...",
                visible=False,
                elem_id="processing",
                size="lg",
                interactive=False,
            )
            complete = gr.Markdown("# 完成！", visible=False, elem_id="complete")

        with gr.Row():
            reward_result = gr.Image(visible=False)

    # actions when player login
    # define args
    send_player_login_info = dict(
        fn=render_finished,
        inputs=[player_activity_tracker, player_info],
        outputs=None,
        queue=False,
    )

    player_info_query_btn.click(
        get_player_info, player_backend_id, player_info, queue=False
    ).then(get_player_logs, player_backend_id, player_logs, queue=False).then(
        lambda: gr.update(visible=True), None, start_make_reward, queue=False
    ).then(
        render_player_data,
        player_info,
        [avatar, pet_gallery, badge_gallery, adventure_log, achievements, adventure],
        queue=False,
    ).then(
        **send_player_login_info
    )

    pull_newest_player_data.submit(
        save_latest_player_data,
        None,
        update_status,
        api_name="pull_newest_player_data",
    )

    def create_visibility_updates(visible, count):
        return tuple(gr.update(visible=visible) for _ in range(count))

    start_make_reward.click(
        check_is_in_completion_reward,
        player_backend_id,
        [
            start_make_reward,
            player_name_title,
            player_name,
            confirm_player_name,
            not_participate,
            not_start,
            already_issued,
        ],
        queue=False,
    )

    set_player_name_args = dict(
        fn=set_player_name,
        inputs=[completion_reward, player_name, player_backend_id],
        outputs=None,
        queue=False,
    )

    def check_plyer_name_length(player_name):
        if len(player_name) > 10:
            return (
                gr.update(visible=True),
                gr.update(visible=False),
                gr.update(visible=True),
            )
        else:
            return (
                gr.update(visible=True),
                gr.update(visible=True),
                gr.update(visible=False),
            )

    confirm_player_name.click(
        lambda: (gr.update(interactive=False), gr.update(visible=False)),
        None,
        [player_name, confirm_player_name],
        queue=False,
    ).then(
        check_plyer_name_length,
        player_name,
        [cancel_player_name, player_name_next_step, player_name_too_long],
        queue=False,
    ).then(
        **set_player_name_args
    )

    cancel_player_name.click(
        lambda: (
            gr.update(interactive=True),
            gr.update(visible=True),
            gr.update(visible=False),
        ),
        None,
        [player_name, confirm_player_name, player_name_too_long],
        queue=False,
    ).then(
        lambda: create_visibility_updates(False, 2),
        None,
        [cancel_player_name, player_name_next_step],
        queue=False,
    )

    player_name_next_step.click(
        lambda: create_visibility_updates(False, 5),
        None,
        [
            player_name,
            player_name_next_step,
            confirm_player_name,
            player_name_title,
            cancel_player_name,
        ],
        queue=False,
    ).then(
        lambda: create_visibility_updates(True, 11),
        None,
        [
            openai_img,
            aws_img,
            google_img,
            mtk_img,
            story_title,
            story_description,
            openai_description,
            aws_description,
            google_description,
            mtk_description,
            start_generate_story,
        ],
        queue=False,
    )

    get_llm_response_args = dict(
        fn=get_llm_response,
        inputs=[completion_reward, player_logs],
        outputs=[bot1, bot2, bot3, bot4],
        queue=False,
    )

    get_first_llm_response_args = dict(
        fn=get_llm_response_once,
        inputs=[completion_reward, player_logs],
        outputs=bot1,
        queue=False,
    )

    get_second_llm_response_args = dict(
        fn=get_llm_response_once,
        inputs=[completion_reward, player_logs],
        outputs=bot2,
        queue=False,
    )

    get_third_llm_response_args = dict(
        fn=get_llm_response_once,
        inputs=[completion_reward, player_logs],
        outputs=bot3,
        queue=False,
    )

    get_fourth_llm_response_args = dict(
        fn=get_llm_response_once,
        inputs=[completion_reward, player_logs],
        outputs=bot4,
        queue=False,
    )

    start_generate_story.click(
        lambda: gr.update(visible=False), None, start_generate_story, queue=False
    ).then(
        lambda: create_visibility_updates(True, 5),
        None,
        [bot1, bot2, bot3, bot4, weaving],
        queue=False,
    ).then(
        **get_first_llm_response_args
    ).then(
        **get_second_llm_response_args
    ).then(
        **get_third_llm_response_args
    ).then(
        **get_fourth_llm_response_args
    ).then(
        lambda: gr.update(visible=True), None, [select_story], queue=False
    ).then(
        lambda: gr.update(visible=False), None, [weaving], queue=False
    )

    select_story.select(
        lambda: gr.update(visible=True), None, confirm_story, queue=False
    )

    set_player_selected_character_args = dict(
        fn=set_player_selected_character,
        inputs=[completion_reward, select_story],
        outputs=None,
        queue=False,
    )

    confirm_story.click(
        lambda: gr.update(interactive=False), None, [select_story], queue=False
    ).then(lambda: gr.update(visible=False), None, [confirm_story], queue=False).then(
        lambda: (gr.update(visible=True), gr.update(visible=True)),
        None,
        [start_generate_certificate, cancel_story],
        queue=False,
    ).then(
        **set_player_selected_character_args
    )

    cancel_story.click(
        lambda: gr.update(interactive=True), None, [select_story], queue=False
    ).then(lambda: gr.update(visible=False), None, [cancel_story], queue=False).then(
        lambda: (gr.update(visible=False), gr.update(visible=True)),
        None,
        [start_generate_certificate, confirm_story],
        queue=False,
    )

    create_certificate_args = dict(
        fn=create_certificate,
        inputs=[completion_reward],
        outputs=reward_result,
        queue=False,
    )

    complete_reward_args = dict(
        fn=complete_reward,
        inputs=[completion_reward],
        outputs=None,
        queue=False,
    )

    start_generate_certificate.click(
        lambda: create_visibility_updates(False, 18),
        None,
        [
            openai_img,
            aws_img,
            google_img,
            mtk_img,
            story_title,
            story_description,
            openai_description,
            aws_description,
            google_description,
            mtk_description,
            bot1,
            bot2,
            bot3,
            bot4,
            select_story,
            processing,
            cancel_story,
            start_generate_certificate,
        ],
        queue=False,
    ).then(lambda: gr.update(visible=True), None, [processing], queue=False).then(
        **create_certificate_args
    ).then(
        lambda: (gr.update(visible=True), gr.update(visible=False)),
        None,
        [complete, processing],
        queue=False,
    ).then(
        **complete_reward_args
    )

if __name__ == "__main__":
    demo.launch()