1188 lines
46 KiB
Python
1188 lines
46 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
TF5 Mixer Telegram Bot - Con profili utente personalizzati
|
|
"""
|
|
import json
|
|
import time
|
|
import sys
|
|
import os
|
|
from typing import Dict, List, Optional
|
|
from pathlib import Path
|
|
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
|
|
import re
|
|
|
|
load_dotenv()
|
|
from config import *
|
|
|
|
|
|
class UserProfile:
|
|
"""Rappresenta il profilo di un utente."""
|
|
|
|
ROLES = {
|
|
"mixerista": {
|
|
"name": "Mixerista",
|
|
"emoji": "🎛️",
|
|
"permissions": ["all"],
|
|
"description": "Accesso completo a tutti i canali e mix"
|
|
},
|
|
"cantante": {
|
|
"name": "Cantante",
|
|
"emoji": "🎤",
|
|
"permissions": ["own_channel", "own_mix"],
|
|
"description": "Gestione del proprio microfono e in-ear"
|
|
},
|
|
"musicista": {
|
|
"name": "Musicista",
|
|
"emoji": "🎸",
|
|
"permissions": ["own_channel", "own_mix"],
|
|
"description": "Gestione del proprio strumento e in-ear"
|
|
},
|
|
"operatore_diretta": {
|
|
"name": "Operatore Diretta",
|
|
"emoji": "📡",
|
|
"permissions": ["own_mix"],
|
|
"description": "Gestione del mix per la diretta streaming/registrazione"
|
|
},
|
|
"traduttore": {
|
|
"name": "Traduttore",
|
|
"emoji": "🌐",
|
|
"permissions": ["own_mix"],
|
|
"description": "Gestione del mix per la traduzione simultanea"
|
|
}
|
|
}
|
|
|
|
def __init__(self, user_id: int, username: str):
|
|
self.user_id = user_id
|
|
self.username = username
|
|
self.role: Optional[str] = None
|
|
self.display_name: Optional[str] = None
|
|
self.channels: List[int] = [] # Canali assegnati
|
|
self.mixes: List[int] = [] # Mix/aux assegnati
|
|
self.setup_completed = False
|
|
self.created_at = time.time()
|
|
self.last_updated = time.time()
|
|
|
|
def to_dict(self) -> Dict:
|
|
"""Converte il profilo in dizionario."""
|
|
return {
|
|
"user_id": self.user_id,
|
|
"username": self.username,
|
|
"role": self.role,
|
|
"display_name": self.display_name,
|
|
"channels": self.channels,
|
|
"mixes": self.mixes,
|
|
"setup_completed": self.setup_completed,
|
|
"created_at": self.created_at,
|
|
"created_at_str": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.created_at)),
|
|
"last_updated": self.last_updated,
|
|
"last_updated_str": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(self.last_updated))
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict) -> 'UserProfile':
|
|
"""Crea un profilo da un dizionario."""
|
|
profile = cls(data["user_id"], data["username"])
|
|
profile.role = data.get("role")
|
|
profile.display_name = data.get("display_name")
|
|
profile.channels = data.get("channels", [])
|
|
profile.mixes = data.get("mixes", [])
|
|
profile.setup_completed = data.get("setup_completed", False)
|
|
profile.created_at = data.get("created_at", time.time())
|
|
profile.last_updated = data.get("last_updated", time.time())
|
|
return profile
|
|
|
|
def get_role_info(self) -> Dict:
|
|
"""Restituisce le info sul ruolo."""
|
|
return self.ROLES.get(self.role, {})
|
|
|
|
def can_access_channel(self, channel: int) -> bool:
|
|
"""Verifica se l'utente può accedere a un canale."""
|
|
if not self.role:
|
|
return False
|
|
|
|
permissions = self.get_role_info().get("permissions", [])
|
|
|
|
if "all" in permissions:
|
|
return True
|
|
|
|
if "own_channel" in permissions:
|
|
return channel in self.channels
|
|
|
|
return False
|
|
|
|
def can_access_mix(self, mix: int) -> bool:
|
|
"""Verifica se l'utente può accedere a un mix."""
|
|
if not self.role:
|
|
return False
|
|
|
|
permissions = self.get_role_info().get("permissions", [])
|
|
|
|
if "all" in permissions:
|
|
return True
|
|
|
|
if "own_mix" in permissions:
|
|
return mix in self.mixes
|
|
|
|
return False
|
|
|
|
def get_summary(self) -> str:
|
|
"""Restituisce un riepilogo del profilo."""
|
|
if not self.setup_completed:
|
|
return "❌ Profilo non configurato. Usa /setup per iniziare."
|
|
|
|
role_info = self.get_role_info()
|
|
msg = f"{role_info['emoji']} **{self.display_name}** ({role_info['name']})\n\n"
|
|
|
|
if self.channels:
|
|
msg += f"🎚️ Canali: {', '.join(map(str, self.channels))}\n"
|
|
|
|
if self.mixes:
|
|
msg += f"🔊 Mix: {', '.join(map(str, self.mixes))}\n"
|
|
|
|
return msg
|
|
|
|
|
|
class ProfileManager:
|
|
"""Gestisce i profili utente."""
|
|
|
|
def __init__(self, profiles_dir: str = "profiles"):
|
|
self.profiles_dir = Path(profiles_dir)
|
|
self.profiles_dir.mkdir(exist_ok=True)
|
|
|
|
def _get_filepath(self, user_id: int) -> Path:
|
|
"""Restituisce il percorso del file per un utente."""
|
|
return self.profiles_dir / f"{user_id}.json"
|
|
|
|
def save_profile(self, profile: UserProfile) -> bool:
|
|
"""Salva il profilo di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(profile.user_id)
|
|
profile.last_updated = time.time()
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(profile.to_dict(), f, ensure_ascii=False, indent=2)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nel salvataggio profilo {profile.user_id}: {e}")
|
|
return False
|
|
|
|
def load_profile(self, user_id: int, username: str) -> UserProfile:
|
|
"""Carica o crea il profilo di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(user_id)
|
|
|
|
if filepath.exists():
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return UserProfile.from_dict(data)
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nel caricamento profilo {user_id}: {e}")
|
|
|
|
# Crea nuovo profilo
|
|
return UserProfile(user_id, username)
|
|
|
|
def delete_profile(self, user_id: int) -> bool:
|
|
"""Elimina il profilo di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(user_id)
|
|
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nell'eliminazione profilo {user_id}: {e}")
|
|
return False
|
|
|
|
def get_all_profiles(self) -> List[UserProfile]:
|
|
"""Restituisce tutti i profili salvati."""
|
|
profiles = []
|
|
|
|
for filepath in self.profiles_dir.glob("*.json"):
|
|
try:
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
profiles.append(UserProfile.from_dict(data))
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore lettura {filepath}: {e}")
|
|
|
|
return profiles
|
|
|
|
|
|
class ConversationManager:
|
|
"""Gestisce il salvataggio delle conversazioni su file JSON."""
|
|
|
|
def __init__(self, conversations_dir: str = "conversations"):
|
|
self.conversations_dir = Path(conversations_dir)
|
|
self.conversations_dir.mkdir(exist_ok=True)
|
|
|
|
def _get_filepath(self, user_id: int) -> Path:
|
|
"""Restituisce il percorso del file per un utente."""
|
|
return self.conversations_dir / f"{user_id}.json"
|
|
|
|
def save_conversation(self, user_id: int, history: List[Dict],
|
|
username: str = "Unknown"):
|
|
"""Salva la conversazione di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(user_id)
|
|
|
|
data = {
|
|
"user_id": user_id,
|
|
"username": username,
|
|
"last_updated": time.time(),
|
|
"last_updated_str": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"message_count": len(history),
|
|
"history": history
|
|
}
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nel salvataggio conversazione {user_id}: {e}")
|
|
return False
|
|
|
|
def load_conversation(self, user_id: int) -> Optional[List[Dict]]:
|
|
"""Carica la conversazione di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(user_id)
|
|
|
|
if not filepath.exists():
|
|
return None
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
return data.get("history", [])
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nel caricamento conversazione {user_id}: {e}")
|
|
return None
|
|
|
|
def delete_conversation(self, user_id: int) -> bool:
|
|
"""Elimina la conversazione di un utente."""
|
|
try:
|
|
filepath = self._get_filepath(user_id)
|
|
|
|
if filepath.exists():
|
|
filepath.unlink()
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore nell'eliminazione conversazione {user_id}: {e}")
|
|
return False
|
|
|
|
|
|
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", "callback_query"]
|
|
}
|
|
|
|
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 escape_markdown_v2(self, text: str) -> str:
|
|
"""Escape dei caratteri speciali per MarkdownV2."""
|
|
escape_chars = r'_*[]()~`>#+-=|{}.!'
|
|
return re.sub(f'([{re.escape(escape_chars)}])', r'\\\1', text)
|
|
|
|
def send_message(self, chat_id: int, text: str,
|
|
parse_mode: Optional[str] = None,
|
|
reply_markup: Optional[Dict] = None) -> Dict:
|
|
"""Invia un messaggio a una chat."""
|
|
params = {
|
|
"chat_id": chat_id,
|
|
"text": text
|
|
}
|
|
|
|
if parse_mode:
|
|
if parse_mode == "MarkdownV2":
|
|
params["text"] = self.escape_markdown_v2(text)
|
|
params["parse_mode"] = parse_mode
|
|
|
|
if reply_markup:
|
|
params["reply_markup"] = reply_markup
|
|
|
|
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)
|
|
|
|
def answer_callback_query(self, callback_query_id: str, text: str = "") -> Dict:
|
|
"""Risponde a una callback query."""
|
|
params = {
|
|
"callback_query_id": callback_query_id,
|
|
"text": text
|
|
}
|
|
return self._make_request("answerCallbackQuery", 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,
|
|
conversations_dir: str = "conversations",
|
|
profiles_dir: str = "profiles"):
|
|
self.bot = TelegramBot(telegram_token)
|
|
self.controller = TF5MixerController(mixer_host, mixer_port)
|
|
self.conversation_manager = ConversationManager(conversations_dir)
|
|
self.profile_manager = ProfileManager(profiles_dir)
|
|
|
|
# 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,
|
|
self.controller.set_channel_to_mix_level,
|
|
self.controller.set_channel_to_mix_on_off,
|
|
self.controller.get_channel_to_mix_info,
|
|
],
|
|
temperature=0,
|
|
)
|
|
|
|
# Storia delle conversazioni per ogni utente (in memoria)
|
|
self.conversations: Dict[int, List[Dict]] = {}
|
|
# Profili utente (in memoria)
|
|
self.profiles: Dict[int, UserProfile] = {}
|
|
# Stati setup utente
|
|
self.setup_states: Dict[int, Dict] = {}
|
|
|
|
# System instruction base
|
|
self.base_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
|
|
|
|
RICHIESTE SEMPLIFICATE (per utenti non mixeristi):
|
|
- "alzami la chitarra" / "più chitarra" → riferimento al MIO mix (quello dell'utente)
|
|
- "abbassa il basso in cuffia" → gestisci il send del basso al MIO in-ear
|
|
- "più voce" → alza la voce nel MIO mix
|
|
- "meno batteria" → abbassa la batteria nel MIO mix
|
|
|
|
SEND AI MIX/AUX:
|
|
- "alza il gelato nell'aux 2" → usa set_channel_to_mix_level
|
|
- "manda più chitarra nel monitor" → aumenta il send
|
|
- "togli la voce dal mix 3" → spegni il send o abbassa molto
|
|
- i mix 9/10, 11/12, 13/14, 15/16, 17/18, 19/20 sono stereo, fai le operazioni su entrambi
|
|
|
|
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_user_system_instruction(self, profile: UserProfile) -> str:
|
|
"""Genera le istruzioni di sistema personalizzate per l'utente."""
|
|
instruction = self.base_system_instruction
|
|
|
|
if not profile.setup_completed:
|
|
return instruction
|
|
|
|
# Aggiungi contesto del profilo
|
|
role_info = profile.get_role_info()
|
|
instruction += f"\n\n{'='*70}\n"
|
|
instruction += f"PROFILO UTENTE CORRENTE:\n"
|
|
instruction += f"{'='*70}\n\n"
|
|
instruction += f"Nome: {profile.display_name}\n"
|
|
instruction += f"Ruolo: {role_info['name']} {role_info['emoji']}\n"
|
|
|
|
if profile.channels:
|
|
instruction += f"\nCanali assegnati: {', '.join(map(str, profile.channels))}\n"
|
|
instruction += "Quando l'utente dice 'il mio canale' o 'il mio mic/strumento', si riferisce a questi.\n"
|
|
|
|
if profile.mixes:
|
|
instruction += f"\nMix assegnati: {', '.join(map(str, profile.mixes))}\n"
|
|
instruction += "Quando l'utente dice 'le mie cuffie', 'il mio monitor', 'in cuffia', si riferisce a questi mix.\n"
|
|
|
|
# Istruzioni specifiche per ruolo
|
|
permissions = role_info.get("permissions", [])
|
|
|
|
if "all" in permissions:
|
|
instruction += "\n🎛️ MIXERISTA - Accesso completo:\n"
|
|
instruction += "- Puoi controllare tutti i canali e mix del mixer\n"
|
|
instruction += "- Puoi caricare scene\n"
|
|
instruction += "- Puoi fare qualsiasi operazione richiesta\n"
|
|
else:
|
|
instruction += f"\n⚠️ LIMITAZIONI DI ACCESSO ({role_info['name']}):\n"
|
|
|
|
if "own_channel" in permissions and profile.channels:
|
|
instruction += f"- Canali controllabili: SOLO {', '.join(map(str, profile.channels))}\n"
|
|
|
|
if "own_mix" in permissions and profile.mixes:
|
|
instruction += f"- Mix controllabili: SOLO {', '.join(map(str, profile.mixes))}\n"
|
|
instruction += f"- Puoi gestire i send di QUALSIASI canale verso i TUOI mix ({', '.join(map(str, profile.mixes))})\n"
|
|
|
|
instruction += "\n❌ NON puoi:\n"
|
|
if "own_channel" not in permissions:
|
|
instruction += "- Modificare il volume/stato dei canali\n"
|
|
if "own_mix" not in permissions:
|
|
instruction += "- Modificare il volume/stato dei mix\n"
|
|
instruction += "- Caricare scene (solo i mixeristi)\n"
|
|
instruction += "- Accedere a canali/mix non assegnati\n"
|
|
|
|
instruction += "\n💡 INTERPRETAZIONE RICHIESTE:\n"
|
|
instruction += "- 'alzami la chitarra' = alza il send della chitarra nel TUO mix\n"
|
|
instruction += "- 'più voce' = alza il send della voce nel TUO mix\n"
|
|
instruction += "- 'meno batteria in cuffia' = abbassa il send della batteria nel TUO mix\n"
|
|
instruction += "- Se l'utente chiede di fare qualcosa fuori dai suoi permessi, spiega gentilmente che non può e suggerisci di chiedere al mixerista\n"
|
|
|
|
instruction += f"\n{'='*70}\n"
|
|
|
|
return instruction
|
|
|
|
def _check_permission(self, profile: UserProfile, operation: str,
|
|
channel: Optional[int] = None,
|
|
mix: Optional[int] = None) -> tuple[bool, str]:
|
|
"""
|
|
Verifica se l'utente ha i permessi per un'operazione.
|
|
Ritorna (permitted, error_message)
|
|
"""
|
|
if not profile.setup_completed:
|
|
return False, "⚠️ Devi prima configurare il tuo profilo con /setup"
|
|
|
|
role_info = profile.get_role_info()
|
|
permissions = role_info.get("permissions", [])
|
|
|
|
# Mixeristi possono fare tutto
|
|
if "all" in permissions:
|
|
return True, ""
|
|
|
|
# Verifica permessi specifici
|
|
if operation in ["set_channel_level", "set_channel_on_off", "set_channel_pan",
|
|
"mute_multiple_channels", "unmute_multiple_channels"]:
|
|
if "own_channel" not in permissions:
|
|
return False, "❌ Solo i mixeristi possono controllare i canali direttamente"
|
|
|
|
if channel and not profile.can_access_channel(channel):
|
|
return False, f"❌ Non hai accesso al canale {channel}. I tuoi canali: {', '.join(map(str, profile.channels))}"
|
|
|
|
if operation in ["set_mix_level", "set_mix_on_off"]:
|
|
if "own_mix" not in permissions:
|
|
return False, "❌ Non hai permessi per controllare i mix"
|
|
|
|
if mix and not profile.can_access_mix(mix):
|
|
return False, f"❌ Non hai accesso al mix {mix}. I tuoi mix: {', '.join(map(str, profile.mixes))}"
|
|
|
|
if operation == "recall_scene":
|
|
return False, "❌ Solo i mixeristi possono caricare scene"
|
|
|
|
# Le operazioni di send ai mix sono permesse se hai accesso al mix di destinazione
|
|
if operation in ["set_channel_to_mix_level", "set_channel_to_mix_on_off"]:
|
|
if "own_mix" not in permissions:
|
|
return False, "❌ Non hai permessi per controllare i send ai mix"
|
|
|
|
if mix and not profile.can_access_mix(mix):
|
|
return False, f"❌ Non hai accesso al mix {mix}. I tuoi mix: {', '.join(map(str, profile.mixes))}"
|
|
|
|
return True, ""
|
|
|
|
def get_profile(self, user_id: int, username: str) -> UserProfile:
|
|
"""Ottiene il profilo di un utente."""
|
|
if user_id not in self.profiles:
|
|
self.profiles[user_id] = self.profile_manager.load_profile(user_id, username)
|
|
return self.profiles[user_id]
|
|
|
|
def _start_setup(self, user_id: int) -> str:
|
|
"""Inizia il processo di setup."""
|
|
roles_text = "🎛️ **Scegli il tuo ruolo:**\n\n"
|
|
|
|
buttons = []
|
|
for role_key, role_info in UserProfile.ROLES.items():
|
|
roles_text += f"{role_info['emoji']} **{role_info['name']}**\n"
|
|
roles_text += f" {role_info['description']}\n\n"
|
|
buttons.append([{
|
|
"text": f"{role_info['emoji']} {role_info['name']}",
|
|
"callback_data": f"setup_role_{role_key}"
|
|
}])
|
|
|
|
keyboard = {"inline_keyboard": buttons}
|
|
|
|
self.bot.send_message(
|
|
user_id,
|
|
roles_text,
|
|
parse_mode="MarkdownV2",
|
|
reply_markup=keyboard
|
|
)
|
|
|
|
self.setup_states[user_id] = {"step": "role"}
|
|
return ""
|
|
|
|
def _handle_setup_callback(self, user_id: int, username: str, data: str) -> str:
|
|
"""Gestisce le callback del processo di setup."""
|
|
profile = self.get_profile(user_id, username)
|
|
state = self.setup_states.get(user_id, {})
|
|
|
|
if data.startswith("setup_role_"):
|
|
role = data.replace("setup_role_", "")
|
|
profile.role = role
|
|
role_info = profile.get_role_info()
|
|
|
|
self.bot.send_message(
|
|
user_id,
|
|
f"✅ Ruolo impostato: {role_info['emoji']} {role_info['name']}\n\n"
|
|
f"📝 Come vuoi che ti chiami? (es: Marco, Chitarra 1, Cantante)"
|
|
)
|
|
|
|
self.setup_states[user_id] = {"step": "name", "role": role}
|
|
return ""
|
|
|
|
return ""
|
|
|
|
def _handle_setup_message(self, user_id: int, username: str, text: str) -> str:
|
|
"""Gestisce i messaggi durante il setup."""
|
|
profile = self.get_profile(user_id, username)
|
|
state = self.setup_states.get(user_id, {})
|
|
|
|
if not state:
|
|
return ""
|
|
|
|
step = state.get("step")
|
|
|
|
# Step: inserimento nome
|
|
if step == "name":
|
|
profile.display_name = text.strip()
|
|
role_info = profile.get_role_info()
|
|
permissions = role_info.get("permissions", [])
|
|
|
|
# Se è mixerista, ha finito
|
|
if "all" in permissions:
|
|
profile.setup_completed = True
|
|
self.profile_manager.save_profile(profile)
|
|
del self.setup_states[user_id]
|
|
|
|
return (f"✅ Configurazione completata!\n\n"
|
|
f"{profile.get_summary()}\n"
|
|
f"Hai accesso completo a tutti i canali e mix del mixer.\n\n"
|
|
f"Inizia a controllare il mixer con comandi come:\n"
|
|
f"• 'Alza il canale 5'\n"
|
|
f"• 'Richiama la scena A10'\n"
|
|
f"• 'Mostrami tutti i canali'")
|
|
|
|
# Altrimenti, chiedi i canali (se necessario)
|
|
if "own_channel" in permissions:
|
|
self.bot.send_message(
|
|
user_id,
|
|
f"🎚️ Quali canali usi?\n\n"
|
|
f"Esempi:\n"
|
|
f"• Un canale: 5\n"
|
|
f"• Più canali: 5, 7, 12\n"
|
|
f"• Range: 5-8 (canali 5, 6, 7, 8)\n"
|
|
f"• Misti: 3, 5-7, 10\n\n"
|
|
f"(Scrivi 0 se non usi canali diretti)"
|
|
)
|
|
self.setup_states[user_id] = {"step": "channels", "role": profile.role}
|
|
return ""
|
|
else:
|
|
# Salta direttamente ai mix
|
|
self.bot.send_message(
|
|
user_id,
|
|
f"🔊 Quali mix/aux usi per le tue cuffie o uscita?\n\n"
|
|
f"Esempi:\n"
|
|
f"• Un mix: 3\n"
|
|
f"• Più mix: 3, 5\n"
|
|
f"• Range: 3-5\n"
|
|
f"• Mix stereo: 9, 10 (in-ear stereo)\n\n"
|
|
f"(Scrivi 0 se non usi mix specifici)"
|
|
)
|
|
self.setup_states[user_id] = {"step": "mixes", "role": profile.role}
|
|
return ""
|
|
|
|
# Step: inserimento canali
|
|
elif step == "channels":
|
|
channels = self._parse_channel_list(text)
|
|
profile.channels = channels
|
|
|
|
role_info = profile.get_role_info()
|
|
permissions = role_info.get("permissions", [])
|
|
|
|
# Chiedi i mix se necessario
|
|
if "own_mix" in permissions:
|
|
self.bot.send_message(
|
|
user_id,
|
|
f"🔊 Quali mix/aux usi per le tue cuffie o uscita?\n\n"
|
|
f"Esempi:\n"
|
|
f"• Un mix: 3\n"
|
|
f"• Più mix: 3, 5\n"
|
|
f"• Range: 3-5\n"
|
|
f"• Mix stereo: 9, 10 (in-ear stereo)\n\n"
|
|
f"(Scrivi 0 se non usi mix specifici)"
|
|
)
|
|
self.setup_states[user_id] = {"step": "mixes", "role": profile.role}
|
|
return ""
|
|
else:
|
|
# Finito
|
|
profile.setup_completed = True
|
|
self.profile_manager.save_profile(profile)
|
|
del self.setup_states[user_id]
|
|
|
|
return (f"✅ Configurazione completata!\n\n"
|
|
f"{profile.get_summary()}\n"
|
|
f"Ora puoi controllare i tuoi canali con comandi come:\n"
|
|
f"• 'Alza il mio canale'\n"
|
|
f"• 'Spegni il mio mic'")
|
|
|
|
# Step: inserimento mix
|
|
elif step == "mixes":
|
|
mixes = self._parse_channel_list(text) # Stessa logica per i mix
|
|
profile.mixes = mixes
|
|
|
|
profile.setup_completed = True
|
|
self.profile_manager.save_profile(profile)
|
|
del self.setup_states[user_id]
|
|
|
|
role_info = profile.get_role_info()
|
|
|
|
example_commands = []
|
|
if profile.channels:
|
|
example_commands.append("• 'Alza il mio canale'")
|
|
if profile.mixes:
|
|
example_commands.extend([
|
|
"• 'Alzami la chitarra' (nel tuo mix)",
|
|
"• 'Più voce in cuffia'",
|
|
"• 'Meno batteria'"
|
|
])
|
|
|
|
return (f"✅ Configurazione completata!\n\n"
|
|
f"{profile.get_summary()}\n"
|
|
f"Esempi di comandi:\n" + "\n".join(example_commands))
|
|
|
|
return ""
|
|
|
|
def _parse_channel_list(self, text: str) -> List[int]:
|
|
"""Parse una lista di canali/mix dal testo utente."""
|
|
channels = []
|
|
|
|
# Rimuovi spazi e split per virgola
|
|
parts = [p.strip() for p in text.split(',')]
|
|
|
|
for part in parts:
|
|
# Range (es: 5-8)
|
|
if '-' in part:
|
|
try:
|
|
start, end = map(int, part.split('-'))
|
|
channels.extend(range(start, end + 1))
|
|
except:
|
|
pass
|
|
# Singolo numero
|
|
else:
|
|
try:
|
|
num = int(part)
|
|
if num > 0: # Ignora 0
|
|
channels.append(num)
|
|
except:
|
|
pass
|
|
|
|
return sorted(list(set(channels))) # Rimuovi duplicati e ordina
|
|
|
|
def _get_file_path(self, file_id: str) -> str:
|
|
"""Richiede a Telegram il percorso di un file."""
|
|
result = self.bot._make_request("getFile", {"file_id": file_id})
|
|
if result.get("ok"):
|
|
return result["result"]["file_path"]
|
|
raise Exception("Impossibile ottenere il file da Telegram")
|
|
|
|
def _download_file(self, url: str) -> bytes:
|
|
"""Scarica un file da Telegram e restituisce i bytes."""
|
|
req = Request(url)
|
|
with urlopen(req, timeout=30) as response:
|
|
return response.read()
|
|
|
|
def _transcribe_audio(self, audio_bytes: bytes) -> Optional[str]:
|
|
"""Trascrive un file audio in testo usando Gemini."""
|
|
try:
|
|
response = self.client.models.generate_content(
|
|
model="gemini-2.5-flash",
|
|
contents=[
|
|
"Trascrivi in testo chiaro ciò che dice questa registrazione:",
|
|
types.Part.from_bytes(
|
|
data=audio_bytes,
|
|
mime_type="audio/ogg"
|
|
),
|
|
]
|
|
)
|
|
return response.text.strip()
|
|
except Exception as e:
|
|
print(f"❌ Errore nella trascrizione: {e}")
|
|
return None
|
|
|
|
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:
|
|
history = self.conversation_manager.load_conversation(user_id)
|
|
self.conversations[user_id] = history if history else []
|
|
|
|
return self.conversations[user_id]
|
|
|
|
def add_to_history(self, user_id: int, role: str, content: str):
|
|
"""Aggiunge un messaggio alla storia e salva."""
|
|
history = self.get_conversation_history(user_id)
|
|
history.append({
|
|
"role": role,
|
|
"content": content,
|
|
"timestamp": time.time(),
|
|
"timestamp_str": time.strftime("%Y-%m-%d %H:%M:%S")
|
|
})
|
|
|
|
if len(history) > 20:
|
|
self.conversations[user_id] = history[-20:]
|
|
|
|
profile = self.profiles.get(user_id)
|
|
username = profile.username if profile else "Unknown"
|
|
self.conversation_manager.save_conversation(user_id, history, username)
|
|
|
|
def clear_history(self, user_id: int):
|
|
"""Cancella la storia di un utente."""
|
|
self.conversations[user_id] = []
|
|
self.conversation_manager.delete_conversation(user_id)
|
|
|
|
def process_message(self, user_id: int, username: str, text: str) -> str:
|
|
"""Elabora un messaggio dell'utente."""
|
|
try:
|
|
profile = self.get_profile(user_id, username)
|
|
|
|
# Comandi speciali
|
|
if text.lower() in ['/start', '/help']:
|
|
return self._get_help_message(profile)
|
|
|
|
if text.lower() == '/setup':
|
|
return self._start_setup(user_id)
|
|
|
|
if text.lower() == '/profile':
|
|
return profile.get_summary()
|
|
|
|
if text.lower() == '/reset':
|
|
self.clear_history(user_id)
|
|
return "🔄 Storia della conversazione cancellata. Ricominciamo da capo!"
|
|
|
|
if text.lower() == '/resetprofile':
|
|
self.profile_manager.delete_profile(user_id)
|
|
self.profiles[user_id] = UserProfile(user_id, username)
|
|
return "🔄 Profilo cancellato. Usa /setup per configurarlo di nuovo."
|
|
|
|
if text.lower() == '/status':
|
|
return self._get_status_message()
|
|
|
|
# Gestione setup in corso
|
|
if user_id in self.setup_states:
|
|
return self._handle_setup_message(user_id, username, text)
|
|
|
|
# Verifica profilo completato
|
|
if not profile.setup_completed:
|
|
return ("⚠️ Prima di iniziare, devi configurare il tuo profilo!\n\n"
|
|
"Usa il comando /setup per iniziare.")
|
|
|
|
# Aggiungi il messaggio alla storia
|
|
self.add_to_history(user_id, "user", text)
|
|
|
|
# Prepara il prompt con la storia e il profilo
|
|
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]
|
|
])
|
|
|
|
system_instruction = self._get_user_system_instruction(profile)
|
|
|
|
full_prompt = f"{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-flash",
|
|
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, profile: UserProfile) -> str:
|
|
"""Restituisce il messaggio di aiuto."""
|
|
msg = "🎛️ **TF5 Mixer Bot**\n\n"
|
|
|
|
if not profile.setup_completed:
|
|
msg += "⚠️ Devi prima configurare il tuo profilo!\n\n"
|
|
msg += "Usa /setup per iniziare la configurazione.\n\n"
|
|
else:
|
|
role_info = profile.get_role_info()
|
|
msg += f"{role_info['emoji']} Sei: **{profile.display_name}** ({role_info['name']})\n\n"
|
|
|
|
if "all" in role_info.get("permissions", []):
|
|
msg += "**Esempi di comandi (accesso completo):**\n"
|
|
msg += "• 'Alza il canale 5 a -10 dB'\n"
|
|
msg += "• 'Spegni i canali dal 1 al 5'\n"
|
|
msg += "• 'Richiama la scena A10'\n"
|
|
msg += "• 'Mostrami lo stato del canale 12'\n"
|
|
else:
|
|
msg += "**Esempi di comandi:**\n"
|
|
if profile.channels:
|
|
msg += "• 'Alza il mio canale'\n"
|
|
msg += "• 'Spegni il mio mic'\n"
|
|
if profile.mixes:
|
|
msg += "• 'Alzami la chitarra' (nel tuo mix)\n"
|
|
msg += "• 'Più voce in cuffia'\n"
|
|
msg += "• 'Meno batteria'\n"
|
|
|
|
msg += "\n**Comandi speciali:**\n"
|
|
msg += "/help - Mostra questo messaggio\n"
|
|
msg += "/setup - Configura/riconfigura profilo\n"
|
|
msg += "/profile - Mostra il tuo profilo\n"
|
|
msg += "/reset - Cancella storia conversazione\n"
|
|
msg += "/resetprofile - Cancella profilo\n"
|
|
msg += "/status - Mostra stato del sistema\n"
|
|
|
|
return msg
|
|
|
|
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"
|
|
|
|
# Statistiche profili
|
|
all_profiles = self.profile_manager.get_all_profiles()
|
|
if all_profiles:
|
|
status += f"\n👥 Utenti registrati: {len(all_profiles)}\n\n"
|
|
|
|
# Conta per ruolo
|
|
role_counts = {}
|
|
for p in all_profiles:
|
|
role = p.get_role_info().get("name", "Unknown")
|
|
role_counts[role] = role_counts.get(role, 0) + 1
|
|
|
|
status += "**Utenti per ruolo:**\n"
|
|
for role, count in role_counts.items():
|
|
status += f"• {role}: {count}\n"
|
|
|
|
return status
|
|
|
|
def handle_update(self, update: Dict):
|
|
"""Gestisce un aggiornamento di Telegram."""
|
|
# Gestione callback query (bottoni inline)
|
|
if "callback_query" in update:
|
|
callback = update["callback_query"]
|
|
user_id = callback["from"]["id"]
|
|
username = callback["from"].get("username", "Unknown")
|
|
data = callback["data"]
|
|
|
|
self.bot.answer_callback_query(callback["id"])
|
|
self._handle_setup_callback(user_id, username, data)
|
|
return
|
|
|
|
if "message" not in update:
|
|
return
|
|
|
|
message = update["message"]
|
|
chat_id = message["chat"]["id"]
|
|
user_id = message["from"]["id"]
|
|
username = message["from"].get("username", "Unknown")
|
|
|
|
# Messaggi vocali
|
|
if "voice" in message or "audio" in message:
|
|
self.bot.send_chat_action(chat_id, "typing")
|
|
file_info = message.get("voice") or message.get("audio")
|
|
file_id = file_info["file_id"]
|
|
|
|
file_path = self._get_file_path(file_id)
|
|
file_url = f"https://api.telegram.org/file/bot{self.bot.token}/{file_path}"
|
|
audio_bytes = self._download_file(file_url)
|
|
|
|
transcript = self._transcribe_audio(audio_bytes)
|
|
if not transcript:
|
|
self.bot.send_message(chat_id, "❌ Non sono riuscito a capire l'audio.")
|
|
return
|
|
|
|
self.bot.send_message(chat_id, f"🗣️ Hai detto:\n> {transcript}")
|
|
|
|
response = self.process_message(user_id, username, transcript)
|
|
self.bot.send_message(chat_id, response, parse_mode='MarkdownV2')
|
|
return
|
|
|
|
# Messaggi testuali
|
|
if "text" not in message:
|
|
return
|
|
|
|
text = message["text"]
|
|
profile = self.get_profile(user_id, username)
|
|
print(f"📨 Messaggio da @{username} ({profile.get_role_info().get('name', 'Non configurato')}): {text}")
|
|
|
|
self.bot.send_chat_action(chat_id, "typing")
|
|
response = self.process_message(user_id, username, text)
|
|
self.bot.send_message(chat_id, response, parse_mode='MarkdownV2')
|
|
|
|
def run(self):
|
|
"""Avvia il bot con long polling."""
|
|
print("=" * 70)
|
|
print("🤖 TF5 Mixer Telegram Bot - Con Profili Utente")
|
|
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 profili registrati
|
|
profiles = self.profile_manager.get_all_profiles()
|
|
if profiles:
|
|
print(f"\n👥 Profili registrati: {len(profiles)}")
|
|
for p in profiles[:5]:
|
|
role = p.get_role_info().get("name", "Unknown")
|
|
print(f" • @{p.username} - {p.display_name} ({role})")
|
|
if len(profiles) > 5:
|
|
print(f" ... e altri {len(profiles) - 5}")
|
|
|
|
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)
|
|
|
|
except KeyboardInterrupt:
|
|
raise
|
|
except Exception as e:
|
|
print(f"❌ Errore nel polling: {e}")
|
|
time.sleep(5)
|
|
|
|
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 con Profili Utente'
|
|
)
|
|
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})')
|
|
parser.add_argument('--conversations-dir', default='conversations',
|
|
help='Directory conversazioni (default: conversations)')
|
|
parser.add_argument('--profiles-dir', default='profiles',
|
|
help='Directory profili (default: profiles)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
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'")
|
|
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'")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
bot = TF5TelegramBot(
|
|
telegram_token=telegram_token,
|
|
mixer_host=args.mixer_host,
|
|
mixer_port=args.mixer_port,
|
|
conversations_dir=args.conversations_dir,
|
|
profiles_dir=args.profiles_dir
|
|
)
|
|
bot.run()
|
|
|
|
except Exception as e:
|
|
print(f"❌ Errore fatale: {e}")
|
|
sys.exit(1)
|
|
finally:
|
|
bot.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |