telegram-bot.py

import logging
import aiohttp
import os,time

from aiohttp import FormData
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup

from telegram.ext import (
    Application, CommandHandler, MessageHandler,
    filters, CallbackQueryHandler, ContextTypes
)

from dotenv import load_dotenv
load_dotenv(dotenv_path='env.env')


#config
TOKEN = "8320924107:AAH505mhHkOxeY3aLk0GObIpO_KCtY9hhLM"
API_ENDPOINT = os.getenv("API_ENDPOINT")
logging.basicConfig(level=logging.INFO)

#category mappings
VIEW_CATEGORIES = {
    "crl": {
        "Maxilla": "mx",
        "Mandible-MDS": "mds", 
        "Mandible-MLS": "mls",
        "Lateral ventricle": "lv",
        "Head": "head",
        "Gestational sac": "gsac",
        "Thorax": "thorax",
        "Abdomen": "ab",
        "Body(Biparietal diameter)": "bd",
        "Rhombencephalon": "rbp",
        "Diencephalon": "dp",
        "NTAPS": "ntaps",
        "Nasal bone": "nb"
    },
    "nt": {
        "Maxilla": "mx",
        "Mandible-MDS": "mds", 
        "Mandible-MLS": "mls",
        "Lateral ventricle": "lv",
        "Head": "head",
        "Thorax": "thorax",
        "Abdomen": "ab",
        "Rhombencephalon": "rbp",
        "Diencephalon": "dp",
        "Nuchal translucency": "nt",
        "NTAPS": "ntaps",
        "Nasal bone": "nb"
    }
}

#start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Send an ultrasound image (JPG/png only) to begin. See /instructions first")

async def instructions(update: Update, context: ContextTypes.DEFAULT_TYPE):
    instruction_message = """
📋 **Instructions**

🔸 **Step 1:** Send a cropped ultrasound image of a structure you want to analyze. See all structures with /list command
🔸 **Step 2:** Select the ultrasound view (CRL or NT)
🔸 **Step 3:** Choose the anatomical category to analyze
🔸 **Step 4:** Wait for the AI analysis results

📌 **Important notes:**
• Only JPG/PNG images are supported
• Ensure the ultrasound image is clear and properly oriented
• Results are for reference only - always consult a medical professional
• Processing may take a few seconds
• Stop bot with /stop

💡 **Tips:**
• Use high-quality, well-lit images for better accuracy
• Make sure the anatomical structure is clearly visible
• Different views (CRL/NT) have different category options

🆘 **Need help?** Contact support if you encounter any issues @d3ikshr.
    """
    
    await update.message.reply_text(instruction_message, parse_mode='Markdown')

async def list_categories(update: Update, context: ContextTypes.DEFAULT_TYPE):
    list_message = """
📋 **Available Categories by View**

🔍 **CRL view categories:**
• Maxilla • Mandible-MDS • Mandible-MLS
• Lateral ventricle • Head • Gestational sac
• Thorax • Abdomen • Body(Biparietal diameter)
• Rhombencephalon • Diencephalon • NTAPS
• Nasal bone

🔍 **NT view categories:**
• Maxilla • Mandible-MDS • Mandible-MLS
• Lateral ventricle • Head • Thorax
• Abdomen • Rhombencephalon • Diencephalon
• Nuchal translucency • NTAPS • Nasal bone

💡 **Note:** Categories will be shown automatically based on your selected view during analysis.
    """
    
    await update.message.reply_text(list_message, parse_mode='Markdown')

async def stop(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("🛑 Bot stopped for this chat. Use /start to begin again.")
    # Clear user data
    context.user_data.clear()

# Receive image
async def handle_image(update: Update, context: ContextTypes.DEFAULT_TYPE):
    photo = update.message.photo[-1]
    file = await photo.get_file()
    file_path = f"{update.message.from_user.id}_ultrasound.jpg"
    await file.download_to_drive(file_path)
    context.user_data["image_path"] = file_path
    
    # get view
    buttons = [
        [InlineKeyboardButton("CRL", callback_data="view:crl"),
         InlineKeyboardButton("NT", callback_data="view:nt")]
    ]
    await update.message.reply_text(
        "Select the ultrasound view:",
        reply_markup=InlineKeyboardMarkup(buttons)
    )

#selcet view
async def handle_view(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()
    view = query.data.split(":")[1]
    context.user_data["selected_view"] = view
    
    #get categories for the selected view
    categories = list(VIEW_CATEGORIES[view].keys())
    buttons = [[InlineKeyboardButton(cat, callback_data=f"category:{cat}")]
               for cat in categories]
    await query.edit_message_text(
        "Select anatomical category:",
        reply_markup=InlineKeyboardMarkup(buttons)
    )

#get category then upload
async def handle_category(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()
    category_display = query.data.split(":")[1]
    context.user_data["selected_category"] = category_display
    
    image_path = context.user_data.get("image_path")
    view = context.user_data.get("selected_view")
    
    if not image_path or not view:
        await query.edit_message_text("Missing image or view.")
        return

    await query.edit_message_text("🔄 Processing image...")
    time.sleep(3)
    await query.edit_message_text("🏥 Building diagnosis, please wait...")

    try:
        #read the image file into memory first
        with open(image_path, "rb") as f:
            image_data = f.read()
        
        #mapping
        category_value = VIEW_CATEGORIES[view][category_display]
        
        #create form data
        form = FormData()
        form.add_field("view", view)
        form.add_field("category", category_value)
        form.add_field("source", "telegram")
        form.add_field("image", image_data, filename="image.jpg", content_type="image/jpeg")
        
        #send request 
        async with aiohttp.ClientSession() as session:
            async with session.post(API_ENDPOINT, data=form) as resp:
                if resp.status == 200:
                    result = await resp.json()
                    
                    message = f"""
📊 **Analysis Results**

🔍 **View:** {result.get('view', view).upper()}
🏥 **Category:** {category_display}

📈 **Confidence:** {result.get('confidence', 0):.2f}%
⚠️ **Reconstruction error:** {result.get('error', 0):.5f}

📋 **Status:** {result.get('comment', 'No comment')}

🩺 **Diagnosis:** {result.get('diagnosis', 'No diagnosis')}*
                    """
                    
                    await query.edit_message_text(message, parse_mode='Markdown')
                     
                else:
                    error_text = await resp.text()
                    await query.edit_message_text(f"❌ Upload failed. Status: {resp.status}\nError: {error_text}")
                    
    except Exception as e:
        logging.error(f"Error processing request: {e}")
        await query.edit_message_text(f"❌ Error: {str(e)}")
        
    finally:
        try:
            if os.path.exists(image_path):
                os.remove(image_path)
                logging.info(f"Cleaned up image file: {image_path}")
        except Exception as e:
            logging.error(f"Error cleaning up file: {e}")

def main():
    app = Application.builder().token(TOKEN).build()
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("instructions", instructions))
    app.add_handler(CommandHandler("list", list_categories))
    app.add_handler(CommandHandler("stop", stop))
    app.add_handler(MessageHandler(filters.PHOTO, handle_image))
    app.add_handler(CallbackQueryHandler(handle_view, pattern="^view:"))
    app.add_handler(CallbackQueryHandler(handle_category, pattern="^category:"))
    app.run_polling()

if __name__ == "__main__":
    main()