Initial

app/commands/exit.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+
+import logging
+
+from webex_bot.models.command import Command
+
+log: logging.Logger = logging.getLogger(__name__)
+
+
+class ExitCommand(Command):
+    def __init__(self) -> None:
+        super().__init__(
+            command_keyword="exit",
+            help_message="Exit",
+            delete_previous_message=True,
+        )
+        self.sender: str = ""
+
+    def pre_execute(self, message, attachment_actions, activity) -> None:
+        pass
+
+    def execute(self, message, attachment_actions, activity) -> None:
+        pass
+
+    def post_execute(self, message, attachment_actions, activity) -> None:
+        pass

app/commands/submit_task.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+
+import logging
+
+from webex_bot.models.command import Command
+from webex_bot.models.response import Response, response_from_adaptive_card
+from webexteamssdk.models.cards import (
+    AdaptiveCard,
+    Column,
+    ColumnSet,
+    Date,
+    FontSize,
+    FontWeight,
+    Text,
+    TextBlock,
+)
+from webexteamssdk.models.cards.actions import Submit
+
+from app.utils.config import config
+from app.utils.n8n import submit_task
+
+log: logging.Logger = logging.getLogger(__name__)
+
+
+class SubmitTaskCommand(Command):
+    def __init__(self) -> None:
+        super().__init__(
+            command_keyword="submit_feedback_dstgmyn",
+            help_message="Submit Task",
+            chained_commands=[SubmitTaskCallback()],
+            delete_previous_message=True,
+        )
+        self.sender: str = ""
+
+    def pre_execute(self, message, attachment_actions, activity) -> None:
+        self.sender = activity.get("actor").get("id")
+
+    def execute(self, message, attachment_actions, activity) -> Response:
+
+        card_body: list = [
+            ColumnSet(
+                columns=[
+                    Column(
+                        items=[
+                            TextBlock(
+                                "Submit Task",
+                                weight=FontWeight.BOLDER,
+                                size=FontSize.MEDIUM,
+                            ),
+                            TextBlock(
+                                f"Add a task to {config.admin_first_name}'s To Do list. All fields are required.",
+                                wrap=True,
+                                isSubtle=True,
+                            ),
+                        ],
+                        width=2,
+                    )
+                ]
+            ),
+            ColumnSet(
+                columns=[
+                    Column(
+                        width=2,
+                        items=[
+                            Text(id="issue_title", placeholder="Summary", maxLength=100),
+                            Text(
+                                id="issue_description",
+                                placeholder="Description",
+                                maxLength=1000,
+                                isMultiline=True,
+                            ),
+                            Date(id="completion_date", placeholder="Completion Date"),
+                        ],
+                    ),
+                ]
+            ),
+        ]
+
+        if self.sender in config.admin_emails:
+            card_body.append(
+                ColumnSet(
+                    columns=[
+                        Column(
+                            width=1,
+                            items=[
+                                Text(
+                                    id="issue_requester",
+                                    placeholder="Requester Email (leave blank to submit for yourself)",
+                                    maxLength=100,
+                                ),
+                            ],
+                        ),
+                    ]
+                ),
+            )
+
+        card: AdaptiveCard = AdaptiveCard(
+            body=card_body,
+            actions=[
+                Submit(
+                    title="Submit",
+                    data={
+                        "callback_keyword": "submit_task_callback_rbamzfyx",
+                        "sender": self.sender,
+                    },
+                ),
+                Submit(title="Cancel", data={"command_keyword": "exit"}),
+            ],
+        )
+        return response_from_adaptive_card(card)
+
+
+class SubmitTaskCallback(Command):
+    def __init__(self) -> None:
+        super().__init__(
+            card_callback_keyword="submit_task_callback_rbamzfyx", delete_previous_message=True
+        )
+        self.msg: str = ""
+
+    def pre_execute(self, message, attachment_actions, activity) -> None:
+        issue_title: str = attachment_actions.inputs.get("issue_title")
+        issue_description: str = attachment_actions.inputs.get("issue_description")
+        completion_date: str = attachment_actions.inputs.get("completion_date")
+
+        sender: str = attachment_actions.inputs.get("sender")
+        issue_requester: str = attachment_actions.inputs.get("issue_requester") or sender
+
+        if not issue_title or not issue_description or not completion_date:
+            self.msg = "Please complete all fields."
+            return
+
+        result: bool = submit_task(
+            requestor=issue_requester,
+            summary=issue_title,
+            description=issue_description,
+            completion_date=completion_date,
+        )
+
+        self.msg = (
+            "Submitting your task..." if result else "Failed to submit task. Please try again."
+        )
+
+    def execute(self, message, attachment_actions, activity) -> str:
+        return self.msg

