mcp-nextcloud/nextcloud_mcp_server/client/__init__.py

import logging
import os
from email.utils import parsedate_to_datetime

from httpx import (
    AsyncBaseTransport,
    AsyncClient,
    Auth,
    BasicAuth,
    Request,
    Response,
    Timeout,
)

from ..controllers.notes_search import NotesSearchController
from ..http import nextcloud_httpx_transport
from .calendar import CalendarClient
from .collectives import CollectivesClient
from .contacts import ContactsClient
from .cookbook import CookbookClient
from .deck import DeckClient
from .groups import GroupsClient
from .news import NewsClient
from .notes import NotesClient
from .sharing import SharingClient
from .tables import TablesClient
from .talk import TalkClient
from .users import UsersClient
from .webdav import WebDAVClient
from .webhooks import WebhooksClient

logger = logging.getLogger(__name__)


async def log_request(request: Request):
    logger.debug(
        "Request event hook: %s %s - Waiting for content",
        request.method,
        request.url,
    )
    logger.debug("Request body: %s", request.content)
    logger.debug("Headers: %s", request.headers)


async def log_response(response: Response):
    await response.aread()
    logger.debug("Response [%s] %s", response.status_code, response.text)


def _normalise_search_result(item: dict) -> dict:
    """Normalise a webdav.search_files item to the get_files_by_tag shape.

    ``WebDAVClient.search_files`` and ``WebDAVClient.get_files_by_tag`` both
    return per-file dicts but with subtly different keys (``file_id`` vs
    ``id``) and path conventions (no leading slash vs leading slash). This
    helper makes a search result interchangeable with a tagged-file result
    so callers (notably the vector scanner) can consume both via one shape.
    """
    path = item.get("path", "")
    if path and not path.startswith("/"):
        path = "/" + path

    last_modified_timestamp = item.get("last_modified_timestamp")
    last_modified = item.get("last_modified")
    if last_modified_timestamp is None and last_modified:
        try:
            last_modified_timestamp = int(
                parsedate_to_datetime(last_modified).timestamp()
            )
        except (TypeError, ValueError):
            last_modified_timestamp = None

    file_id = item.get("file_id") if item.get("file_id") is not None else item.get("id")

    return {
        "id": file_id,
        "path": path,
        "name": item.get("name") or (path.rsplit("/", 1)[-1] if path else ""),
        "size": item.get("size", 0),
        "content_type": item.get("content_type", ""),
        "last_modified": last_modified,
        "last_modified_timestamp": last_modified_timestamp,
        "etag": item.get("etag"),
        "is_directory": item.get("is_directory", False),
    }


class AsyncDisableCookieTransport(AsyncBaseTransport):
    """This Transport disable cookies from accumulating in the httpx AsyncClient

    Thanks to: https://github.com/encode/httpx/issues/2992#issuecomment-2133258994
    """

    def __init__(self, transport: AsyncBaseTransport):
        self.transport = transport

    async def handle_async_request(self, request: Request) -> Response:
        response = await self.transport.handle_async_request(request)
        response.headers.pop("set-cookie", None)
        return response


