Feature idea: trigger one-off from webhook

[pierre-leon]

I had an idea and wanted to check if there's interest do add this functionality.

Context:

I'm using the bot with a docker image hosted on EC2, sending notifications to telegram

Idea:

Trigger on-off dry run by sending a message to the telegram bot e.g. /today

Pros:

1. Can make sure the bot is running
2. Can check how I'm doing so far today to get an idea if a switch is likely.
3. Only making changes in scheduler.py or a new file e.g. telegram_server.py - no changes in main functionality files

Potentials Cons:

1. Would have to have a public url (https)
2. Each channel (e.g. telegram, discord) would require separate set up as there isn't an apprise "inbound" equivalent that I'm aware of.
3. KISS

Thoughts?

If there's no interest, I might still do it, in a fork.

[pierre-leon]

something like this (or could use flask)

import http.server
import socketserver
import json
import requests
import threading
import time

from urllib.parse import parse_qs


# Telegram Bot Config
TELEGRAM_TOKEN = <BOT_TOKEN> 
CHAT_ID = <CHAT_ID>
WEBHOOK_URL = <EC2_public_URL>
PORT = 5000

def setup_webhook():
    """Configure Telegram webhook"""
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/setWebhook"
    params = {"url": WEBHOOK_URL}
    try:
        response = requests.get(url, params=params)
        return response.json()
    except Exception as e:
        return None

def process_command(text):
    """Process commands received from Telegram"""
    if text.startswith('/today'):
        # Set dry run to true temporarily
        original_dry_run = config.DRY_RUN
        config.DRY_RUN = true
        
        try:
            run_tariff_compare()
        finally:
            # Restore original dry run setting
            config.DRY_RUN = original_dry_run
            
    elif text.startswith('/status'):
        # Send status info
        now = datetime.now()
        next_run = config.EXECUTION_TIME
        send_notification(f"Server is running normally.\nCurrent time: {now.strftime('%H:%M')}\nNext scheduled run: {next_run}")
    else:
        # Help 
        send_telegram_message("Available commands:\n/run - Run tariff comparison (dry run)\n/status - Check server status")


class TelegramWebhookHandler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        """Handle POST requests to the webhook endpoint"""
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length)
        
        try:
            update = json.loads(post_data.decode('utf-8'))            
            # Check if this is a message with text
            if 'message' in update and 'text' in update['message']:
                chat_id = update['message']['chat']['id']
                text = update['message']['text']
                
                # Only process commands from your chat ID for security
                if str(chat_id) == str(CHAT_ID):
                    # Process in a separate thread to avoid blocking
                    threading.Thread(target=process_command, args=(text,)).start()
                else:
                    logger.warning(f"Received message from unauthorized chat ID: {chat_id}")
            
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps({"status": "success"}).encode('utf-8'))
            
        except Exception as e:
            self.send_response(500)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(json.dumps({"status": "error", "message": str(e)}).encode('utf-8'))

def run_server():
    """Run the HTTP server"""
    handler = TelegramWebhookHandler
    httpd = socketserver.TCPServer(("", PORT), handler)
    httpd.serve_forever()


if __name__ == '__main__':
    setup_webhook()
    
    # Start the scheduling in a background thread 
    schedule_thread = threading.Thread(target=scheduler)
    schedule_thread.daemon = True
    schedule_thread.start()
    
    run_server()

[eelmafia]

This is an interesting idea. I was thinking recently of potentially adding some kind of admin dashboard where one could see statistics and data. I think your idea would be a better fit for that, plus it would remove the need to have a different solution each for discord, telegram, WhatsApp, etc.

I probably won't have much free time to work on this idea for the time being as I'm very busy with some life stuff so if you're still keen on sending commands via Telegram you should do it in a fork for now.

[pierre-leon]

Any thoughts on using flask vs http.server & socketserver?
The rational was keeping the image size small

[eelmafia]

Any thoughts on using flask vs http.server & socketserver? The rational was keeping the image size small

I don't have a strong opinion. Unless the image size would increase dramatically I think using a framework would be better because it's easier and faster.