Add GitLab connector and update main.py to support multiple destinations

.env.example

@@ -1,4 +1,7 @@
 MATTERMOST_ACCESS_TOKEN=
 MATTERMOST_SERVER_URL=https://my-mattermost.host
 
+GITLAB_ACCESS_TOKEN=
+GITLAB_SERVER_URL=https://gitlab.com
+
 SOURCE=autodetect # or apple_music or spotify or random

connectors/gitlab.py

@@ -0,0 +1,25 @@
+import requests
+
+class GitlabConnector:
+    def __init__(self, connection_config) -> None:
+        self.url = connection_config.get("url", "https://gitlab.com")
+        self.token = connection_config.get("token", "")
+        self.connect()
+
+    def connect(self):
+        self.headers = {
+            "PRIVATE-TOKEN": f"{self.token}",
+            "Content-Type": "application/json",
+        }
+        self.url = f"{self.url}/api/v4/user/status"
+
+    def send(self, data):
+        request_data = {
+            "emoji": data["emoji"],
+            "message": data["text"],
+            "availability": "not_set",
+            "clear_status_after": "30_minutes",
+        }
+        response = requests.patch(self.url, headers=self.headers, json=request_data)
+        if response.status_code != 200:
+            raise Exception(f"Failed to set GitLab status: {response.content!r}")

main.py

@@ -20,25 +20,21 @@ def get_status(source: str | None = None) -> dict:
     return {}
 
 
-def send_user_status(**kwargs) -> bool:
+def send_user_status(destination: str |None = None, status = {}) -> bool:
     try:
-        now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+        StatusSend(destination=destination).set_status(**status)
-        print(
-            f"{now} Setting Mattermost status to {kwargs.get('emoji',{}).get('icon','')} {kwargs.get('text')} ⏱️ for {kwargs.get('duration')} seconds"
-        )
-        StatusSend().set_status(**kwargs)
         return True
     except Exception as e:
         print(e)
         return False
 
 
-def main(source: str = "autodetect"):
+def main(source: str = "autodetect", destination: str = "mattermost"):
     status_curr = {"status": None}
     while True:
         status = get_status(source)
         if status.get("text", "") != status_curr.get("text", ""):
-            status_result = send_user_status(**status)
+            status_result = send_user_status(destination=destination, status=status)
             status_curr = status
             if not status_result:
                 continue
@@ -47,11 +43,15 @@ def main(source: str = "autodetect"):
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()
-    # Source can be "autodetect", "spotify", "apple_music", "random"
+    # Source can be "env", "autodetect", "spotify", "apple_music", "random"
+    parser.add_argument("--source", help="source to use for connector", default="env")
+    # Destination can be "env", "mattermost", "gitlab"
     parser.add_argument(
-        "--source", help="source to use for connector", default="autodetect"
+        "--destination", help="destination to use for connector", default="env"
     )
     args = parser.parse_args()
     if args.source == "env":
         args.source = os.getenv("SOURCE", "autodetect")
-    main(args.source)
+    if args.destination == "env":
+        args.destination = os.getenv("DESTINATION", "mattermost")
+    main(args.source, args.destination)

processors/music.py

@@ -4,7 +4,7 @@ from connectors.spotify import Spotify
 
 
 class MusicProcessor:
-    def __init__(self, source: str | None = None):
+    def __init__(self, source: str | None = None) -> None:
         self.source: str | None = source
         self.connector: Spotify | AppleMusic | None = self.get_connector()
 

processors/status_send.py

@@ -1,38 +1,58 @@
 import os
+import time
 from datetime import datetime, timedelta, timezone
 from dotenv import load_dotenv
 from connectors.mattermost import MattermostConnector
+from connectors.gitlab import GitlabConnector
 
 load_dotenv()
 
 
 class StatusSend:
-    def __init__(self):
+    def __init__(self, destination: str | None = None) -> None:
-        conf = {
+        self.destination: str | None = destination
-            "url": os.getenv("MATTERMOST_SERVER_URL"),
+        self.connector: MattermostConnector | GitlabConnector | None = self.get_connector()
-            "token": os.getenv("MATTERMOST_ACCESS_TOKEN"),
+
-        }
+    def get_connector(self) -> MattermostConnector | GitlabConnector | None:
-        self.connector = MattermostConnector(conf)
+        if self.destination == "mattermost":
+            conf = {
+                "url": os.getenv("MATTERMOST_SERVER_URL"),
+                "token": os.getenv("MATTERMOST_ACCESS_TOKEN"),
+            }
+            return MattermostConnector(conf)
+        elif self.destination == "gitlab":
+            conf = {
+                "url": os.getenv("GITLAB_SERVER_URL"),
+                "token": os.getenv("GITLAB_ACCESS_TOKEN"),
+            }
+            return GitlabConnector(conf)
+        else:
+            print("Invalid source")
+            return None
 
     def set_status(self, text, emoji, duration=None, **kwargs):
-        if duration is not None:
+        if self.connector:
-            expires_at = datetime.now(timezone.utc) + timedelta(seconds=duration)
+            if duration is not None:
-        if isinstance(emoji, dict):
+                expires_at = datetime.now(timezone.utc) + timedelta(seconds=duration)
-            emoji_name = emoji.get("name")
+            if isinstance(emoji, dict):
-        else:
+                emoji_name = emoji.get("name")
-            emoji_name = emoji
+            else:
-        data = {
+                emoji_name = emoji
-            "emoji": emoji_name,
+            data = {
-            "text": text,
+                "emoji": emoji_name,
-            "expires_at": expires_at.isoformat() if expires_at else None,
+                "text": text,
-        }
+                "expires_at": expires_at.isoformat() if expires_at else None,
-        try:
+            }
-            self.connector.send(data=data)
+            now = time.strftime("%H:%M:%S", time.localtime())
-        except Exception as e:
+            print(f"{now} Setting status to {self.destination.capitalize()} of {emoji.get("icon")} {text} ⏱️ for {duration} seconds")
-            print(e)
+            try:
+                self.connector.send(data=data)
+            except Exception as e:
+                print(e)
 
     def clear_status(self):
-        try:
+        if self.connector:
-            self.connector.send(data={"emoji": "", "text": "", "expires_at": ""})
+            try:
-        except Exception as e:
+                self.connector.send(data={"emoji": "", "text": "", "expires_at": ""})
-            print(e)
+            except Exception as e:
+                print(e)