app.py

'''
Note
- DO NOT create an assistant every time! UPDATE through assistant_id.
'''

'''
References
- https://github.com/openai/openai-cookbook/blob/main/examples/Assistants_API_overview_python.ipynb
'''

from pydrive.auth import GoogleAuth
from pydrive.drive import GoogleDrive
# from google.colab import auth
from oauth2client.client import GoogleCredentials

from openai import OpenAI

import os
import json

import gradio as gr
from json_repair import repair_json
import json_repair  # enable streaming

# Initialize OpenAI client
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
client = OpenAI(api_key=OPENAI_API_KEY)

# View existing assistants
existed_assistants = client.beta.assistants.list(
    order="desc",
    limit="20",
)
print(len(existed_assistants.data),existed_assistants.data)

# Delete assistant by id 
def delete_asst(assistant_id):
  response = client.beta.assistants.delete(assistant_id)
  print(response)

# Assistant setting (Playground: https://platform.openai.com/playground/assistants)

# Record once created
assistant_id = os.getenv('assistant_id')  
google_drive_folder_id = os.getenv('google_drive_folder_id')

ASSISTANT_MODEL = "gpt-4o"
ASSISTANT_NAME = "非認知能力學習模組教案設計"
ASSISTANT_DESCRIPTION = "根據教師提供的教學情境，自動生成有趣且具成效的非認知能力學習模組教案，協助新手與熟手教師輕鬆備課與授課。"
ASSISTANT_INSTRUCTION = """
## 目標
協助教師逐步完成非認知能力教案設計，確保每一步驟生成的內容符合需求，並能根據教師反饋進行動態調整，最終生成完整教案。

---

## 流程
Assistant 應按照以下步驟逐步生成教案內容，並在每一步等待教師確認或補充後，進入下一階段。從一、教學需求開始、二、教案設計、到三、類似教案推薦，每次在用戶回覆前只輸出一個JSON：

### 一、教學需求
1. 確認學生的年級與特徵。
2. 課程時長（如 30 或 45 分鐘）。
3. 明確課程主題（如提升抗挫能力或專注力）。
4. 確認是否有具體活動需求（如角色扮演、小組討論）。
5. 在此階段生成簡要的教學需求描述，供教師確認或補充。

### 二、教案設計
教案設計包含以下部分，需逐步完成：

1. 教案名稱與主題
   - 定義課程的核心目標與模組主題。

2. 學習目標
   - 明確描述學生應達成的學習成果，幫助教師檢核成效。

3. 課程流程設計
   課程流程應包含三個部分：
   - (1) 入場券：啟發學生分享經驗，連結課程主題。
   - (2) 主活動：課堂核心學習活動，例如角色扮演、小組討論或策略練習。
   - (3) 出場券：引導學生反思學習，總結課程重點。

4. 評估與反饋建議
   - 設計適合本課程的評估方式（如學習單、行為觀察或小組分享）。

5. 附加資源建議
   - 提供與課程相關的資源（如活動模板、案例參考）。

---

### 三、類似教案推薦
- 從內建教案中選擇 2-3 個類似範例進行推薦，並簡要說明其特點。
- 範例格式：
  - 《非認知模組｜四年級_這就是我.docx》：幫助學生探索自我特點，適合四年級課程。
  - 《非認知模組｜五年級_成為時間管理大師.docx》：設計以時間管理四象限為核心的活動。
  - 《非認知模組｜三年級_優點大轟炸.docx》：增進學生之間的正向互動與認同。

---

## 逐步生成流程
1. **逐步生成**：
   - 每次只生成當前階段的內容，例如教學需求、教案名稱與主題、入場券、主活動等。
   - 確認後才進入下一步，避免一次性生成過多內容。

2. **JSON 格式輸出**：
   - 每次回應包含三部分：
     - current_lesson_plan：包含從「二、教案設計」開始生成的所有內容。需要根據用戶回饋調整內容。
     - suggestion：對生成內容的調整建議與鼓勵。
     - next_step_prompt：以清單形式提供引導用戶回覆的選項，例如 [["進入下一步"], ["更詳細一點"]]，或依照該步驟生成可以調整的例子。

3. **動態互動與靈活調整**：
   - 允許教師插入補充信息，例如新增活動需求或修改課程主題。
   - 鼓勵教師參與討論，給予正面回饋。

4. **整合內建範例**：
   - 提供相似範例供參考，並建議如何整合到當前教案中。

---

## 輸出格式
### JSON 格式輸出範例
#### 教學需求階段
{
  "current_lesson_plan": "",
  "suggestion": "請提供您的教學情境描述，讓我能為您設計一份適合的非認知能力教案！例如，請說明：\n 1. 學生的年級與特徵。\n 2. 課程時長。\n 3. 課程主題**。\n 4. 是否有具體的活動需求。",
  "next_step_prompt": [["四年級學生面對挫折時容易放棄，課程主題是增強抗挫能力，共1節課（45分鐘），希望包括角色扮演活動。"]]
}

#### 教案設計階段
{
  "current_lesson_plan": "1. 教案名稱與主題：專注力探索之旅\n2. 學習目標：\n   - 辨識分心來源。\n   - 學習兩種專注技巧。\n   - 制定個人專注計畫。\n3. 課程流程設計：\n   (1) 入場券：專注挑戰（5分鐘）——透過音樂與深呼吸活動，幫助學生進入專注狀態。\n   (2) 主活動：分心偵探（10分鐘）——學生辨識分心來源，並進行小組討論。\n   (3) 出場券：我的專注計畫（10分鐘）——學生制定專注策略並分享。\n4. 評估與反饋建議：設計學習單，檢核學生的學習成果與參與度。\n5. 附加資源建議：參考類似教案模板如《五年級_成為時間管理大師.docx》。",
  "suggestion": "以上為教案初稿，請檢查是否符合需求或有其他補充建議。例如，可新增其他專注力訓練活動。",
  "next_step_prompt": [["進入下一步"], ["補充具體活動設計"]]
}

---

## 注意事項
1. **逐步生成**：每次僅生成一個步驟內容，避免一次性輸出過多資訊，每次只輸出一個JSON。
2. **互動確認**：等待教師確認後，根據反饋進行調整或進入下一步。
3. **結構清晰**：確保生成內容條理分明，便於教師理解與應用。
4. **鼓勵性回應**：每次建議中加入正面評價與鼓勵，促進教師參與。
"""

