Generate filtered permalinks for Superset dashboards via MCP

WEBDRIVER_BASEURL_USER_FRIENDLY = "https://<fqdn>:<port>"

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""
Create dashboard permalink FastMCP tool

This module contains the FastMCP tool for generating a persistent,
shareable permalink to a dashboard with a specific filter state applied
(most commonly a time range, e.g. "April 2026" or "2026-01-01 : 2026-03-15"),
replicating the "Share > Copy permalink" button in the UI.
"""

import json
import logging
from datetime import datetime, timezone
from typing import Any

from fastmcp import Context
from pydantic import BaseModel, Field
from superset_core.mcp.decorators import tool, ToolAnnotations

from superset.commands.dashboard.permalink.create import CreateDashboardPermalinkCommand
from superset.dashboards.permalink.types import DashboardPermalinkState
from superset.extensions import event_logger
from superset.mcp_service.dashboard.schemas import DashboardError
from superset.utils.urls import get_url_path  # fallback only, see _build_permalink_url

logger = logging.getLogger(__name__)


class CreateDashboardPermalinkRequest(BaseModel):
    identifier: str = Field(
        ..., description="Dashboard ID, UUID, or slug"
    )
    time_range: str | None = Field(
        default=None,
        description=(
            "Time range to apply, in Superset's native time-range grammar. "
            "Accepts human phrases ('April 2026', 'last month', "
            "'previous calendar year') or explicit ranges "
            "('2026-01-01 : 2026-03-16'). Applied to the dashboard's native "
            "time filter (auto-detected unless time_filter_id is given)."
        ),
    )
    time_filter_id: str | None = Field(
        default=None,
        description=(
            "Native filter id to apply time_range to. Only needed if the "
            "dashboard has more than one time filter and auto-detection "
            "should be overridden. Use get_dashboard_info to inspect "
            "native filters."
        ),
    )
    data_mask: dict[str, Any] | None = Field(
        default=None,
        description=(
            "Optional raw dataMask entries for other native filters "
            "(value filters, numerical ranges, etc.), in the format "
            "{filter_id: {id, extraFormData, filterState}}. Merged with "
            "the time filter entry if time_range is also given."
        ),
    )
    active_tabs: list[str] | None = Field(
        default=None, description="IDs of active tabs, if the dashboard has tabs"
    )


class DashboardPermalinkInfo(BaseModel):
    key: str
    url: str
    dashboard_id: int
    applied_filters: dict[str, Any]


def _get_native_filter_configuration(dashboard: Any) -> list[dict[str, Any]]:
    """Extract native_filter_configuration from a dashboard's json_metadata."""
    if not dashboard.json_metadata:
        return []
    try:
        metadata = json.loads(dashboard.json_metadata)
    except (TypeError, ValueError):
        return []
    return metadata.get("native_filter_configuration", []) or []


_MONTH_NAMES = {
    # English
    "january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6,
    "july": 7, "august": 8, "september": 9, "october": 10, "november": 11,
    "december": 12,
    # Italiano
    "gennaio": 1, "febbraio": 2, "marzo": 3, "aprile": 4, "maggio": 5,
    "giugno": 6, "luglio": 7, "agosto": 8, "settembre": 9, "ottobre": 10,
    "novembre": 11, "dicembre": 12,
}


def _normalize_time_range(time_range: str) -> str:
    """Normalize a "Month YYYY" phrase into an explicit ISO range.

    Superset's native time-range grammar reliably parses explicit ranges
    ("2026-05-01 : 2026-06-01") and specific relative phrases ("last month",
    "previous calendar month"), but does NOT reliably parse a bare
    "<Month> <Year>" phrase (e.g. "May 2026"). When it fails to parse, the
    native time filter silently falls back to no restriction instead of
    raising an error - so we convert month+year phrases to an explicit range
    ourselves rather than relying on the native parser for that case.

    Anything else (already an explicit range, or a relative phrase like
    "last month") is passed through untouched.
    """
    if ":" in time_range:
        return time_range  # already an explicit range

    parts = time_range.strip().split()
    if len(parts) == 2 and parts[1].isdigit():
        month_name, year_str = parts[0].lower(), parts[1]
        month = _MONTH_NAMES.get(month_name)
        if month:
            year = int(year_str)
            start = datetime(year, month, 1)
            end_year, end_month = (year + 1, 1) if month == 12 else (year, month + 1)
            end = datetime(end_year, end_month, 1)
            return f"{start:%Y-%m-%d} : {end:%Y-%m-%d}"

    return time_range


def _find_time_filter_id(native_filters: list[dict[str, Any]]) -> str | None:
    """Return the id of the first native time-range filter found, if any."""
    for native_filter in native_filters:
        if native_filter.get("filterType") == "filter_time":
            return native_filter.get("id")
    return None


