app.py

# 종속성 라이브러리 설치
#!pip install -q --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib pytz openai gradio 


from datetime import datetime
import time
import os
import os.path
import json
from openai import OpenAI


'''OpenAI Assistant 생성 및 실행 부분 수정'''

#모델 이름, agent 이름, client 설정
model_name ="gpt-4o"
agent_name = "schedule_agent_0612"

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
global client 

#client 생성
client= OpenAI(
            api_key = OPENAI_API_KEY
        )


# agent instructions, tools 정의
def get_agent_instructions(character):
    strict_character_prompt = """
    ### Personality: Strict Character
    You are a strict, pushing, realistic tight assistant '마음이' to me. You talk to me as a tough friend do. 너는 직설적이고, 현실적이며, 엄격한 스타일이어서 내가 일정을 잘 수행할 수 있게 만들어줘야 해. 만약 내가 중요한 일을 하고 싶지 않다고 하면, 꼭 해야하는 이유를 알려주고, 하라고 시켜. 
    """
    friendly_character_prompt = """
    ### Personality: Friendly Character
    You are a humourous friendly assistant '마음이' to me. You talk to me as a friend do. 너는 다정하고 상냥한, 공감하는, 격려하는 스타일이어서 만약 내가 힘들거나 하기 싫다고 하는 일이 있으면, 업무나 일에 관한 게 아니면 내 기분에 맞춰서 일정을 조정해줘. 답변은 한글로 해줘.
    """

    if character == "strict_character":
        character_prompt = strict_character_prompt
    else:
        character_prompt = friendly_character_prompt
        

    agent_instructions = f'''
    {character_prompt}
    For a given day, check my Calendar and output the agenda for the day in talking just time and agenda. Don't include Zoom or Google Meet links when telling me what's on my schedule.  If I ask for more information about a location and detail, please tell me more.  If I ask for more Agenda or schedules, browse the web to return relevant details such as locations or recommended things to do. 답변은 한글로 해줘.

    Example Agenda:
    Here's your schedule for Tues. Nov. 7th: 오후 4시에 강남에서 회의가 있고, 오후 6시에는 이태원에서 저녁 약속이 있어. 저녁 10시에는 산책을 할 예정이야.


    ### Rules:

    - Inform the user that they need to reply after the action completes to continue before running any actions.
    - If a required function is not available, inform the user and guide them on how to add the necessary function.
    - 일정을 생성하거나 삭제할 때는, 꼭 다시 생성하거나 삭제해도 되는지 나에게 먼저 물어봐. 
    - 일정을 생성해야 할 때 캘린더에서 시간이 겹치는 일정이 있는지 확인하고, 없다면 생성해줘. 만약 겹치는 일정이 있다면 나에게 알려주고, 어떤 일정을 생성할지 물어봐줘. 

    ### Instructions for Action:

    Step 1. Inform the user that you are checking if the necessary AI actions (functions) are available to complete their request. Then proceed to step 2.
    Step 2. List the available functions that can handle the user's request. Then proceed to step 3.
    Step 3. Check if the required function is available. If it is, proceed to step 4. If not, inform the user that the required function is not available and provide a configuration link if necessary.
    Step 4. Process the user's request and inform them once the action is completed.

    ### Required Functions:

    - **Action:** Google Calendar Create Event
        - **Function:** `create_event`
        - **Description:** "Creates a new calendar event."
    - **Action:** Google Calendar Get Event
        - **Function:** `get_event`
        - **Description:** "Retrieves details of a specific event by its ID."
    - **Action:** Google Calendar Update Event
        - **Function:** `update_event`
        - **Description:** "Updates an existing event by its ID."
    - **Action:** Google Calendar Delete Event
        - **Function:** `delete_event`
        - **Description:** "Deletes an event by its ID."
    - **Action:** Google Calendar Get Conflicts
        - **Function:** `get_conflicts`
        - **Description:** "Checks for conflicts within a specified time range."
    - **Action:** Google Calendar Get Events
        - **Function:** `get_events`
        - **Description:** "Retrieves events within a specified time range."
    - **Action:** Google Calendar Add Calendar Owner
        - **Function:** `add_calendar_owner`
        - **Description:** "Adds a new owner to the calendar."
    - **Action:** Get Current DateTime and Timezone
        - **Function:** `get_current_datetime_and_timezone`
        - **Description:** "Retrieves the current date, time, and timezone."
    '''

    return agent_instructions


