Simple git repo/user activity tracker with Telegram notification

git-tracker.py

import requests
import time
import json
import logging
import os
import urllib.parse

# --- REQUIRED ---
TELEGRAM_BOT_TOKEN = ""
TELEGRAM_CHAT_ID = ""

# --- OPTIONAL TOKENS ---
GITHUB_TOKEN = "" 
GITLAB_TOKEN = ""

# --- GITHUB TRACKING ---
TRACKED_GITHUB_REPOS = [
    "v12-security/pocs"
]
TRACKED_GITHUB_USERS = [
    "0xdeadbeefnetwork"
]

# --- GITLAB TRACKING ---
GITLAB_URL = "https://gitlab.com" # Change this if they use a self-hosted GitLab instance

TRACKED_GITLAB_REPOS = [
    # "nightmare-eclipse/some-repo"
]
TRACKED_GITLAB_USERS = [
    "nightmare-eclipse"
]

CHECK_INTERVAL = 600
STATE_FILE = "git_advanced_state.json"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()]
)

# Caches to minimize GitLab API requests
GITLAB_USER_ID_CACHE = {}
GITLAB_PROJECT_CACHE = {}

# ==========================================
# UTILITIES & TELEGRAM
# ==========================================

def get_github_headers():
    headers = {"Accept": "application/vnd.github.v3+json"}
    if GITHUB_TOKEN:
        headers["Authorization"] = f"token {GITHUB_TOKEN}"
    return headers

def get_gitlab_headers():
    headers = {}
    if GITLAB_TOKEN:
        headers["PRIVATE-TOKEN"] = GITLAB_TOKEN
    return headers

def send_telegram_message(text):
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": text,
        "parse_mode": "HTML",
        "disable_web_page_preview": True
    }
    try:
        response = requests.post(url, json=payload, timeout=10)
        response.raise_for_status()
    except Exception as e:
        logging.error(f"Telegram API error: {e}")

# ==========================================
# GITHUB FUNCTIONS
# ==========================================

def get_new_github_commits(repo, last_sha):
    url = f"https://api.github.com/repos/{repo}/commits"
    try:
        response = requests.get(url, headers=get_github_headers(), timeout=10)
        if response.status_code != 200:
            logging.error(f"Access error for GitHub repo {repo}: HTTP {response.status_code}")
            return [], last_sha

        commits = response.json()
        if not commits:
            return [], last_sha

        new_commits = []
        for commit in commits:
            if commit["sha"] == last_sha:
                break
            new_commits.append({
                "sha": commit["sha"],
                "message": commit["commit"]["message"].split("\n")[0],
                "author": commit["commit"]["author"]["name"],
                "url": commit["html_url"]
            })
        
        latest_sha = commits[0]["sha"]
        new_commits.reverse()
        return new_commits, latest_sha

    except Exception as e:
        logging.error(f"Error while checking GitHub commits for {repo}: {e}")
        return [], last_sha

def get_new_github_user_events(username, last_event_id):
    url = f"https://api.github.com/users/{username}/events/public"
    try:
        response = requests.get(url, headers=get_github_headers(), timeout=10)
        if response.status_code != 200:
            logging.error(f"Access error for GitHub user {username}: HTTP {response.status_code}")
            return [], last_event_id

        events = response.json()
        if not events:
            return [], last_event_id

        new_events = []
        for event in events:
            if str(event["id"]) == str(last_event_id):
                break
            if event["type"] in ["CreateEvent", "PushEvent"]:
                if event["type"] == "CreateEvent" and event["payload"].get("ref_type") != "repository":
                    continue
                new_events.append(event)
        
        latest_event_id = str(events[0]["id"])
        new_events.reverse()
        return new_events, latest_event_id

    except Exception as e:
        logging.error(f"Error while checking GitHub events for {username}: {e}")
        return [], last_event_id

# ==========================================
# GITLAB FUNCTIONS
# ==========================================