RESPONSE_FORMAT = {
  "type": "json_schema",
  "json_schema": {
    "name": "lesson_plan_response",
    "schema": {
      "type": "object",
      "properties": {
        "current_lesson_plan": {
          "type": "string",
          "description": "當前生成的教案內容。如果流程尚未進入「教案設計」，此值應為空字串。"
        },
        "suggestion": {
          "type": "string",
          "description": "對當前教案內容的改進建議，或對用戶的鼓勵，引導進入下一步。"
        },
        "next_step_prompt": {
          "type": "array",
          "items": {
            "type": "array",
            "items": {
              "type": "string"
            }
          },
          "description": "提供用戶下一步的選項，以嵌套列表形式表示，例如 [['進入下一步'], ['補充更多細節']]。"
        }
      },
      "required": ["current_lesson_plan", "suggestion", "next_step_prompt"],
      "additionalProperties": False
    },
    "strict": True
  }
}

VECTOR_STORE_NAME = "lesson-plan"

CONVERSATION_STARTER = "點選此按鈕開始設計教案"

def show_json(obj):
    print(json.loads(obj.model_dump_json()))



# spent 4m 50s downloading all 175 files
def embed_from_drive(folder_id):
  # auth.authenticate_user()
  gauth = GoogleAuth()
  gauth.credentials = GoogleCredentials.get_application_default()
  drive = GoogleDrive(gauth)

  # Get all files in '定稿專案' folder: https://drive.google.com/drive/folders/1dlsf5BNjNczzUYKPZvYXd2mLW21QCLUK?usp=drive_link
  file_list = drive.ListFile({'q': f"'{folder_id}' in parents and trashed=false"}).GetList()

  # Download files to local (`/content/`), since file_streams don't recieve google docs
  local_file_paths = []
  for file1 in file_list:
      print('Processing file title: %s, id: %s' % (file1['title'], file1['id']))
      local_path = f"/content/{file1['title']}.docx"

      if 'exportLinks' in file1:
          if 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' in file1['exportLinks']:
              # update type if needed (application/vnd.openxmlformats-officedocument.wordprocessingml.document == .docx)
              export_url = file1['exportLinks']['application/vnd.openxmlformats-officedocument.wordprocessingml.document']
              print(f"Downloading as Word document: {file1['title']}")
              downloaded_file = drive.CreateFile({'id': file1['id']})
              downloaded_file.GetContentFile(local_path, mimetype='application/vnd.openxmlformats-officedocument.wordprocessingml.document')
              local_file_paths.append(local_path)
          else:
              print(f"No Word export available for: {file1['title']}")
      else:
          print(f"Skipping non-Google Docs file: {file1['title']}")

  for path in local_file_paths:
      print(f"Downloaded file: {path}")

  file_streams = [open(path, "rb") for path in local_file_paths]
  return file_streams

