tutorial

Building a NFT minting bot with Cardano, Telegram, and Python - part 5

Part 5: Finishing the conversation

Finishing the conversation

Outline

Intro

OK, now on to the fun part. We created out database, created scripts for working with IPFS and the Cardano CLI. Now we will put it all together in a bot you can use.

I took the "creepy Mark Zuckerberg" version of conversationbot.py example from part 1 and expanded it into a more functional bot.

Code walk through

This is another long walk through and you should get a cup of coffee before starting out.

As usual we bring in our imports, but this time I am bringing in the functions from the other scripts to give the bot something to work with. We also setup the main logging functionality.

# app.py

import os
from telegram import Update
from telegram.ext import Updater, CommandHandler, MessageHandler, \
    Filters, CallbackContext, ConversationHandler, Defaults
from uuid import uuid4
import logging

# Local Imports
from db_model import Session, Tokens
from ipfs_utils import create_ipfs, pin_ipfs
from token_utils import pre_mint, mint, get_tx_details, check_wallet_utxo

logging.basicConfig(
    level=logging.INFO,
    filename='logging.log',
    filemode='w',
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logging.info('****** cNFT TESTNET BOT START ******')

The bot script uses a newer form of python coding that I personally have yet to start using. I do like it as I then know what a function is expecting as a return type.

We pull in the bot Telegram API team we received from the BotFather. We also setup a form of states that we will expand on later on in the main() function.

API_TOKEN = os.getenv('BOT_API_TOKEN')

# Conversation STATES
PHOTO, TICKER, NAME, DESCRIPTION, NUMBER, PRE_MINT = range(6)

This little function get_chat_info takes some of the data available and puts it in a nice dictionary for use later on.


def get_chat_info(chat_update):
    """ Helper function to gather some chat details """
    msg_dict = {
        'username': chat_update.message.from_user.username,
        'user_id': chat_update.message.from_user.id,
        'message_id': chat_update.message.message_id,
        'is_bot': chat_update.message.from_user.is_bot
    }
    logging.info("******* CHAT INFO ********")
    logging.info(msg_dict)
    logging.info("***************")
    return msg_dict

This is where the conversation starts, it gives the user some warm fuzzies and sets them up to upload their image via the Telegram client Photo upload. It then sends the user to the PHOTO state.


def start(update: Update, context: CallbackContext) -> int:
    """ Starts the whole process """
    chat_info = get_chat_info(chat_update=update)
    context.bot.send_message(
        chat_id=update.effective_chat.id,
        text=f"Welcome @{chat_info['username']} to the *cNFT TESTNET Bot*."
    )
    context.bot.send_message(
        chat_id=update.effective_chat.id,
        text="We will be preforming the following actions in this session: \n"
             " - Uploading and Pinning a photo to IPFS \n"
             " - Gathering the token metadata from you (Ticker, Name, etc.)\n"
             " - Funding the token creator account to create the token (5 ADA) \n"
             " - Minting the token and returning it back to the address sending the initial funds \n"
    )
    context.bot.send_message(
        chat_id=update.effective_chat.id,
        text="_Note: The 5 ADA will be returned to the address "
             "sending the initial funds subtracting the required "
             "token minting fees (~2-3 ADA)._"
    )
    context.bot.send_message(
        chat_id=update.effective_chat.id,
        text="Please begin by uploading a *PHOTO* using the PHOTO "
             "upload feature in your Telegram client to start the minting process."
             "\n I will then upload it to IPFS, pin it, and return the IPFS hash."
    )
    return PHOTO

The PHOTO state takes the image, downloads it, then uploads it and pins it to IPFS. It uses the create_ipfs function from the previously built ipfs_utils.py we covered in part 3.

Once successfully uploaded and pinned, it kicks the user over to the TICKER state to start gathering token metadata.


def photo(update: Update, context: CallbackContext) -> int:
    chat_info = get_chat_info(chat_update=update)
    photo_file = update.message.photo[-1].get_file()
    logging.info(update.message.photo[-1])
    file_name = f"{chat_info['username']}_{chat_info['user_id']}_{chat_info['message_id']}.jpg"
    photo_file.download(file_name)
    # Upload to BF
    res = create_ipfs(image=file_name)
    if res:
        # Pin it
        pin_response = pin_ipfs(ipfs_hash=res['ipfs_hash'])
        if pin_response:
            update.message.reply_text(
                "Hooray! Your image has been uploaded to IPFS and pinned:"
            )
            update.message.reply_text(
                f"Here is a handy link for you to keep. \n"
                f"https://gateway.ipfs.io/ipfs/{res['ipfs_hash']}"
            )
            context.user_data["token_ipfs_hash"] = res['ipfs_hash']
        else:
            update.message.reply_text(
                'Something failed here? Pin to blockfrost.io failed.'
            )
            return ConversationHandler.END
    else:
        update.message.reply_text(
            'Something failed here? Upload to blockfrost.io failed.'
        )
        return ConversationHandler.END
    # Remove left overs
    os.remove(file_name)
    # Respond to get ticker
    context.bot.send_message(
        chat_id=update.effective_chat.id,
        text="What will the token ticker be for your NFT? (examples: TEST0, DOGE1) \n"
             "Please enter that now."
    )
    return TICKER

The user can /skip the photo upload if they choose to. I left it in from the earlier example and was going to expand it for all states later on.


def skip_photo(update: Update, _: CallbackContext) -> int:
    """ User has chosen to not upload an image, End the session. """
    user = update.message.from_user
    logging.info(f"User {user.username} did not send a photo.")
    update.message.reply_text(
        'Process Ended. Use /start to try again.'
    )
    return ConversationHandler.END

The metadata states are all the same so I will lump them together. Each one gathers the raw data from the user, puts it in the session user_data key/value store, tells them what they typed, and moves on to the next state.


def put_ticker(update: Update, context: CallbackContext) -> int:
    """ PUT the TICKER to the users context """
    logging.info(f'RAW DATA ENTERED: {update.message.text}')
    # TODO add check for 4-5 characters
    # TODO Set to uppercase for ticker
    context.user_data["token_ticker"] = update.message.text
    update.message.reply_text(
        f'Great! Your TICKER will be: {update.message.text}')
    update.message.reply_text(
        'Thank you, what will the name of'
        ' your token be? (Example: TestToken0) \n'
        'Please enter that now.'
    )
    return NAME


def put_token_name(update: Update, context: CallbackContext) -> int:
    """ PUT the TOKEN NAME to the users context """
    logging.info(f'RAW DATA ENTERED: {update.message.text}')
    context.user_data["token_name"] = update.message.text
    update.message.reply_text(f'Awesome! Your TOKEN NAME will be: {update.message.text}')
    update.message.reply_text(
        'Thank you, what will the description of your token be? (example: Just a test) \n'
        'Please enter that now.'
    )
    return DESCRIPTION


def put_token_desc(update: Update, context: CallbackContext) -> int:
    """ PUT the TOKEN DESCRIPTION to the users context """
    logging.info(f'RAW DATA ENTERED: {update.message.text}')
    context.user_data["token_desc"] = update.message.text
    update.message.reply_text(f'Sweet! Your TOKEN DESCRIPTION will be: {update.message.text}')
    update.message.reply_text(
        'Thank you, if this is a series of tokens, \n'
        'what will the number of your token be? (example: 0, 1, 2, 3, etc) \n'
        'Please enter that now.'
    )
    return NUMBER


def put_token_number(update: Update, context: CallbackContext) -> int:
    """ PUT the TOKEN NUMBER to the users context """
    # TODO check for int and if no int, set to zero
    logging.info(f'RAW DATA ENTERED: {update.message.text}')
    context.user_data["token_number"] = update.message.text
    update.message.reply_text(f'Rad! Your TOKEN NUMBER will be: {update.message.text}')
    update.message.reply_text('Now we will begin the pre-minting process. \n')
    update.message.reply_text('Type OK to begin.')
    return PRE_MINT

Now we are to the PRE_MINT state and ask the user to type 'OK' to start the pre-minting process. We create the session's UUID and send the metadata to the token_utils.py pre_mint() function.

At the end of this state if the pre-minting works, it will have generated a bot ADA address that the user can send the 5 ADA to pay for the token minting fees.


def put_pre_mint(update: Update, context: CallbackContext) -> int:
    """ Starts pre-minting a token, returns an address to send funds to """
    auth = update.message.text
    if auth.lower() == 'ok':
        logging.info(f"We are a GO!")
        # Gather data
        logging.info(context.user_data)
        if context.user_data:
            session_uuid = str(uuid4())
            logging.info(session_uuid)
            context.user_data['session_uuid'] = session_uuid
            user = get_chat_info(update)
            nft_data = {
                'session_uuid': session_uuid,
                'creator_username': user['username'],
                'token_ticker': context.user_data.get('token_ticker', 'Not found'),
                'token_name': context.user_data.get('token_name', 'Not found'),
                'token_desc': context.user_data.get('token_desc', 'Not found'),
                'token_number': context.user_data.get('token_number', 'Not found'),
                'token_ipfs_hash': context.user_data.get('token_ipfs_hash', 'Not found'),
            }
            update.message.reply_text(
                "Here is the token metadata data we will be using to create a NFT. \n"
            )
            update.message.reply_text(
                f'*token_ticker*: {nft_data["token_ticker"]} \n'
                f'*token_name*: {nft_data["token_name"]} \n'
                f'*token_desc*: {nft_data["token_desc"]} \n'
                f'*token_number*: {nft_data["token_number"]} \n'
                f'*token_ipfs_hash*: {nft_data["token_ipfs_hash"]} \n'
            )
            update.message.reply_text("Setting up the NFT metadata...")
            pre_minted = pre_mint(**nft_data)
            if pre_minted:
                # Means we updated the DB and have a addr to send funds to
                # Start DB Session to get addr
                session = Session()
                sesh_exists = session.query(Tokens).filter(
                    Tokens.session_uuid == session_uuid).scalar() is not None
                if sesh_exists:
                    token_data = session.query(Tokens).filter(
                        Tokens.session_uuid == session_uuid).one()
                    logging.info(f'Session Data Created: {token_data}')
                    logging.info(f'Bot Address: {token_data.bot_payment_addr}')
                    # At this point we need the bot_payment_addr to have UTXO to burn
                    update.message.reply_text(f'Please deposit 5 ADA in the following address:')
                    update.message.reply_text(f'*{token_data.bot_payment_addr}*')
                    update.message.reply_text(
                        f'Once transaction has been confirmed in your wallet, \n'
                        f'run the /MINT command.')
                    return ConversationHandler.END
                else:
                    update.message.reply_text("No Session Data yet. /start to begin.")
                    return ConversationHandler.END
            else:
                update.message.reply_text("Pre-mint failed, check bot logs.")
                return ConversationHandler.END
        else:
            update.message.reply_text("No Data yet. /start to begin.")
            return ConversationHandler.END
    update.message.reply_text("No OK?!? /start to begin again.")
    return ConversationHandler.END

Now the big moment, the user confirms their ADA transaction in their own wallet and then they enter /MINT. The bot then takes all the previous work and mints a fresh NFT from the data and sends it back to the user with change.

I wanted to add a spinning waiting animation, but not really sure how to do that logically with the Telegram API.


def put_mint(update: Update, context: CallbackContext) -> int:
    """ Returns the token data to user """
    session_uuid = context.user_data['session_uuid']
    user = get_chat_info(update)
    creator_username = user['username']

    # Start DB Session to get addr
    session = Session()
    sesh_exists = session.query(Tokens).filter(
        Tokens.session_uuid == session_uuid).scalar() is not None
    if sesh_exists:
        # Add a check for the UTXO, bail if not found
        token_data = session.query(Tokens).filter(
            Tokens.session_uuid == session_uuid).one()
        logging.info(f'Looking for UTXOs in bot address: {token_data}')
        bot_payment_addr = token_data.bot_payment_addr
        update.message.reply_text("Checking for transaction confirmation.")
        utxo = check_wallet_utxo(bot_payment_addr)
        if utxo:
            sesh = {'session_uuid': session_uuid}
            update.message.reply_text("OK, I found the Transaction, thank you.")
            update.message.reply_text(
                "Please be patient as I build your NFT "
                "and send it back to you with your change in ADA."
            )
            update.message.reply_text("Starting NFT Minting process......")
            minted = mint(**sesh)
            if minted:
                update.message.reply_text(
                    f"Awesome Sauce! \n I finished minting your token, @{creator_username}."
                )
                update.message.reply_text(
                    f"The token should arrive in your wallet shortly."
                )
                update.message.reply_text(
                    f"Thank you for using the *cNFT TESTNET Bot*. \n Have an ADA day."
                )
                return ConversationHandler.END
            else:
                update.message.reply_text(
                    f"Something failed, please try not to panic, "
                    f"but you may have hit a bug. Sorry."
                )
                return ConversationHandler.END
        else:
            update.message.reply_text(
                f"Sorry, but there is no UTXO to use yet. "
                f"Transaction not found."
                f"Please try running /MINT again in a few moments."
            )
            return ConversationHandler.END
    update.message.reply_text(
        f"Sorry, but there is no PRE_MINT session yet. "
        f"Please try /start again in a few moments."
    )
    return ConversationHandler.END

We finish off with some helper functions that make for some usefulness while developing the bot. I am not sure people would actually know to look up UTXOs, but I added it anyway.

I also kept the /cancel state incase the user wanted to bail at any moment.


#
# Some Helper Chat Commands
#


def get_token_data(update: Update, context: CallbackContext) -> int:
    """ Returns the token data to user """
    if context.user_data:
        nft_data = {
            'token_ticker': context.user_data.get('token_ticker', 'Not found'),
            'token_name': context.user_data.get('token_name', 'Not found'),
            'token_desc': context.user_data.get('token_desc', 'Not found'),
            'token_number': context.user_data.get('token_number', 'Not found'),
            'token_ipfs_hash': context.user_data.get('token_ipfs_hash', 'Not found'),
        }
        update.message.reply_text(
            f'*token_ticker*: {nft_data["token_ticker"]} \n'
            f'*token_name*: {nft_data["token_name"]} \n'
            f'*token_desc*: {nft_data["token_desc"]} \n'
            f'*token_number*: {nft_data["token_number"]} \n'
            f'*token_ipfs_hash*: {nft_data["token_ipfs_hash"]} \n'
        )
    else:
        update.message.reply_text("No Data yet. Enter /start to begin.")
    return ConversationHandler.END


def get_tx(update: Update, context: CallbackContext) -> int:
    """ Returns the transaction data to user """
    # Need user to send tx hash
    tx_hash = context.args[0]
    logging.info(f"tx_hash: {tx_hash}")
    data = get_tx_details(tx_hash=tx_hash)
    logging.info(data)
    update.message.reply_text("Transaction found.")
    update.message.reply_text(
        f"{data}"
    )
    return ConversationHandler.END


def get_utxo(update: Update, context: CallbackContext) -> int:
    """ Returns the UTXO data to user """
    logging.info(context.args)
    update.message.reply_text("Enter an *address* to look up a transaction.")
    ada_addr = context.args[0]
    update.message.reply_text("Checking Transaction confirmation.")
    utxo = check_wallet_utxo(ada_addr)
    if utxo:
        update.message.reply_text("UTXO found.")
        update.message.reply_text(
            f"{utxo}"
        )
    else:
        update.message.reply_text("No transaction found.")
    return ConversationHandler.END


def cancel(update: Update, _: CallbackContext) -> int:
    """ User opted to end the conversation """
    user = update.message.from_user
    logging.info(f"User {user.username} canceled the conversation.")
    update.message.reply_text('Bye!')
    return ConversationHandler.END

The main() function of the bot app sets up the bot by giving it the API token, and setting the output to my preference of Markdown in the responses.

I just modified the ConversationHandler to work with the states created above.


def main() -> None:
    """Start the bot."""
    # Create the Updater and pass it your token and private key
    updater = Updater(
        token=API_TOKEN,
        use_context=True,
        defaults=Defaults(parse_mode='Markdown')
    )

    # Get the dispatcher to register handlers
    dispatcher = updater.dispatcher

    # Handlers
    # Add conversation handler with the states PHOTO, TICKER, NAME, DESCRIPTION, NUMBER, and PRE_MINT
    # TODO add skip functions for each state
    conv_handler = ConversationHandler(
        entry_points=[CommandHandler('start', start)],
        states={
            PHOTO: [MessageHandler(Filters.photo, photo), CommandHandler('skip', skip_photo)],
            TICKER: [MessageHandler(Filters.text & ~Filters.command, put_ticker)],
            NAME: [MessageHandler(Filters.text & ~Filters.command, put_token_name)],
            DESCRIPTION: [MessageHandler(Filters.text & ~Filters.command, put_token_desc)],
            NUMBER: [MessageHandler(Filters.text & ~Filters.command, put_token_number)],
            PRE_MINT: [MessageHandler(Filters.text & ~Filters.command, put_pre_mint)]
        },
        fallbacks=[CommandHandler('cancel', cancel)],
    )
    dispatcher.add_handler(conv_handler)
    dispatcher.add_handler(CommandHandler('get_token_data', get_token_data))
    dispatcher.add_handler(CommandHandler('get_tx', get_tx))
    dispatcher.add_handler(CommandHandler('get_utxo', get_utxo))
    dispatcher.add_handler(CommandHandler('MINT', put_mint))

    # Start the Bot
    updater.start_polling()

    # Run the bot until you press Ctrl-C or the process receives SIGINT,
    # SIGTERM or SIGABRT. This should be used most of the time, since
    # start_polling() is non-blocking and will stop the bot gracefully.
    updater.idle()


if __name__ == '__main__':
    main()