payment-checker.py

#!/usr/bin/env python3

import calendar
import datetime
import pyotp
from googleapiclient.discovery import build                                                                                                                                                             
import oauth2client.client                                                                                                                                                                              
from google.oauth2 import service_account
import yaml
import time
from common import *
from hetzner import hetzner
from redswitches import redswitches
from vsys import vsys
from scaleway import scaleway
from knownsrv import knownsrv
from prahost import prahost
from ovh import ovh
from worldstream import worldstream
import os
import requests
import textwrap
import click

SCOPES = ['https://www.googleapis.com/auth/spreadsheets']

def clear_google_sheet(sa_secrets_file, spreadsheet_id, sheet_name):

    credentials = service_account.Credentials.from_service_account_file(sa_secrets_file, scopes=SCOPES)
    sheets_service = build('sheets', 'v4', credentials=credentials)

    # Clear the sheet starting from the second row
    request = sheets_service.spreadsheets().values().clear(spreadsheetId=spreadsheet_id, range=sheet_name + "!A2:Z")
    response = request.execute()
    print("Cleared the Google Sheet")

def append_google_sheet(sa_secrets_file, spreadsheet_id, sheet_name, row_list):

    credentials = service_account.Credentials.from_service_account_file(sa_secrets_file, scopes=SCOPES)
    sheets_service = build('sheets', 'v4', credentials=credentials)

    # Append the row to the sheet
    request = sheets_service.spreadsheets().values().append(spreadsheetId=spreadsheet_id, range=sheet_name + "!A:Z", valueInputOption="USER_ENTERED", body={"values": [row_list]})
    response = request.execute()
    print("Appended the row to the Google Sheet")

def remove_screenshots():

    import os

    # Remove all screenshots from the screenshots directory
    # png or html files
    for file in os.listdir("screenshots"):
        if file.endswith(".png") or file.endswith(".html"):
            os.remove("screenshots/" + file)

def alert_telegram(chat_id, token, date, type, login, details, days, status, header, attempts=None):

    # Markdown alert text template
    attempts_text = ""
    if attempts is not None:
        attempts_text = f"Attempts Used: {attempts}\n        "

    text = textwrap.dedent(
        """
        {header}
        Date Checked: {date}
        Account Type: {type}
        Login: {login}
        Account Details: {details}
        Days Remaining: {days}
        Payment Status: {status}
        {attempts_text}"""
    ).format(
        header=header,
        date=date,
        type=type,
        login=login,
        details=details,
        days=days,
        status=status,
        attempts_text=attempts_text
    )

    # Telegram API URL
    url = "https://api.telegram.org/bot{token}/sendMessage?chat_id={chat_id}&text={text}".format(token=token, chat_id=chat_id, text=text)
    requests.get(url)