def get_new_gitlab_commits(repo, last_sha):
    # GitLab requires the project path to be URL-encoded (e.g., namespace%2Frepo)
    encoded_repo = urllib.parse.quote(repo, safe="")
    url = f"{GITLAB_URL}/api/v4/projects/{encoded_repo}/repository/commits"
    try:
        response = requests.get(url, headers=get_gitlab_headers(), timeout=10)
        if response.status_code != 200:
            logging.error(f"Access error for GitLab repo {repo}: HTTP {response.status_code}")
            return [], last_sha

        commits = response.json()
        if not commits:
            return [], last_sha

        new_commits = []
        for commit in commits:
            if commit["id"] == last_sha:
                break
            new_commits.append({
                "sha": commit["id"],
                "message": commit["title"],
                "author": commit["author_name"],
                "url": commit["web_url"]
            })
        
        latest_sha = commits[0]["id"]
        new_commits.reverse()
        return new_commits, latest_sha

    except Exception as e:
        logging.error(f"Error while checking GitLab commits for {repo}: {e}")
        return [], last_sha

def get_gitlab_user_id(username):
    if username in GITLAB_USER_ID_CACHE:
        return GITLAB_USER_ID_CACHE[username]
    url = f"{GITLAB_URL}/api/v4/users?username={username}"
    try:
        response = requests.get(url, headers=get_gitlab_headers(), timeout=10)
        data = response.json()
        if data:
            user_id = data[0]["id"]
            GITLAB_USER_ID_CACHE[username] = user_id
            return user_id
    except Exception as e:
        logging.error(f"Failed to resolve GitLab User ID for {username}: {e}")
    return None

def get_gitlab_project_info(project_id):
    if project_id in GITLAB_PROJECT_CACHE:
        return GITLAB_PROJECT_CACHE[project_id]
    url = f"{GITLAB_URL}/api/v4/projects/{project_id}"
    try:
        response = requests.get(url, headers=get_gitlab_headers(), timeout=10)
        if response.status_code == 200:
            data = response.json()
            info = {
                "name": data.get("path_with_namespace", f"Project {project_id}"),
                "url": data.get("web_url", f"{GITLAB_URL}/projects/{project_id}")
            }
            GITLAB_PROJECT_CACHE[project_id] = info
            return info
    except:
        pass
    return {"name": f"Project {project_id}", "url": f"{GITLAB_URL}/projects/{project_id}"}

def get_new_gitlab_user_events(username, last_event_id):
    user_id = get_gitlab_user_id(username)
    if not user_id:
        return [], last_event_id

    url = f"{GITLAB_URL}/api/v4/users/{user_id}/events"
    try:
        response = requests.get(url, headers=get_gitlab_headers(), timeout=10)
        if response.status_code != 200:
            logging.error(f"Access error for GitLab user events ({username}): HTTP {response.status_code}")
            return [], last_event_id

        events = response.json()
        if not events:
            return [], last_event_id

        new_events = []
        for event in events:
            if str(event["id"]) == str(last_event_id):
                break
            
            action = event.get("action_name", "")
            target_type = event.get("target_type")
            if action.startswith("push") or (action in ["created", "imported"] and target_type == "Project"):
                new_events.append(event)
        
        latest_event_id = str(events[0]["id"])
        new_events.reverse()
        return new_events, latest_event_id

    except Exception as e:
        logging.error(f"Error while checking GitLab events for {username}: {e}")
        return [], last_event_id

# ==========================================
# MAIN ROUTINE
# ==========================================

def load_state():
    default_state = {"github_repos": {}, "github_users": {}, "gitlab_repos": {}, "gitlab_users": {}}
    if os.path.exists(STATE_FILE):
        try:
            with open(STATE_FILE, "r", encoding="utf-8") as f:
                data = json.load(f)
                # Backward compatibility migration layer for old state format
                if "repos" in data:
                    return {
                        "github_repos": data.get("repos", {}), 
                        "github_users": data.get("users", {}), 
                        "gitlab_repos": {}, 
                        "gitlab_users": {}
                    }
                return data
        except Exception:
            pass
    return default_state

def save_state(state):
    tmp_file = f"{STATE_FILE}.tmp"
    try:
        with open(tmp_file, "w", encoding="utf-8") as f:
            json.dump(state, f, indent=4)
        os.replace(tmp_file, STATE_FILE)
    except Exception as e:
        logging.error(f"Failed to save state atomically: {e}")

