docs_hierarchy.py

Code Hygiene Score: 90

Keine Issues gefunden.

Dependencies 9

Funktionen 2

Code

"""Hierarchy operations for documentation management."""

import json
import time
from datetime import datetime
from typing import Optional

from config import Config
from infrastructure.docs_repository import get_repository
from shared.domain import LogEntry
from shared.infrastructure import get_logger

from .constants import MS_PER_SECOND


def _log_operation(
    logger,
    tool_name: str,
    request_data: dict,
    start_time: float,
    status: str = "success",
    error: Optional[str] = None,
) -> None:
    """Log an operation with timing information."""
    logger.log(
        LogEntry(
            timestamp=datetime.now(),
            client_name="mcp-docs",
            tool_name=tool_name,
            request=json.dumps(request_data),
            status=status,
            error_message=error,
            duration_ms=int((time.time() - start_time) * MS_PER_SECOND),
        )
    )


def register_hierarchy_tools(mcp) -> None:
    """Register hierarchy documentation tools with the MCP server."""

    @mcp.tool()
    def docs_hierarchy() -> dict:
        """Retrieve the complete documentation tree."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            hierarchy = repo.get_hierarchy()
            _log_operation(logger, "docs_hierarchy", {}, start_time)
            return {"success": True, "hierarchy": hierarchy}

        except Exception as e:
            _log_operation(
                logger, "docs_hierarchy", {}, start_time, "error", str(e)
            )
            return {"success": False, "error": str(e)}

    @mcp.tool()
    def docs_move(
        id: int,
        new_parent_id: Optional[int] = None,
        new_sort_order: Optional[int] = None,
    ) -> dict:
        """Move a document to a new parent."""
        start_time = time.time()
        logger = get_logger("mcp-docs", Config)
        repo = get_repository()

        try:
            doc = repo.find_by_id(id)
            if not doc:
                return {
                    "success": False,
                    "error": f"Document with ID {id} not found",
                }

            success = repo.move(id, new_parent_id, new_sort_order)
            if success:
                updated_doc = repo.find_by_id(id)
                _log_operation(
                    logger,
                    "docs_move",
                    {"id": id, "new_parent_id": new_parent_id},
                    start_time,
                )
                return {
                    "success": True,
                    "doc": updated_doc.to_dict_compact() if updated_doc else None,
                    "message": f"Document #{id} moved",
                }
            else:
                return {"success": False, "error": "Move failed"}

        except Exception as e:
            _log_operation(
                logger, "docs_move", {"id": id}, start_time, "error", str(e)
            )
            return {"success": False, "error": str(e)}
← Übersicht