def _build_permalink_url(key: str) -> str:
    """Build a publicly reachable permalink URL for the given key.

    get_url_path() resolves against the Flask app's default SERVER_NAME,
    which inside a container/background task context (no live HTTP request)
    can resolve to the internal service hostname (e.g. "superset_app")
    instead of the public-facing one. We prefer an explicit public base URL
    from config if available, and fall back to get_url_path() otherwise.

    Adjust the config keys below to whichever your instance actually sets -
    check how the dashboard "url" field returned by list_dashboards/
    get_dashboard_info is built (likely via request.host during a real HTTP
    request) and align this helper to the same source of truth.
    """
    from flask import current_app

    public_base = (
        current_app.config.get("SUPERSET_WEBSERVER_BASE_URL")
        or current_app.config.get("WEBDRIVER_BASEURL_USER_FRIENDLY")
    )
    if public_base:
        return f"{public_base.rstrip('/')}/superset/dashboard/p/{key}/?standalone=1"

    # Fallback: may resolve to an internal hostname in non-request contexts.
    return f"{get_url_path('Superset.dashboard_permalink', key=key)}?standalone=1"


@tool(
    tags=["mutate"],
    class_permission_name="Dashboard",
    annotations=ToolAnnotations(
        title="Create dashboard permalink",
        readOnlyHint=False,
        destructiveHint=False,
    ),
)
async def create_dashboard_permalink(
    request: CreateDashboardPermalinkRequest, ctx: Context
) -> DashboardPermalinkInfo | DashboardError:
    """
    Generate a Superset permalink (/superset/dashboard/p/<key>/) for a
    dashboard with a specific filter state applied - typically a time
    range like "April 2026" or "2026-01-01 : 2026-03-15".

    Typical usage - "give me the link to dashboard X filtered to April":
    ```json
    {
        "identifier": "123",
        "time_range": "April 2026"
    }
    ```

    Explicit date range:
    ```json
    {
        "identifier": "123",
        "time_range": "2026-01-01 : 2026-03-16"
    }
    ```

    Combined with another native filter (e.g. a region value filter),
    passed as raw dataMask:
    ```json
    {
        "identifier": "123",
        "time_range": "last month",
        "data_mask": {
            "NATIVE_FILTER-abc123": {
                "id": "NATIVE_FILTER-abc123",
                "extraFormData": {"filters": [{"col": "region", "op": "IN", "val": ["EMEA"]}]},
                "filterState": {"value": ["EMEA"]}
            }
        }
    }
    ```

    If the dashboard has no native time filter and time_range is provided,
    returns a DashboardError explaining that a data_mask must be supplied
    manually instead.
    """
    await ctx.info(
        "Creating dashboard permalink: identifier=%s, time_range=%s, "
        "time_filter_id=%s"
        % (request.identifier, request.time_range, request.time_filter_id)
    )

    try:
        from superset.daos.dashboard import DashboardDAO

        with event_logger.log_context(action="mcp.create_dashboard_permalink.lookup"):
            dashboard = DashboardDAO.get_by_id_or_slug(request.identifier)

        if not dashboard:
            await ctx.warning(
                "Dashboard not found: identifier=%s" % (request.identifier,)
            )
            return DashboardError(
                error=f"Dashboard not found: {request.identifier}",
                error_type="DashboardNotFoundError",
                timestamp=datetime.now(timezone.utc),
            )

        combined_data_mask: dict[str, Any] = dict(request.data_mask or {})

        if request.time_range:
            filter_id = request.time_filter_id
            if not filter_id:
                native_filters = _get_native_filter_configuration(dashboard)
                filter_id = _find_time_filter_id(native_filters)

            if not filter_id:
                await ctx.warning(
                    "No native time filter found on dashboard id=%s and "
                    "time_filter_id not provided" % (dashboard.id,)
                )
                return DashboardError(
                    error=(
                        "This dashboard has no native time filter to apply "
                        "time_range to. Pass time_filter_id explicitly, or "
                        "supply a full data_mask instead."
                    ),
                    error_type="NoTimeFilterFoundError",
                    timestamp=datetime.now(timezone.utc),
                )

            normalized_time_range = _normalize_time_range(request.time_range)
            if normalized_time_range != request.time_range:
                await ctx.debug(
                    "Normalized time_range '%s' -> '%s'"
                    % (request.time_range, normalized_time_range)
                )

            combined_data_mask[filter_id] = {
                "id": filter_id,
                "extraFormData": {"time_range": normalized_time_range},
                "filterState": {"value": normalized_time_range},
            }

        dashboard_state: DashboardPermalinkState = {
            "dataMask": combined_data_mask,
            "activeTabs": request.active_tabs or [],
            "anchor": "",
            "urlParams": [],
        }

        with event_logger.log_context(action="mcp.create_dashboard_permalink.create"):
            key = CreateDashboardPermalinkCommand(
                dashboard_id=str(dashboard.id),
                state=dashboard_state,
            ).run()

        url = _build_permalink_url(key)

        await ctx.info(
            "Dashboard permalink created: dashboard_id=%s, key=%s, url=%s"
            % (dashboard.id, key, url)
        )

        return DashboardPermalinkInfo(
            key=key,
            url=url,
            dashboard_id=dashboard.id,
            applied_filters=combined_data_mask,
        )

    except Exception as e:
        await ctx.error(
            "Dashboard permalink creation failed: identifier=%s, error=%s, "
            "error_type=%s" % (request.identifier, str(e), type(e).__name__)
        )
        return DashboardError(
            error=f"Failed to create dashboard permalink: {str(e)}",
            error_type="InternalError",
            timestamp=datetime.now(timezone.utc),
        )