add telegram bot
This commit is contained in:
Nick
3
.env
3
.env
@@ -1 +1,2 @@
|
||||
GEMINI_API_KEY='AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M'
|
||||
GEMINI_API_KEY='AIzaSyCdjM5XE0hv9O1Ecx-uspvI8UInSzK1U9M'
|
||||
TELEGRAM_BOT_TOKEN='8252062619:AAGL7Fr338vvHJ2bUqmURw9bxw9LL_kDjIM'
|
||||
456
telegram_bot.py
Normal file
456
telegram_bot.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user