syncs my global status to all 3rd party status (slack, github)

sync_status.py

# /// script
# requires-python = ">=3.12"
# dependencies = ["aiohttp", "pydantic-settings"]
# ///

"""
Status Syndication Script
Syncs status from status.zzstoatzz.io to various third-party services
"""

import asyncio
import logging
from datetime import datetime
from typing import Any

import aiohttp
from aiohttp import ClientSession
from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class Settings(BaseSettings):
    """Application settings with sensible defaults"""
    model_config = SettingsConfigDict(
        env_file='.env',
        env_file_encoding='utf-8',
        extra='ignore'
    )
    
    github_token: SecretStr | None = Field(default=None, description="GitHub personal access token")
    slack_token: SecretStr | None = Field(default=None, description="Slack user token")
    status_url: str = Field(
        default="https://status.zzstoatzz.io/json",
        description="Status endpoint URL"
    )
    sync_interval: int = Field(default=60, description="Seconds between syncs", gt=0)
    oneshot: bool = Field(default=False, description="Run once and exit")
    
    @field_validator('github_token', 'slack_token')
    @classmethod
    def validate_tokens(cls, v: SecretStr | None) -> SecretStr | None:
        if v and len(v.get_secret_value()) < 10:
            raise ValueError("Token seems too short to be valid")
        return v


async def fetch_status(session: ClientSession, url: str) -> dict[str, Any] | None:
    """Fetch current status from source"""
    try:
        async with session.get(url) as response:
            if response.status == 200:
                return await response.json()
            logger.error(f"Failed to fetch status: HTTP {response.status}")
    except Exception as e:
        logger.error(f"Error fetching status: {e}")
    return None


async def update_github_status(
    session: ClientSession,
    token: str,
    text: str,
    emoji: str | None = None,
    expires: str | None = None
) -> bool:
    """Update GitHub status via GraphQL API"""
    
    api_url = "https://api.github.com/graphql"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    try:
        # Get user ID first
        user_query = """
        query {
            viewer {
                id
            }
        }
        """
        
        async with session.post(
            api_url,
            json={"query": user_query},
            headers=headers
        ) as response:
            data = await response.json()
            user_id = data['data']['viewer']['id']
        
        # Update status
        mutation = """
        mutation($input: ChangeUserStatusInput!) {
            changeUserStatus(input: $input) {
                status {
                    message
                    emoji
                }
            }
        }
        """
        
        # Log what we're sending
        logger.debug(f"Sending to GitHub - Message: '{text}', Emoji: '{emoji}', Expires: '{expires}'")
        
        # GitHub doesn't support custom emojis - use a fallback
        github_emoji = ":speech_balloon:"
        if emoji:
            if emoji.startswith("custom:"):
                # Custom emoji not supported by GitHub
                github_emoji = ":speech_balloon:"
            elif emoji.startswith(":") and emoji.endswith(":"):
                # Already in GitHub format
                github_emoji = emoji
            elif len(emoji) <= 2:
                # Unicode emoji
                github_emoji = emoji
            else:
                # Try wrapping in colons
                github_emoji = f":{emoji}:"
        
        variables = {
            "input": {
                "message": text,
                "emoji": github_emoji,
                "expiresAt": expires
            }
        }
        
        logger.debug(f"GitHub mutation variables: {variables}")
        
        async with session.post(
            api_url,
            json={"query": mutation, "variables": variables},
            headers=headers
        ) as response:
            if response.status == 200:
                result = await response.json()
                logger.debug(f"GitHub API response: {result}")
                if 'errors' not in result:
                    # Log the actual status that was set
                    if 'data' in result and 'changeUserStatus' in result['data']:
                        status_info = result['data']['changeUserStatus']['status']
                        logger.info(f"✅ GitHub status updated - Message: '{status_info.get('message')}', Emoji: '{status_info.get('emoji')}'")
                    else:
                        logger.info(f"✅ GitHub status updated: {text}")
                        logger.warning(f"Unexpected response structure: {result}")
                    return True
                logger.error(f"GitHub GraphQL errors: {result['errors']}")
            else:
                logger.error(f"GitHub API returned {response.status}")
                body = await response.text()
                logger.error(f"Response body: {body}")
                
    except Exception as e:
        logger.error(f"Failed to update GitHub status: {e}")
    
    return False


