import os import aiohttp import asyncio from typing import Type from mautrix.client import Client from maubot import Plugin, MessageEvent from maubot.handlers import command, event from mautrix.types import EventType, MessageType, RelationType, TextMessageEventContent, Format,RelatesTo,InReplyTo from synology_api.filestation import FileStation from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper import pyotp class Config(BaseProxyConfig): def do_update(self, helper: ConfigUpdateHelper) -> None: helper.copy("synology_host") helper.copy("synology_port") helper.copy("use_ssl") helper.copy("username") helper.copy("password") helper.copy("default_folder") helper.copy("room_folders") helper.copy("api_version") helper.copy("otp_secret_key") class SynologyPhotoPlugin(Plugin): # def __init__(self, bot): # super().__init__(bot) # self.synology = None @classmethod def get_config_class(cls) -> Type[BaseProxyConfig]: return Config def save_config(self) -> None: self.config.save() async def update_config(self, new_config: dict) -> None: self.config.update(new_config) self.save_config() self.log.debug("Configuration updated and saved") async def start(self): await super().start() # Récupérer la configuration self.config.load_and_update() self.synology_host = self.config['synology_host'] self.synology_port = self.config['synology_port'] self.use_ssl = self.config['use_ssl'] self.username = self.config['username'] self.password = self.config['password'] self.default_folder = self.config['default_folder'] self.room_folders = self.config['room_folders'] self.api_version = self.config['api_version'] self.otp_secret_key = self.config['otp_secret_key'] self.synology = None otp = pyotp.TOTP(self.otp_secret_key) self.otp_code = otp.now() self.log.info("SynologyPhotoPlugin démarré") # Initialiser la connexion Synology try: self.synology = FileStation( ip_address=self.config["synology_host"], port=self.config["synology_port"], username=self.config["username"], password=self.config["password"], secure=bool(self.config["use_ssl"]), cert_verify=False, dsm_version=int(self.config["api_version"]), debug=True, otp_code=self.otp_code ) self.log.info("Connexion à Synology réussie") except Exception as e: self.log.error(f"Échec de la connexion à Synology : {e}") @event.on(EventType.ROOM_MESSAGE) async def on_event(self, event): await self.client.set_typing(event.room_id, timeout=0) # Vérifier si l'événement est un message contenant du contenu multimédia msgtype = str(event.content.msgtype) if msgtype in ["m.image", "m.video"]: self.log.info(f"Reçu un {event.content.msgtype} dans le canal {event.room_id}") self.log.error(event) await self.handle_media(event) # else: # self.log.error("ELSE") # self.log.error("'" + str(event.content.msgtype) + "'") async def handle_media(self, event): # Extraire l'URL du contenu multimédia content = event.content url = content.get("url") if not url: self.log.warning("URL du média non trouvée") return # Télécharger le fichier file_bytes = await self.client.download_media(url) filename = str(event.content.filename) if not filename or filename == "None": filename = str(event.content.body) if not file_bytes: self.log.error("Échec du téléchargement du média") return with open('/tmp/'+ filename, "wb") as binary_file: # Write bytes to file binary_file.write(file_bytes) # Déterminer le dossier cible basé sur la room Matrix room_id = event.room_id target_folder = self.room_folders.get(room_id, self.default_folder) self.log.info(f"Dossier cible pour la room {room_id} : {target_folder}") # Télécharger le média sur Synology Photos upload_success = await self.upload_to_synology('/tmp/' + filename, target_folder) if upload_success: self.log.info(f"Média téléchargé avec succès sur Synology Photos : {filename} dans {target_folder}") message = f"Média téléchargé avec succès sur Synology Photos : {filename} dans {target_folder}" # Optionnel : supprimer le fichier local après téléchargement os.remove('/tmp/' + filename) else: self.log.error("Échec du téléchargement sur Synology Photos") message = "Échec du téléchargement sur Synology Photos" content = TextMessageEventContent( msgtype=MessageType.TEXT, body=message, format=Format.HTML, formatted_body=message ) in_reply_to = InReplyTo(event_id=event.event_id) if event.content.relates_to and event.content.relates_to.rel_type == RelationType.THREAD: await event.respond(content, in_thread=True) else: content.relates_to = RelatesTo( in_reply_to=in_reply_to ) await event.respond(content) async def get_access_token(self): # Utiliser le jeton d'accès de Maubot pour télécharger le média # Vous pouvez personnaliser cette méthode selon votre configuration token = self.bot.token return token async def download_media(self, url, access_token): headers = { "Authorization": f"Bearer {access_token}" } async with aiohttp.ClientSession() as session: async with session.get(url, headers=headers) as resp: if resp.status == 200: data = await resp.read() # Déterminer le nom du fichier filename = url.split("/")[-1].split("?")[0] # Enlever les paramètres URL file_path = os.path.join("/tmp/", filename) with open(file_path, "wb") as f: f.write(data) return file_path, filename else: self.log.error(f"Erreur lors du téléchargement du média : {resp.status}") return None, None async def upload_to_synology(self, file_full_path, target_folder): if not self.synology: self.log.error("Connexion Synology non initialisée") return False try: # Vérifier si le dossier cible existe, sinon le créer ### TODO: Gérer les dossiers imbriqués # folders = self.synology.get_list() # folder_titles = [folder['title'] for folder in folders['data']['items']] # if target_folder not in folder_titles: # self.synology.create_folder(target_folder) # self.log.info(f"Dossier '{target_folder}' créé sur Synology Photos") # Télécharger le fichier with open(file_full_path, 'rb') as f: file_data = f.read() # Télécharger le fichier dans le dossier cible self.synology.upload_file( dest_path=target_folder, file_path=file_full_path, # file_data=file_data, # path=target_folder, create_parents=True, # filename=filename ) return True except Exception as e: self.log.error(f"Erreur lors de l'upload sur Synology Photos : {e}") return False async def stop(self): if self.synology: try: self.synology.logout() self.log.info("Déconnexion de Synology réussie") except Exception as e: self.log.error(f"Erreur lors de la déconnexion de Synology : {e}") await super().stop()