ManualTrader/utils/get_telegram_chat_id.py

#!/usr/bin/env python3
"""
Get Telegram Chat ID Helper

This script helps you find your Telegram Chat ID for bot configuration.
"""

import sys
import asyncio
import logging
import signal
from pathlib import Path

# Add src directory to Python path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

try:
    from telegram import Update
    from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
except ImportError:
    print("❌ python-telegram-bot not installed!")
    print("💡 Run: pip install python-telegram-bot")
    sys.exit(1)

# Set up logging to reduce noise
logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', 
    level=logging.WARNING  # Reduce log noise
)
logger = logging.getLogger(__name__)

class ChatIDBot:
    """Simple bot to get your Chat ID."""
    
    def __init__(self, token: str):
        self.token = token
        self.application = None
        self.found_chat_ids = set()
        self.running = True
    
    async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """Handle any message and show chat information."""
        try:
            chat_id = update.effective_chat.id
            user = update.effective_user
            message_text = update.message.text if update.message else "No text"
            
            # Store found chat ID
            self.found_chat_ids.add(chat_id)
            
            # Respond with chat information
            response = f"""
🎯 <b>Found Your Chat Information!</b>

🆔 <b>Chat ID:</b> <code>{chat_id}</code>
👤 <b>User:</b> {user.first_name} {user.last_name or ''}
📧 <b>Username:</b> @{user.username or 'None'}
💬 <b>Message:</b> {message_text}

✅ <b>Copy this Chat ID to your .env file:</b>
<code>TELEGRAM_CHAT_ID={chat_id}</code>

🛑 <b>Press Ctrl+C to stop this script</b>
            """
            
            await update.message.reply_text(response.strip(), parse_mode='HTML')
            
            # Print to console
            print(f"\n🎉 SUCCESS! Found Chat ID: {chat_id}")
            print(f"👤 User: {user.first_name} {user.last_name or ''}")
            print(f"📧 Username: @{user.username or 'None'}")
            print(f"\n📋 Add this to your .env file:")
            print(f"TELEGRAM_CHAT_ID={chat_id}")
            print(f"\n✅ You can now stop this script (Ctrl+C) and start your trading bot!")
            print(f"💬 Or send another message to test again...")
            
        except Exception as e:
            print(f"❌ Error handling message: {e}")
    
    async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
        """Handle the /start command."""
        await self.handle_message(update, context)
    
    async def stop_bot(self):
        """Gracefully stop the bot."""
        if self.application:
            try:
                await self.application.stop()
                await self.application.shutdown()
            except Exception as e:
                print(f"⚠️ Warning during shutdown: {e}")
    
    async def run(self):
        """Run the Chat ID finder bot."""
        try:
            # Build application
            self.application = Application.builder().token(self.token).build()
            
            # Add handlers for any message or /start command
            self.application.add_handler(CommandHandler("start", self.start_command))
            self.application.add_handler(MessageHandler(filters.ALL, self.handle_message))
            
            print("🤖 Chat ID finder bot is running...")
            print("💬 Now go to Telegram and send ANY message to your bot")
            print("🛑 Press Ctrl+C when done\n")
            
            # Initialize and run
            await self.application.initialize()
            await self.application.start()
            await self.application.updater.start_polling()
            
            # Keep running until interrupted
            while self.running:
                await asyncio.sleep(1)
                
        except KeyboardInterrupt:
            print(f"\n👋 Chat ID finder stopped by user")
        except Exception as e:
            print(f"\n❌ Error running bot: {e}")
        finally:
            # Cleanup
            try:
                if self.application:
                    if self.application.updater.running:
                        await self.application.updater.stop()
                    await self.application.stop()
                    await self.application.shutdown()
            except Exception as e:
                pass  # Ignore cleanup errors
            
            # Show results
            if self.found_chat_ids:
                print(f"\n📋 Found Chat IDs: {', '.join(map(str, self.found_chat_ids))}")
                print(f"✅ Copy any of these Chat IDs to your .env file")
            else:
                print("\n❌ No Chat IDs found - make sure you messaged your bot!")

def signal_handler(bot):
    """Handle shutdown signals."""
    def handler(signum, frame):
        print(f"\n🛑 Received shutdown signal...")
        bot.running = False
    return handler

def main():
    """Main function."""
    print("🔍 Telegram Chat ID Finder\n")
    
    # Get bot token
    token = input("🤖 Enter your Telegram Bot Token: ").strip()
    
    if not token:
        print("❌ No token provided!")
        return
    
    if ':' not in token:
        print("❌ Invalid token format! Should look like: 123456789:ABCdefGHIjklMNOPqrs")
        return
    
    print(f"\n✅ Token looks valid!")
    print(f"🚀 Starting Chat ID finder bot...")
    print(f"\n📱 Instructions:")
    print(f"1. Open Telegram")
    print(f"2. Find your bot (search for the username you gave it)")
    print(f"3. Send ANY message to your bot")
    print(f"4. Come back here to see your Chat ID")
    print(f"5. Copy the Chat ID to your .env file")
    
    try:
        # Create bot instance
        bot = ChatIDBot(token)
        
        # Set up signal handler for graceful shutdown
        signal.signal(signal.SIGINT, signal_handler(bot))
        signal.signal(signal.SIGTERM, signal_handler(bot))
        
        # Handle existing event loop
        try:
            loop = asyncio.get_running_loop()
            print("⚠️ Running in existing event loop")
            # If there's already a loop, create a task
            task = loop.create_task(bot.run())
            loop.run_until_complete(task)
        except RuntimeError:
            # No existing loop, create new one
            asyncio.run(bot.run())
            
    except KeyboardInterrupt:
        print(f"\n👋 Stopped by user")
    except Exception as e:
        print(f"❌ Error: {e}")
        print("💡 Troubleshooting tips:")
        print("  - Make sure your bot token is correct")
        print("  - Check your internet connection")
        print("  - Verify the bot exists in @BotFather")

if __name__ == "__main__":
    main()