Merge pull request #9 from ttpss930141011/development

feat: add google canlendar

requirements.txt

@@ -21,4 +21,4 @@ psycopg2==2.9.7
 openai==0.27.8
 tiktoken==0.4.0
 google-search-results==2.4.2
-flake8==6.1.0
+flake8==6.1.0

src/app/flask_postgresql/api/response.py

@@ -1,15 +1,29 @@
+from typing import List
+
 from linebot.v3.messaging import (
     ApiClient,
     Configuration,
     MessagingApi,
     ReplyMessageRequest,
-    TextMessage,
 )
 from linebot.v3.messaging.api_response import ApiResponse
+from linebot.v3.messaging.models.message import Message
+
+
+def create_response(
+    configuration: Configuration, reply_token: str, messages: List[Message]
+) -> ApiResponse:
+    """
+    Creates a response using the Line Bot API.
 
+    Args:
+        configuration (Configuration): The configuration object for the API client.
+        reply_token (str): The token used to identify the reply message.
+        messages (List[Message]): The list of messages to be included in the response.
 
-def create_response(configuration: Configuration, reply_token: str, *results) -> ApiResponse:
-    messages = [TextMessage(text=result) for result in results]
+    Returns:
+        ApiResponse: The API response object.
+    """
 
     with ApiClient(configuration) as api_client:
         line_bot_api = MessagingApi(api_client)

src/app/flask_postgresql/presenters/message_reply_presenter.py

@@ -1,14 +1,18 @@
 """ Module for the EventPresenter
 """
 
+from typing import List
+
+from linebot.v3.messaging.models.message import Message
+
 from src.interactor.dtos.event_dto import EventOutputDto
 from src.interactor.interfaces.presenters.message_reply_presenter import EventPresenterInterface
 
 
 class EventPresenter(EventPresenterInterface):
     """Class for the EventPresenter"""
 