@click.command()
@click.option("--config", required=True, help="Path to the config.yaml file")
def main(config):

    with open(config, "r") as stream:
        config = yaml.safe_load(stream)

    # Change working directory to the cd from config file
    os.chdir(config["cd"])

    # Current date in the format of YYYY-MM-DD
    current_date = datetime.datetime.now().strftime("%Y-%m-%d")

    # Clear the Google Sheet
    try:
        clear_google_sheet(config["google_sheets"]["sa_secrets_file"], config["google_sheets"]["spreadsheet_id"], config["google_sheets"]["sheet_name"])
    except Exception as e:
        print("Error:", e)
        alert_telegram(config["telegram"]["chat_id"], config["telegram"]["token"], current_date, "---", "---", "---", "---", "---", "ERROR: Google Sheet Clearing Failed")
        exit(1)

    # Remove all screenshots from the screenshots directory
    remove_screenshots()

    # Set this to False if any account has a payment issue
    no_issues = True

    # Iterate over the list of accounts
    item_number = 0
    for account in config["accounts"]:

        item_number += 1

        print("Checking account type {type} for the login {login}".format(**account))

        # Unset the variables
        account_details = "None"
        days_remaining = "None"
        payment_status = "None"
        attempts_made = 0

        # Set the proxy
        if "proxy" in account:
            proxy = account["proxy"]
        else:
            proxy = None

        # If there is no 2FA secret, then set it to None
        if "2fa" not in account:
            account["2fa"] = None

        # Get retry configuration with defaults
        max_attempts = config.get("attempts", 1)
        delay_between_attempts = config.get("attempt_delay", 0)

        # Try multiple attempts
        for attempt in range(1, max_attempts + 1):
            try:
                print(f"Attempt {attempt} of {max_attempts}")

                if account["type"] == "Hetzner":
                    account_details, days_remaining, payment_status = hetzner(account["login"], account["password"], account["2fa"], item_number, proxy)
                elif account["type"] == "RedSwitches":
                    account_details, days_remaining, payment_status = redswitches(account["login"], account["password"], account["2fa"], item_number, proxy)
                elif account["type"] == "VSys":
                    account_details, days_remaining, payment_status = vsys(account["login"], account["password"], account["2fa"], item_number, proxy)
                elif account["type"] == "Scaleway":
                    account_details, days_remaining, payment_status = scaleway(account["login"], account["password"], item_number, proxy)
                elif account["type"] == "KnownSRV":
                    account_details, days_remaining, payment_status = knownsrv(account["login"], account["password"], account["2fa"], item_number, proxy)
                elif account["type"] == "PraHost":
                    account_details, days_remaining, payment_status = prahost(account["login"], account["password"], account["2fa"], item_number, proxy)
                elif account["type"] == "OVH":
                    account_details, days_remaining, payment_status = ovh(account["login"], account["password"], account["consumer_key"], item_number, proxy)
                elif account["type"] == "Worldstream":
                    account_details, days_remaining, payment_status = worldstream(account["login"], item_number, proxy)

                # Check if the provider function returned error values
                if account_details == "Error" or payment_status == "Error":
                    print(f"Provider returned error values on attempt {attempt}")

                    # If this is not the last attempt, wait before retrying
                    if attempt < max_attempts:
                        print(f"Waiting {delay_between_attempts} seconds before next attempt...")
                        time.sleep(delay_between_attempts)
                        continue
                    else:
                        # Last attempt failed with error values
                        print(f"All {max_attempts} attempts failed")
                        attempts_made = attempt
                else:
                    # Success - we got valid values
                    attempts_made = attempt
                    print(f"Successfully checked account on attempt {attempt}")
                    break

            except Exception as e:
                print(f"Exception on attempt {attempt}: {e}")

                # If this is not the last attempt, wait before retrying
                if attempt < max_attempts:
                    print(f"Waiting {delay_between_attempts} seconds before next attempt...")
                    time.sleep(delay_between_attempts)
                else:
                    # Last attempt failed, set error values
                    print(f"All {max_attempts} attempts failed with exception")
                    account_details = "Error"
                    days_remaining = "Error"
                    payment_status = "Error"
                    attempts_made = attempt

        # Prepare the row to be written to the Google Sheet
        row_list = [
            current_date,
            account["type"],
            account["login"],
            account_details,
            days_remaining,
            payment_status
        ]

        # Write the data to the Google Sheet
        append_google_sheet(config["google_sheets"]["sa_secrets_file"], config["google_sheets"]["spreadsheet_id"], config["google_sheets"]["sheet_name"], row_list)

        # Send an alert to the Telegram if status is not True
        if payment_status != True:
            alert_telegram(config["telegram"]["chat_id"], config["telegram"]["token"], current_date, account["type"], account["login"], account_details, days_remaining, payment_status, "WARNING: Payment Issue Detected", attempts_made)
            # Also set the no_issues to False
            no_issues = False

        print("---")

        # Take a pause between the accounts just in case
        time.sleep(5)

    # If there are no issues, then send a message to the Telegram
    if no_issues:
        alert_telegram(config["telegram"]["chat_id"], config["telegram"]["token"], current_date, "---", "---", "---", "---", "All accounts are OK", "INFO: All Accounts OK")

if __name__ == "__main__":
    main()