erigon-tg-monitor.py

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from telegram.constants import ParseMode  # Corrected import here
import subprocess
import re
import os

def get_log_status():
    log_file = '/tmp/erigon.log'
    try:
        with open(log_file, 'r') as file:
            lines = file.readlines()
            for line in reversed(lines):
                if 'progress' in line:
                    match = re.search(r'progress="([^"]+)"', line)
                    if match:
                        return match.group(1)
            return "No progress information found."
    except Exception as e:
        return str(e)

def get_disk_usage():
    directory = '/data/erigon-data'
    try:
        result = subprocess.run(['du', '-sh', directory], capture_output=True, text=True)
        if result.returncode == 0:
            return result.stdout.strip()
        else:
            return "Error getting disk usage: " + result.stderr
    except Exception as e:
        return str(e)

async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
    progress_info = get_log_status()
    disk_usage = get_disk_usage()
    message = f"Download Progress: {progress_info}\nDisk Usage: {disk_usage}"
    await update.message.reply_text(message)

async def logs(update: Update, context: ContextTypes.DEFAULT_TYPE):
    log_file = '/tmp/erigon.log'
    try:
        with open(log_file, 'r') as file:
            lines = file.readlines()
            last_lines = ''.join(lines[-10:])  # Get the last 10 lines
            formatted_lines = f"```\n{last_lines}\n```"  # Format as code block
            await update.message.reply_text(formatted_lines, parse_mode=ParseMode.MARKDOWN_V2)
    except Exception as e:
        await update.message.reply_text(f"Error reading log file: {str(e)}")

def main():
    token = os.getenv('TELEGRAM_BOT_TOKEN')
    if not token:
        print("Token not found in environment variables")
        return
    
    application = Application.builder().token(token).build()
    
    status_handler = CommandHandler('status', status)
    logs_handler = CommandHandler('logs', logs)
    application.add_handler(status_handler)
    application.add_handler(logs_handler)
    
    application.run_polling()

if __name__ == '__main__':
    main()