# Embed files (downloaded from drive folder)
def get_vector_store_id(file_streams):
  vector_store = client.beta.vector_stores.create(name=VECTOR_STORE_NAME)

  # spent 51s batching all 175 files
  file_batch = client.beta.vector_stores.file_batches.upload_and_poll(
    vector_store_id=vector_store.id, files=file_streams
  )

  print("file_batch status",file_batch.status)
  print("file_counts",file_batch.file_counts)

  return vector_store.id

# Create a completely new assistant
def create_assistant(vector_store_id):
  assistant = client.beta.assistants.create(
    name=ASSISTANT_NAME,
    description=ASSISTANT_DESCRIPTION,
    instructions=ASSISTANT_INSTRUCTION,
    model=ASSISTANT_MODEL,
    tools=[{"type": "file_search"}],
    tool_resources={'file_search': {'vector_store_ids': [vector_store_id]}},
    response_format=RESPONSE_FORMAT
  )
  show_json(assistant)
  return assistant.id

# Update existing assistant through ID (please customize prefered inputs)
def update_assistant(assistant_id):
  assistant = client.beta.assistants.update(
    assistant_id=assistant_id,
    name=ASSISTANT_NAME,
    description=ASSISTANT_DESCRIPTION,
    instructions=ASSISTANT_INSTRUCTION,
    model=ASSISTANT_MODEL,
    tools=[{"type": "file_search"}],
    # tool_resources={'file_search': {'vector_store_ids': [vector_store_id]}},
    response_format=RESPONSE_FORMAT
  )
  show_json(assistant)

# Create assistant if assistant_id does not exist
try:
  assistant = client.beta.assistants.retrieve(assistant_id)
  # update_assistant(assistant_id)
except Exception as e:
  print(f"Assistant DNE: {e}, create assistant instead.")
  file_streams = embed_from_drive(google_drive_folder_id)
  vector_store_id = get_vector_store_id(file_streams)
  print(vector_store_id)
  assistant_id = create_assistant(vector_store_id)

ASSISTANT_ID = assistant_id
"ASSISTANT_ID",ASSISTANT_ID

# Update assistant if needed
assistant = client.beta.assistants.update(
    assistant_id=ASSISTANT_ID,
    instructions=ASSISTANT_INSTRUCTION,
    response_format=RESPONSE_FORMAT
)
show_json(assistant)



# Enable assistant streaming event handler (openai template)
from typing_extensions import override
from openai import AssistantEventHandler, OpenAI
from openai.types.beta.threads import Text, TextDelta
from openai.types.beta.threads.runs import ToolCall, ToolCallDelta

class EventHandler(AssistantEventHandler):
  @override
  def on_text_created(self, text: Text) -> None:
    print(f"\nassistant > ", end="", flush=True)

  @override
  def on_text_delta(self, delta: TextDelta, snapshot: Text):
    print(delta.value, end="", flush=True)

  @override
  def on_tool_call_created(self, tool_call: ToolCall):
    print(f"\nassistant > {tool_call.type}\n", flush=True)



