From d5780106d282fa091997d6634a63213e57baa137 Mon Sep 17 00:00:00 2001 From: Killian Larcher <98161034+KillianLarcher@users.noreply.github.com> Date: Tue, 20 Oct 2026 20:45:35 +0200 Subject: [PATCH] feat(contacts): add create and update tools. --- ex_app/lib/all_tools/contacts.py | 327 +++++++++++++++++++++++++------ 1 file changed, 269 insertions(+), 58 deletions(-) diff --git a/ex_app/lib/all_tools/contacts.py b/ex_app/lib/all_tools/contacts.py index eaa17a21..c2f8c231 100644 --- a/ex_app/lib/all_tools/contacts.py +++ b/ex_app/lib/all_tools/contacts.py @@ -2,16 +2,156 @@ # SPDX-License-Identifier: AGPL-3.0-or-later import json import typing +import uuid +from xml.sax.saxutils import escape from langchain_core.tools import tool from nc_py_api import AsyncNextcloudApp import xml.etree.ElementTree as ET import vobject -from ex_app.lib.all_tools.lib.impulse import ImpulseRadius, impulse +from ex_app.lib.all_tools.lib.audience import principal_radius +from ex_app.lib.all_tools.lib.impulse import ImpulseRadius, destructive, impulse + + +def _format_address(value) -> str | None: + parts = [getattr(value, field, "") for field in ("box", "extended", "street", "city", "region", "code", "country")] + formatted = ", ".join(str(part).strip() for part in parts if str(part).strip()) + return formatted or None + + +def _contact_fields(vcard) -> dict[str, typing.Any]: + return { + "uid": getattr(vcard, "uid", None).value if hasattr(vcard, "uid") else None, + "full_name": getattr(vcard, "fn", None).value if hasattr(vcard, "fn") else None, + "email": getattr(vcard.email, "value", None) if hasattr(vcard, "email") else None, + "phone": getattr(vcard.tel, "value", None) if hasattr(vcard, "tel") else None, + "address": _format_address(vcard.adr.value) if hasattr(vcard, "adr") else None, + } async def get_tools(nc: AsyncNextcloudApp): + ADDRESSBOOK_PROPFIND = ( + '' + '' + '' + '' + ) + + async def list_addressbooks() -> list[dict[str, typing.Any]]: + username = await nc._session.user + response = await nc._session._create_adapter(True).request( + "PROPFIND", + f"{nc.app_cfg.endpoint}/remote.php/dav/addressbooks/users/{username}/", + headers={ + "Content-Type": "application/xml; charset=utf-8", + "Depth": "1", + }, + data=ADDRESSBOOK_PROPFIND, + ) + if response.status_code != 207: + raise ValueError(f"Unable to list address books: HTTP {response.status_code}") + + namespaces = { + "DAV": "DAV:", + "CARD": "urn:ietf:params:xml:ns:carddav", + "OC": "http://owncloud.org/ns", + } + root = ET.fromstring(response.text) # noqa: S314 - trusted CardDAV server response + addressbooks = [] + for item in root.findall("DAV:response", namespaces): + if item.find(".//CARD:addressbook", namespaces) is None: + continue + href = item.findtext("DAV:href", default="", namespaces=namespaces).strip() + display_name = item.findtext(".//DAV:displayname", default="", namespaces=namespaces).strip() + if href: + radius = ImpulseRadius.SELF + owner = item.find(".//DAV:owner/DAV:href", namespaces) + if owner is not None and (owner.text or "").rstrip("/").rsplit("/", 1)[-1] != username: + radius = max(radius, ImpulseRadius.INDIVIDUALS) + for sharee in item.findall(".//OC:invite/OC:user/DAV:href", namespaces): + radius = max(radius, principal_radius(sharee.text)) + addressbooks.append( + { + "href": href, + "name": display_name or href.rstrip("/").rsplit("/", 1)[-1], + "radius": radius, + } + ) + return addressbooks + + def select_addressbook(addressbooks, addressbook_name): + if addressbook_name is None: + if not addressbooks: + raise ValueError("No writable address book was found") + return addressbooks[0] + addressbook = next( + (book for book in addressbooks if book["name"].casefold() == addressbook_name.casefold()), + None, + ) + if addressbook is None: + raise ValueError(f"Address book not found: {addressbook_name}") + return addressbook + + async def addressbook_radius(addressbook_name: str | None = None): + return select_addressbook(await list_addressbooks(), addressbook_name)["radius"] + + async def search_contacts(name: str) -> list[dict[str, typing.Any]]: + found_contacts = [] + for addressbook in await list_addressbooks(): + xml_body = ( + '' + '' + '' + '' + '' + f"{escape(name)}" + "" + "" + ) + response = await nc._session._create_adapter(True).request( + "REPORT", + f"{nc.app_cfg.endpoint}{addressbook['href']}", + headers={ + "Content-Type": "application/xml; charset=utf-8", + "Depth": "1", + }, + data=xml_body, + ) + + if response.status_code != 207: + raise ValueError(f"Unable to search contacts: HTTP {response.status_code}") + + namespaces = { + "DAV": "DAV:", + "CARD": "urn:ietf:params:xml:ns:carddav", + } + root = ET.fromstring(response.text) # noqa: S314 - trusted CardDAV server response + for item in root.findall("DAV:response", namespaces): + vcard_element = item.find(".//CARD:address-data", namespaces) + if vcard_element is None or not vcard_element.text: + continue + vcard = vobject.readOne(vcard_element.text.strip()) + found_contacts.append( + { + **_contact_fields(vcard), + "href": item.findtext("DAV:href", default="", namespaces=namespaces).strip(), + "etag": item.findtext(".//DAV:getetag", default="", namespaces=namespaces).strip(), + "addressbook": addressbook["name"], + "radius": addressbook["radius"], + "_vcard": vcard, + } + ) + return found_contacts + + async def contact_radius(name: str): + matches = await search_contacts(name) + if not matches: + raise ValueError(f"No contact found for name: {name}") + if len(matches) > 1: + raise ValueError(f"Multiple contacts found for name: {name}") + return matches[0]["radius"] + @tool @impulse(ImpulseRadius.SELF) async def find_person_in_users(search_term: str): @@ -38,61 +178,10 @@ async def find_person_in_contacts(name: str) -> list[dict[str, typing.Any]]: :param name: the name to search for :return: a dictionary with the person's email, phone and address """ - username = await nc._session.user - response = await nc._session._create_adapter(True).request('PROPFIND', f"{nc.app_cfg.endpoint}/remote.php/dav/addressbooks/users/{username}/", headers={ - "Content-Type": "application/xml; charset=utf-8", - }) - namespace = {"DAV": "DAV:"} # Define the namespace - root = ET.fromstring(response.text) - hrefs = root.findall(".//DAV:href", namespace) - - contacts = [] - - for href in hrefs: - link = href.text.strip() - if not link.startswith(f"/remote.php/dav/addressbooks/users/{username}/") or not link != f"/remote.php/dav/addressbooks/users/{username}/": - continue - - # XML body for the search - xml_body = """ - - - - - - - - {NAME} - - - - """.replace('{NAME}', name) - response = await nc._session._create_adapter(True).request('REPORT', f"{nc.app_cfg.endpoint}{link}", headers={ - "Content-Type": "application/xml; charset=utf-8", - "Depth": "1", - }, data=xml_body) - - if response.status_code != 207: # Multi-Status - raise Exception(f"Error: {response.status_code} - {response.reason_phrase}") - - # Parse the XML response to extract vCard data - namespace = {"DAV": "urn:ietf:params:xml:ns:carddav"} # Define the namespace - root = ET.fromstring(response.text) - vcard_elements = root.findall(".//DAV:address-data", namespace) - # Parse vCard strings into dictionaries - for vcard_element in vcard_elements: - vcard_text = vcard_element.text.strip() # Get the raw vCard data - vcard = vobject.readOne(vcard_text) # Parse vCard using vobject - - # Extract fields and add to the contacts list - contact = { - "full_name": getattr(vcard, "fn", None).value if hasattr(vcard, "fn") else None, - "email": getattr(vcard.email, "value", None) if hasattr(vcard, "email") else None, - "phone": getattr(vcard.tel, "value", None) if hasattr(vcard, "tel") else None, - "address": getattr(vcard.adr, "value", None) if hasattr(vcard, "adr") else None, - } - contacts.append(contact) - return contacts + return [ + {key: contact[key] for key in ("full_name", "email", "phone", "address")} + for contact in await search_contacts(name) + ] @tool @impulse(ImpulseRadius.SELF) @@ -104,15 +193,137 @@ async def find_details_of_current_user() -> dict[str, typing.Any]: return await nc.ocs('GET', '/ocs/v2.php/cloud/user') + @tool + @impulse(addressbook_radius) + async def create_contact( + full_name: str, + email: str | None = None, + phone: str | None = None, + address: str | None = None, + addressbook_name: str | None = None, + ) -> dict[str, str]: + """ + Create a contact in one of the current user's address books + :param full_name: the contact's full name + :param email: optional email address + :param phone: optional phone number + :param address: optional street address + :param addressbook_name: optional address book name; the first available address book is used by default + :return: the created contact's UID, name, and address book + """ + if not full_name.strip(): + raise ValueError("The contact's full name cannot be empty") + + addressbook = select_addressbook(await list_addressbooks(), addressbook_name) + + contact_uid = str(uuid.uuid4()) + vcard = vobject.vCard() + vcard.add("uid").value = contact_uid + vcard.add("fn").value = full_name.strip() + vcard.add("n").value = vobject.vcard.Name(given=full_name.strip()) + if email: + vcard.add("email").value = email + if phone: + vcard.add("tel").value = phone + if address: + vcard.add("adr").value = vobject.vcard.Address(street=address) + + href = f"{addressbook['href'].rstrip('/')}/{contact_uid}.vcf" + response = await nc._session._create_adapter(True).request( + "PUT", + f"{nc.app_cfg.endpoint}{href}", + headers={ + "Content-Type": "text/vcard; charset=utf-8", + "If-None-Match": "*", + }, + data=vcard.serialize(), + ) + if response.status_code not in (201, 204): + raise ValueError(f"Unable to create contact: HTTP {response.status_code}") + + return { + "uid": contact_uid, + "full_name": full_name.strip(), + "addressbook": addressbook["name"], + } + + @tool + @impulse(contact_radius) + @destructive + async def update_contact( + name: str, + full_name: str | None = None, + email: str | None = None, + phone: str | None = None, + address: str | None = None, + ) -> dict[str, typing.Any]: + """ + Update a uniquely matching contact while preserving unspecified fields + :param name: the current contact name to search for + :param full_name: optional replacement full name + :param email: optional replacement email address + :param phone: optional replacement phone number + :param address: optional replacement street address + :return: the updated contact + """ + if all(value is None for value in (full_name, email, phone, address)): + raise ValueError("At least one contact field must be provided") + + matches = await search_contacts(name) + if not matches: + raise ValueError(f"No contact found for name: {name}") + if len(matches) > 1: + raise ValueError(f"Multiple contacts found for name: {name}") + + contact = matches[0] + vcard = contact["_vcard"] + if not contact["uid"]: + raise ValueError("The matching contact has no UID") + if not contact["href"] or not contact["etag"]: + raise ValueError("The matching contact has no CardDAV identity") + + if full_name is not None: + if not full_name.strip(): + raise ValueError("The contact's full name cannot be empty") + vcard.fn.value = full_name.strip() + if hasattr(vcard, "n"): + vcard.n.value = vobject.vcard.Name(given=full_name.strip()) + if email is not None: + (vcard.email if hasattr(vcard, "email") else vcard.add("email")).value = email + if phone is not None: + (vcard.tel if hasattr(vcard, "tel") else vcard.add("tel")).value = phone + if address is not None: + (vcard.adr if hasattr(vcard, "adr") else vcard.add("adr")).value = vobject.vcard.Address(street=address) + + response = await nc._session._create_adapter(True).request( + "PUT", + f"{nc.app_cfg.endpoint}{contact['href']}", + headers={ + "Content-Type": "text/vcard; charset=utf-8", + "If-Match": contact["etag"], + }, + data=vcard.serialize(), + ) + if response.status_code == 412: + raise ValueError("The contact changed since it was found; search again before updating it") + if response.status_code not in (201, 204): + raise ValueError(f"Unable to update contact: HTTP {response.status_code}") + + return { + **_contact_fields(vcard), + "addressbook": contact["addressbook"], + } return [ find_person_in_users, find_person_in_contacts, - find_details_of_current_user + find_details_of_current_user, + create_contact, + update_contact, ] def get_category_name(): return "Contacts" async def is_available(nc: AsyncNextcloudApp): - return True \ No newline at end of file + return True