Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 121 additions & 0 deletions ex_app/lib/all_tools/files.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
# SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
import xml.etree.ElementTree as ET
from urllib.parse import unquote
import niquests
from langchain_core.tools import tool
from nc_py_api import AsyncNextcloudApp
Expand Down Expand Up @@ -215,6 +217,122 @@ async def copy_file(source_path: str, destination_path: str):

return {"status": "success", "from": source_path, "to": destination_path}

@tool
@safe_tool
async def get_file_id_by_path(file_path: str) -> int:
"""
Resolve a file or folder path to its Nextcloud file ID.
:param file_path: the path of the file/folder (e.g., "/Documents/file.docx")
:return: the numeric file ID
"""
user_id = await nc.user

propfind_body = (
'<?xml version="1.0" encoding="UTF-8"?>'
'<d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">'
'<d:prop><oc:fileid/></d:prop>'
'</d:propfind>'
)
propfind_response = await nc._session._create_adapter(True).request(
'PROPFIND',
f"{nc.app_cfg.endpoint}/remote.php/dav/files/{user_id}/{file_path.lstrip('/')}",
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "0"},
data=propfind_body,
)

root = ET.fromstring(propfind_response.text)
fileid_element = root.find(".//{http://owncloud.org/ns}fileid")
if fileid_element is None or not fileid_element.text:
raise Exception(f"Could not resolve file ID for path: {file_path}")
return int(fileid_element.text)

@tool
@safe_tool
async def get_file_path_by_id(file_id: int) -> str:
"""
Resolve a Nextcloud file ID to its path (relative to the user's files root).
:param file_id: the numeric file ID
:return: the path of the file/folder
"""
user_id = await nc.user

search_body = (
'<?xml version="1.0" encoding="UTF-8"?>'
'<d:searchrequest xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">'
'<d:basicsearch>'
'<d:select><d:prop><d:displayname/></d:prop></d:select>'
f'<d:from><d:scope><d:href>/files/{user_id}</d:href><d:depth>infinity</d:depth></d:scope></d:from>'
'<d:where>'
'<d:eq>'
'<d:prop><oc:fileid/></d:prop>'
f'<d:literal>{file_id}</d:literal>'
'</d:eq>'
'</d:where>'
'<d:orderby/>'
'</d:basicsearch>'
'</d:searchrequest>'
)
search_response = await nc._session._create_adapter(True).request(
'SEARCH',
f"{nc.app_cfg.endpoint}/remote.php/dav/",
headers={"Content-Type": "text/xml; charset=utf-8"},
data=search_body,
)

root = ET.fromstring(search_response.text)
href_element = root.find(".//{DAV:}response/{DAV:}href")
if href_element is None or not href_element.text:
raise Exception(f"Could not resolve path for file ID: {file_id}")

href = unquote(href_element.text)
prefix = f"/remote.php/dav/files/{user_id}"
if href.startswith(prefix):
return href[len(prefix):] or "/"
return href

@tool
@dangerous_tool
async def convert_file(source_path: str, target_mime_type: str, destination_path: str | None = None):
"""
Convert a file from one MIME type to another (e.g., docx to pdf, jpg to png).
Uses the Nextcloud files conversion API.
:param source_path: the path of the file to convert (e.g., "/Documents/file.docx")
:param target_mime_type: the target MIME type (e.g., "application/pdf")
:param destination_path: optional destination path for the converted file
:return: information about the converted file
"""
user_id = await nc.user

propfind_body = (
'<?xml version="1.0" encoding="UTF-8"?>'
'<d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">'
'<d:prop><oc:fileid/></d:prop>'
'</d:propfind>'
)
propfind_response = await nc._session._create_adapter(True).request(
'PROPFIND',
f"{nc.app_cfg.endpoint}/remote.php/dav/files/{user_id}/{source_path.lstrip('/')}",
headers={"Content-Type": "application/xml; charset=utf-8", "Depth": "0"},
data=propfind_body,
)

root = ET.fromstring(propfind_response.text)
fileid_element = root.find(".//{http://owncloud.org/ns}fileid")
if fileid_element is None or not fileid_element.text:
raise Exception(f"Could not resolve file ID for path: {source_path}")
file_id = int(fileid_element.text)

payload = {'fileId': file_id, 'targetMimeType': target_mime_type}
if destination_path:
payload['destination'] = destination_path

return await nc.ocs(
'POST',
'/ocs/v2.php/apps/files/api/v1/convert',
json=payload,
response_type='json',
)

@tool
@dangerous_tool
async def delete_file(path: str):
Expand All @@ -237,11 +355,14 @@ async def delete_file(path: str):
get_file_content_by_file_link,
get_file_tree,
get_folder_tree,
get_file_id_by_path,
get_file_path_by_id,
create_public_sharing_link,
upload_file,
create_folder,
move_file,
copy_file,
convert_file,
delete_file
]

Expand Down
Loading