xch_telegram.py

import subprocess
import time

from typing import Optional

import requests


TELEGRAM_BOT_API_KEY = '<telegram_api_key>'
TELEGRAM_CHAT_ID = '<telegram_chat_id>'
TELEGRAM_SEND_MESSAGE_URL = "https://api.telegram.org/bot{}/sendMessage".format(
    TELEGRAM_BOT_API_KEY
)
MINUTES_TO_SLEEP = 5


def run_chia_farm_summary() -> str:
    return str(subprocess.check_output(['chia', 'farm', 'summary']))


def parse_total_chia_farmed(farm_summary: str) -> float:
    return float(str(farm_summary).split('\\n')[1].split(':')[1])


def get_total_chia_farmed() -> Optional[float]:
    try:
        farm_summary = run_chia_farm_summary()
    except FileNotFoundError:
        print('ERROR: Cannot run chia!')
        return None

    if 'Farming status' not in farm_summary:
        print('ERROR: Unexpected Chia output')
        return None
    return parse_total_chia_farmed(farm_summary)

def send_telegram_message(text, chat_id):
    requests.post(TELEGRAM_SEND_MESSAGE_URL, json={'chat_id': chat_id, 'text': text})

def main():
    previous_chia_balance = None

    while True:
        print('Getting Chia balance...')
        current_chia_balance = get_total_chia_farmed()
        print(f'Current balance: {current_chia_balance} XCH')

        if current_chia_balance and current_chia_balance != previous_chia_balance:
            print('Nice! You farmed new XCH, sending message to Telegram!')
            message = f'Congrats! New Chia farmed. Total balance: {current_chia_balance} XCH'
            send_telegram_message(message, TELEGRAM_CHAT_ID)

        previous_chia_balance = current_chia_balance
        print(f'Sleeping for {MINUTES_TO_SLEEP} minutes...')
        time.sleep(MINUTES_TO_SLEEP * 60)

if __name__ == '__main__':
    main()