# -*- coding: utf-8 -*- import requests import os from pydantic import ( BaseModel, Field, ) from xml.etree import ( ElementTree, ) import caldav from caldav.lib.error import NotFoundError, DAVError from datetime import ( datetime, timedelta, timezone, date, ) import icalendar import urllib.parse import uuid class Tools: def __init__(self): self.app_name = "OpenWebUI Nextcloud Tool" self.nextcloud_url = os.environ.get("NEXTCLOUD_URL") self.username = os.environ.get("NEXTCLOUD_USERNAME") self.app_password = os.environ.get("NEXTCLOUD_PASSWORD") self.caldav_client = None self.caldav_principal = None self.caldav_error = None if not self.nextcloud_url or not self.username or not self.app_password: self.caldav_error = "Advertencia: Las credenciales de Nextcloud (NEXTCLOUD_URL, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD) no se encontraron en las variables de entorno. Las funciones fallarán." print(self.caldav_error) print( "Las funciones que requieran autenticación fallarán hasta que las variables se configuren." ) else: self.caldav_base_url = f"{self.nextcloud_url.rstrip('/')}/remote.php/dav/" try: self.caldav_client = caldav.DAVClient( url=self.caldav_base_url, username=self.username, password=self.app_password, ) self.caldav_principal = self.caldav_client.principal() print("Cliente CalDAV inicializado correctamente.") except DAVError as e: self.caldav_error = f"Error al inicializar el cliente CalDAV (DAVError): {e}. Verifica URL ({self.caldav_base_url}) y credenciales." print(self.caldav_error) self.caldav_principal = None except requests.exceptions.RequestException as e: self.caldav_error = f"Error de conexión al inicializar CalDAV: {e}." print(self.caldav_error) self.caldav_principal = None except Exception as e: self.caldav_error = f"Error inesperado al inicializar CalDAV: {e}." print(self.caldav_error) self.caldav_principal = None def _check_credentials(self) -> bool: if not self.nextcloud_url or not self.username or not self.app_password: print("Error interno: Falta configuración de credenciales.") return False return True def _check_caldav(self) -> bool: if not self.caldav_principal: print( f"Error interno: Cliente CalDAV no inicializado. Razón: {self.caldav_error or 'Desconocida'}" ) return False return True def authenticate_nextcloud(self) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." try: api_url = f"{self.nextcloud_url.rstrip('/')}/remote.php/dav/files/{self.username}/" response = requests.request( "PROPFIND", api_url, auth=(self.username, self.app_password), headers={ "Depth": "0", "Content-Type": "application/xml; charset=utf-8", }, timeout=10, ) if response.status_code == 207: return "Autenticación verificada con Nextcloud (WebDAV OK)!" elif response.status_code == 401: return "Error 401: Credenciales de Nextcloud incorrectas." else: return f"Error al verificar autenticación WebDAV: Código {response.status_code}. Respuesta: {response.text[:200]}..." except requests.exceptions.Timeout: return ( "Error: Timeout al conectar con Nextcloud para verificar autenticación." ) except requests.exceptions.RequestException as e: return f"Error de conexión al verificar autenticación: {str(e)}" except Exception as e: return f"Error inesperado al verificar autenticación: {str(e)}" def list_calendars(self) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not self._check_caldav(): return f"Error: Cliente CalDAV no operativo. {self.caldav_error}" try: calendars_info = [] calendars = self.caldav_principal.calendars() if not calendars: return "No se encontraron calendarios para este usuario." for calendar in calendars: friendly_name = getattr(calendar, "name", None) if friendly_name is None: friendly_name = getattr(calendar, "display_name", "Sin nombre") technical_id = getattr(calendar, "id", "ID Desconocido") calendars_info.append(f"- {friendly_name} (ID técnico: {technical_id})") if calendars_info: return "Lista de calendarios disponibles:\n" + "\n".join(calendars_info) else: return "No se encontraron calendarios listables (quizás sólo el home set sin nombre explícito)." except DAVError as e: return f"Error DAV al listar calendarios: {str(e)}. Verifica permisos y configuración en Nextcloud." except requests.exceptions.RequestException as e: return f"Error de conexión al listar calendarios: {str(e)}" except Exception as e: print(f"Error inesperado al listar calendarios: {e}") import traceback traceback.print_exc() return f"Error inesperado al listar calendarios: {str(e)}" def create_calendar( self, calendar_name: str = Field( ..., description="El ID técnico del calendario (URL-safe, sin espacios, ej: 'trabajo-proyectos'). Requerido.", ), displayname: str = Field( ..., description="El nombre amigable que se mostrará en Nextcloud. Requerido.", ), description: str = Field( None, description="Descripción opcional para el calendario." ), ) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not self._check_caldav(): return f"Error: Cliente CalDAV no operativo. {self.caldav_error}" if not calendar_name: return "Error: El nombre técnico (ID) no puede estar vacío y debe ser URL-safe (letras, números, guiones, guiones bajos)." try: new_calendar = self.caldav_principal.make_calendar( name=displayname, cal_id=calendar_name, ) if description and hasattr(new_calendar, "set_properties"): try: props = { "{urn:ietf:params:xml:ns:caldav}calendar-description": description } new_calendar.set_properties(props=props) print(f"Descripción del calendario '{displayname}' establecida.") except Exception as prop_e: print( f"Advertencia: No se pudo establecer la descripción del calendario '{displayname}': {prop_e}" ) return f"Calendario '{displayname}' (ID técnico: {calendar_name}) creado con éxito!" except DAVError as e: error_str = str(e) if "already exists" in error_str or "405 Method Not Allowed" in error_str: return f"Error: El calendario con ID técnico '{calendar_name}' ya parece existir." elif "Forbidden" in error_str or "403 Forbidden" in error_str: return f"Error 403: Permiso denegado para crear el calendario. Verifica los permisos del usuario '{self.username}' en Nextcloud." else: return f"Error DAV al crear calendario: {error_str}" except requests.exceptions.RequestException as e: return f"Error de conexión al crear calendario: {str(e)}" except Exception as e: print(f"Ocurrió un error inesperado al crear calendario: {e}") import traceback traceback.print_exc() return f"Ocurrió un error inesperado al crear calendario: {str(e)}" def _find_calendar_by_id(self, calendar_id: str) -> caldav.objects.Calendar | None: if not self._check_caldav(): return None try: for calendar in self.caldav_principal.calendars(): if hasattr(calendar, "id") and calendar.id == calendar_id: return calendar return None except Exception as e: print(f"Error al buscar calendario por ID '{calendar_id}': {e}") import traceback traceback.print_exc() return None def get_events_from_calendar( self, calendar_name: str = Field( ..., description="El ID técnico (ej: 'personal', 'trabajo-proyectos') del calendario.", ), start_date: str = Field( None, description="Fecha de inicio (YYYY-MM-DD). Por defecto: hoy." ), end_date: str = Field( None, description="Fecha de fin (YYYY-MM-DD). Por defecto: 30 días desde inicio.", ), ) -> list: """ Obtiene los eventos del calendario y los devuelve como una lista de diccionarios. """ if not self._check_credentials(): return ["Error: Credenciales no configuradas."] if not self._check_caldav(): return [f"Error: Cliente CalDAV no operativo. {self.caldav_error}"] try: target_calendar = self._find_calendar_by_id(calendar_name) if not target_calendar: return [ f"Error: Calendario con ID técnico '{calendar_name}' no encontrado o no accesible." ] try: tz = timezone.utc if start_date and isinstance(start_date, str): dt_start = datetime.strptime(start_date, "%Y-%m-%d").replace( tzinfo=tz ) else: dt_start = datetime.now(tz).replace( hour=0, minute=0, second=0, microsecond=0 ) if end_date and isinstance(end_date, str): dt_end = datetime.strptime(end_date, "%Y-%m-%d").replace( hour=23, minute=59, second=59, tzinfo=tz ) else: dt_end = dt_start + timedelta(days=30) dt_end = dt_end.replace(hour=23, minute=59, second=59) if dt_end < dt_start: return [ "Error: La fecha de fin debe ser igual o posterior a la fecha de inicio." ] except ValueError: return ["Error: Formato de fecha inválido. Usa YYYY-MM-DD."] except Exception as date_parse_error: return [f"Error inesperado al parsear fechas: {date_parse_error}"] events_found = target_calendar.search( start=dt_start, end=dt_end, event=True, expand=True, ) friendly_calendar_name = getattr( target_calendar, "name", getattr(target_calendar, "display_name", calendar_name), ) if not events_found: return [ f"No se encontraron eventos entre {dt_start.strftime('%Y-%m-%d')} y {dt_end.strftime('%Y-%m-%d')} en '{friendly_calendar_name}'." ] event_list = [] for event_obj in events_found: try: cal = icalendar.Calendar.from_ical(event_obj.data) for component in cal.walk(): if component.name == "VEVENT": summary = str(component.get("summary", "Sin Título")) dtstart = component.get("dtstart") dtend = component.get("dtend") location = str(component.get("location", "")) description = str(component.get("description", "")) start_str = "Fecha/Hora desconocida" if dtstart: if isinstance(dtstart.dt, datetime): start_str = dtstart.dt.astimezone(tz).strftime( "%Y-%m-%d %H:%M:%S %Z" ) elif isinstance(dtstart.dt, date): start_str = dtstart.dt.strftime("%Y-%m-%d") elif isinstance(dtstart, (datetime, date)): start_str = dtstart.strftime( "%Y-%m-%d %H:%M:%S %Z" if isinstance(dtstart, datetime) else "%Y-%m-%d" ) end_str = "" if dtend: if isinstance(dtend.dt, datetime): end_str = dtend.dt.astimezone(tz).strftime( "%Y-%m-%d %H:%M:%S %Z" ) elif isinstance(dtend.dt, date): try: end_date_obj = dtend.dt - timedelta(days=1) end_str = end_date_obj.strftime("%Y-%m-%d") except Exception: end_str = dtend.dt.strftime( "%Y-%m-%d (ajustado)" ) elif isinstance(dtend, (datetime, date)): end_str = dtend.strftime( "%Y-%m-%d %H:%M:%S %Z" if isinstance(dtend, datetime) else "%Y-%m-%d" ) if dtstart and dtend: if isinstance(dtstart.dt, date) and isinstance( dtend.dt, date ): start_date_only = dtstart.dt end_date_only = dtend.dt - timedelta(days=1) if start_date_only == end_date_only: start_end_range_str = f"Día completo: {start_date_only.strftime('%Y-%m-%d')}" else: start_end_range_str = f"Día completo: {start_date_only.strftime('%Y-%m-%d')} al {end_date_only.strftime('%Y-%m-%d')}" elif isinstance(dtstart.dt, datetime) and isinstance( dtend.dt, datetime ): start_end_range_str = ( f"Inicio: {start_str}\n Fin: {end_str}" ) else: start_end_range_str = ( f"Inicio: {start_str}\n Fin: {end_str}" ) elif dtstart: start_end_range_str = f"Inicio: {start_str}" else: start_end_range_str = "Fechas/Horas desconocidas" event_list.append( { "summary": summary, "start": start_end_range_str, "location": location, "description": description, } ) except Exception as parse_error: event_list.append( f"Error al parsear un evento ({event_obj.url if hasattr(event_obj, 'url') else 'URL desconocida'}): {parse_error}" ) return event_list except NotFoundError: return [ f"Error NotFound: No se pudo encontrar el calendario '{calendar_name}' durante la búsqueda de eventos." ] except DAVError as e: return [f"Error DAV al obtener eventos: {str(e)}. Verifica permisos."] except requests.exceptions.RequestException as e: return [f"Error de conexión al obtener eventos: {str(e)}"] except Exception as e: print(f"Error inesperado al obtener eventos: {e}") import traceback traceback.print_exc() return [f"Error inesperado al obtener eventos: {str(e)}"] def create_event( self, calendar_name: str = Field( ..., description="ID técnico del calendario donde crear el evento." ), summary: str = Field(..., description="Título del evento."), start_time: str = Field( ..., description="Fecha/Hora inicio (YYYY-MM-DD o YYYY-MM-DDTHH:MM:SS[Z/+-HH:MM]).", ), end_time: str = Field( ..., description="Fecha/Hora fin (YYYY-MM-DD o YYYY-MM-DDTHH:MM:SS[Z/+-HH:MM]).", ), description: str = Field(None, description="Descripción opcional."), location: str = Field(None, description="Ubicación opcional."), ) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not self._check_caldav(): return f"Error: Cliente CalDAV no operativo. {self.caldav_error}" try: target_calendar = self._find_calendar_by_id(calendar_name) if not target_calendar: return f"Error: Calendario con ID técnico '{calendar_name}' no encontrado o no accesible." event = icalendar.Event() event.add("summary", summary) event.add("uid", str(uuid.uuid4())) event.add("dtstamp", datetime.now(timezone.utc)) try: is_all_day = False try: dtstart_obj = datetime.fromisoformat(start_time) if dtstart_obj.tzinfo is None: dtstart_obj = dtstart_obj.replace(tzinfo=timezone.utc) event.add("dtstart", dtstart_obj) dtend_obj = datetime.fromisoformat(end_time) if dtend_obj.tzinfo is None: dtend_obj = dtend_obj.replace(tzinfo=timezone.utc) event.add("dtend", dtend_obj) except ValueError: try: dstart_obj = datetime.strptime(start_time, "%Y-%m-%d").date() dend_obj = datetime.strptime(end_time, "%Y-%m-%d").date() event.add("dtstart", icalendar.vDate(dstart_obj)) if dend_obj >= dstart_obj: dtend_exclusive = dend_obj + timedelta(days=1) event.add("dtend", icalendar.vDate(dtend_exclusive)) is_all_day = True else: return "Error: Para eventos de día completo (YYYY-MM-DD), la fecha de fin debe ser igual o posterior a la de inicio." except ValueError: return f"Error: Formato de fecha/hora inválido para '{start_time}' o '{end_time}'. Usa YYYY-MM-DD para día completo o YYYY-MM-DDTHH:MM:SS[Z/+-HH:MM] para fechas con hora." if not is_all_day: if dtend_obj <= dtstart_obj: return f"Error: La fecha/hora de fin ({end_time}) debe ser posterior a la de inicio ({start_time})." except Exception as date_e: return f"Error interno al procesar fechas/horas: {date_e}" if description: event.add("description", description) if location: event.add("location", location) saved_event = target_calendar.save_event(event.to_ical()) friendly_calendar_name = getattr( target_calendar, "name", getattr(target_calendar, "display_name", calendar_name), ) return f"Evento '{summary}' creado/actualizado con éxito en '{friendly_calendar_name}' (ID: {calendar_name})!" except NotFoundError: return f"Error NotFound: No se pudo encontrar el calendario '{calendar_name}' al intentar guardar el evento." except DAVError as e: error_str = str(e) if "409 Conflict" in error_str: return f"Error 409: Conflicto al crear el evento. Podría haber un problema en el servidor." elif "403 Forbidden" in error_str: return f"Error 403: Permiso denegado para crear eventos en el calendario '{calendar_name}'. Verifica los permisos del usuario." else: return f"Error DAV al crear evento: {error_str}" except requests.exceptions.RequestException as e: return f"Error de conexión al crear evento: {str(e)}" except Exception as e: print(f"Error inesperado al crear evento: {e}") import traceback traceback.print_exc() return f"Error inesperado al crear evento: {str(e)}" def list_documents( self, path: str = Field( "/", description="Ruta de la carpeta (ej: /Documentos/). Debe empezar y terminar con /.", ), ) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not path.startswith("/"): path = "/" + path if not path.endswith("/"): path += "/" safe_path_segment = urllib.parse.quote(path.lstrip("/")) files_url = f"{self.nextcloud_url.rstrip('/')}/remote.php/dav/files/{self.username}/{safe_path_segment}" try: headers = { "Depth": "1", "Content-Type": "application/xml; charset=utf-8", } xml_body = """<?xml version="1.0" encoding="utf-8" ?> <d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns"> <d:prop> <d:displayname/> <d:resourcetype/> <d:getcontentlength/> <d:getlastmodified/> <d:getcontenttype/> <oc:fileid/> <oc:size/> </d:prop> </d:propfind> """ response = requests.request( "PROPFIND", files_url, auth=(self.username, self.app_password), headers=headers, data=xml_body.encode("utf-8"), timeout=15, ) if response.status_code == 207: root = ElementTree.fromstring(response.content) namespaces = {"d": "DAV:", "oc": "http://owncloud.org/ns"} files_list = [] base_url_path = f"/remote.php/dav/files/{self.username}{path}" for resp_elem in root.findall("d:response", namespaces): href_elem = resp_elem.find("d:href", namespaces) propstat_elem = resp_elem.find("d:propstat", namespaces) if href_elem is None or propstat_elem is None: continue prop_elem = propstat_elem.find("d:prop", namespaces) if prop_elem is None: continue href = urllib.parse.unquote(href_elem.text) if href.rstrip("/") == base_url_path.rstrip("/"): continue displayname_elem = prop_elem.find("d:displayname", namespaces) name = ( displayname_elem.text if displayname_elem is not None else href.split("/")[-1] ) resourcetype_elem = prop_elem.find("d:resourcetype", namespaces) is_collection = ( resourcetype_elem is not None and resourcetype_elem.find("d:collection", namespaces) is not None ) size_elem = prop_elem.find("oc:size", namespaces) if size_elem is None: size_elem = prop_elem.find("d:getcontentlength", namespaces) size = ( int(size_elem.text) if size_elem is not None and size_elem.text is not None else None ) size_str = "" if size is not None and not is_collection: if size < 1024: size_str = f"{size} B" elif size < 1024**2: size_str = f"{size/1024:.1f} KB" elif size < 1024**3: size_str = f"{size/1024**2:.1f} MB" else: size_str = f"{size/1024**3:.1f} GB" file_type_icon = "carpeta" if is_collection else "archivo" files_list.append( f"{file_type_icon} {name}" + (f" ({size_str})" if size_str else "") ) if files_list: return f"Contenido de '{path}':\n" + "\n".join(files_list) else: return f"La carpeta '{path}' está vacía." elif response.status_code == 404: return f"Error 404: La ruta '{path}' no fue encontrada." elif response.status_code == 401: return "Error 401: Autenticación fallida. Verifica el usuario y la contraseña de aplicación." elif response.status_code == 403: return f"Error 403: Permiso denegado para acceder a '{path}'. Verifica los permisos del usuario '{self.username}'." else: return f"Error {response.status_code} al listar documentos en '{path}'. Respuesta: {response.text[:200]}..." except requests.exceptions.Timeout: return f"Timeout al listar documentos en '{path}'. La conexión tardó demasiado." except requests.exceptions.RequestException as e: return f"Error de conexión al listar documentos: {e}" except ElementTree.ParseError as e: return f"Error al parsear la respuesta XML de Nextcloud: {e}. La respuesta del servidor no tiene el formato esperado." except Exception as e: print(f"Error inesperado al listar documentos: {e}") import traceback traceback.print_exc() return f"Error inesperado al listar documentos: {str(e)}" def create_document( self, path: str = Field( "/", description="Ruta de la carpeta donde crear el archivo (ej: /Documentos/). Debe empezar y terminar con /. ", ), filename: str = Field( ..., description="Nombre del archivo a crear (ej: nota.txt)." ), content: str = Field("", description="Contenido del archivo (texto plano)."), ) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not filename: return "Error: Se requiere un nombre de archivo." if not path.startswith("/"): path = "/" + path if not path.endswith("/"): path += "/" if "/" in filename: return "Error: El nombre de archivo no puede contener barras '/'." safe_path_segment = urllib.parse.quote(path.lstrip("/")) safe_filename = urllib.parse.quote(filename) file_url = f"{self.nextcloud_url.rstrip('/')}/remote.php/dav/files/{self.username}/{safe_path_segment}{safe_filename}" try: response = requests.put( file_url, auth=(self.username, self.app_password), data=content.encode("utf-8"), headers={"Content-Type": "text/plain; charset=utf-8"}, timeout=20, ) if response.status_code == 201: return f"Archivo '{filename}' creado con éxito en '{path}'!" elif response.status_code == 204: return f"Archivo '{filename}' actualizado con éxito en '{path}'!" elif response.status_code == 401: return "Error 401: Autenticación fallida. Verifica el usuario y la contraseña de aplicación." elif response.status_code == 403: return f"Error 403: Permiso denegado para crear/escribir en '{path}'. Verifica los permisos del usuario." elif response.status_code == 404: return f"Error 404: La ruta padre '{path}' no parece existir." elif response.status_code == 409: return f"Error 409: Conflicto al crear '{filename}'. ¿Ya existe una carpeta con ese nombre en '{path}'?" else: return f"Error {response.status_code} al crear/actualizar archivo '{filename}'. Respuesta: {response.text[:200]}..." except requests.exceptions.Timeout: return f"Timeout al crear/actualizar archivo '{filename}'. La conexión tardó demasiado." except requests.exceptions.RequestException as e: return f"Error de conexión al crear/actualizar archivo: {e}" except Exception as e: print(f"Error inesperado al crear/actualizar archivo: {e}") import traceback traceback.print_exc() return f"Error inesperado al crear/actualizar archivo: {str(e)}" def read_file(self, path: str, filename: str) -> str: if not self._check_credentials(): return "Error: Credenciales no configuradas." if not path.startswith("/"): path = "/" + path if not path.endswith("/"): path += "/" safe_path_segment = urllib.parse.quote(path.lstrip("/")) safe_filename = urllib.parse.quote(filename) file_url = f"{self.nextcloud_url.rstrip('/')}/remote.php/dav/files/{self.username}/{safe_path_segment}{safe_filename}" try: response = requests.get( file_url, auth=(self.username, self.app_password), timeout=10, ) if response.status_code == 200: return response.text elif response.status_code == 404: return f"Error 404: Archivo '{filename}' no encontrado en '{path}'." elif response.status_code == 401: return "Error 401: Autenticación fallida. Verifica el usuario y la contraseña de aplicación." elif response.status_code == 403: return f"Error 403: Permiso denegado para leer '{filename}' en '{path}'. Verifica los permisos del usuario." else: return f"Error {response.status_code}: No se pudo leer '{filename}'. {response.text[:100]}..." except requests.exceptions.Timeout: return f"Timeout al leer archivo '{filename}'. La conexión tardó demasiado." except requests.exceptions.RequestException as e: return f"Error de conexión al leer archivo: {e}" except Exception as e: print(f"Error inesperado al leer archivo: {e}") import traceback traceback.print_exc() return f"Error inesperado al leer archivo: {str(e)}"