github_monitor_bot.py

import os
import time
import requests
import telegram
from telegram.ext import Updater, CommandHandler
from dotenv import load_dotenv
import random

# Load environment variables from .env file
load_dotenv()

# Configuration
GITHUB_TOKEN = os.getenv('GITHUB_TOKEN')  # GitHub personal access token
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')  # Telegram bot token
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')  # Telegram chat ID (channel or user)
REPOS = [
    {'owner': 'vercel', 'name': 'next.js'},
    {'owner': 'LayerZero-Labs', 'name': 'LayerZero-v2'},
    # Add more repositories as needed
]
IN_SCOPE_FILES = []  # Empty list means every file is in-scope
POLL_INTERVAL = 60  # Polling interval in seconds

# Initialize the Telegram bot
bot = telegram.Bot(token=TELEGRAM_TOKEN)

# Function to fetch the latest commit SHA for a repository
def get_latest_commit_sha(repo_owner, repo_name):
    url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/commits'
    headers = {'Authorization': f'token {GITHUB_TOKEN}'} if GITHUB_TOKEN else {}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        commits = response.json()
        if commits:
            return commits[0]['sha']
    return None

# Function to check if a commit modifies any in-scope files
def is_commit_in_scope(repo_owner, repo_name, commit_sha):
    url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/commits/{commit_sha}'
    headers = {'Authorization': f'token {GITHUB_TOKEN}'} if GITHUB_TOKEN else {}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        commit_data = response.json()
        for file in commit_data.get('files', []):
            file_path = file.get('filename', '')
            if not IN_SCOPE_FILES or any(scope_file in file_path for scope_file in IN_SCOPE_FILES):
                return True
    return False

# Function to send a notification to Telegram
def send_notification(repo_owner, repo_name, commit_sha):
    commit_url = f'https://github.com/{repo_owner}/{repo_name}/commit/{commit_sha}'
    url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/commits/{commit_sha}'
    headers = {'Authorization': f'token {GITHUB_TOKEN}'} if GITHUB_TOKEN else {}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        commit_data = response.json()
        commit_message = commit_data.get('commit', {}).get('message', 'No commit message')
        commit_author = commit_data.get('commit', {}).get('author', {}).get('name', 'Unknown')
        commit_date = commit_data.get('commit', {}).get('author', {}).get('date', 'Unknown')
        files_changed = commit_data.get('files', [])
        file_changes = '\n'.join([f"- {file.get('filename')} ({file.get('status')})" for file in files_changed])
        timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
        random_id = random.randint(1000, 9999)
        message = (
            f'🔔 *New Commit Detected!* (ID: {random_id})\n\n'
            f'*Repository:* {repo_owner}/{repo_name}\n'
            f'*Commit:* `{commit_sha}`\n'
            f'*Author:* {commit_author}\n'
            f'*Date:* {commit_date}\n'
            f'*Message:* {commit_message}\n\n'
            f'*Files Changed:*\n{file_changes}\n\n'
            f'*Link:* [View Commit]({commit_url})\n'
            f'*Notification Time:* {timestamp}'
        )
        bot.send_message(chat_id=TELEGRAM_CHAT_ID, text=message, parse_mode='Markdown')

# Function to start the monitoring process
def start_monitoring(update, context):
    update.message.reply_text('GitHub monitoring started!')
    for repo in REPOS:
        latest_sha = get_latest_commit_sha(repo['owner'], repo['name'])
        if latest_sha:
            context.user_data[f"{repo['owner']}/{repo['name']}"] = latest_sha
            update.message.reply_text(f"Latest commit SHA for {repo['owner']}/{repo['name']}: {latest_sha}")
        else:
            update.message.reply_text(f"Failed to fetch the latest commit SHA for {repo['owner']}/{repo['name']}.")

# Function to check for new commits
def check_commits(context):
    for repo in REPOS:
        latest_sha = get_latest_commit_sha(repo['owner'], repo['name'])
        if not latest_sha:
            continue

        repo_key = f"{repo['owner']}/{repo['name']}"
        if repo_key not in context.user_data:
            context.user_data[repo_key] = latest_sha
            continue

        if latest_sha != context.user_data[repo_key]:
            if is_commit_in_scope(repo['owner'], repo['name'], latest_sha):
                send_notification(repo['owner'], repo['name'], latest_sha)
            context.user_data[repo_key] = latest_sha

# Main function to run the bot
def main():
    updater = Updater(token=TELEGRAM_TOKEN, use_context=True)
    dispatcher = updater.dispatcher

    # Add command handler for /start
    dispatcher.add_handler(CommandHandler('start', start_monitoring))

    # Start the job to check for new commits
    job_queue = updater.job_queue
    job_queue.run_repeating(check_commits, interval=POLL_INTERVAL, first=0)

    # Start the bot
    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()