Added Repository
This commit is contained in:
178
bot.py
Normal file
178
bot.py
Normal file
@@ -0,0 +1,178 @@
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime, time
|
||||
from dotenv import load_dotenv
|
||||
from telegram import Update
|
||||
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
|
||||
|
||||
# Load environment variables (keep token secure)
|
||||
load_dotenv()
|
||||
|
||||
# Enable logging
|
||||
logging.basicConfig(
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
level=logging.INFO
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ===== CONFIGURATION =====
|
||||
BOT_TOKEN = os.getenv('BOT_TOKEN', 'YOUR_BOT_TOKEN_HERE') # Better: use environment variable
|
||||
ADMIN_USER_ID = 123456789 # Replace with YOUR Telegram user ID
|
||||
|
||||
# Set your "off-hours" schedule
|
||||
AFTER_HOURS_START = time(18, 0) # 6:00 PM
|
||||
AFTER_HOURS_END = time(9, 0) # 9:00 AM
|
||||
WEEKEND_AUTO_REPLY = True # Auto-reply on weekends?
|
||||
|
||||
# Customizable auto-reply messages
|
||||
AFTER_HOURS_MESSAGE = """🤖 Automated Reply:
|
||||
|
||||
Thanks for your message. I'm currently offline and not checking messages.
|
||||
|
||||
⏰ My hours: 9 AM - 6 PM, Monday to Friday
|
||||
📝 I'll review your message during my next working session."""
|
||||
|
||||
NORMAL_HOURS_MESSAGE = """🤖 Quick Reply:
|
||||
|
||||
I've received your message and will respond when I'm able.
|
||||
|
||||
Please ensure you've:
|
||||
1. Checked the documentation
|
||||
2. Tried at least 2 solutions
|
||||
|
||||
For urgent matters, use our emergency protocol."""
|
||||
|
||||
# ===== HELPER FUNCTIONS =====
|
||||
def is_after_hours():
|
||||
"""Check if current time is outside working hours"""
|
||||
now = datetime.now()
|
||||
current_time = now.time()
|
||||
|
||||
# Check if weekend
|
||||
if WEEKEND_AUTO_REPLY and now.weekday() >= 5: # 5=Saturday, 6=Sunday
|
||||
return True
|
||||
|
||||
# Check if outside working hours
|
||||
if current_time >= AFTER_HOURS_START or current_time <= AFTER_HOURS_END:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def should_auto_reply(user_id):
|
||||
"""Add specific logic here for certain users"""
|
||||
# Example: Always auto-reply to specific problematic users
|
||||
# problematic_users = [999888777, 666555444] # Their Telegram IDs
|
||||
# if user_id in problematic_users:
|
||||
# return True
|
||||
|
||||
# For now, auto-reply to everyone during off-hours
|
||||
return is_after_hours()
|
||||
|
||||
# ===== BOT COMMANDS =====
|
||||
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Send a message when the command /start is issued."""
|
||||
user = update.effective_user
|
||||
welcome_msg = f"""Hello {user.first_name}! 👋
|
||||
|
||||
I'm an automated assistant. Messages here are logged and reviewed periodically.
|
||||
|
||||
📍 Please note:
|
||||
• I'm not monitored 24/7
|
||||
• For quickest response, message during business hours
|
||||
• Include all relevant details in your first message"""
|
||||
|
||||
await update.message.reply_text(welcome_msg)
|
||||
|
||||
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Send a message when the command /help is issued."""
|
||||
help_text = """Available commands:
|
||||
/start - Welcome message
|
||||
/help - This help message
|
||||
/status - Check if I'm currently in auto-reply mode
|
||||
/contact - How to reach me urgently"""
|
||||
|
||||
await update.message.reply_text(help_text)
|
||||
|
||||
async def status(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Check if bot is in after-hours mode"""
|
||||
if is_after_hours():
|
||||
status_msg = "🔕 Currently in AFTER-HOURS mode (auto-reply enabled)"
|
||||
else:
|
||||
status_msg = "🔔 Currently in BUSINESS HOURS mode"
|
||||
|
||||
await update.message.reply_text(status_msg)
|
||||
|
||||
# ===== MESSAGE HANDLER =====
|
||||
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Handle incoming messages"""
|
||||
user_id = update.effective_user.id
|
||||
username = update.effective_user.username or "No username"
|
||||
first_name = update.effective_user.first_name or "Unknown"
|
||||
message_text = update.message.text or "No text"
|
||||
|
||||
# Log the message
|
||||
logger.info(f"Message from {first_name} (@{username}, ID: {user_id}): {message_text[:50]}...")
|
||||
|
||||
# Check if we should auto-reply
|
||||
if should_auto_reply(user_id):
|
||||
# Send auto-reply
|
||||
if is_after_hours():
|
||||
await update.message.reply_text(AFTER_HOURS_MESSAGE)
|
||||
else:
|
||||
await update.message.reply_text(NORMAL_HOURS_MESSAGE)
|
||||
|
||||
# Forward message to you (the admin) for review
|
||||
forward_msg = f"📥 New message from {first_name} (@{username}):\n\n{message_text}\n\n⏰ Received during {'off-hours' if is_after_hours() else 'business hours'}"
|
||||
|
||||
try:
|
||||
await context.bot.send_message(
|
||||
chat_id=ADMIN_USER_ID,
|
||||
text=forward_msg
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to forward message: {e}")
|
||||
|
||||
else:
|
||||
# During business hours, just forward without auto-reply
|
||||
forward_msg = f"📞 Immediate attention from {first_name} (@{username}):\n\n{message_text}"
|
||||
|
||||
try:
|
||||
await context.bot.send_message(
|
||||
chat_id=ADMIN_USER_ID,
|
||||
text=forward_msg
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to forward message: {e}")
|
||||
|
||||
# ===== ERROR HANDLER =====
|
||||
async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Log errors"""
|
||||
logger.warning(f"Update {update} caused error {context.error}")
|
||||
|
||||
# ===== MAIN FUNCTION =====
|
||||
def main():
|
||||
"""Start the bot"""
|
||||
# Create the Application
|
||||
application = Application.builder().token(BOT_TOKEN).build()
|
||||
|
||||
# Register command handlers
|
||||
application.add_handler(CommandHandler("start", start))
|
||||
application.add_handler(CommandHandler("help", help_command))
|
||||
application.add_handler(CommandHandler("status", status))
|
||||
|
||||
# Register message handler
|
||||
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
|
||||
|
||||
# Register error handler
|
||||
application.add_error_handler(error_handler)
|
||||
|
||||
# Start the bot
|
||||
print("🤖 Bot is starting...")
|
||||
print(f"⏰ After-hours: {AFTER_HOURS_START.strftime('%H:%M')} to {AFTER_HOURS_END.strftime('%H:%M')}")
|
||||
print(f"📅 Weekend auto-reply: {'Yes' if WEEKEND_AUTO_REPLY else 'No'}")
|
||||
|
||||
# Run bot until interrupted
|
||||
application.run_polling(allowed_updates=Update.ALL_TYPES)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user