dogetipbot-telegram/dogetipbot_telegram.py

from telegram.ext import Updater, CommandHandler, CallbackContext
from telegram import ParseMode, Update
from pycoin.symbols.doge import network
from pycoin.encoding.hexbytes import b2h_rev, h2b, h2b_rev
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
import logging
import os
import io
import requests
import json
import argparse

# Parsing arguments

parser = argparse.ArgumentParser(description="Dogetipbot telegram")
parser.add_argument("--telegram-api-key", default=os.environ.get("TELEGRAM_API_KEY"))
parser.add_argument("--private-key", default=os.environ.get("DOGE_WALLET_KEY"))
parser.add_argument("--db-path")

args = parser.parse_args()

TELEGRAM_API_KEY = args.telegram_api_key

# Dogecoin Wallet

DERIVATION_PATH = "44/3"
TX_FEE_PER_THOUSAND_BYTES = 100000000
DOGE_WALLET = network.keys.bip32_seed(args.private_key.encode())

# SQL

engine = create_engine(f"sqlite:///{args.db_path}")
sessionM = sessionmaker(bind=engine)
SESSION = sessionM()
Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)

    @hybrid_property
    def private_key(self):
        return DOGE_WALLET.subkey_for_path(f"{DERIVATION_PATH}/{self.id}")

    @hybrid_property
    def address(self):
        return self.private_key.address()

    @hybrid_property
    def balance(self):
        address_infos = requests.get(
            f"https://api.blockcypher.com/v1/doge/main/addrs/{self.address}"
        ).json()
        return (address_infos["final_balance"], address_infos["unconfirmed_balance"])

    @hybrid_property
    def unspent(self):
        unspent_outputs = requests.get(
            f"https://api.blockcypher.com/v1/doge/main/addrs/{self.address}?unspentOnly=true&includeScript=true"
        ).json()
        if unspent_outputs["final_balance"] > 0:
            tx_refs = []
            if "txrefs" in unspent_outputs:
                tx_refs.extend(unspent_outputs["txrefs"])
            if "unconfirmed_txrefs" in unspent_outputs:
                tx_refs.extend(unspent_outputs["unconfirmed_txrefs"])
            tx_ins = []
            for output in tx_refs:
                tx_ins.append(
                    network.Tx.Spendable(
                        output["value"],
                        h2b(output["script"]),
                        h2b_rev(output["tx_hash"]),
                        output["tx_output_n"],
                    )
                )
            return tx_ins, unspent_outputs["final_balance"]

    def __repr__(self):
        return f"{self.name}: {self.address}"


Base.metadata.create_all(engine)

# Logging

logging.basicConfig(
    level=logging.ERROR,
    format="%(asctime)s - %(name)s - %(levelname)s - \
                    %(message)s",
)

# Exceptions


class NoAccountError(Exception):
    pass


class NotEnoughDoge(Exception):
    pass


class AccountExisting(Exception):
    pass


class NotValidUnit(Exception):
    pass


# Core functions


def get_user(account):
    query = SESSION.query(User).filter_by(name=account)
    if query.count() > 0:
        return query.first()
    else:
        raise NoAccountError(account)


# API broken
# def get_value(amount):
#     data = requests.get("https://api.coinmarketcap.com/v1/ticker/dogecoin/?convert=EUR").json()
#     return round(float(data[0]['price_eur'])*amount, 2)


