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,
)

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

        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("個人化戰報"):
        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("完賽獎勵"):
        with gr.Row():
            start_make_reward = gr.Button(
                "開始製作完賽獎勵！", visible=True, 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,
            )

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

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

        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,
    )

    player_info_query_btn.click(get_player_info, player_backend_id, player_info).then(
        get_player_logs, player_backend_id, player_logs
    ).then(
        render_player_data,
        player_info,
        [avatar, pet_gallery, badge_gallery, adventure_log, achievements, adventure],
    ).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,
    )

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

    cancel_player_name.click(
        lambda: (gr.update(interactive=True), gr.update(visible=True)),
        None,
        [player_name, confirm_player_name],
        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,
    )

    start_generate_story.click(
        lambda: gr.update(visible=False), None, start_generate_story, queue=False
    ).then(
        lambda: create_visibility_updates(True, 4),
        None,
        [bot1, bot2, bot3, bot4],
        queue=False,
    ).then(
        **get_llm_response_args
    ).then(
        lambda: gr.update(visible=True), None, [select_story], 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()