class NextcloudClient:
    """Main Nextcloud client that orchestrates all app clients."""

    def __init__(
        self,
        base_url: str,
        username: str,
        auth: Auth | None = None,
        *,
        password: str | None = None,
        token: str | None = None,
    ):
        self.username = username
        self._client = AsyncClient(
            base_url=base_url,
            auth=auth,
            transport=AsyncDisableCookieTransport(nextcloud_httpx_transport()),
            event_hooks={"request": [log_request], "response": [log_response]},
            timeout=Timeout(timeout=30, connect=5),
        )

        # Initialize app clients
        self.notes = NotesClient(self._client, username)
        self.webdav = WebDAVClient(self._client, username)
        self.tables = TablesClient(self._client, username)
        # CalendarClient takes raw credentials so caldav (which uses niquests as
        # its preferred backend in v3.x) builds a backend-compatible auth object
        # itself — passing httpx.BasicAuth here breaks under niquests (#731).
        self.calendar = CalendarClient(
            base_url, username, password=password, token=token
        )
        self.contacts = ContactsClient(self._client, username)
        self.cookbook = CookbookClient(self._client, username)
        self.collectives = CollectivesClient(self._client, username)
        self.deck = DeckClient(self._client, username)
        self.news = NewsClient(self._client, username)
        self.talk = TalkClient(self._client, username)
        self.users = UsersClient(self._client, username)
        self.groups = GroupsClient(self._client, username)
        self.sharing = SharingClient(self._client, username)
        self.webhooks = WebhooksClient(self._client, username)

        # Initialize controllers
        self._notes_search = NotesSearchController()

    @classmethod
    def from_env(cls):
        logger.info("Creating NC Client using env vars")

        host = os.environ["NEXTCLOUD_HOST"]
        username = os.environ["NEXTCLOUD_USERNAME"]
        password = os.environ["NEXTCLOUD_PASSWORD"]
        # Pass username to constructor
        return cls(
            base_url=host,
            username=username,
            auth=BasicAuth(username, password),
            password=password,
        )

    @classmethod
    def from_token(cls, base_url: str, token: str, username: str):
        """Create NextcloudClient with OAuth bearer token.

        Args:
            base_url: Nextcloud base URL
            token: OAuth access token
            username: Nextcloud username

        Returns:
            NextcloudClient configured with bearer token authentication
        """
        from ..auth import BearerAuth  # noqa: PLC0415

        logger.info(f"Creating NC Client for user '{username}' using OAuth token")
        return cls(
            base_url=base_url,
            username=username,
            auth=BearerAuth(token),
            token=token,
        )

    async def capabilities(self):
        response = await self._client.get(
            "/ocs/v2.php/cloud/capabilities",
            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
        )
        response.raise_for_status()

        return response.json()

    async def notes_search_notes(self, *, query: str):
        """Search notes using token-based matching with relevance ranking."""
        all_notes = self.notes.get_all_notes()
        return await self._notes_search.search_notes(all_notes, query)

    async def find_files_by_tag(
        self, tag_name: str, mime_type_filter: str | None = None
    ) -> list[dict]:
        """Find files by system tag name, optionally filtered by MIME type.

        This method coordinates tag lookup and file retrieval via WebDAV:
        1. Look up the tag ID by name
        2. Get all entries (files and directories) with that tag via REPORT
        3. For each tagged directory, walk descendants matching ``mime_type_filter``
           via WebDAV SEARCH (``Depth: infinity``) so a tag on a folder applies
           to every matching file beneath it. Mirrors the directory semantics
           of :mod:`nextcloud_mcp_server.server.tag_exclusion` (issue #710).
        4. Dedupe by file id — a file directly tagged AND living under a
           tagged ancestor is returned once.

        Directory expansion only runs when ``mime_type_filter`` is set:
        without it, expanding a tagged folder would dump the user's entire
        tree into the caller, which is almost never what the operator
        wanted.

        Args:
            tag_name: Name of the system tag to search for (e.g., "vector-index")
            mime_type_filter: Optional MIME type filter (e.g., "application/pdf").
                When set, also enables directory expansion.

        Returns:
            List of file dictionaries with WebDAV properties (path, size, content_type, etc.)

        Raises:
            RuntimeError: If tag lookup or the initial file query fails. A
            failure walking one tagged directory is logged and skipped — other
            directly-tagged files are still returned.

        Examples:
            # Find all files with "vector-index" tag (no directory expansion)
            files = await nc_client.find_files_by_tag("vector-index")

            # Find only PDFs with the tag, including PDFs under any folder
            # that carries the tag
            pdfs = await nc_client.find_files_by_tag("vector-index", "application/pdf")
        """
        tag = await self.webdav.get_tag_by_name(tag_name)
        if not tag:
            logger.debug("Tag %r not found, returning empty list", tag_name)
            return []

        items = await self.webdav.get_files_by_tag(tag["id"])
        if not items:
            logger.debug("No items found with tag %r", tag_name)
            return []

        logger.debug(
            "Found %d directly-tagged item(s) with tag %r", len(items), tag_name
        )

        # Split into directly-tagged files vs tagged directories.
        by_id: dict[int, dict] = {}
        tagged_dirs: list[dict] = []
        for item in items:
            if item.get("is_directory"):
                tagged_dirs.append(item)
                continue
            if mime_type_filter and not item.get("content_type", "").startswith(
                mime_type_filter
            ):
                continue
            file_id = item.get("id")
            if file_id is None:
                continue
            by_id[file_id] = item

        # Expand each tagged directory into its descendant files matching
        # the MIME filter. Skip when no MIME filter is set — see docstring.
        if mime_type_filter and tagged_dirs:
            for dir_info in tagged_dirs:
                dir_path = dir_info.get("path", "").strip("/")
                try:
                    descendants = await self.webdav.find_by_type(
                        mime_type_filter, scope=dir_path
                    )
                except Exception as e:
                    logger.warning(
                        "Tag-based directory walk failed for %r (tag %r): %s; "
                        "skipping descendants",
                        dir_path,
                        tag_name,
                        e,
                    )
                    continue

                added = 0
                for d in descendants:
                    if d.get("is_directory"):
                        continue
                    file_id = d.get("file_id") or d.get("id")
                    if file_id is None:
                        continue
                    if file_id in by_id:
                        # Directly-tagged entry already wins; keeps the
                        # canonical shape from get_files_by_tag.
                        continue
                    by_id[file_id] = _normalise_search_result(d)
                    added += 1

                logger.debug(
                    "Tag %r: directory %r expanded to %d descendant %s file(s)",
                    tag_name,
                    dir_path,
                    added,
                    mime_type_filter,
                )

        files = list(by_id.values())
        if mime_type_filter:
            logger.info(
                "Returning %d file(s) with tag %r (mime_type=%s, "
                "%d directly-tagged folder(s) expanded)",
                len(files),
                tag_name,
                mime_type_filter,
                len(tagged_dirs),
            )
        else:
            logger.info("Returning %d file(s) with tag %r", len(files), tag_name)
        return files

    def _get_webdav_base_path(self) -> str:
        """Helper to get the base WebDAV path for the authenticated user."""
        return f"/remote.php/dav/files/{self.username}"

    async def __aenter__(self):
        """Async context manager entry."""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit - closes all clients."""
        await self.close()
        return False  # Don't suppress exceptions

    async def close(self):
        """Close the HTTP client and CalDAV client."""
        await self._client.aclose()
        await self.calendar.close()