app/main.py

@@ -1,8 +1,32 @@
-#!/usr/local/bin/python3
+#!/usr/bin/env python3
+
+from webex_bot.webex_bot import WebexBot
+
+from app.commands.exit import ExitCommand
+from app.commands.submit_task import SubmitTaskCommand
+from app.utils.config import config
+
+
+def create_bot() -> WebexBot:
+    # Create a Bot Object
+    webex_bot: WebexBot = WebexBot(
+        bot_name=config.bot_name,
+        teams_bot_token=config.webex_token,
+        approved_domains=["cisco.com"],
+    )
+    webex_bot.commands.clear()
+    webex_bot.add_command(SubmitTaskCommand())
+    webex_bot.add_command(ExitCommand())
+    webex_bot.help_command = SubmitTaskCommand()
+    webex_bot.help_command.delete_previous_message = True
+
+    return webex_bot
 
-def main():
-    # Commands here
-    
 
 if __name__ == "__main__":
-    main()
+    try:
+        bot: WebexBot = create_bot()
+        bot.run()
+    except KeyboardInterrupt:
+        print("Shutting down bot...")
+        exit()

app/utils/config.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+
+
+import os
+
+
+class Config:
+    def __init__(self) -> None:
+        self.__bot_name: str = os.environ["BOT_NAME"]
+        self.__webex_token: str = os.environ["WEBEX_API_KEY"]
+        self.__admin_first_name: str = os.environ["ADMIN_FIRST_NAME"]
+        self.__admin_emails: list = os.environ["ADMIN_EMAIL"].split(",")
+        self.__n8n_webhook_url: str = os.environ["N8N_WEBHOOK_URL"]
+
+    @property
+    def bot_name(self) -> str:
+        return self.__bot_name
+
+    @property
+    def webex_token(self) -> str:
+        return self.__webex_token
+
+    @property
+    def admin_first_name(self) -> str:
+        return self.__admin_first_name
+
+    @property
+    def admin_emails(self) -> list:
+        return self.__admin_emails
+
+    @property
+    def n8n_webhook_url(self) -> str:
+        return self.__n8n_webhook_url
+
+
+config: Config = Config()

app/utils/datetime.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python3
+
+from datetime import datetime
+
+
+def timestamp_to_date(timestamp: int) -> str:
+    return datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")

app/utils/n8n.py

@@ -0,0 +1,26 @@
+import requests
+
+from app.utils.config import config
+
+
+def __n8n_post(data: dict) -> bool:
+    headers: dict = {"Content-Type": "application/json"}
+    resp: requests.Response = requests.post(
+        url=config.n8n_webhook_url,
+        headers=headers,
+        json=data,
+        timeout=10,
+        verify=False,
+    )
+    return bool(resp.status_code == 200)
+
+
+def submit_task(summary, description, completion_date, requestor) -> bool:
+    print(f"submit_task: {completion_date=}")
+    data: dict = {
+        "requestor": requestor,
+        "title": summary,
+        "description": description,
+        "completiondate": completion_date,
+    }
+    return __n8n_post(data=data)