Telegram Bot plugin
Posted: Monday 20 January 2025 18:26
Hello,
I build a Telegram plugin with ChatGPT in french, for send message from Telegram to Domoticz
You need to rename it to plugin.py and move it in a folder
I build a Telegram plugin with ChatGPT in french, for send message from Telegram to Domoticz
You need to rename it to plugin.py and move it in a folder
Code: Select all
# -*- coding: utf-8 -*-
import Domoticz
import urllib.request
import json
import time # Pour gérer les délais
# Contenu du fichier XML intégré dans le plugin
PLUGIN_XML = """
<plugin key="TelegramBot" name="Telegram Bot" author="Dyter" version="1.0.0" wikilink="https://github.com/your-repo" externallink="https://github.com/your-repo">
<params>
<!-- Champ pour le token Telegram -->
<param field="Mode1" label="Token Telegram" width="300px" required="true" default=""/>
<!-- Champ pour l'intervalle de récupération (en secondes) -->
<param field="Mode2" label="Intervalle de récupération (secondes)" width="100px" required="true" default="5"/>
<!-- Option pour activer/désactiver les logs de débogage -->
<param field="Mode6" label="Debug" width="75px">
<options>
<option label="True" value="Debug"/>
<option label="False" value="Normal" default="true"/>
</options>
</param>
</params>
</plugin>
"""
class BasePlugin:
def __init__(self):
self.token = "" # Token Telegram (configuré par l'utilisateur)
self.update_interval = 5 # Intervalle de récupération en secondes (configuré par l'utilisateur)
self.last_update_id = None # Dernier ID de message traité
self.text_device_id = None # ID du capteur texte dans Domoticz
self.debug = False
self.last_check_time = 0 # Temps de la dernière vérification
def onStart(self):
Domoticz.Log("Plugin Telegram Bot démarré")
self.token = Parameters["Mode1"] # Récupère le token depuis la configuration
self.update_interval = int(Parameters["Mode2"]) # Récupère l'intervalle depuis la configuration
if Parameters["Mode6"] == "Debug":
self.debug = True
Domoticz.Debugging(1)
DumpConfigToLog()
else:
Domoticz.Debugging(0)
# Crée un capteur texte dans Domoticz s'il n'existe pas déjà
if 1 not in Devices:
Domoticz.Device(Name="Telegram Messages", Unit=1, TypeName="Text").Create()
Domoticz.Log("Capteur texte créé.")
self.text_device_id = 1
Domoticz.Heartbeat(self.update_interval)
def onStop(self):
Domoticz.Log("Plugin Telegram Bot arrêté")
def onHeartbeat(self):
current_time = time.time()
if current_time - self.last_check_time >= self.update_interval:
self.get_updates()
self.last_check_time = current_time # Met à jour le temps de la dernière vérification
def get_updates(self):
# Récupère les nouveaux messages via l'API Telegram
url = f"https://api.telegram.org/bot{self.token}/getUpdates"
params = {"offset": self.last_update_id + 1 if self.last_update_id else None} # Pas de timeout
url_with_params = url + "?" + urllib.parse.urlencode(params)
try:
with urllib.request.urlopen(url_with_params) as response:
if response.status == 200:
data = response.read().decode("utf-8") # Décode la réponse en UTF-8
self.handle_telegram_response(data)
else:
if self.debug:
Domoticz.Log(f"Erreur lors de la récupération des messages : {response.status}")
except urllib.error.URLError as e:
if self.debug:
Domoticz.Log(f"Erreur lors de la connexion à l'API Telegram : {str(e)}")
def handle_telegram_response(self, response_data):
# Gère la réponse de l'API Telegram
try:
data = json.loads(response_data)
if data["ok"] and data["result"]:
for update in data["result"]:
self.last_update_id = update["update_id"]
message_text = update["message"]["text"]
self.update_text_device(message_text)
except Exception as e:
if self.debug:
Domoticz.Log(f"Erreur lors du traitement de la réponse : {str(e)}")
def update_text_device(self, message):
# Met à jour le capteur texte dans Domoticz
Devices[self.text_device_id].Update(nValue=0, sValue=message)
if self.debug:
Domoticz.Log(f"Message reçu : {message}")
# Fonction pour retourner le contenu XML
def onGetXML():
return PLUGIN_XML
# Crée une instance du plugin
global _plugin
_plugin = BasePlugin()
# Fonctions requises par Domoticz
def onStart():
global _plugin
_plugin.onStart()
def onStop():
global _plugin
_plugin.onStop()
def onHeartbeat():
global _plugin
_plugin.onHeartbeat()
# Fonction pour afficher la configuration dans les logs
def DumpConfigToLog():
for x in Parameters:
if Parameters[x] != "":
Domoticz.Debug(f"'{x}': '{str(Parameters[x])}'")
Domoticz.Debug(f"Nombre de dispositifs : {str(len(Devices))}")
for x in Devices:
Domoticz.Debug(f"Dispositif : {str(x)} - {str(Devices[x])}")
Domoticz.Debug(f"ID du dispositif : '{str(Devices[x].ID)}'")
Domoticz.Debug(f"Nom du dispositif : '{Devices[x].Name}'")
Domoticz.Debug(f"Valeur nValue : {str(Devices[x].nValue)}")
Domoticz.Debug(f"Valeur sValue : '{Devices[x].sValue}'")