def transaction(sender, receiver, amount=None):
    pkey = get_user(sender).private_key
    unspent, balance = get_user(sender).unspent
    s = io.BytesIO()
    if amount is None:
        tx = network.tx_utils.create_signed_tx(unspent, [receiver], wifs=[pkey.wif()])
        tx.stream(s)
        tx_byte_count = len(s.getvalue())
        tx = network.tx_utils.create_signed_tx(
            unspent,
            [receiver],
            wifs=[pkey.wif()],
            fee=TX_FEE_PER_THOUSAND_BYTES * ((999 + tx_byte_count) // 1000),
        )
    elif balance >= amount:
        tx = network.tx_utils.create_signed_tx(
            unspent, [(receiver, amount), pkey.address()], wifs=[pkey.wif()]
        )
        tx.stream(s)
        tx_byte_count = len(s.getvalue())
        tx = network.tx_utils.create_signed_tx(
            unspent,
            [(receiver, amount), pkey.address()],
            wifs=[pkey.wif()],
            fee=TX_FEE_PER_THOUSAND_BYTES * ((999 + tx_byte_count) // 1000),
        )
    else:
        raise NotEnoughDoge
    push_obj = {"tx": tx.as_hex()}
    # print(push_obj)
    pushed_tx = requests.post(
        "https://api.blockcypher.com/v1/doge/main/txs/push", json=push_obj
    )
    pushed_tx.raise_for_status()
    return pushed_tx.json()["tx"]["hash"]
    # return "null"


# Telegram functions


def start(bot, update):
    bot.send_message(
        chat_id=update.message.chat_id,
        text="Bark ! Je suis un tipbot Dogecoin ! \n\n \
                     Pour commencer envoyez moi /register",
    )


def dogetip(update: Update, context: CallbackContext):
    try:
        montant = int(context.args[0])
        unit = context.args[1]
        destinataire = context.args[2][1:]
    except (IndexError, ValueError):
        context.bot.send_message(
            chat_id=update.message.chat_id,
            text="Syntaxe : /dogetip xxx doge @destinataire",
        )
    else:
        try:
            if unit == "doge":
                txid = transaction(
                    update.message.from_user.username,
                    get_user(destinataire).address,
                    montant * 100000000,
                )
            else:
                raise NotValidUnit(unit)
        except NotEnoughDoge:
            message = "Pas assez de doge @" + update.message.from_user.username
        except NoAccountError as e:
            message = (
                "Vous n'avez pas de compte @"
                + str(e)
                + "\n\n"
                + "Utilisez /register pour démarrer"
            )
        except NotValidUnit as e:
            message = str(e) + " n'est pas une unité valide"
        else:
            message = (
                "🚀 Transaction effectuée 🚀\n\n"
                + f"{str(montant)} DOGE\n"
                + f"@{update.message.from_user.username} → @{destinataire}\n\n"
                + f'<a href="https://blockchair.com/dogecoin/transaction/{txid}">Voir la transaction</a>'
            )

        context.bot.send_message(
            chat_id=update.message.chat_id, parse_mode=ParseMode.HTML, text=message
        )


def register(update: Update, context: CallbackContext):
    query = SESSION.query(User).filter_by(name=update.message.from_user.username)
    if query.count() > 0:
        update.message.reply_text("Vous avez déjà un compte")
    else:
        user = User(name=update.message.from_user.username)
        SESSION.add(user)
        SESSION.commit()
        update.message.reply_text(get_user(update.message.from_user.username).address)


def infos(update: Update, context: CallbackContext):
    try:
        address = get_user(update.message.from_user.username).address
        balance, unconfirmed_balance = get_user(
            update.message.from_user.username
        ).balance
    except NoAccountError as e:
        update.message.reply_text(
            "Vous n'avez pas de compte @"
            + str(e)
            + "\n\n"
            + "Utilisez /register pour démarrer"
        )
    else:
        update.message.reply_text(
            address
            + "\n\n"
            + str(balance / 100000000)
            + " DOGE\n"
            + str(unconfirmed_balance / 100000000)
            + " DOGE unconfirmed"
        )


def withdraw(update: Update, context: CallbackContext):
    try:
        unit = context.args[1]
        address = context.args[2]
    except (IndexError, ValueError):
        context.bot.send_message(
            chat_id=update.message.chat_id, text="Syntaxe : /withdraw xxx doge adresse"
        )
    else:
        if unit == "doge":
            if context.args[0] == "all":
                txid = transaction(update.message.from_user.username, address)
            else:
                montant = int(context.args[0])
                txid = transaction(
                    update.message.from_user.username, address, montant * 100000000
                )

        context.bot.send_message(
            chat_id=update.message.chat_id,
            parse_mode=ParseMode.MARKDOWN,
            text="Transaction effectuée !\n"
            + f"[tx](https://blockchair.com/dogecoin/transaction/{txid})",
        )


def users(update: Update, context: CallbackContext):
    query = SESSION.query(User).all()
    reply = ""
    for user in query:
        reply += (
            f"\n{repr(user)}\thttps://blockchair.com/dogecoin/address/{user.address}"
        )
    update.message.reply_text(reply)


# Telegram initialisation

updater = Updater(token=TELEGRAM_API_KEY, use_context=True)
dispatcher = updater.dispatcher

start_handler = CommandHandler("start", start)
dispatcher.add_handler(start_handler)

dogetip_handler = CommandHandler("dogetip", dogetip)
dispatcher.add_handler(dogetip_handler)

register_handler = CommandHandler("register", register)
dispatcher.add_handler(register_handler)

infos_handler = CommandHandler("infos", infos)
dispatcher.add_handler(infos_handler)

infos_handler = CommandHandler("users", users)
dispatcher.add_handler(infos_handler)

withdraw_handler = CommandHandler("withdraw", withdraw)
dispatcher.add_handler(withdraw_handler)


def main():
    updater.start_polling()
    updater.idle()