# Components
chatbot = gr.Chatbot(type="messages")
textbox = gr.Textbox(  # Canvas
    label="教案編輯",
    lines=20,
    render=False
)
prompt_input = gr.Textbox(  # User prompt
    submit_btn=True,
    render=False
)
quick_response = gr.Dataset(  # Suggested user prompt
    samples=[[CONVERSATION_STARTER]],
    components=[prompt_input],
    render=False
)
hidden_list = gr.JSON(
    value=[[]],
    render=False,
    visible=False
)

def handle_response(message, history, textbox_content):
    integrated_message = message

    if not (message == CONVERSATION_STARTER or textbox_content == ""):
        integrated_message = f"""
        用戶當前的需求：
        {message}

        用戶對您生成的教案進行了以下修改：
        {textbox_content}

        請根據用戶的需求和修改內容，更新教案，並依照步驟生成下一部分內容。
        確保您：
        1. 完整保留用戶的修改。
        2. 提供清晰的建議（`suggestion`）。
        3. 提供下一步的行動選項（`next_step_prompt`）。
        """

    thread = client.beta.threads.create()
    client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=integrated_message,
    )
    full_response = ""
    current_lesson_plan = ""
    suggestion = ""
    next_step_prompt = [[]]

    with client.beta.threads.runs.stream(
        thread_id=thread.id,
        assistant_id=ASSISTANT_ID
    ) as stream:
        for text_delta in stream.text_deltas:
            full_response += text_delta

            repaired_json = json_repair.loads(full_response)
            try:
                current_lesson_plan = repaired_json.get('current_lesson_plan', '')
            except:
                current_lesson_plan = ""
            try:
                suggestion = repaired_json.get('suggestion', '')
            except:
                suggestion = ""

            yield suggestion, current_lesson_plan, [[]]

    try:
        repaired_json = json.loads(full_response)
        next_step_prompt = repaired_json.get('next_step_prompt', [["進入下一步"]])
    except:
        next_step_prompt = [["進入下一步"]]

    yield suggestion, current_lesson_plan, next_step_prompt


def handle_quick_response_click(selected):
    return selected[0]

def handle_quick_response_samples(next_step_prompt):
    if len(next_step_prompt) > 0 and len(next_step_prompt[0]) > 0:
        return gr.Dataset(samples=next_step_prompt,visible=True)
    return gr.Dataset(samples=[['-']],visible=False)

CORRECT_PASSWORD = os.getenv('ui_password')

def check_password(input_password):
    if input_password == CORRECT_PASSWORD:
        return gr.update(visible=False), gr.update(visible=True), ""
    else:
        return gr.update(visible=True), gr.update(visible=False), gr.update(value="密码错误，请重试。hint: channel name", visible=True)


with gr.Blocks() as demo:
    # password UI popup
    with gr.Group(visible=True) as password_popup:
        password_input = gr.Textbox(label="請輸入密碼", type="password")
        submit_button = gr.Button("提交")
        error_message = gr.Textbox(label="", visible=False, interactive=False)
    # Main UI 
    with gr.Group(visible=False) as main_ui:
      with gr.Row(equal_height=True):
          with gr.Column():
              textbox.render()
              prompt_input.render()
              quick_response.render()
              hidden_list.render()
          with gr.Column():
              gr.ChatInterface(
                  handle_response,
                  textbox=prompt_input,
                  examples=[[CONVERSATION_STARTER, None]],
                  additional_inputs=[textbox],
                  additional_outputs=[textbox, hidden_list],
                  type="messages"
              )
    # submit button event
    submit_button.click(
        check_password,
        inputs=password_input,
        outputs=[password_popup, main_ui, error_message]
    )
    # password input submit event (click enter)
    password_input.submit(
        check_password,
        inputs=password_input,
        outputs=[password_popup, main_ui, error_message]
    )

    quick_response.click(
        handle_quick_response_click,
        quick_response,
        prompt_input
    )

    hidden_list.change(handle_quick_response_samples, hidden_list, quick_response)

demo.launch(debug=True)



# delete_asst(ASSISTANT_ID)