diff --git a/.tf5_mixer_cache/channels_cache.json b/.tf5_mixer_cache/channels_cache.json index c5e2bc4..2155302 100644 --- a/.tf5_mixer_cache/channels_cache.json +++ b/.tf5_mixer_cache/channels_cache.json @@ -11,13 +11,13 @@ "channel": 2, "name": "Gelato", "on": false, - "level_db": 0.0, + "level_db": null, "pan": 0 }, "3": { "channel": 3, "name": "Talkback", - "on": true, + "on": false, "level_db": null, "pan": 0 }, @@ -32,15 +32,15 @@ "channel": 5, "name": "Vox1", "on": true, - "level_db": 0.65, + "level_db": -1.1, "pan": 0 }, "6": { "channel": 6, - "name": "ON", - "on": false, - "level_db": 0.01, - "pan": 200 + "name": "Vox2", + "on": true, + "level_db": -0.15, + "pan": 0 }, "7": { "channel": 7, @@ -60,49 +60,49 @@ "channel": 9, "name": "Kick", "on": true, - "level_db": null, + "level_db": -5.2, "pan": 0 }, "10": { "channel": 10, "name": "Snare", "on": true, - "level_db": null, + "level_db": -9.95, "pan": 0 }, "11": { "channel": 11, "name": "Tom 1", "on": true, - "level_db": null, + "level_db": -16.6, "pan": 0 }, "12": { "channel": 12, "name": "Tom 2", "on": true, - "level_db": null, + "level_db": -18.0, "pan": 0 }, "13": { "channel": 13, "name": "Tom3", "on": true, - "level_db": null, + "level_db": -24.0, "pan": 0 }, "14": { "channel": 14, "name": "Pan SX", "on": true, - "level_db": null, + "level_db": -15.08, "pan": 0 }, "15": { "channel": 15, "name": "Pan dx", "on": true, - "level_db": null, + "level_db": -14.68, "pan": 0 }, "16": { @@ -129,15 +129,15 @@ "19": { "channel": 19, "name": "Vox 5", - "on": false, - "level_db": null, + "on": true, + "level_db": -0.1, "pan": 0 }, "20": { "channel": 20, "name": "Tast", "on": true, - "level_db": -3.55, + "level_db": 0.0, "pan": 0 }, "21": { @@ -158,21 +158,21 @@ "channel": 23, "name": " Vox3", "on": true, - "level_db": 3.8, + "level_db": -0.2, "pan": 0 }, "24": { "channel": 24, "name": "Chit cnt", "on": true, - "level_db": -3.95, + "level_db": -1.75, "pan": 0 }, "25": { "channel": 25, "name": "Chit dx", - "on": true, - "level_db": -9.45, + "on": false, + "level_db": null, "pan": 0 }, "26": { @@ -185,8 +185,8 @@ "27": { "channel": 27, "name": "Vox 4", - "on": false, - "level_db": -132.0, + "on": true, + "level_db": null, "pan": 0 }, "28": { @@ -199,7 +199,7 @@ "29": { "channel": 29, "name": "Pad", - "on": true, + "on": false, "level_db": null, "pan": 0 }, @@ -228,14 +228,14 @@ "channel": 33, "name": "PC", "on": true, - "level_db": -19.8, + "level_db": -20.4, "pan": -63 }, "34": { "channel": 34, "name": "PC", "on": true, - "level_db": -19.8, + "level_db": -20.4, "pan": 63 }, "35": { @@ -284,9 +284,9 @@ "mixes": { "1": { "mix": 1, - "name": "Sinistro", + "name": "Aux 1", "on": true, - "level_db": null + "level_db": -1.55 }, "2": { "mix": 2, @@ -310,17 +310,17 @@ "mix": 5, "name": "batteria", "on": true, - "level_db": 1.1 + "level_db": 1.35 }, "6": { "mix": 6, "name": "Aux 6", "on": true, - "level_db": 0.9 + "level_db": -6.65 }, "7": { "mix": 7, - "name": "Destro", + "name": "Aux 7", "on": true, "level_db": -2.7 }, @@ -328,7 +328,7 @@ "mix": 8, "name": "Aux 8", "on": true, - "level_db": -0.15 + "level_db": -2.65 }, "9": { "mix": 9, @@ -403,5 +403,5 @@ "level_db": 6.0 } }, - "timestamp": 1761592662.917339 + "timestamp": 1762193941.3288717 } \ No newline at end of file diff --git a/conversations/4762864.json b/conversations/4762864.json index 9c1d891..2e4b56f 100644 --- a/conversations/4762864.json +++ b/conversations/4762864.json @@ -1,9 +1,9 @@ { "user_id": 4762864, "username": "unnikked", - "last_updated": 1761595431.5373232, - "last_updated_str": "2025-10-27 21:03:51", - "message_count": 6, + "last_updated": 1762194065.7709692, + "last_updated_str": "2025-11-03 19:21:05", + "message_count": 13, "history": [ { "role": "user", @@ -40,6 +40,48 @@ "content": "Gelato spento! βœ…\nNon ho trovato nessun canale chiamato \"Cappuccino\", sei sicuro del nome? πŸ€”", "timestamp": 1761595431.5373232, "timestamp_str": "2025-10-27 21:03:51" + }, + { + "role": "user", + "content": "Okay, aprimi il gelato allora!", + "timestamp": 1762193935.418264, + "timestamp_str": "2025-11-03 19:18:55" + }, + { + "role": "user", + "content": "Gli alzi il gelato?", + "timestamp": 1762193958.1497903, + "timestamp_str": "2025-11-03 19:19:18" + }, + { + "role": "user", + "content": "Aprimi il giubbotto.", + "timestamp": 1762194045.3195145, + "timestamp_str": "2025-11-03 19:20:45" + }, + { + "role": "assistant", + "content": "Non posso aprirti il giubbotto, non Γ¨ un canale o un mix del mixer! πŸ˜„\n\nPosso aiutarti con i canali audio, i microfoni o i monitor, se vuoi.", + "timestamp": 1762194046.8080468, + "timestamp_str": "2025-11-03 19:20:46" + }, + { + "role": "user", + "content": "Aprimi il gelato", + "timestamp": 1762194053.0858698, + "timestamp_str": "2025-11-03 19:20:53" + }, + { + "role": "assistant", + "content": "Certo! Gelato acceso! πŸ¦βœ…", + "timestamp": 1762194054.6322277, + "timestamp_str": "2025-11-03 19:20:54" + }, + { + "role": "user", + "content": "È ancora chiuso", + "timestamp": 1762194065.7709692, + "timestamp_str": "2025-11-03 19:21:05" } ] } \ No newline at end of file diff --git a/profiles/4762864.json b/profiles/4762864.json new file mode 100644 index 0000000..6ed9a29 --- /dev/null +++ b/profiles/4762864.json @@ -0,0 +1,13 @@ +{ + "user_id": 4762864, + "username": "unnikked", + "role": "mixerista", + "display_name": "Nicola", + "channels": [], + "mixes": [], + "setup_completed": true, + "created_at": 1762193898.87283, + "created_at_str": "2025-11-03 19:18:18", + "last_updated": 1762193916.4840386, + "last_updated_str": "2025-11-03 19:18:36" +} \ No newline at end of file diff --git a/telegram_bot.py b/telegram_bot.py index a907689..23cc0c7 100644 --- a/telegram_bot.py +++ b/telegram_bot.py @@ -1,7 +1,7 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- """ -TF5 Mixer Telegram Bot - Con salvataggio conversazioni +TF5 Mixer Telegram Bot - Con profili utente personalizzati """ import json import time @@ -22,6 +22,206 @@ 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.""" @@ -88,27 +288,6 @@ class ConversationManager: except Exception as e: print(f"❌ Errore nell'eliminazione conversazione {user_id}: {e}") return False - - def get_all_users(self) -> List[Dict]: - """Restituisce la lista di tutti gli utenti con conversazioni salvate.""" - users = [] - - for filepath in self.conversations_dir.glob("*.json"): - try: - with open(filepath, 'r', encoding='utf-8') as f: - data = json.load(f) - - users.append({ - "user_id": data["user_id"], - "username": data.get("username", "Unknown"), - "message_count": data.get("message_count", 0), - "last_updated": data.get("last_updated_str", "Unknown") - }) - - except Exception as e: - print(f"❌ Errore lettura {filepath}: {e}") - - return users class TelegramBot: @@ -150,7 +329,7 @@ class TelegramBot: params = { "offset": self.offset, "timeout": timeout, - "allowed_updates": ["message"] + "allowed_updates": ["message", "callback_query"] } result = self._make_request("getUpdates", params) @@ -165,12 +344,12 @@ class TelegramBot: def escape_markdown_v2(self, text: str) -> str: """Escape dei caratteri speciali per MarkdownV2.""" - # Caratteri da escapare in 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) -> Dict: + parse_mode: Optional[str] = None, + reply_markup: Optional[Dict] = None) -> Dict: """Invia un messaggio a una chat.""" params = { "chat_id": chat_id, @@ -178,11 +357,13 @@ class TelegramBot: } if parse_mode: - # Se usi MarkdownV2, applica l'escape automatico 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: @@ -192,16 +373,27 @@ class TelegramBot: "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"): + 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") @@ -237,11 +429,13 @@ class TF5TelegramBot: # Storia delle conversazioni per ogni utente (in memoria) self.conversations: Dict[int, List[Dict]] = {} - # Mappa user_id -> username per il salvataggio - self.usernames: Dict[int, str] = {} + # Profili utente (in memoria) + self.profiles: Dict[int, UserProfile] = {} + # Stati setup utente + self.setup_states: Dict[int, Dict] = {} - # System instruction - self.system_instruction = """Sei un assistente per il controllo del mixer audio Yamaha TF5. + # 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: @@ -289,6 +483,12 @@ IDENTIFICAZIONE CANALI E MIX: - "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 @@ -321,6 +521,309 @@ 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}) @@ -343,7 +846,7 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. "Trascrivi in testo chiaro ciΓ² che dice questa registrazione:", types.Part.from_bytes( data=audio_bytes, - mime_type="audio/ogg" # Telegram voice Γ¨ in OGG/Opus + mime_type="audio/ogg" ), ] ) @@ -355,7 +858,6 @@ 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: - # Prova a caricare dal file history = self.conversation_manager.load_conversation(user_id) self.conversations[user_id] = history if history else [] @@ -371,12 +873,11 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. "timestamp_str": time.strftime("%Y-%m-%d %H:%M:%S") }) - # Mantieni solo gli ultimi 20 messaggi per evitare context overflow if len(history) > 20: self.conversations[user_id] = history[-20:] - # Salva su file - username = self.usernames.get(user_id, "Unknown") + 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): @@ -387,34 +888,52 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. def process_message(self, user_id: int, username: str, text: str) -> str: """Elabora un messaggio dell'utente.""" try: - # Salva username - self.usernames[user_id] = username + profile = self.get_profile(user_id, username) # Comandi speciali if text.lower() in ['/start', '/help']: - return self._get_help_message() + 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() - if text.lower() == '/export': - return self._export_conversation(user_id) + # 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 + # 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] # Escludi l'ultimo (giΓ  incluso sotto) + for msg in history[:-1] ]) - full_prompt = f"{self.system_instruction}\n\n" + 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}" @@ -436,49 +955,43 @@ Devi essere veloce, chiaro e capire anche richieste approssimative. except Exception as e: return f"❌ Errore nell'elaborazione: {e}" - def _export_conversation(self, user_id: int) -> str: - """Genera un riepilogo della conversazione.""" - history = self.get_conversation_history(user_id) + def _get_help_message(self, profile: UserProfile) -> str: + """Restituisce il messaggio di aiuto.""" + msg = "πŸŽ›οΈ **TF5 Mixer Bot**\n\n" - if not history: - return "πŸ“­ Non ci sono messaggi nella conversazione." + 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" - username = self.usernames.get(user_id, "Unknown") - filepath = self.conversation_manager._get_filepath(user_id) - - msg = f"πŸ’Ύ **Conversazione salvata**\n\n" - msg += f"πŸ‘€ Utente: @{username} (ID: {user_id})\n" - msg += f"πŸ’¬ Messaggi: {len(history)}\n" - msg += f"πŸ“ File: {filepath}\n\n" - msg += "La conversazione viene salvata automaticamente dopo ogni messaggio." + 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_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 -/export - Info sul file della conversazione - -Scrivi semplicemente cosa vuoi fare e ci penso io! 🎡 - -πŸ’Ύ Ogni conversazione viene salvata automaticamente.""" - def _get_status_message(self) -> str: """Restituisce lo stato del sistema.""" cache_age = time.time() - self.controller._cache.get("timestamp", 0) @@ -495,77 +1008,81 @@ Scrivi semplicemente cosa vuoi fare e ci penso io! 🎡 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}\n\n" - - # Lista utenti con conversazioni salvate - all_users = self.conversation_manager.get_all_users() - if all_users: - status += "πŸ“ **Conversazioni salvate:**\n" - for user in all_users[:5]: # Mostra solo i primi 5 - status += f"β€’ @{user['username']}: {user['message_count']} msg ({user['last_updated']})\n" + # Statistiche profili + all_profiles = self.profile_manager.get_all_profiles() + if all_profiles: + status += f"\nπŸ‘₯ Utenti registrati: {len(all_profiles)}\n\n" - if len(all_users) > 5: - status += f"... e altri {len(all_users) - 5} utenti" + # 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") - # 🎀 Se Γ¨ un messaggio vocale o audio + # 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"] - # Ottieni l'URL del file Telegram file_path = self._get_file_path(file_id) file_url = f"https://api.telegram.org/file/bot{self.bot.token}/{file_path}" - - # Scarica il file audio_bytes = self._download_file(file_url) - # Trascrivi con Gemini transcript = self._transcribe_audio(audio_bytes) if not transcript: - self.bot.send_message(chat_id, "❌ Non sono riuscito a capire l’audio.") + self.bot.send_message(chat_id, "❌ Non sono riuscito a capire l'audio.") return - # Conferma trascrizione self.bot.send_message(chat_id, f"πŸ—£οΈ Hai detto:\n> {transcript}") - # Processa il testo come se fosse un messaggio response = self.process_message(user_id, username, transcript) self.bot.send_message(chat_id, response, parse_mode='MarkdownV2') return - # 🎯 Messaggi testuali classici + # Messaggi testuali if "text" not in message: return text = message["text"] - print(f"πŸ“¨ Messaggio da @{username}: {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") + print("πŸ€– TF5 Mixer Telegram Bot - Con Profili Utente") print("=" * 70) # Verifica connessione @@ -579,21 +1096,15 @@ Scrivi semplicemente cosa vuoi fare e ci penso io! 🎡 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") - - # Mostra conversazioni salvate - saved_convs = self.conversation_manager.get_all_users() - if saved_convs: - print(f"\nπŸ“ Conversazioni salvate: {len(saved_convs)}") - print(f" Directory: {self.conversation_manager.conversations_dir}") + # 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") @@ -608,17 +1119,16 @@ Scrivi semplicemente cosa vuoi fare e ci penso io! 🎡 except Exception as e: print(f"❌ Errore nell'elaborazione update: {e}") - time.sleep(0.1) # Piccola pausa per non sovraccaricare + time.sleep(0.1) except KeyboardInterrupt: raise except Exception as e: print(f"❌ Errore nel polling: {e}") - time.sleep(5) # Attendi prima di riprovare + time.sleep(5) except KeyboardInterrupt: print("\n\nπŸ‘‹ Bot terminato") - print(f"πŸ’Ύ Conversazioni salvate in: {self.conversation_manager.conversations_dir}") def close(self): """Chiude le connessioni.""" @@ -630,24 +1140,24 @@ def main(): import argparse parser = argparse.ArgumentParser( - description='TF5 Mixer Telegram Bot' + 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 per salvare le conversazioni (default: conversations)') + help='Directory conversazioni (default: conversations)') + parser.add_argument('--profiles-dir', default='profiles', + help='Directory profili (default: profiles)') 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") @@ -655,7 +1165,6 @@ def main(): 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: @@ -663,7 +1172,8 @@ def main(): telegram_token=telegram_token, mixer_host=args.mixer_host, mixer_port=args.mixer_port, - conversations_dir=args.conversations_dir + conversations_dir=args.conversations_dir, + profiles_dir=args.profiles_dir ) bot.run()