-    def present(self, output_dto: EventOutputDto) -> str:
+    def present(self, output_dto: EventOutputDto) -> List[Message]:
         """
         Present the output DTO.
 

src/infrastructure/tools/__init__.py

@@ -3,6 +3,7 @@
 
 from src.app.flask_postgresql.configs import Config
 
+from .google_calendar import GoogleCalendarTool
 from .stock import CurrentStockPriceTool, StockPerformanceTool
 
 # initialize LangChain services
@@ -16,4 +17,5 @@
     ),
     CurrentStockPriceTool(),
     StockPerformanceTool(),
+    GoogleCalendarTool(),
 ]

src/infrastructure/tools/google_calendar.py

@@ -0,0 +1,70 @@
+import datetime
+import urllib
+from typing import Optional, Type
+
+from langchain.tools import BaseTool
+from pydantic import BaseModel, Field
+
+
+def create_gcal_url(
+    title="What?", date="20230524T180000/20230524T220000", location="Where?", description=""
+):
+    """
+    Generate a Google Calendar URL for creating a new event.
+
+    Args:
+        title (str): The title of the event. Defaults to "What?".
+        date (str): The date and time of the event in the format "yyyyMMddTHHmmss/yyyyMMddTHHmmss".
+                    Defaults to "20230524T180000/20230524T220000".
+        location (str): The location of the event. Defaults to "Where?".
+        description (str): The description of the event. Defaults to an empty string.
+
+    Returns:
+        str: The URL for creating a new event in Google Calendar.
+
+    """
+
+    base_url = "https://www.google.com/calendar/render?action=TEMPLATE"
+    event_url = f"{base_url}&text={urllib.parse.quote(title)}&dates={date}\
+            &location={urllib.parse.quote(location)}&details={urllib.parse.quote(description)}"
+
+    return event_url + "&openExternalBrowser=1"
+
+
+class GoogleCalendarGeneratorInput(BaseModel):
+    """Input for Google Calendar Generator."""
+
+    dates: str = Field(
+        ...,
+        description=f"Datetime symbol if text contained. format should be \
+        'YYYYMMDDTHHMMSS/YYYYMMDDTHHMMSS'. Current time is {datetime.date.today()}",
+    )
+    title: str = Field(..., description="Calendar Title symbol for reserve schedule.")
+    description: str = Field(
+        ..., description="Calendar Summary text symbol for schedule description."
+    )
+    location: str = Field(..., description="Calendar location symbol for reservation.")
+
+
+class GoogleCalendarTool(BaseTool):
+    name = "google_calendar_reservation"
+    description = "Generate Google Calendar url from user text first when containing time, date."
+    args_schema: Optional[Type[BaseModel]] = GoogleCalendarGeneratorInput
+
+    def _run(self, dates: str, title: str, description: str, location: str):
+        """
+        Generates a Google Calendar URL based on the given parameters.
+
+        Args:
+            dates (str): The dates of the event.
+            title (str): The title of the event.
+            description (str): The description of the event.
+            location (str): The location of the event.
+
+        Returns:
+            str: The generated Google Calendar URL.
+        """
+
+        result = self.create_gcal_url(title, dates, location, description)
+
+        return result

src/interactor/dtos/event_dto.py

@@ -3,7 +3,9 @@
 
 
 from dataclasses import asdict, dataclass
-from typing import Dict
+from typing import Dict, List
+
+from linebot.v3.messaging.models.message import Message
 
 
 @dataclass
@@ -24,7 +26,7 @@ class EventOutputDto:
 
     window: Dict
     user_input: str
-    response: str
+    response: List[Message]
 
     def to_dict(self):
         """Convert data into dictionary"""

src/interactor/interfaces/presenters/message_reply_presenter.py

@@ -3,7 +3,9 @@
 
 
 from abc import ABC, abstractmethod
-from typing import Dict
+from typing import List
+
+from linebot.v3.messaging.models.message import Message
 
 from src.interactor.dtos.event_dto import EventOutputDto
 
@@ -12,5 +14,5 @@ class EventPresenterInterface(ABC):
     """Class for the Interface of the WindowPresenter"""
 
     @abstractmethod
-    def present(self, output_dto: EventOutputDto) -> Dict:
+    def present(self, output_dto: EventOutputDto) -> List[Message]:
         """Present the Window"""

src/interactor/use_cases/create_text_message_reply.py

@@ -1,6 +1,10 @@
 """ This module is responsible for creating a new window.
 """
+from typing import List
+
 from langchain.agents import AgentExecutor
+from linebot.v3.messaging.models import TextMessage
+from linebot.v3.messaging.models.message import Message
 
 from src.interactor.dtos.event_dto import EventInputDto, EventOutputDto
 from src.interactor.interfaces.logger.logger import LoggerInterface
@@ -43,15 +47,30 @@ def _get_agent_executor(self, input_dto: EventInputDto) -> AgentExecutor:
             )
         return agent_executor
 
-    def execute(self, input_dto: EventInputDto) -> str:
+    def execute(self, input_dto: EventInputDto):
+        """
+        Executes the given event input DTO.
+
+        Args:
+            input_dto (EventInputDto): The event input DTO containing the necessary information for execution.
+
+        Returns:
+            EventOutputDto: The output DTO containing the result of the execution.
+
+        Raises:
+            None.
+        """
         validator = EventInputDtoValidator(input_dto.to_dict())
         validator.validate()
 
+        response: List[Message] = []
+
         if input_dto.window.get("is_muting") is True:
-            response = "靜悄悄的，什麼都沒有發生。"
+            response.append(TextMessage(text="靜悄悄的，什麼都沒有發生。"))
         else:
             agent_executor = self._get_agent_executor(input_dto)
-            response = agent_executor.run(input=input_dto.user_input)
+            result = agent_executor.run(input=input_dto.user_input)
+            response.append(TextMessage(text=result))
 
         output_dto = EventOutputDto(
             window=input_dto.window,

src/interactor/use_cases/create_text_message_reply_test.py

@@ -1,5 +1,7 @@
 from unittest import mock
 
+from linebot.v3.messaging.models import TextMessage
+
 from src.interactor.dtos.event_dto import EventInputDto, EventOutputDto
 from src.interactor.interfaces.logger.logger import LoggerInterface
 from src.interactor.interfaces.presenters.message_reply_presenter import EventPresenterInterface
@@ -35,7 +37,7 @@ def test_new_user_create_text_message_reply(mocker: mock, fixture_window):
     logger_mock.log_info.assert_called_once_with("Create reply successfully")
 
     output_dto = EventOutputDto(
-        window=fixture_window, user_input="Test input", response="Test output"
+        window=fixture_window, user_input="Test input", response=[TextMessage(text="Test output")]
     )
     presenter_mock.present.assert_called_once_with(output_dto)
     assert result == "Test output"
@@ -69,7 +71,7 @@ def test_regular_create_text_message_reply(mocker: mock, fixture_window):
     logger_mock.log_info.assert_called_once_with("Create reply successfully")
 
     output_dto = EventOutputDto(
-        window=fixture_window, user_input="Test input", response="Test output"
+        window=fixture_window, user_input="Test input", response=[TextMessage(text="Test output")]
     )
     presenter_mock.present.assert_called_once_with(output_dto)
     assert result == "Test output"
@@ -104,7 +106,7 @@ def test_create_text_message_reply_if_window_is_muting(mocker: mock, fixture_win
     logger_mock.log_info.assert_called_once_with("Create reply successfully")
 
     output_dto = EventOutputDto(
-        window=fixture_window, user_input="Test input", response="靜悄悄的，什麼都沒有發生。"
+        window=fixture_window, user_input="Test input", response=[TextMessage(text="靜悄悄的，什麼都沒有發生。")]
     )
     presenter_mock.present.assert_called_once_with(output_dto)
     assert result == "Test output"