diff --git a/.env b/.env index d7da4a6..a9d4ca7 100644 --- a/.env +++ b/.env @@ -1 +1,2 @@ -GEMINI_API_KEY='AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M' \ No newline at end of file +GEMINI_API_KEY='AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M' +TELEGRAM_BOT_TOKEN='8252062619:AAGL7Fr338vvHJ2bUqmURw9bxw9LL_kDjIM' \ No newline at end of file diff --git a/telegram_bot.py b/telegram_bot.py new file mode 100644 index 0000000..d744f78 --- /dev/null +++ b/telegram_bot.py @@ -0,0 +1,456 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +TF5 Mixer Telegram Bot - Controllo del mixer Yamaha TF5 tramite Telegram +usando Google Gemini con function calling e long polling. +""" +import json +import time +import sys +import os +from typing import Dict, List, Optional +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError +from urllib.parse import urlencode +from google import genai +from google.genai import types +from dotenv import load_dotenv +from mixer_controller import TF5MixerController + +load_dotenv() + +from config import * + + +class TelegramBot: + """Bot Telegram base con long polling.""" + + def __init__(self, token: str): + self.token = token + self.base_url = f"https://api.telegram.org/bot{token}" + self.offset = 0 + + def _make_request(self, method: str, params: Optional[Dict] = None) -> Dict: + """Esegue una richiesta all'API di Telegram.""" + url = f"{self.base_url}/{method}" + + try: + if params: + data = json.dumps(params).encode('utf-8') + headers = {'Content-Type': 'application/json'} + req = Request(url, data=data, headers=headers) + else: + req = Request(url) + + with urlopen(req, timeout=30) as response: + return json.loads(response.read().decode('utf-8')) + + except HTTPError as e: + error_body = e.read().decode('utf-8') + print(f"❌ HTTP Error {e.code}: {error_body}") + return {"ok": False, "description": error_body} + except URLError as e: + print(f"❌ URL Error: {e.reason}") + return {"ok": False, "description": str(e.reason)} + except Exception as e: + print(f"❌ Error: {e}") + return {"ok": False, "description": str(e)} + + def get_updates(self, timeout: int = 30) -> List[Dict]: + """Ottiene gli aggiornamenti usando long polling.""" + params = { + "offset": self.offset, + "timeout": timeout, + "allowed_updates": ["message"] + } + + result = self._make_request("getUpdates", params) + + if result.get("ok"): + updates = result.get("result", []) + if updates: + self.offset = updates[-1]["update_id"] + 1 + return updates + + return [] + + def send_message(self, chat_id: int, text: str, + parse_mode: Optional[str] = None) -> Dict: + """Invia un messaggio a una chat.""" + params = { + "chat_id": chat_id, + "text": text + } + + if parse_mode: + params["parse_mode"] = parse_mode + + return self._make_request("sendMessage", params) + + def send_chat_action(self, chat_id: int, action: str = "typing") -> Dict: + """Invia un'azione (es. "typing").""" + params = { + "chat_id": chat_id, + "action": action + } + return self._make_request("sendChatAction", params) + + +class TF5TelegramBot: + """Bot Telegram per controllare il mixer TF5.""" + + def __init__(self, telegram_token: str, mixer_host: str = DEFAULT_HOST, + mixer_port: int = DEFAULT_PORT): + self.bot = TelegramBot(telegram_token) + self.controller = TF5MixerController(mixer_host, mixer_port) + + # Configura Gemini + api_key = os.getenv("GEMINI_API_KEY") + if not api_key: + raise ValueError("GEMINI_API_KEY non trovata") + + self.client = genai.Client(api_key=api_key) + + # Configurazione Gemini con function calling + self.config = types.GenerateContentConfig( + tools=[ + self.controller.recall_scene, + self.controller.set_channel_level, + self.controller.set_channel_on_off, + self.controller.set_channel_pan, + self.controller.set_mix_level, + self.controller.set_mix_on_off, + self.controller.mute_multiple_channels, + self.controller.unmute_multiple_channels, + self.controller.get_channel_info, + self.controller.get_mix_info, + self.controller.search_channels_by_name, + self.controller.search_mixes_by_name, + self.controller.get_all_channels_summary, + self.controller.get_all_mixes_summary, + self.controller.refresh_cache, + ], + temperature=0, + ) + + # Storia delle conversazioni per ogni utente + self.conversations: Dict[int, List[Dict]] = {} + + # System instruction + self.system_instruction = """Sei un assistente per il controllo del mixer audio Yamaha TF5. +Parli in modo semplice e diretto, come un tecnico del suono esperto che aiuta i musicisti sul palco. + +Il mixer ha: +- 40 canali (microfoni, strumenti, ecc.) +- 20 mix/aux (monitor, effetti, ecc.) +- Ogni canale ha volume (da silenzio a +10 dB), acceso/spento, e bilanciamento sinistra/destra +- Scene memorizzate nei banchi A e B (da 0 a 99) + +IMPORTANTE - Sistema di Cache: +- Le info sui canali e mix sono salvate per 60 minuti per non sovraccaricare il mixer +- Usa search_channels_by_name e search_mixes_by_name per cercare velocemente +- Usa get_all_channels_summary e get_all_mixes_summary per vedere tutto +- Quando modifichi un canale/mix, la cache viene aggiornata automaticamente +- Quando si carica una scena, i dati vengono invalidati e aggiornati alla prossima richiesta +- Puoi fare refresh_cache solo se l'utente lo chiede esplicitamente + +Come interpretare le richieste: + +VOLUME/LIVELLO: +- "alza/abbassa/aumenta/diminuisci" → cambia il volume +- "più/meno forte/volume" → cambia il volume +- "al massimo" → +10 dB +- "un po' più alto" → +3 dB circa +- "metti a zero" o "unity" → 0 dB +- "abbassa di poco" → -3 dB +- "metti basso" → -20 dB +- "silenzio/muto" → spegni il canale + +ON/OFF: +- "accendi/attiva/apri" → canale/mix ON +- "spegni/muta/chiudi/stacca" → canale/mix OFF +- "muto" può significare sia spegnere che abbassare molto + +BILANCIAMENTO (PAN): +- "a sinistra/left" → pan -63 +- "a destra/right" → pan +63 +- "al centro" → pan 0 +- "un po' a sinistra" → pan -30 circa + +IDENTIFICAZIONE CANALI E MIX: +- Accetta sia numeri ("canale 5", "mix 3") che nomi ("il microfono del cantante", "monitor palco") +- Se non trovi un canale/mix per nome, cerca usando search_channels_by_name o search_mixes_by_name +- "il mio mic/microfono" → cerca tra i canali chi è sul palco +- "le chitarre/i vox/le tastiere" → cerca per strumento +- "tutti i mic/tutte le chitarre" → cerca e gestisci multipli +- "il monitor" / "l'aux 2" → cerca tra i mix + +SCENE: +- "carica/richiama/vai alla scena X" → recall_scene +- Accetta "A5", "scena A 5", "la cinque del banco A", ecc. + +GRUPPI DI CANALI: +- "i canali dal 3 al 7" → canali 3,4,5,6,7 +- "spegni tutto tranne..." → muta tutti gli altri +- "solo i microfoni" → attiva solo quelli, spegni il resto + +CASI PARTICOLARI: +- Se la richiesta è ambigua, chiedi chiarimenti in modo colloquiale +- Se serve cercare un canale/mix, usa prima la cache (search_channels_by_name / search_mixes_by_name) +- Conferma sempre cosa hai fatto con un messaggio breve e chiaro +- Usa emoji occasionalmente per rendere le risposte più amichevoli (✅ ❌ 🎤 🎸 🔊) +- Se qualcosa non funziona, spiega il problema in modo semplice + +Rispondi sempre in modo: +- Diretto e colloquiale +- Senza troppi tecnicismi +- Confermando chiaramente l'azione eseguita +- Suggerendo alternative se qualcosa non è possibile + +Ricorda: chi ti parla è spesso sul palco, con le mani occupate da uno strumento. +Devi essere veloce, chiaro e capire anche richieste approssimative. +""" + + def get_conversation_history(self, user_id: int) -> List[Dict]: + """Ottiene la storia della conversazione per un utente.""" + if user_id not in self.conversations: + self.conversations[user_id] = [] + return self.conversations[user_id] + + def add_to_history(self, user_id: int, role: str, content: str): + """Aggiunge un messaggio alla storia.""" + history = self.get_conversation_history(user_id) + history.append({"role": role, "content": content}) + + # Mantieni solo gli ultimi 20 messaggi per evitare context overflow + if len(history) > 20: + self.conversations[user_id] = history[-20:] + + def clear_history(self, user_id: int): + """Cancella la storia di un utente.""" + self.conversations[user_id] = [] + + def process_message(self, user_id: int, username: str, text: str) -> str: + """Elabora un messaggio dell'utente.""" + try: + # Comandi speciali + if text.lower() in ['/start', '/help']: + return self._get_help_message() + + if text.lower() == '/reset': + self.clear_history(user_id) + return "🔄 Storia della conversazione cancellata. Ricominciamo da capo!" + + if text.lower() == '/status': + return self._get_status_message() + + # Aggiungi il messaggio alla storia + self.add_to_history(user_id, "user", text) + + # Prepara il prompt con la storia + history = self.get_conversation_history(user_id) + conversation_context = "\n".join([ + f"{'Utente' if msg['role'] == 'user' else 'Assistente'}: {msg['content']}" + for msg in history[:-1] # Escludi l'ultimo (già incluso sotto) + ]) + + full_prompt = f"{self.system_instruction}\n\n" + if conversation_context: + full_prompt += f"Conversazione precedente:\n{conversation_context}\n\n" + full_prompt += f"Utente: {text}" + + # Genera risposta con Gemini + response = self.client.models.generate_content( + model="gemini-2.5-pro", + contents=full_prompt, + config=self.config, + ) + + assistant_response = response.text + + # Aggiungi la risposta alla storia + self.add_to_history(user_id, "assistant", assistant_response) + + return assistant_response + + except Exception as e: + return f"❌ Errore nell'elaborazione: {e}" + + def _get_help_message(self) -> str: + """Restituisce il messaggio di aiuto.""" + return """🎛️ **TF5 Mixer Bot** + +Controlla il mixer Yamaha TF5 con comandi in linguaggio naturale! + +**Esempi di comandi:** +• "Alza il canale 5 a -10 dB" +• "Spegni i canali dal 1 al 5" +• "Richiama la scena A10" +• "Muta i canali 2, 4 e 6" +• "Alza il mix 3 di 5 dB" +• "Mostrami lo stato del canale 12" +• "Cerca i monitor" + +**Comandi speciali:** +/help - Mostra questo messaggio +/reset - Cancella la storia della conversazione +/status - Mostra stato del sistema + +Scrivi semplicemente cosa vuoi fare e ci penso io! 🎵""" + + def _get_status_message(self) -> str: + """Restituisce lo stato del sistema.""" + cache_age = time.time() - self.controller._cache.get("timestamp", 0) + cache_valid = self.controller._is_cache_valid() + + status = "📊 **Status del Sistema**\n\n" + status += f"🎛️ Mixer: Connesso a {self.controller.host}:{self.controller.port}\n" + + if cache_valid: + channels_count = len(self.controller._cache.get("channels", {})) + mixes_count = len(self.controller._cache.get("mixes", {})) + status += f"💾 Cache: Valida (età: {int(cache_age)}s)\n" + status += f"📊 Dati: {channels_count} canali, {mixes_count} mix\n" + else: + status += "💾 Cache: Non disponibile\n" + + users_count = len(self.conversations) + total_messages = sum(len(hist) for hist in self.conversations.values()) + status += f"👥 Utenti attivi: {users_count}\n" + status += f"💬 Messaggi totali: {total_messages}" + + return status + + def handle_update(self, update: Dict): + """Gestisce un aggiornamento di Telegram.""" + if "message" not in update: + return + + message = update["message"] + + # Ignora messaggi non testuali + if "text" not in message: + return + + chat_id = message["chat"]["id"] + user_id = message["from"]["id"] + username = message["from"].get("username", "Unknown") + text = message["text"] + + print(f"📨 Messaggio da @{username} (ID: {user_id}): {text}") + + # Mostra "typing..." + self.bot.send_chat_action(chat_id, "typing") + + # Elabora il messaggio + response = self.process_message(user_id, username, text) + + # Invia la risposta + self.bot.send_message(chat_id, response) + + print(f"✅ Risposta inviata a @{username}") + + def run(self): + """Avvia il bot con long polling.""" + print("=" * 70) + print("🤖 TF5 Mixer Telegram Bot") + print("=" * 70) + + # Verifica connessione + me = self.bot._make_request("getMe") + if me.get("ok"): + bot_info = me["result"] + print(f"\n✅ Bot avviato: @{bot_info['username']}") + print(f" Nome: {bot_info['first_name']}") + print(f" ID: {bot_info['id']}") + else: + print("❌ Errore nella verifica del bot") + return + + # Mostra stato cache + cache_age = time.time() - self.controller._cache.get("timestamp", 0) + if self.controller._is_cache_valid(): + channels_count = len(self.controller._cache.get("channels", {})) + mixes_count = len(self.controller._cache.get("mixes", {})) + print(f"\n💾 Cache disponibile (età: {int(cache_age)}s)") + print(f" 📊 {channels_count} canali, {mixes_count} mix") + else: + print("\n💾 Cache non disponibile, verrà creata al primo utilizzo") + + print("\n🔄 In attesa di messaggi... (Ctrl+C per terminare)\n") + + try: + while True: + try: + updates = self.bot.get_updates() + + for update in updates: + try: + self.handle_update(update) + except Exception as e: + print(f"❌ Errore nell'elaborazione update: {e}") + + time.sleep(0.1) # Piccola pausa per non sovraccaricare + + except KeyboardInterrupt: + raise + except Exception as e: + print(f"❌ Errore nel polling: {e}") + time.sleep(5) # Attendi prima di riprovare + + except KeyboardInterrupt: + print("\n\n👋 Bot terminato") + + def close(self): + """Chiude le connessioni.""" + self.controller.close() + + +def main(): + """Funzione principale.""" + import argparse + + parser = argparse.ArgumentParser( + description='TF5 Mixer Telegram Bot' + ) + parser.add_argument('--mixer-host', default=DEFAULT_HOST, + help=f'IP del mixer (default: {DEFAULT_HOST})') + parser.add_argument('--mixer-port', type=int, default=DEFAULT_PORT, + help=f'Porta mixer (default: {DEFAULT_PORT})') + + args = parser.parse_args() + + # Verifica variabili d'ambiente + telegram_token = os.getenv("TELEGRAM_BOT_TOKEN") + if not telegram_token: + print("❌ Errore: TELEGRAM_BOT_TOKEN non trovata") + print("\nPer impostare il token:") + print(" export TELEGRAM_BOT_TOKEN='il-tuo-token'") + print("\nOttieni un token da @BotFather su Telegram") + sys.exit(1) + + gemini_key = os.getenv("GEMINI_API_KEY") + if not gemini_key: + print("❌ Errore: GEMINI_API_KEY non trovata") + print("\nPer impostare la chiave API:") + print(" export GEMINI_API_KEY='la-tua-chiave-api'") + print("\nOttieni una chiave su: https://aistudio.google.com/apikey") + sys.exit(1) + + try: + bot = TF5TelegramBot( + telegram_token=telegram_token, + mixer_host=args.mixer_host, + mixer_port=args.mixer_port + ) + bot.run() + + except Exception as e: + print(f"❌ Errore fatale: {e}") + sys.exit(1) + finally: + bot.close() + + +if __name__ == '__main__': + main() \ No newline at end of file