init: Initial commit

bot/msteams/adaptivecard/card.py

@@ -0,0 +1,14 @@
+class AdaptiveCard:
+    def __init__(self, body: list[dict] | None = None, actions: list[dict] | None = None) -> None:
+        self.__body = body if body else []
+        self.__actions = actions if actions else []
+        
+    @property
+    def data(self) -> dict:
+        return {
+            "type": "AdaptiveCard",
+            "body": self.__body,
+            "actions": self.__actions,
+            "version": "1.6",
+            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json"
+        }

bot/msteams/adaptivecard/template.py

@@ -0,0 +1,25 @@
+import os
+import json
+import re
+
+class Template:
+    def __init__(self, name: str) -> None:
+        self.__name = name
+        self.__path = "templates/" + name + ".json"
+        self.__template = json.load(open(self.__path))
+        
+    def __str__(self) -> str:
+        return json.dumps(self.__template)
+    
+    def replace_placeholders(self, data: dict) -> dict:
+        """Return the template as a dictionary with placeholders replaced."""
+        template_str = json.dumps(self.__template)
+        
+        for key, value in data.items():
+            template_str = template_str.replace(f'{{{{{key}}}}}', str(value))
+        
+        return json.loads(template_str)
+    
+    def render(self, data: dict) -> str:
+        """Return the template as a JSON string with placeholders replaced."""
+        return json.dumps(self.replace_placeholders(data))

bot/msteams/adaptivecard/webhook_card.py

@@ -0,0 +1,13 @@
+import os
+import json
+
+from bot.msteams.adaptivecard.card import AdaptiveCard
+from bot.msteams.adaptivecard.template import Template
+
+class WebhookCard(AdaptiveCard):
+    def __init__(self, data: dict, template: Template) -> None:
+        super().__init__(body=template.replace_placeholders(data)["body"], actions=template.replace_placeholders(data)["actions"])
+        
+    @property
+    def data(self) -> dict:
+        return super().data

bot/msteams/auth.py

@@ -0,0 +1,29 @@
+import requests
+
+from bot.msteams.token import Token
+
+class Auth:
+    def __init__(self, client_id: str, client_secret: str, tenant_id: str) -> None:
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.tenant_id = tenant_id  
+        self.__token = Token(self.__get_token_data())
+
+    def __str__(self) -> str:
+        if self.__token.is_expired:
+            self.__token = Token(self.__get_token_data())
+        return self.__token.token
+
+    def __get_token_data(self) -> dict:
+        url = f"https://login.microsoftonline.com/{self.tenant_id}/oauth2/v2.0/token"
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded"
+        }
+        data = {
+            "grant_type": "client_credentials",
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "scope": "https://api.botframework.com/.default"
+        }
+        response = requests.post(url, headers=headers, data=data)
+        return response.json()

bot/msteams/conversation.py

@@ -0,0 +1,58 @@
+import requests
+
+from bot.msteams.auth import Auth
+from bot.msteams.adaptivecard.card import AdaptiveCard
+#
+class Conversation:
+    def __init__(self, auth: Auth) -> None:
+        self.auth = auth
+
+    def new_conversation(self, channel_id: str, message: str) -> None:
+        access_token = str(self.auth)
+        url = "https://smba.trafficmanager.net/emea/v3/conversations"
+        headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json"
+        }
+        data = {
+            "isGroup": True,
+            "channelData": {
+                "channel": {
+                    "id": channel_id
+                }
+            },
+            "activity": {
+                "type": "message",
+                "text": message
+            }
+        }
+        response = requests.post(url, headers=headers, json=data)
+        return response.json()
+    
+    def new_conversation_with_card(self, channel_id: str, card: AdaptiveCard) -> None:
+        access_token = str(self.auth)
+        url = "https://smba.trafficmanager.net/emea/v3/conversations"
+        headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json"
+        }
+        data = {
+            "isGroup": True,
+            "channelData": {
+                "channel": {
+                    "id": channel_id,
+                }
+            },
+            "activity": {
+                "type": "message",
+                "text": "",
+                "attachments": [
+                    {
+                        "contentType": "application/vnd.microsoft.card.adaptive",
+                        "content": card.data
+                    }
+                ]
+            }
+        }
+        response = requests.post(url, headers=headers, json=data)
+        return response.json()

bot/msteams/token.py

@@ -0,0 +1,20 @@
+from datetime import datetime, timedelta
+
+class Token:
+    def __init__(self, data: dict) -> None:
+        self.data = data
+        self.__token = data["access_token"]
+        self.__expires_in = data["expires_in"]
+        self.__expires_at = datetime.now() + timedelta(seconds=self.__expires_in)
+        
+        
+    def __str__(self) -> str:
+        return self.__token
+    
+    @property
+    def is_expired(self) -> bool:
+        return datetime.now() > self.__expires_at
+    
+    @property
+    def token(self) -> str:
+        return self.__token