github_commit_bot.py

#!/usr/bin/env python3
"""
GitHub Commit Monitor Bot
Monitors a GitHub repository for new commits and sends notifications to Slack.
"""

import os
import time
import json
import requests
from datetime import datetime
from typing import Optional, Dict, List
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('github_bot.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


class GitHubCommitBot:
    """Bot that monitors GitHub commits and sends Slack notifications."""
    
    def __init__(self, config_file: str = 'bot_config.json'):
        """Initialize the bot with configuration."""
        self.config = self._load_config(config_file)
        self.github_repo = self.config.get('github_repo')
        self.github_token = self.config.get('github_token', '')
        self.slack_webhook_url = self.config.get('slack_webhook_url')
        self.check_interval = self.config.get('check_interval', 60)  # seconds
        self.state_file = self.config.get('state_file', 'last_commit_state.json')
        self.last_commit_sha = self._load_last_commit()
        
        # Validate configuration
        if not self.github_repo:
            raise ValueError("github_repo must be specified in config")
        if not self.slack_webhook_url:
            raise ValueError("slack_webhook_url must be specified in config")
        
        logger.info(f"Initialized bot for repository: {self.github_repo}")
        logger.info(f"Check interval: {self.check_interval} seconds")
    
    def _load_config(self, config_file: str) -> Dict:
        """Load configuration from JSON file."""
        if not os.path.exists(config_file):
            logger.warning(f"Config file {config_file} not found. Creating default config.")
            default_config = {
                "github_repo": "owner/repo-name",
                "github_token": "",
                "slack_webhook_url": "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
                "check_interval": 60,
                "state_file": "last_commit_state.json"
            }
            with open(config_file, 'w') as f:
                json.dump(default_config, f, indent=2)
            logger.info(f"Please edit {config_file} with your settings")
            raise ValueError(f"Please configure {config_file} first")
        
        with open(config_file, 'r') as f:
            return json.load(f)
    
    def _load_last_commit(self) -> Optional[str]:
        """Load the last known commit SHA from state file."""
        if os.path.exists(self.state_file):
            try:
                with open(self.state_file, 'r') as f:
                    state = json.load(f)
                    return state.get('last_commit_sha')
            except Exception as e:
                logger.warning(f"Error loading state file: {e}")
        return None
    
    def _save_last_commit(self, commit_sha: str):
        """Save the last known commit SHA to state file."""
        state = {
            'last_commit_sha': commit_sha,
            'last_check': datetime.now().isoformat()
        }
        with open(self.state_file, 'w') as f:
            json.dump(state, f, indent=2)
    
    def _get_latest_commit(self) -> Optional[Dict]:
        """Fetch the latest commit from GitHub API."""
        url = f"https://api.github.com/repos/{self.github_repo}/commits"
        headers = {
            'Accept': 'application/vnd.github.v3+json',
        }
        
        if self.github_token:
            headers['Authorization'] = f'token {self.github_token}'
        else:
            logger.warning("No GitHub token provided. If repository is private, authentication is required.")
        
        try:
            response = requests.get(url, headers=headers, params={'per_page': 1})
            
            if response.status_code == 404:
                logger.error(f"Repository '{self.github_repo}' not found (404). Possible reasons:")
                logger.error("  1. Repository doesn't exist or name is incorrect")
                logger.error("  2. Repository is private and requires authentication (add github_token)")
                logger.error("  3. Repository name format should be 'owner/repo-name'")
                logger.error(f"  Check: https://github.com/{self.github_repo}")
                return None
            elif response.status_code == 403:
                logger.error(f"Access forbidden (403) for repository '{self.github_repo}'. Possible reasons:")
                logger.error("  1. Repository is private - add a GitHub token with 'repo' scope")
                logger.error("  2. Rate limit exceeded - add a GitHub token to increase limits")
                if response.headers.get('X-RateLimit-Remaining') == '0':
                    reset_time = response.headers.get('X-RateLimit-Reset', 'unknown')
                    logger.error(f"  Rate limit resets at: {reset_time}")
                return None
            elif response.status_code == 401:
                logger.error(f"Unauthorized (401) for repository '{self.github_repo}'.")
                logger.error("  Invalid or expired GitHub token. Please check your token.")
                return None
            
            response.raise_for_status()
            commits = response.json()
            
            if commits and len(commits) > 0:
                return commits[0]
            return None
        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching commits from GitHub: {e}")
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_detail = e.response.json()
                    if 'message' in error_detail:
                        logger.error(f"GitHub API error: {error_detail['message']}")
                except:
                    pass
            return None
    
    def _format_slack_message(self, commit: Dict) -> Dict:
        """Format commit information for Slack message."""
        sha = commit['sha'][:7]
        author = commit['commit']['author']['name']
        message = commit['commit']['message'].split('\n')[0]  # First line only
        url = commit['html_url']
        date = commit['commit']['author']['date']
        
        # Format date
        try:
            dt = datetime.fromisoformat(date.replace('Z', '+00:00'))
            formatted_date = dt.strftime('%Y-%m-%d %H:%M:%S UTC')
        except:
            formatted_date = date
        
        slack_message = {
            "text": f"New commit detected in {self.github_repo}!",
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": "🚀 New Commit Detected"
                    }
                },
                {
                    "type": "section",
                    "fields": [
                        {
                            "type": "mrkdwn",
                            "text": f"*Repository:*\n{self.github_repo}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Commit SHA:*\n`{sha}`"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Author:*\n{author}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Date:*\n{formatted_date}"
                        }
                    ]
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*Commit Message:*\n```{message}```"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"<{url}|View Commit on GitHub>"
                    }
                }
            ]
        }
        
        return slack_message
    
    def _send_slack_notification(self, commit: Dict) -> bool:
        """Send notification to Slack channel."""
        message = self._format_slack_message(commit)
        
        try:
            response = requests.post(
                self.slack_webhook_url,
                json=message,
                headers={'Content-Type': 'application/json'}
            )
            response.raise_for_status()
            logger.info(f"Successfully sent Slack notification for commit {commit['sha'][:7]}")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f"Error sending Slack notification: {e}")
            return False
    
    def check_for_new_commits(self) -> bool:
        """Check for new commits and send notification if found."""
        logger.info("Checking for new commits...")
        
        commit = self._get_latest_commit()
        if not commit:
            logger.warning("No commits found or error fetching commits")
            return False
        
        current_sha = commit['sha']
        
        # If we have a last commit and it's the same, no new commit
        if self.last_commit_sha and self.last_commit_sha == current_sha:
            logger.debug("No new commits detected")
            return False
        
        # New commit detected!
        logger.info(f"New commit detected: {current_sha[:7]}")
        
        # Send Slack notification
        if self._send_slack_notification(commit):
            # Update last commit SHA
            self.last_commit_sha = current_sha
            self._save_last_commit(current_sha)
            return True
        else:
            logger.error("Failed to send notification, will retry on next check")
            return False
    
    def run(self):
        """Run the bot in a continuous loop."""
        logger.info("Starting GitHub Commit Monitor Bot...")
        logger.info(f"Monitoring repository: {self.github_repo}")
        logger.info(f"Press Ctrl+C to stop")
        
        # Do an initial check
        if not self.last_commit_sha:
            logger.info("First run - checking for latest commit...")
            commit = self._get_latest_commit()
            if commit:
                self.last_commit_sha = commit['sha']
                self._save_last_commit(commit['sha'])
                logger.info(f"Initialized with commit: {commit['sha'][:7]}")
        
        try:
            while True:
                self.check_for_new_commits()
                time.sleep(self.check_interval)
        except KeyboardInterrupt:
            logger.info("Bot stopped by user")
        except Exception as e:
            logger.error(f"Unexpected error: {e}", exc_info=True)


def main():
    """Main entry point."""
    import sys
    
    config_file = sys.argv[1] if len(sys.argv) > 1 else 'bot_config.json'
    
    try:
        bot = GitHubCommitBot(config_file)
        bot.run()
    except Exception as e:
        logger.error(f"Failed to start bot: {e}")
        sys.exit(1)


if __name__ == '__main__':
    main()