31_slack_status.py

#!/usr/bin/env python3
"""
Slack Status Updater - Sets Slack presence based on working hours
Automatically manages your Slack presence based on UK working hours,
accounting for weekends and bank holidays.

This script continuously monitors your Slack status and enforces the correct
presence based on working hours, even if manually changed within Slack.
"""

import logging
import os
import signal
import sys
import time
from datetime import datetime, date
from pathlib import Path

import dotenv
import pytz
import requests
from requests.exceptions import RequestException

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler(Path(__file__).parent / "slack_status.log"),
    ],
)
logger = logging.getLogger("slack_status")

# Load environment variables
script_dir = Path(__file__).parent
dotenv.load_dotenv()
env_local = script_dir / ".env.local"
if env_local.exists():
    logger.info(f"Loading environment from {env_local}")
    dotenv.load_dotenv(env_local, override=True)
else:
    logger.warning(f".env.local file not found at {env_local}")

# Configuration
SLACK_USER_OAUTH_TOKEN = os.getenv("SLACK_USER_OAUTH_TOKEN")
if not SLACK_USER_OAUTH_TOKEN:
    logger.critical("SLACK_USER_OAUTH_TOKEN not found in environment variables")
    sys.exit(1)

UK_TZ = pytz.timezone("Europe/London")
ACTIVE_HOURS = [(9, 0), (18, 0)]  # UK working hours (9 AM to 6 PM)
CHECK_INTERVAL = 300  # seconds (5 minutes)
MAX_RETRIES = 3
RETRY_DELAY = 5  # seconds

# Bank holidays list
BANK_HOLIDAYS = {
    date(2025, 4, 18),  # Good Friday
    date(2025, 4, 21),  # Easter Monday
    date(2025, 5, 5),  # Early May Bank Holiday
    date(2025, 5, 26),  # Spring Bank Holiday
    date(2025, 8, 25),  # Summer Bank Holiday
    date(2025, 12, 25),  # Christmas Day
    date(2025, 12, 26),  # Boxing Day
    date(2026, 1, 1),  # New Year's Day
    date(2026, 4, 3),  # Good Friday
    date(2026, 4, 6),  # Easter Monday
    date(2026, 5, 4),  # Early May Bank Holiday
    date(2026, 5, 25),  # Spring Bank Holiday
    date(2026, 8, 31),  # Summer Bank Holiday
    date(2026, 12, 25),  # Christmas Day
    date(2026, 12, 28),  # Boxing Day (Substitute)
}


class SlackStatusUpdater:
    def __init__(self):
        self.last_status = None
        self.run = True
        self.setup_signal_handlers()
        logger.info("Slack Status Updater initialized")

    def setup_signal_handlers(self):
        """Set up signal handlers for graceful shutdown"""
        signal.signal(signal.SIGINT, self.handle_shutdown)
        signal.signal(signal.SIGTERM, self.handle_shutdown)
        logger.debug("Signal handlers configured")

    def handle_shutdown(self, signum, frame):
        """Handle shutdown signals gracefully"""
        signal_name = signal.Signals(signum).name
        logger.info(f"Received signal {signal_name}, shutting down gracefully")
        self.run = False

    def get_current_presence(self):
        """Get the current Slack presence status"""
        logger.debug("Checking current Slack presence")

        for attempt in range(1, MAX_RETRIES + 1):
            try:
                response = requests.get(
                    "https://slack.com/api/users.getPresence",
                    headers={
                        "Authorization": f"Bearer {SLACK_USER_OAUTH_TOKEN}",
                        "Content-Type": "application/x-www-form-urlencoded",
                    },
                    timeout=10,
                )
                response.raise_for_status()
                result = response.json()

                if result.get("ok"):
                    presence = result.get("presence")
                    logger.debug(f"Current Slack presence is: {presence}")
                    return presence
                else:
                    error = result.get("error", "Unknown error")
                    logger.error(f"Slack API error while getting presence: {error}")

            except RequestException as e:
                logger.error(
                    f"Request failed (attempt {attempt}/{MAX_RETRIES}): {str(e)}"
                )

            if attempt < MAX_RETRIES:
                logger.debug(f"Retrying get presence in {RETRY_DELAY} seconds...")
                time.sleep(RETRY_DELAY)

        logger.error(f"Failed to get current presence after {MAX_RETRIES} attempts")
        return None

    def set_presence(self, presence):
        """Set Slack presence with retry mechanism"""
        # First check current presence
        current_presence = self.get_current_presence()

        # Only update if current presence doesn't match desired presence
        if current_presence == presence:
            logger.debug(f"Presence already set to {presence}, no action needed")
            self.last_status = presence
            return True

        logger.info(f"Current presence is {current_presence}, changing to: {presence}")

        for attempt in range(1, MAX_RETRIES + 1):
            try:
                response = requests.post(
                    "https://slack.com/api/users.setPresence",
                    headers={
                        "Authorization": f"Bearer {SLACK_USER_OAUTH_TOKEN}",
                        "Content-Type": "application/x-www-form-urlencoded",
                    },
                    data={"presence": presence},
                    timeout=10,
                )
                response.raise_for_status()
                result = response.json()

                if result.get("ok"):
                    logger.info(f"Successfully set presence to: {presence}")
                    self.last_status = presence
                    return True
                else:
                    error = result.get("error", "Unknown error")
                    logger.error(f"Slack API error: {error}")

            except RequestException as e:
                logger.error(
                    f"Request failed (attempt {attempt}/{MAX_RETRIES}): {str(e)}"
                )

            if attempt < MAX_RETRIES:
                logger.info(f"Retrying in {RETRY_DELAY} seconds...")
                time.sleep(RETRY_DELAY)

        logger.error(f"Failed to set presence after {MAX_RETRIES} attempts")
        return False

    def is_working_hours(self, now_uk):
        """Check if current time is during working hours"""
        today = now_uk.date()

        # Skip weekends and bank holidays
        if today.weekday() >= 5:
            logger.debug(f"Today is a weekend ({today.strftime('%A')})")
            return False

        if today in BANK_HOLIDAYS:
            logger.debug(f"Today is a bank holiday ({today})")
            return False

        # Check if within working hours
        start = now_uk.replace(
            hour=ACTIVE_HOURS[0][0], minute=ACTIVE_HOURS[0][1], second=0, microsecond=0
        )
        end = now_uk.replace(
            hour=ACTIVE_HOURS[1][0], minute=ACTIVE_HOURS[1][1], second=0, microsecond=0
        )

        return start <= now_uk <= end

    def run_loop(self):
        """Main execution loop"""
        logger.info("Starting Slack status updater service")

        while self.run:
            try:
                now_uk = datetime.now(UK_TZ)
                logger.debug(
                    f"Current time (UK): {now_uk.strftime('%Y-%m-%d %H:%M:%S')}"
                )

                if self.is_working_hours(now_uk):
                    self.set_presence("active")
                else:
                    self.set_presence("away")

            except Exception as e:
                logger.exception(f"Unexpected error: {str(e)}")

            # Sleep and check for any shutdown signals
            for _ in range(int(CHECK_INTERVAL / 1)):
                if not self.run:
                    break
                time.sleep(1)

        logger.info("Slack status updater service stopped")


if __name__ == "__main__":
    updater = SlackStatusUpdater()
    updater.run_loop()