def main():
    logging.info("Starting Git Tracker...")
    state = load_state()
    is_first_run = not os.path.exists(STATE_FILE)

    while True:
        state_changed = False

        # --- GITHUB REPOS ---
        for repo in TRACKED_GITHUB_REPOS:
            last_sha = state["github_repos"].get(repo)
            new_commits, latest_sha = get_new_github_commits(repo, last_sha)
            if latest_sha != last_sha:
                state["github_repos"][repo] = latest_sha
                state_changed = True
                if not is_first_run and new_commits:
                    for commit in new_commits:
                        msg = (f"🐙 <b>GitHub: New commit!</b>\n\n"
                               f"<b>Repo:</b> {repo}\n"
                               f"<b>Author:</b> {commit['author']}\n"
                               f"<b>Message:</b> <i>{commit['message']}</i>\n\n"
                               f"<a href='{commit['url']}'>View Commit</a>")
                        send_telegram_message(msg)
                        time.sleep(1)

        # --- GITHUB USERS ---
        for user in TRACKED_GITHUB_USERS:
            last_event = state["github_users"].get(user)
            new_events, latest_event = get_new_github_user_events(user, last_event)
            if latest_event != last_event:
                state["github_users"][user] = latest_event
                state_changed = True
                if not is_first_run and new_events:
                    for event in new_events:
                        repo_name = event["repo"]["name"]
                        if event["type"] == "CreateEvent":
                            msg = (f"🐙 <b>GitHub: Repo created!</b>\n\n"
                                   f"<b>User:</b> {user}\n"
                                   f"<b>Repo:</b> {repo_name}\n"
                                   f"<a href='https://github.com/{repo_name}'>View Repo</a>")
                            send_telegram_message(msg)
                        elif event["type"] == "PushEvent":
                            commits_count = len(event["payload"].get("commits", []))
                            commits_info = f"<b>New commits:</b> {commits_count}\n" if commits_count > 0 else ""
                            msg = (f"🐙 <b>GitHub: Repo updated (Push)</b>\n\n"
                                   f"<b>User:</b> {user}\n"
                                   f"<b>Repo:</b> {repo_name}\n{commits_info}"
                                   f"<a href='https://github.com/{repo_name}'>View Repo</a>")
                            send_telegram_message(msg)
                        time.sleep(1)

        # --- GITLAB REPOS ---
        for repo in TRACKED_GITLAB_REPOS:
            last_sha = state["gitlab_repos"].get(repo)
            new_commits, latest_sha = get_new_gitlab_commits(repo, last_sha)
            if latest_sha != last_sha:
                state["gitlab_repos"][repo] = latest_sha
                state_changed = True
                if not is_first_run and new_commits:
                    for commit in new_commits:
                        msg = (f"🦊 <b>GitLab: New commit!</b>\n\n"
                               f"<b>Repo:</b> {repo}\n"
                               f"<b>Author:</b> {commit['author']}\n"
                               f"<b>Message:</b> <i>{commit['message']}</i>\n\n"
                               f"<a href='{commit['url']}'>View Commit</a>")
                        send_telegram_message(msg)
                        time.sleep(1)

        # --- GITLAB USERS ---
        for user in TRACKED_GITLAB_USERS:
            last_event = state["gitlab_users"].get(user)
            new_events, latest_event = get_new_gitlab_user_events(user, last_event)
            if latest_event != last_event:
                state["gitlab_users"][user] = latest_event
                state_changed = True
                if not is_first_run and new_events:
                    for event in new_events:
                        project_id = event.get("project_id")
                        proj_info = get_gitlab_project_info(project_id) if project_id else {"name": "Unknown", "url": GITLAB_URL}
                        
                        action = event.get("action_name", "")
                        if action in ["created", "imported"]:
                            msg = (f"🦊 <b>GitLab: Repo created!</b>\n\n"
                                   f"<b>User:</b> {user}\n"
                                   f"<b>Repo:</b> {proj_info['name']}\n"
                                   f"<a href='{proj_info['url']}'>View Repo</a>")
                            send_telegram_message(msg)
                        elif action.startswith("push"):
                            push_data = event.get("push_data", {})
                            commits_count = push_data.get("commit_count", 0)
                            commits_info = f"<b>New commits:</b> {commits_count}\n" if commits_count > 0 else ""
                            msg = (f"🦊 <b>GitLab: Repo updated (Push)</b>\n\n"
                                   f"<b>User:</b> {user}\n"
                                   f"<b>Repo:</b> {proj_info['name']}\n{commits_info}"
                                   f"<a href='{proj_info['url']}'>View Repo</a>")
                            send_telegram_message(msg)
                        time.sleep(1)

        if state_changed:
            save_state(state)
            
        if is_first_run:
            is_first_run = False
            logging.info("First run completed. State generated.")

        logging.info("Verification completed. Going to sleep...")
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    main()