async def update_slack_status(
    session: ClientSession,
    token: str,
    text: str,
    emoji: str | None = None,
    expires: str | None = None
) -> bool:
    """Update Slack user status"""
    
    api_url = "https://slack.com/api/users.profile.set"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # Calculate expiration if provided
    status_expiration = 0
    if expires:
        try:
            exp_dt = datetime.fromisoformat(expires.replace('Z', '+00:00'))
            status_expiration = int(exp_dt.timestamp())
        except:
            pass
    
    # Handle custom emoji format - Slack needs :shortcode: format
    slack_emoji = ":speech_balloon:"  # default
    if emoji:
        if emoji.startswith("custom:"):
            # Extract the part after custom: and wrap in colons
            emoji_name = emoji[7:]
            # Try the custom emoji first, with fallback
            slack_emoji = f":{emoji_name}:"
        elif emoji.startswith(":") and emoji.endswith(":"):
            # Already in Slack format
            slack_emoji = emoji
        elif len(emoji) <= 2:
            # Unicode emoji - Slack handles these directly
            slack_emoji = emoji
        else:
            # Try wrapping in colons
            slack_emoji = f":{emoji}:"
    
    # First attempt with the parsed emoji
    payload = {
        "profile": {
            "status_text": text,
            "status_emoji": slack_emoji,
            "status_expiration": status_expiration
        }
    }
    
    try:
        async with session.post(
            api_url,
            json=payload,
            headers=headers
        ) as response:
            if response.status == 200:
                data = await response.json()
                if data.get('ok'):
                    logger.info(f"✅ Slack status updated: {text}")
                    return True
                elif data.get('error') == 'profile_status_set_failed_not_emoji_syntax':
                    # Emoji not recognized, retry with fallback
                    logger.warning(f"Emoji '{slack_emoji}' not recognized, using fallback")
                    payload['profile']['status_emoji'] = ":speech_balloon:"
                    
                    async with session.post(
                        api_url,
                        json=payload,
                        headers=headers
                    ) as retry_response:
                        if retry_response.status == 200:
                            retry_data = await retry_response.json()
                            if retry_data.get('ok'):
                                logger.info(f"✅ Slack status updated with fallback emoji: {text}")
                                return True
                        logger.error(f"Slack API retry failed: {retry_data.get('error', 'Unknown')}")
                else:
                    logger.error(f"Slack API error: {data.get('error', 'Unknown')}")
            else:
                logger.error(f"Slack API returned {response.status}")
                
    except Exception as e:
        logger.error(f"Failed to update Slack status: {e}")
    
    return False


async def sync_status(settings: Settings, last_status: dict[str, Any] | None = None) -> dict[str, Any] | None:
    """Sync status to all configured providers"""
    
    async with aiohttp.ClientSession() as session:
        # Fetch current status
        current = await fetch_status(session, settings.status_url)
        
        if not current:
            logger.warning("Could not fetch current status")
            return last_status
        
        # Check if status changed
        if last_status and last_status.get('text') == current.get('text'):
            logger.debug("Status unchanged, skipping sync")
            return last_status
        
        logger.info(f"📡 New status detected: {current.get('text')}")
        
        # Prepare status fields
        text = current.get('text', '')
        emoji = current.get('emoji')
        expires = current.get('expires')
        
        # Update all configured providers concurrently
        tasks = []
        
        if settings.github_token:
            tasks.append(update_github_status(
                session,
                settings.github_token.get_secret_value(),
                text,
                emoji,
                expires
            ))
        
        if settings.slack_token:
            tasks.append(update_slack_status(
                session,
                settings.slack_token.get_secret_value(),
                text,
                emoji,
                expires
            ))
        
        if tasks:
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Log any exceptions
            for i, result in enumerate(results):
                if isinstance(result, Exception):
                    logger.error(f"Provider {i} failed: {result}")
        
        return current


async def main():
    """Main entry point"""
    
    # Load and validate settings
    try:
        settings = Settings()
    except Exception as e:
        logger.error(f"Invalid configuration: {e}")
        return
    
    # Check if any providers are configured
    if not settings.github_token and not settings.slack_token:
        logger.error("No providers configured! Set GITHUB_TOKEN and/or SLACK_TOKEN")
        return
    
    # Log configured providers
    if settings.github_token:
        logger.info("✓ GitHub provider configured")
    if settings.slack_token:
        logger.info("✓ Slack provider configured")
    
    # Run once or forever
    if settings.oneshot:
        await sync_status(settings)
    else:
        logger.info(f"🚀 Starting status syndication (interval: {settings.sync_interval}s)")
        last_status = None
        
        while True:
            try:
                last_status = await sync_status(settings, last_status)
            except Exception as e:
                logger.error(f"Unexpected error in sync loop: {e}")
            
            await asyncio.sleep(settings.sync_interval)


if __name__ == "__main__":
    asyncio.run(main())