chore: cleanup

slack_status_cli/client.py

@@ -0,0 +1,80 @@
+import urllib.request
+import urllib.parse
+import typing
+import json
+
+import logger as log
+
+logger = log.get_logger(__name__)
+
+
+class SlackClient:
+    """
+    Lightweight abstraction around Slack's REST API.
+    """
+
+    _token: str
+
+    def __init__(self, token: str):
+        self._token = token
+
+    @property
+    def headers(self):
+        return {
+            "Authorization": f"Bearer {self._token}",
+        }
+
+    def _post(self, url: str, payload):
+        request = urllib.request.Request(
+            url,
+            urllib.parse.urlencode(payload).encode(),
+            method="POST",
+            headers=self.headers,
+        )
+
+        response = urllib.request.urlopen(request)
+        response_status = response.status
+        response_data = response.read()
+
+        logger.debug("API request: %s", str(payload))
+        logger.debug("API response: %s", str(response_data))
+
+        if response_status != 200:
+            raise Exception("Failed due to an API error.")
+
+        response_data = json.loads(response_data)
+
+        if not response_data["ok"]:
+            raise Exception("Failed due to an API error.")
+
+    def update_status(
+        self,
+        status: str,
+        emoticon: typing.Optional[str] = None,
+        expiration: typing.Optional[int] = None,
+    ):
+        """
+        Sets the Slack status of the given user to <status>, optionally with <emoticon> if provided.
+        If an expiration is provided, the status is set to expire after this time.
+
+        Reference: https://api.slack.com/methods/users.profile.set
+        """
+        payload = {
+            "profile": {
+                "status_text": status,
+                "status_emoji": emoticon or "",
+                "status_expiration": expiration or 0,
+            }
+        }
+
+        self._post("https://slack.com/api/users.profile.set", payload)
+
+    def set_do_not_disturb(self, duration_minutes: int):
+        """
+        Silences notifications, potentially with the specified duration.
+
+        Reference: https://api.slack.com/methods/dnd.setSnooze
+        """
+        payload = {"num_minutes": duration_minutes}
+
+        self._post("https://slack.com/api/dnd.setSnooze", payload)

slack_status_cli/logger.py

@@ -0,0 +1,20 @@
+import logging
+import os
+
+DEBUG = bool(os.environ.get("DEBUG", False))
+
+
+def get_logger(name: str) -> logging.Logger:
+    """
+    Prepares a standardized logger with the given name.
+    """
+    logging.basicConfig()
+    logger = logging.getLogger(name)
+    logger.propagate = False
+    logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)
+    log_handler = logging.StreamHandler()
+    log_handler.setLevel(level=logging.DEBUG if DEBUG else logging.INFO)
+    log_handler.setFormatter(logging.Formatter(fmt="%(message)s"))
+    logger.addHandler(log_handler)
+
+    return logger

slack_status_cli/main.py

@@ -13,11 +13,8 @@ With preset:
 SLACK_TOKEN=XXX slack-status-cli set --preset <preset-name> --duration <duration_description>
 """
 
-import urllib.request
-import urllib.parse
 import typing
 import os
-import logging
 import argparse
 import datetime
 import collections
@@ -25,27 +22,18 @@ import re
 import json
 import pathlib
 import sys
+import math
 
-# Debug mode modifies the log level used for reporting. If truthy,
+import client as slack_client
-# extra information is included in each run to diagnose common
+import logger as log
-# issues.
-DEBUG = bool(os.environ.get("DEBUG", False))
 
-# Logger setup
+logger = log.get_logger(__name__)
-logging.basicConfig()
-logger = logging.getLogger(__name__)
-logger.propagate = False
-logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)
-log_handler = logging.StreamHandler()
-log_handler.setLevel(level=logging.DEBUG if DEBUG else logging.INFO)
-log_handler.setFormatter(logging.Formatter(fmt="%(message)s"))
-logger.addHandler(log_handler)
 
 ParsedUserInput = collections.namedtuple(
-    "ParsedUserInput", ["text", "icon", "duration", "preset"]
+    "ParsedUserInput", ["text", "icon", "duration", "preset", "quiet"]
 )
 
-StatusPreset = collections.namedtuple("StatusPreset", ["text", "icon"])
+StatusPreset = collections.namedtuple("StatusPreset", ["text", "icon", "quiet"])
 Defaults = collections.namedtuple(
     "Defaults",
     [("icon"), ("duration")],
@@ -61,50 +49,6 @@ Configuration = collections.namedtuple(
 )
 
 
-def update_status(
-    token: str,
-    status: str,
-    emoticon: typing.Optional[str] = None,
-    expiration: typing.Optional[int] = None,
-):
-    """
-    Sets the Slack status of the given user to <status>, optionally with <emoticon> if provided.
-    If an expiration is provided, the status is set to expire after this time.
-    """
-    payload = {
-        "profile": {
-            "status_text": status,
-            "status_emoji": emoticon or "",
-            "status_expiration": expiration or 0,
-        }
-    }
-    headers = {
-        "Authorization": f"Bearer {token}",
-    }
-    request = urllib.request.Request(
-        "https://slack.com/api/users.profile.set",
-        urllib.parse.urlencode(payload).encode(),
-        method="POST",
-    )
-    for header_key, header_value in headers.items():
-        request.add_header(header_key, header_value)
-
-    response = urllib.request.urlopen(request)
-    response_status = response.status
-    response_data = response.read()
-
-    logger.debug("API request: %s", str(payload))
-    logger.debug("API response: %s", str(response_data))
-
-    if response_status != 200:
-        raise Exception("Failed to set status due to an API error.")
-
-    response_data = json.loads(response_data)
-
-    if not response_data["ok"]:
-        raise Exception("Failed to set status due to an API error.")
-
-
 def parse_input(known_presets: typing.List[str]) -> ParsedUserInput:
     """
     Handles command-line argument parsing and help text display.
@@ -130,11 +74,17 @@ def parse_input(known_presets: typing.List[str]) -> ParsedUserInput:
     set_parser.add_argument(
         "--preset", type=str, default=None, choices=known_presets, help="Preset to use"
     )
-
+    set_parser.add_argument(
+        "--quiet", type=bool, default=False, help="Silences notifications"
+    )
     args = parser.parse_args()
 
     return ParsedUserInput(
-        text=args.text, icon=args.icon, duration=args.duration, preset=args.preset
+        text=args.text,
+        icon=args.icon,
+        duration=args.duration,
+        preset=args.preset,
+        quiet=args.quiet,
     )
 
 
@@ -216,24 +166,38 @@ def run():
         if not token:
             raise Exception("Slack token not provided.")
 
+        client = slack_client.SlackClient(token=token)
+
         status_text = args.text
         status_icon = args.icon
         status_expiration = get_expiration(
             args.duration or configuration.defaults.duration
         )
+        quiet = args.quiet
 
         if args.preset:
             preset = configuration.presets[args.preset]
             status_text = preset.text
             status_icon = preset.icon
 
-        update_status(
+        client.update_status(
-            token,
             status_text,
             status_icon or configuration.defaults.icon,
             status_expiration,
         )
 
+        if quiet:
+            quiet_duration = math.ceil(
+                (
+                    datetime.datetime.fromtimestamp(status_expiration)
+                    - datetime.datetime.now()
+                ).seconds
+                / 60
+            )
+            client.set_do_not_disturb(quiet_duration)
+        else:
+            client.set_do_not_disturb(0)
+
         new_status = (
             "%s %s" % (status_icon, status_text) if status_icon else status_text
         )