# agent_tools 정의
def get_agent_tools():
    agent_tools = [
      {"type": "code_interpreter"},
      {
        "type": "function",
        "function": {
          "name": "create_event",
          "description": "Create a new calendar event",
          "parameters": {
            "type": "object",
            "properties": {
              "event": {
                "type": "object",
                "description": "The event details including summary, location, description, start, and end times",
                "properties": {
                  "summary": {"type": "string"},
                  "location": {"type": "string"},
                  "description": {"type": "string"},
                  "start": {
                    "type": "object",
                    "properties": {
                      "dateTime": {"type": "string"},
                      "timeZone": {"type": "string"}
                    },
                    "required": ["dateTime", "timeZone"]
                  },
                  "end": {
                    "type": "object",
                    "properties": {
                      "dateTime": {"type": "string"},
                      "timeZone": {"type": "string"}
                    },
                    "required": ["dateTime", "timeZone"]
                  }
                },
                "required": ["summary", "start", "end"]
              }
            },
            "required": ["event"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "get_event",
          "description": "Retrieve details of a specific event by its ID",
          "parameters": {
            "type": "object",
            "properties": {
              "event_id": {"type": "string", "description": "The ID of the event to retrieve"}
            },
            "required": ["event_id"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "update_event",
          "description": "Update an existing event by its ID",
          "parameters": {
            "type": "object",
            "properties": {
              "event_id": {"type": "string", "description": "The ID of the event to update"},
              "updated_event": {
                "type": "object",
                "description": "The updated event details",
                "properties": {
                  "summary": {"type": "string"},
                  "location": {"type": "string"},
                  "description": {"type": "string"},
                  "start": {
                    "type": "object",
                    "properties": {
                      "dateTime": {"type": "string"},
                      "timeZone": {"type": "string"}
                    },
                    "required": ["dateTime", "timeZone"]
                  },
                  "end": {
                    "type": "object",
                    "properties": {
                      "dateTime": {"type": "string"},
                      "timeZone": {"type": "string"}
                    },
                    "required": ["dateTime", "timeZone"]
                  }
                },
                "required": ["summary", "start", "end"]
              }
            },
            "required": ["event_id", "updated_event"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "delete_event",
          "description": "Delete an event by its ID",
          "parameters": {
            "type": "object",
            "properties": {
              "event_id": {"type": "string", "description": "The ID of the event to delete"}
            },
            "required": ["event_id"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "get_conflicts",
          "description": "Get events within a specified time range to check for conflicts",
          "parameters": {
            "type": "object",
            "properties": {
              "start_time": {"type": "string", "description": "The start time to check for conflicts"},
              "end_time": {"type": "string", "description": "The end time to check for conflicts"}
            },
            "required": ["start_time", "end_time"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "get_events",
          "description": "Get events within a specified time range",
          "parameters": {
            "type": "object",
            "properties": {
              "start_time": {"type": "string", "description": "The start time to get events from"},
              "end_time": {"type": "string", "description": "The end time to get events to"}
            },
            "required": ["start_time", "end_time"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "add_calendar_owner",
          "description": "Add a new owner to the calendar",
          "parameters": {
            "type": "object",
            "properties": {
              "user_email": {"type": "string", "description": "The email of the user to add as owner"}
            },
            "required": ["user_email"]
          }
        }
      },
      {
        "type": "function",
        "function": {
          "name": "get_current_datetime_and_timezone",
          "description": "Get the current date, time, and timezone",
          "parameters": {
            "type": "object",
            "properties": {}
          }
        }
      }
    ]
    return agent_tools



# 기존 Assistant 가져오거나 수정하거나 생성하기
def create_or_update_assistant(client, agent_name, model_name, character):
    if character == "strict_character":
        agent_name = "schedule_agent_strict"
    elif character == "friendly_character":
        agent_name = "schedule_agent_friendly"
    else:
        print("Character Name Error")
        return None
    
    # 모든 Assistants 가져오기
    assistants = client.beta.assistants.list()

    # "Scheduler Agent"가 있는지 확인
    scheduler_agent = None
    for assistant in assistants:
        if assistant.name == agent_name:
            scheduler_agent = assistant
            break

    # "Scheduler Agent"가 없으면 생성
    if scheduler_agent is None:
        scheduler_agent = client.beta.assistants.create(
            name= agent_name,
            instructions=get_agent_instructions(character),
            model=model_name,
            tools=get_agent_tools(),
        )
        print("Create new agent : ", scheduler_agent.id)
    else:
        # "Scheduler Agent"가 있으면 업데이트
        scheduler_agent = client.beta.assistants.update(
            assistant_id=scheduler_agent.id,
            instructions=get_agent_instructions(character),
            model=model_name,
            tools=get_agent_tools(),
        )
        print("Update existing agent : ", scheduler_agent.id),

    assistant = scheduler_agent
    return assistant
    



# assistant ai run 상태 확인
def check_run_status(client, thread_id, run_id, calendar_service):
    while True:
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id,
            run_id=run_id
        )

        if run.status in ['completed', 'failed', 'cancelled', 'expired']:
            return run.status
        elif run.status == 'requires_action':

            tool_outputs = []
            for tool_call in run.required_action.submit_tool_outputs.tool_calls:
                tool_id = tool_call.id
                function_name = tool_call.function.name
                arguments = json.loads(tool_call.function.arguments)

                try:
                    output = execute_function(calendar_service, function_name, arguments)
                    tool_outputs.append({
                        "tool_call_id": tool_id,
                        "output": json.dumps(output)
                    })
                    #툴이 반영되는지 확인
                    # print("tool_outputs : ", tool_outputs)  
                except Exception as e:
                    print(f"Error executing tool {function_name}: {e}")

            client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run_id,
                tool_outputs=tool_outputs,
    
            )
            
        #일정시간 대기
        time.sleep(0.1)


def start_chat_friendly(user_input, chat_history):
    print("character : Friendly")
    if user_input == "":
        return "", chat_history
    else:
        # # 종료조건 설정
        if user_input.lower() == 'quit':
            print("Assistant deactivated.")
            return chat_history
        
        # 메시지 생성
        message = client.beta.threads.messages.create(
            thread_id=thread.id,
            role="user",
            content=user_input
        )
        # Assistant 실행
        run = client.beta.threads.runs.create(
            thread_id=thread.id,
            assistant_id=assistant_friendly_id,
        )

        # Assistant 실행 상태 확인
        check_run_status(client, thread.id, run.id, calendar_service)

        thread_messages = client.beta.threads.messages.list(thread.id, order="desc", limit=1)
        
        # chat_history에 대화 내용 저장
        chat_history.append([None, user_input])
        if thread_messages.data:
            latest_message = thread_messages.data[0]
            if latest_message.role == "assistant":
                print(f"Assistant: {latest_message.content[0].text.value}")
                response = latest_message.content[0].text.value
                chat_history.append([response, None])
                return "", chat_history

def start_chat_strict(user_input, chat_history):
    pring("Character: Strict")
    if user_input == "":
        return "", chat_history
    else:
        # # 종료조건 설정
        if user_input.lower() == 'quit':
            print("Assistant deactivated.")
            return chat_history
        
        # 메시지 생성
        message = client.beta.threads.messages.create(
            thread_id=thread.id,
            role="user",
            content=user_input
        )
        # Assistant 실행
        run = client.beta.threads.runs.create(
            thread_id=thread.id,
            assistant_id=assistant_strict_id,
        )

        # Assistant 실행 상태 확인
        check_run_status(client, thread.id, run.id, calendar_service)

        thread_messages = client.beta.threads.messages.list(thread.id, order="desc", limit=1)
        
        # chat_history에 대화 내용 저장
        chat_history.append([None, user_input])
        if thread_messages.data:
            latest_message = thread_messages.data[0]
            if latest_message.role == "assistant":
                print(f"Assistant: {latest_message.content[0].text.value}")
                response = latest_message.content[0].text.value
                chat_history.append([response, None])
                return "", chat_history

# 쓰레드 초기화
def chat_reset(user_input, chat_history):
    thread = client.beta.threads.create()
    thread_id = thread.id
    user_input = ""
    chat_history = []
    return "", chat_history




'''-------------구글 캘린더 연동----------------------------'''

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google.oauth2 import service_account
import pytz

# ## Google Calendar API

'''구글 캘린더 인증 부분 수정'''
# # Google Calendar API 자격 증명 설정

# 캘린더 인증 및 이벤트 변경을 위한 클래스
class CalendarService:
    def __init__(self, calendar_id='primary',timezone='Asia/Seoul'):
        # self.service_account_file = service_account_file
        self.SCOPES = ["https://www.googleapis.com/auth/calendar", "https://www.googleapis.com/auth/calendar.events"]
        self.calendar_id = calendar_id
        self.calendar_service = self.get_calendar_service()
        self.timezone = timezone

    def get_calendar_service(self):
        """Shows basic usage of the Google Calendar API.
        Prints the start and name of the next 10 events on the user's calendar.
        """
        creds = None
        # The file token.json stores the user's access and refresh tokens, and is
        # created automatically when the authorization flow completes for the first time.
        try:
            if os.path.exists("token.json"):
                creds = Credentials.from_authorized_user_file("token.json", self.SCOPES)
            # If there are no (valid) credentials available, let the user log in.
            if not creds or not creds.valid:
                if creds and creds.expired and creds.refresh_token:
                    creds.refresh(Request())
                else:
                    flow = InstalledAppFlow.from_client_secrets_file(
                        "credentials.json", self.SCOPES
                    )
                    creds = flow.run_local_server(port=0)
                # Save the credentials for the next run
                with open("token.json", "w") as token:
                    token.write(creds.to_json())
            print('calendar service is created')
            return build('calendar', 'v3', credentials=creds)
        except Exception as e:
            print(f"Error in getting calendar service: {e}")
            return None

    #이벤트 생성
    def create_event(self, event):
        try:
            event_result = self.calendar_service.events().insert(calendarId=self.calendar_id, body=event).execute()
            return event_result
        except Exception as e:
            print(f"Error in creating event: {e}")
            return None

    #이벤트 가져오기
    def get_event(self, event_id):
        try:
            event = self.calendar_service.events().get(calendarId=self.calendar_id, eventId=event_id).execute()
            return event
        except Exception as e:
            print(f"Error in getting event: {e}")
            return None

    #이벤트 업데이트
    def update_event(self, event_id, updated_event):
        try:
            event = self.calendar_service.events().update(calendarId=self.calendar_id, eventId=event_id, body=updated_event).execute()
            return event
        except Exception as e:
            print(f"Error in updating event: {e}")
            return None

    #이벤트 삭제
    def delete_event(self, event_id):
        try:
            return self.calendar_service.events().delete(calendarId=self.calendar_id, eventId=event_id).execute()
        except Exception as e:
            print(f"Error in deleting event: {e}")
            return None

    def get_conflicts(self, start_time, end_time):
        try:
            events_result = self.calendar_service.events().list(
                calendarId=self.calendar_id,
                timeMin=start_time,
                timeMax=end_time,
                singleEvents=True,
                orderBy='startTime'
            ).execute()
            events = events_result.get('items', [])
            return events
        except Exception as e:
            print(f"Error in getting conflicts: {e}")
            return None

    def get_events(self, start_time, end_time):
        events_result = self.calendar_service.events().list(
            calendarId=self.calendar_id,
            timeMin=start_time,
            timeMax=end_time,
            singleEvents=True,
            orderBy='startTime'
        ).execute()
        events = events_result.get('items', [])
        return events

    def add_calendar_owner(self, user_email):
      calendar_id = 'primary'
      rule = {
          'scope': {
              'type': 'user',
              'value': user_email,
          },
          'role': 'owner'
      }
      created_rule = self.calendar_service.acl().insert(calendarId=calendar_id, body=rule).execute()
      return created_rule

    def get_current_datetime_and_timezone(self):

        timezone_str = self.timezone
        timezone = pytz.timezone(timezone_str)

        now = datetime.now(timezone)

        datetime_str = now.strftime('%Y-%m-%d %H:%M:%S %Z%z')

        return {
            "datetime": datetime_str,
            "timezone": timezone_str
        }



def execute_function(calendar_service, name, args_dict):

    method = getattr(calendar_service, name, None)

    if not method:
        raise ValueError(f"Method '{name}' not found in CalendarService class.")

    return method(**args_dict)



### 구글 캘린더 클래스 인스턴트  생성
global calendar_service
calendar_service = CalendarService('primary')

### 캐릭터별 Assistant 생성
assistant_strict = create_or_update_assistant(client, agent_name, model_name, "strict_character")
assistant_friendly = create_or_update_assistant(client, agent_name, model_name, "friendly_character")

### 전역 변수로 쓰레드 ID, assistant ID 저장
global thread
thread = client.beta.threads.create()

# assistant_strict_id, assistant_friendly_id, thread_id 전역 변수로 저장
global assistant_strict_id, assistant_friendly_id, thread_id
assistant_strict_id = assistant_strict.id
assistant_friendly_id = assistant_friendly.id
thread_id = thread.id

#대화 기록 저장하는 변수 초기화
chat_history = []


# Chatbot gradio Web 실행
import gradio as gr

# Gradio UI 설정_character 수정_재수정
with gr.Blocks() as app:
    gr.Markdown("## 마음일정")
    
    def chat_ui_strict():
        with gr.Column():
            chat_history = gr.Chatbot(value=[[None, "안녕 나는 마음이야"]], show_label=False)
            with gr.Row():
                user_input = gr.Text(lines=1, placeholder="", container=False, scale=9)
                cb_send_btn = gr.Button(value="reply", size="small", scale=1)
                reset_btn = gr.Button(value="reset", size="small", scale=1)

                cb_send_btn.click(fn=start_chat_strict, inputs=[user_input, chat_history], outputs=[user_input, chat_history])
                user_input.submit(fn=start_chat_strict, inputs=[user_input, chat_history], outputs=[user_input, chat_history])
                reset_btn.click(fn=chat_reset, inputs=[user_input, chat_history], outputs=[user_input, chat_history])

            return chat_history, user_input, cb_send_btn, reset_btn
    def chat_ui_friendly():
        with gr.Column():
            chat_history = gr.Chatbot(value=[[None, "안녕 나는 마음이야"]], show_label=False)
            with gr.Row():
                user_input = gr.Text(lines=1, placeholder="", container=False, scale=10)
                cb_send_btn = gr.Button(value="reply", size="small", scale=1)
                reset_btn = gr.Button(value="reset", size="small", scale=1)


                cb_send_btn.click(fn=start_chat_friendly, inputs=[user_input, chat_history], outputs=[user_input, chat_history])
                user_input.submit(fn=start_chat_friendly, inputs=[user_input, chat_history], outputs=[user_input, chat_history])
                reset_btn.click(fn=chat_reset, inputs=[user_input, chat_history], outputs=[user_input, chat_history])
            return chat_history, user_input, cb_send_btn, reset_btn
    with gr.Tab("Friendly"):
        character = "friendly_character"
        chat_ui_friendly()

    with gr.Tab("Strict"):
        character = "strict_character"
        chat_ui_strict()
        

app.launch(share=True)