diff --git a/.env.example b/.env.example index 7ba7199..6aa1b2c 100644 --- a/.env.example +++ b/.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 \ No newline at end of file diff --git a/connectors/gitlab.py b/connectors/gitlab.py new file mode 100644 index 0000000..8ba20fc --- /dev/null +++ b/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}") diff --git a/main.py b/main.py index 1ec5302..b7bdc0d 100644 --- a/main.py +++ b/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()) - print( - f"{now} Setting Mattermost status to {kwargs.get('emoji',{}).get('icon','')} {kwargs.get('text')} ⏱️ for {kwargs.get('duration')} seconds" - ) - StatusSend().set_status(**kwargs) + StatusSend(destination=destination).set_status(**status) 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) diff --git a/processors/music.py b/processors/music.py index d9cf0b2..68c7ae0 100644 --- a/processors/music.py +++ b/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() diff --git a/processors/status_send.py b/processors/status_send.py index 3c846b2..d5de72b 100644 --- a/processors/status_send.py +++ b/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): - conf = { - "url": os.getenv("MATTERMOST_SERVER_URL"), - "token": os.getenv("MATTERMOST_ACCESS_TOKEN"), - } - self.connector = MattermostConnector(conf) + def __init__(self, destination: str | None = None) -> None: + self.destination: str | None = destination + self.connector: MattermostConnector | GitlabConnector | None = self.get_connector() + + def get_connector(self) -> MattermostConnector | GitlabConnector | None: + 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: - expires_at = datetime.now(timezone.utc) + timedelta(seconds=duration) - if isinstance(emoji, dict): - emoji_name = emoji.get("name") - else: - emoji_name = emoji - data = { - "emoji": emoji_name, - "text": text, - "expires_at": expires_at.isoformat() if expires_at else None, - } - try: - self.connector.send(data=data) - except Exception as e: - print(e) + if self.connector: + if duration is not None: + expires_at = datetime.now(timezone.utc) + timedelta(seconds=duration) + if isinstance(emoji, dict): + emoji_name = emoji.get("name") + else: + emoji_name = emoji + data = { + "emoji": emoji_name, + "text": text, + "expires_at": expires_at.isoformat() if expires_at else None, + } + now = time.strftime("%H:%M:%S", time.localtime()) + print(f"{now} Setting status to {self.destination.capitalize()} of {emoji.get("icon")} {text} ⏱️ for {duration} seconds") + try: + self.connector.send(data=data) + except Exception as e: + print(e) def clear_status(self): - try: - self.connector.send(data={"emoji": "", "text": "", "expires_at": ""}) - except Exception as e: - print(e) + if self.connector: + try: + self.connector.send(data={"emoji": "", "text": "", "expires_at": ""}) + except Exception as e: + print(e)