325 lines
9.5 KiB
Python
Executable File
325 lines
9.5 KiB
Python
Executable File
from telegram.ext import Updater, CommandHandler, CallbackContext, Application
|
|
from telegram.constants import ParseMode
|
|
from telegram import Update, Bot
|
|
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
|
|
import asyncio
|
|
|
|
# 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 = 1000000
|
|
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
|
|
|
|
|
|
async def start(update: Update, context: CallbackContext):
|
|
await update.message.reply_text(
|
|
"Bark ! Je suis un tipbot Dogecoin ! \n\n \
|
|
Pour commencer envoyez moi /register",
|
|
)
|
|
|
|
|
|
async 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
|
|
)
|
|
|
|
|
|
async 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)
|
|
|
|
|
|
async 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"
|
|
)
|
|
|
|
|
|
async 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})",
|
|
)
|
|
|
|
|
|
async 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
|
|
|
|
application = Application.builder().token(TELEGRAM_API_KEY).build()
|
|
|
|
start_handler = CommandHandler("start", start)
|
|
application.add_handler(start_handler)
|
|
|
|
dogetip_handler = CommandHandler("dogetip", dogetip)
|
|
application.add_handler(dogetip_handler)
|
|
|
|
register_handler = CommandHandler("register", register)
|
|
application.add_handler(register_handler)
|
|
|
|
infos_handler = CommandHandler("infos", infos)
|
|
application.add_handler(infos_handler)
|
|
|
|
infos_handler = CommandHandler("users", users)
|
|
application.add_handler(infos_handler)
|
|
|
|
withdraw_handler = CommandHandler("withdraw", withdraw)
|
|
application.add_handler(withdraw_handler)
|
|
|
|
|
|
def main():
|
|
application.run_polling(1.0)
|