"""Fabric Service Konnektor – verbindet sich über die Power BI REST API."""
from __future__ import annotations

import logging
import time
from typing import Any

import msal
import requests

from core.connectors.base import IConnector
from core.models import (
    AuthenticationError,
    ColumnInfo,
    ConnectionConfig,
    DataSourceInfo,
    DependentReportInfo,
    MeasureInfo,
    MetadataExtractionError,
    PBIConnectionError,
    PowerQueryInfo,
    RefreshScheduleInfo,
    RelationshipInfo,
    ReportMetadata,
    RLSRoleInfo,
    TableInfo,
    VisualInfo,
    WorkspaceItemInfo,
    WorkspaceMetadata,
    WorkspaceUserInfo,
)

logger = logging.getLogger(__name__)

_BASE_URL = "https://api.powerbi.com/v1.0/myorg"
_SCOPE = ["https://analysis.windows.net/powerbi/api/.default"]
_AUTHORITY_TEMPLATE = "https://login.microsoftonline.com/{tenant_id}"
_MAX_RETRIES = 3
_BACKOFF_BASE = 1.0  # seconds

# SSAS/Tabular DataType-Enum → lesbare Namen (identisch zum Local-Connector).
_DATA_TYPE_MAP: dict[int, str] = {
    2: "string",
    6: "int64",
    8: "double",
    9: "dateTime",
    10: "decimal",
    11: "boolean",
    17: "binary",
}


def _resolve_data_type(*values: Any) -> str:
    """Ermittelt einen lesbaren Datentyp aus mehreren Kandidatenwerten.

    Akzeptiert sowohl numerische SSAS-Enum-Werte (z. B. aus ``INFO.COLUMNS()``
    ``[ExplicitDataType]``/``[InferredDataType]``) als auch bereits lesbare
    Strings (z. B. aus ``INFO.VIEW.COLUMNS()`` ``[Data Type]``). Der erste
    nicht-leere Kandidat gewinnt.

    Wichtig: Importierte Spalten haben häufig ein leeres ``ExplicitDataType``
    (der Typ wird *inferred*); daher muss ``InferredDataType`` als Fallback
    berücksichtigt werden, sonst bleibt die Datentyp-Spalte leer.
    """
    for value in values:
        if value is None or value == "":
            continue
        # Numerischer Enum-Wert (int oder numerischer String)?
        if isinstance(value, bool):
            continue
        if isinstance(value, int):
            return _DATA_TYPE_MAP.get(value, str(value))
        text = str(value).strip()
        if not text:
            continue
        if text.isdigit():
            return _DATA_TYPE_MAP.get(int(text), text)
        return text
    return ""


def _get_param(params: dict, *keys: str) -> str:
    """Liest den ersten nicht-leeren Wert aus ``params`` für die gegebenen Schlüssel.

    Unterstützt sowohl snake_case- als auch camelCase-Schlüssel, sodass
    Parameter aus dem Frontend und aus Tests gleichermaßen funktionieren.
    """
    for key in keys:
        value = params.get(key)
        if value:
            return str(value)
    return ""


class FabricConnector(IConnector):
    """Konnektor für den Microsoft Fabric / Power BI Service."""

    def __init__(self) -> None:
        self._access_token: str | None = None
        self._workspace_id: str = ""
        self._dataset_id: str = ""
        self._report_id: str = ""

    # ------------------------------------------------------------------
    # IConnector interface
    # ------------------------------------------------------------------

    def connect(self, config: ConnectionConfig) -> None:
        """Authenticate and store the access token.

        Supports three auth modes:
        - ``sso``: Device Code Flow (default, no client credentials needed)
        - ``service_principal``: Client ID + Secret
        - ``interactive``: Browser popup
        """
        params = config.params

        # Akzeptiert Parameter-Schlüssel sowohl in snake_case (Backend/Tests)
        # als auch in camelCase (Frontend), analog zu connect_workspace().
        self._workspace_id = _get_param(params, "workspace_id", "workspaceId")
        self._dataset_id = _get_param(params, "dataset_id", "datasetId")
        self._report_id = _get_param(params, "report_id", "reportId")

        if not self._workspace_id or not self._dataset_id:
            raise PBIConnectionError(
                "workspace_id und dataset_id müssen angegeben werden."
            )

        self._access_token = self._authenticate(params)

    def connect_workspace(self, config: ConnectionConfig) -> None:
        """Authenticate for workspace-level operations (no dataset required).

        Wird für den Bulk-/Workspace-Export verwendet. Es genügt eine
        ``workspace_id`` – einzelne Datasets werden anschließend über
        :meth:`set_dataset` gesetzt und mit :meth:`extract_metadata`
        dokumentiert.
        """
        params = config.params

        self._workspace_id = _get_param(params, "workspace_id", "workspaceId")
        self._dataset_id = ""
        self._report_id = ""

        if not self._workspace_id:
            raise PBIConnectionError("workspace_id muss angegeben werden.")

        self._access_token = self._authenticate(params)

    def set_dataset(self, dataset_id: str, report_id: str = "") -> None:
        """Setzt das aktuell zu dokumentierende Semantic Model / den Bericht.

        Ermöglicht die Wiederverwendung derselben authentifizierten Verbindung
        für mehrere Datasets im Workspace-Export.
        """
        self._dataset_id = dataset_id
        self._report_id = report_id

    def _authenticate(self, params: dict[str, str]) -> str:
        """Führt die Authentifizierung anhand des ``auth_mode`` durch.

        Akzeptiert Parameter-Schlüssel sowohl in snake_case (Backend/Tests)
        als auch in camelCase (Frontend).
        """
        tenant_id = _get_param(params, "tenant_id", "tenantId")
        if not tenant_id:
            raise PBIConnectionError("Tenant-ID muss angegeben werden.")
        auth_mode = _get_param(params, "auth_mode", "authMode") or "sso"
        authority = _AUTHORITY_TEMPLATE.format(tenant_id=tenant_id)

        try:
            if auth_mode == "service_principal":
                client_id = _get_param(params, "client_id", "clientId")
                client_secret = _get_param(params, "client_secret", "clientSecret")
                if not client_id or not client_secret:
                    raise PBIConnectionError(
                        "client_id und client_secret werden für Service-Principal benötigt."
                    )
                return self._authenticate_service_principal(
                    client_id, client_secret, authority
                )
            else:
                # SSO via Interactive Browser Flow
                client_id = _get_param(params, "client_id", "clientId")
                return self._authenticate_device_code(client_id, authority)
        except AuthenticationError:
            raise
        except PBIConnectionError:
            raise
        except Exception as exc:
            raise PBIConnectionError(
                f"Verbindung zum Fabric Service fehlgeschlagen: {exc}"
            ) from exc

    def extract_metadata(self) -> ReportMetadata:
        """Extract metadata via Power BI REST API."""
        if not self._access_token:
            raise PBIConnectionError("Nicht verbunden – bitte zuerst connect() aufrufen.")

        report_name = self._extract_report_name()
        tables, columns, measures, relationships, rls_roles, power_queries = (
            self._extract_model_metadata()
        )
        data_sources = self._extract_data_sources()

        # Interne PBI-Tabellen filtern (analog zum Local Connector)
        internal_names = {
            t.name for t in tables
            if t.name.startswith("LocalDateTable_")
            or t.name.startswith("DateTableTemplate_")
        }
        if internal_names:
            tables = [t for t in tables if t.name not in internal_names]
            columns = [c for c in columns if c.table_name not in internal_names]
            relationships = [
                r for r in relationships
                if r.from_table not in internal_names
                and r.to_table not in internal_names
            ]
            power_queries = [
                pq for pq in power_queries
                if pq.target_table not in internal_names
                and pq.name not in internal_names
            ]

        # Tabellenklassifikation (Fakten-/Dimensions-/Hilfstabelle)
        from core.connectors.pbip_connector import PBIPConnector
        PBIPConnector._classify_tables(tables, relationships, measures)

        # Enrich data sources with table names from the model
        if data_sources and tables:
            table_names = [t.name for t in tables if not t.is_hidden]
            if len(data_sources) == 1:
                # Single data source: assign all tables
                data_sources[0].tables = table_names
            else:
                # Multiple data sources: distribute tables based on power queries
                self._assign_tables_to_sources(data_sources, power_queries, tables)

        refresh_schedule = self._extract_refresh_schedule()
        dependent_reports = self._extract_dependent_reports()
        visuals = self._extract_visuals()
        report_users = self._extract_report_users()

        return ReportMetadata(
            report_name=report_name,
            tables=tables,
            columns=columns,
            measures=measures,
            relationships=relationships,
            data_sources=data_sources,
            power_queries=power_queries,
            rls_roles=rls_roles,
            visuals=visuals,
            dependent_reports=dependent_reports,
            refresh_schedule=refresh_schedule,
            report_users=report_users,
            raw_metadata={},
        )

    def disconnect(self) -> None:
        """Clear stored token."""
        self._access_token = None

    def test_connection(self) -> bool:
        """Verify the token works with a simple API call."""
        if not self._access_token:
            return False
        try:
            url = f"{_BASE_URL}/groups/{self._workspace_id}/datasets/{self._dataset_id}"
            resp = self._api_get(url, retries=1)
            return resp is not None
        except Exception:
            return False

    # ------------------------------------------------------------------
    # Workspace enumeration (Bulk-/Workspace-Export)
    # ------------------------------------------------------------------

    def list_workspace_contents(self) -> WorkspaceMetadata:
        """Enumerate all artifacts in the workspace via the Power BI REST API.

        Ruft Berichte, Semantic Models, Zugriffsberechtigungen sowie sonstige
        Artefakte (Dashboards, Dataflows, Paginated Reports) ab und liefert
        sie gebündelt als :class:`WorkspaceMetadata` zurück.
        """
        if not self._access_token:
            raise PBIConnectionError(
                "Nicht verbunden – bitte zuerst connect_workspace() aufrufen."
            )

        from datetime import datetime

        ws_info = self._get_workspace_info()
        reports, paginated = self._list_reports()
        datasets = self._list_datasets()
        other_items: list[WorkspaceItemInfo] = list(paginated)
        other_items.extend(self._list_other_items())
        users = self._extract_workspace_users()

        return WorkspaceMetadata(
            workspace_id=self._workspace_id,
            workspace_name=ws_info.get("name", "") or self._workspace_id,
            description=ws_info.get("description", "") or "",
            capacity_id=ws_info.get("capacityId", "") or "",
            workspace_type=ws_info.get("type", "") or "",
            is_on_dedicated_capacity=bool(ws_info.get("isOnDedicatedCapacity", False)),
            users=users,
            reports=reports,
            datasets=datasets,
            other_items=other_items,
            extracted_at=datetime.now().isoformat(timespec="seconds"),
        )

    def _get_workspace_info(self) -> dict[str, Any]:
        """Fetch workspace (group) metadata. Degrades gracefully."""
        # Non-Admin-API: Gruppenliste nach ID filtern
        url = f"{_BASE_URL}/groups?$filter=id%20eq%20'{self._workspace_id}'"
        try:
            data = self._api_get(url)
            if data:
                values = data.get("value", [])
                if values:
                    return values[0]
        except Exception as exc:
            logger.warning("Workspace-Infos konnten nicht abgerufen werden: %s", exc)
        return {}

    def _list_reports(
        self,
    ) -> tuple[list[WorkspaceItemInfo], list[WorkspaceItemInfo]]:
        """List reports; separate Power BI reports from paginated reports."""
        url = f"{_BASE_URL}/groups/{self._workspace_id}/reports"
        pbi_reports: list[WorkspaceItemInfo] = []
        paginated: list[WorkspaceItemInfo] = []
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Berichte konnten nicht aufgelistet werden: %s", exc)
            return pbi_reports, paginated
        if not data:
            return pbi_reports, paginated
        for item in data.get("value", []):
            report_type = item.get("reportType", "PowerBIReport")
            info = WorkspaceItemInfo(
                id=item.get("id", ""),
                name=item.get("name", ""),
                item_type=(
                    "Report" if report_type == "PowerBIReport" else "PaginatedReport"
                ),
                dataset_id=item.get("datasetId", "") or "",
                web_url=item.get("webUrl", "") or "",
                description=item.get("description", "") or "",
                endorsement=self._read_endorsement(item),
            )
            if report_type == "PowerBIReport":
                pbi_reports.append(info)
            else:
                paginated.append(info)
        return pbi_reports, paginated

    def _list_datasets(self) -> list[WorkspaceItemInfo]:
        """List semantic models (datasets) in the workspace."""
        url = f"{_BASE_URL}/groups/{self._workspace_id}/datasets"
        datasets: list[WorkspaceItemInfo] = []
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Semantic Models konnten nicht aufgelistet werden: %s", exc)
            return datasets
        if not data:
            return datasets
        for item in data.get("value", []):
            datasets.append(
                WorkspaceItemInfo(
                    id=item.get("id", ""),
                    name=item.get("name", ""),
                    item_type="SemanticModel",
                    web_url=item.get("webUrl", "") or "",
                    description=item.get("description", "") or "",
                    endorsement=self._read_endorsement(item),
                    configured_by=item.get("configuredBy", "") or "",
                )
            )
        return datasets

    def _list_other_items(self) -> list[WorkspaceItemInfo]:
        """List dashboards and dataflows (best-effort, graceful degradation)."""
        items: list[WorkspaceItemInfo] = []
        # Dashboards
        try:
            data = self._api_get(
                f"{_BASE_URL}/groups/{self._workspace_id}/dashboards"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Dashboard",
                            web_url=item.get("webUrl", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Dashboards nicht verfügbar: %s", exc)
        # Dataflows
        try:
            data = self._api_get(
                f"{_BASE_URL}/groups/{self._workspace_id}/dataflows"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("objectId", item.get("id", "")),
                            name=item.get("name", ""),
                            item_type="Dataflow",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Dataflows nicht verfügbar: %s", exc)
        return items

    def _extract_workspace_users(self) -> list[WorkspaceUserInfo]:
        """List users/groups with access to the workspace.

        Erfordert Admin-/Member-Rechte auf dem Workspace. Ist der Zugriff
        nicht möglich, wird eine leere Liste zurückgegeben (graceful).
        """
        url = f"{_BASE_URL}/groups/{self._workspace_id}/users"
        users: list[WorkspaceUserInfo] = []
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.info("Workspace-Benutzer nicht verfügbar (Rechte?): %s", exc)
            return users
        if not data:
            return users
        for item in data.get("value", []):
            users.append(
                WorkspaceUserInfo(
                    display_name=(
                        item.get("displayName", "") or item.get("identifier", "")
                    ),
                    identifier=item.get("identifier", "") or "",
                    principal_type=item.get("principalType", "") or "",
                    access_right=item.get("groupUserAccessRight", "") or "",
                )
            )
        return users

    @staticmethod
    def _read_endorsement(item: dict[str, Any]) -> str:
        """Read the endorsement level from an item's endorsementDetails."""
        details = item.get("endorsementDetails") or {}
        return details.get("endorsement", "") or ""

    def extract_report_visuals(self, report_id: str) -> list[VisualInfo]:
        """Extract visuals for a specific report (used by the workspace export)."""
        self._report_id = report_id
        return self._extract_visuals()

    def extract_report_users(self, report_id: str) -> list[WorkspaceUserInfo]:
        """Extract report access (users/groups) for a specific report.

        Verwendet die Admin-API und degradiert bei fehlenden Rechten graceful.
        """
        self._report_id = report_id
        return self._extract_report_users()

    def _extract_report_users(self) -> list[WorkspaceUserInfo]:
        """List users/groups with access to the current report (Admin API).

        Berichtszugriffe sind nur über die Admin-API abrufbar. Ohne
        entsprechende Rechte wird eine leere Liste zurückgegeben (graceful).
        """
        if not self._report_id:
            return []
        url = f"{_BASE_URL}/admin/reports/{self._report_id}/users"
        try:
            data = self._api_get(url)
        except Exception as exc:  # noqa: BLE001
            logger.info("Berichtszugriffe nicht verfügbar (Admin-Rechte?): %s", exc)
            return []
        if not data:
            return []
        users: list[WorkspaceUserInfo] = []
        for item in data.get("value", []):
            users.append(
                WorkspaceUserInfo(
                    display_name=(
                        item.get("displayName", "") or item.get("identifier", "")
                    ),
                    identifier=item.get("identifier", "") or "",
                    principal_type=item.get("principalType", "") or "",
                    access_right=(
                        item.get("reportUserAccessRight", "")
                        or item.get("accessRight", "")
                        or ""
                    ),
                )
            )
        return users

    # ------------------------------------------------------------------
    # Authentication helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _authenticate_service_principal(
        client_id: str, client_secret: str, authority: str
    ) -> str:
        """Authenticate using a service principal (client credentials)."""
        app = msal.ConfidentialClientApplication(
            client_id,
            authority=authority,
            client_credential=client_secret,
        )
        result = app.acquire_token_for_client(scopes=_SCOPE)
        if "access_token" in result:
            return result["access_token"]
        error_desc = result.get("error_description", result.get("error", "Unbekannter Fehler"))
        raise AuthenticationError(
            f"Service-Principal-Authentifizierung fehlgeschlagen: {error_desc}"
        )

    @staticmethod
    def _authenticate_interactive(client_id: str, authority: str) -> str:
        """Authenticate interactively (browser popup)."""
        app = msal.PublicClientApplication(client_id, authority=authority)
        result = app.acquire_token_interactive(scopes=_SCOPE)
        if "access_token" in result:
            return result["access_token"]
        error_desc = result.get("error_description", result.get("error", "Unbekannter Fehler"))
        raise AuthenticationError(
            f"Interaktive Authentifizierung fehlgeschlagen: {error_desc}"
        )

    @staticmethod
    def _authenticate_device_code(client_id: str, authority: str) -> str:
        """Authenticate via Interactive Browser Flow (SSO).

        Öffnet den Browser mit dem Microsoft Login — der Benutzer meldet
        sich mit seinem eigenen Account an. Kein Code nötig.
        """
        # Well-known Microsoft Power BI first-party client ID
        if not client_id:
            client_id = "23d8f6bd-1eb0-4cc2-a08c-7bf525c67bcd"

        app = msal.PublicClientApplication(client_id, authority=authority)

        # Versuche zuerst einen gecachten Token
        accounts = app.get_accounts()
        if accounts:
            result = app.acquire_token_silent(scopes=_SCOPE, account=accounts[0])
            if result and "access_token" in result:
                return result["access_token"]

        # Interactive Browser Flow — öffnet Login direkt
        result = app.acquire_token_interactive(
            scopes=_SCOPE,
            prompt="select_account",
        )

        if "access_token" in result:
            return result["access_token"]

        error_desc = result.get("error_description", result.get("error", "Unbekannter Fehler"))
        raise AuthenticationError(
            f"SSO-Authentifizierung fehlgeschlagen: {error_desc}"
        )

    # ------------------------------------------------------------------
    # HTTP helpers with retry
    # ------------------------------------------------------------------

    def _headers(self) -> dict[str, str]:
        return {"Authorization": f"Bearer {self._access_token}"}

    def _api_get(
        self, url: str, *, retries: int = _MAX_RETRIES
    ) -> dict[str, Any] | None:
        """GET request with exponential backoff retry."""
        last_exc: Exception | None = None
        for attempt in range(retries):
            try:
                resp = requests.get(url, headers=self._headers(), timeout=30)
                resp.raise_for_status()
                return resp.json()
            except requests.exceptions.HTTPError as exc:
                # Don't retry client errors (4xx) except 429
                if exc.response is not None and 400 <= exc.response.status_code < 500 and exc.response.status_code != 429:
                    raise
                last_exc = exc
            except requests.exceptions.RequestException as exc:
                last_exc = exc
            if attempt < retries - 1:
                wait = _BACKOFF_BASE * (2 ** attempt)
                logger.debug("Retry %d/%d nach %.1fs …", attempt + 1, retries, wait)
                time.sleep(wait)
        if last_exc is not None:
            raise last_exc
        return None

    def _api_post(
        self, url: str, json_body: dict, *, retries: int = _MAX_RETRIES
    ) -> dict[str, Any] | None:
        """POST request with exponential backoff retry."""
        last_exc: Exception | None = None
        for attempt in range(retries):
            try:
                resp = requests.post(
                    url, headers=self._headers(), json=json_body, timeout=30
                )
                resp.raise_for_status()
                return resp.json()
            except requests.exceptions.HTTPError as exc:
                if exc.response is not None and 400 <= exc.response.status_code < 500 and exc.response.status_code != 429:
                    raise
                last_exc = exc
            except requests.exceptions.RequestException as exc:
                last_exc = exc
            if attempt < retries - 1:
                wait = _BACKOFF_BASE * (2 ** attempt)
                logger.debug("Retry %d/%d nach %.1fs …", attempt + 1, retries, wait)
                time.sleep(wait)
        if last_exc is not None:
            raise last_exc
        return None

    # ------------------------------------------------------------------
    # Metadata extraction helpers
    # ------------------------------------------------------------------

    def _dataset_url(self, suffix: str = "") -> str:
        return f"{_BASE_URL}/groups/{self._workspace_id}/datasets/{self._dataset_id}{suffix}"

    def _extract_report_name(self) -> str:
        """Get dataset/report name from the REST API."""
        try:
            data = self._api_get(self._dataset_url())
            if data:
                return data.get("name", "")
        except Exception as exc:
            logger.warning("Berichtname konnte nicht extrahiert werden: %s", exc)
            raise MetadataExtractionError(
                f"Berichtname konnte nicht extrahiert werden: {exc}"
            ) from exc
        return ""

    def _execute_dax_query(self, dax: str) -> list[dict[str, Any]]:
        """Execute a DAX query via the executeQueries endpoint.

        The Power BI REST API executeQueries endpoint supports DAX queries
        including INFO.* functions (e.g. INFO.TABLES(), INFO.MEASURES(),
        INFO.PARTITIONS()) which are the DAX equivalents of DMV queries.
        """
        url = self._dataset_url("/executeQueries")
        body = {
            "queries": [{"query": dax}],
            "serializerSettings": {"includeNulls": True},
        }
        try:
            data = self._api_post(url, body)
        except requests.exceptions.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else "?"
            body_text = ""
            if exc.response is not None:
                try:
                    body_text = exc.response.text[:300]
                except Exception:
                    pass
            logger.warning(
                "DAX-Abfrage fehlgeschlagen (HTTP %s): %s – %s",
                status, dax[:80], body_text,
            )
            return []
        except Exception as exc:
            logger.warning("DAX-Abfrage fehlgeschlagen: %s – %s", dax[:80], exc)
            return []
        if not data:
            return []
        # Check for error in response
        results = data.get("results", [])
        if not results:
            error = data.get("error", {})
            if error:
                logger.warning(
                    "DAX-Abfrage Fehler: %s – %s",
                    error.get("code", ""), dax[:80],
                )
            return []
        # Check for per-query error
        query_result = results[0]
        if query_result.get("error"):
            err = query_result["error"]
            logger.warning(
                "DAX-Abfrage Ergebnis-Fehler: %s – %s",
                err.get("message", err.get("code", "")), dax[:80],
            )
            return []
        tables = query_result.get("tables", [])
        if not tables:
            return []
        return tables[0].get("rows", [])

    @staticmethod
    def _row_get(row: dict[str, Any], key: str, default: Any = "") -> Any:
        """Get a value from a DAX query result row.

        The executeQueries API returns column names in brackets like [Name]
        when using SELECTCOLUMNS with renamed columns. This helper tries
        both formats: "Key" and "[Key]".

        Empty strings are treated as missing values and return the default,
        since the API sometimes returns "" instead of null for absent fields.
        """
        val = row.get(key)
        if val is not None and val != "":
            return val
        val = row.get(f"[{key}]")
        if val is not None and val != "":
            return val
        return default

    def _extract_model_metadata(
        self,
    ) -> tuple[
        list[TableInfo],
        list[ColumnInfo],
        list[MeasureInfo],
        list[RelationshipInfo],
        list[RLSRoleInfo],
        list[PowerQueryInfo],
    ]:
        """Extract tables, columns, measures, relationships, roles, and power queries via DAX INFO functions.

        Uses DAX INFO.* functions (INFO.TABLES, INFO.COLUMNS, INFO.MEASURES,
        INFO.PARTITIONS, INFO.RELATIONSHIPS, INFO.ROLES) which are the DAX
        equivalents of DMV queries and work with the executeQueries REST API.
        Falls back to INFO.VIEW.* functions if the primary queries fail.
        """
        tables: list[TableInfo] = []
        all_columns: list[ColumnInfo] = []
        measures: list[MeasureInfo] = []
        relationships: list[RelationshipInfo] = []
        rls_roles: list[RLSRoleInfo] = []
        power_queries: list[PowerQueryInfo] = []

        try:
            tables, all_columns = self._extract_tables_and_columns()
        except Exception as exc:
            logger.warning("Tabellen/Spalten-Extraktion fehlgeschlagen: %s", exc)

        try:
            measures = self._extract_measures()
        except Exception as exc:
            logger.warning("Measures-Extraktion fehlgeschlagen: %s", exc)

        try:
            relationships = self._extract_relationships()
        except Exception as exc:
            logger.warning("Beziehungs-Extraktion fehlgeschlagen: %s", exc)

        try:
            rls_roles = self._extract_rls_roles()
        except Exception as exc:
            logger.warning("RLS-Rollen-Extraktion fehlgeschlagen: %s", exc)

        try:
            power_queries = self._extract_power_queries()
        except Exception as exc:
            logger.warning("Power-Query-Extraktion fehlgeschlagen: %s", exc)

        # Per-Kategorie-Fallback auf INFO.VIEW.* : In manchen Tenants sind die
        # INFO.*-Funktionen blockiert (HTTP 400 / DatasetExecuteQueriesError),
        # während die INFO.VIEW.*-Varianten funktionieren. Da bereits Tabellen
        # über INFO.VIEW.TABLES() gefunden werden können, greift der globale
        # INFO.VIEW-Fallback weiter unten nicht – daher hier gezielt nachziehen.
        if not measures:
            try:
                measures = self._extract_measures_via_view()
                if measures:
                    logger.info(
                        "Measures über INFO.VIEW.MEASURES()-Fallback: %d", len(measures)
                    )
            except Exception as exc:
                logger.warning("INFO.VIEW.MEASURES()-Fallback fehlgeschlagen: %s", exc)

        if not relationships:
            try:
                relationships = self._extract_relationships_via_view()
                if relationships:
                    logger.info(
                        "Beziehungen über INFO.VIEW.RELATIONSHIPS()-Fallback: %d",
                        len(relationships),
                    )
            except Exception as exc:
                logger.warning(
                    "INFO.VIEW.RELATIONSHIPS()-Fallback fehlgeschlagen: %s", exc
                )

        # Fallback: If INFO.PARTITIONS() returned no power queries but we have
        # tables, generate power query entries from the table list.
        # Each table in the model corresponds to at least one partition/query.
        if not power_queries and tables:
            logger.info(
                "INFO.PARTITIONS() lieferte keine Daten – "
                "generiere Abfragen aus %d Tabellen.", len(tables),
            )
            for t in tables:
                if t.is_hidden:
                    continue
                power_queries.append(
                    PowerQueryInfo(
                        name=t.name,
                        source="Fabric Service",
                        target_table=t.name,
                        m_expression="",
                        description=t.description or "",
                    )
                )

        # Fallback: If primary INFO functions returned no tables, try INFO.VIEW variants
        if not tables and not measures:
            logger.info("INFO-Funktionen lieferten keine Tabellen – versuche INFO.VIEW Fallback…")
            try:
                fb_tables, fb_columns, fb_measures, fb_relationships = self._extract_via_info_view()
                if fb_tables:
                    tables = fb_tables
                if fb_columns and not all_columns:
                    all_columns = fb_columns
                if fb_measures:
                    measures = fb_measures
                if fb_relationships and not relationships:
                    relationships = fb_relationships
            except Exception as exc:
                logger.warning("INFO.VIEW Fallback fehlgeschlagen: %s", exc)

        return tables, all_columns, measures, relationships, rls_roles, power_queries

    def _extract_via_info_view(
        self,
    ) -> tuple[list[TableInfo], list[ColumnInfo], list[MeasureInfo], list[RelationshipInfo]]:
        """Fallback extraction using INFO.VIEW.* functions.

        INFO.VIEW functions use friendly names (table name instead of ID)
        and are more broadly supported in some environments.
        """
        tables: list[TableInfo] = []
        all_columns: list[ColumnInfo] = []
        measures: list[MeasureInfo] = []
        relationships: list[RelationshipInfo] = []

        # Tables
        table_rows = self._execute_dax_query("EVALUATE INFO.VIEW.TABLES()")
        for row in table_rows:
            name = row.get("[Name]", row.get("Name", ""))
            if not name:
                continue
            tables.append(
                TableInfo(
                    name=name,
                    description=row.get("[Description]", row.get("Description", "")) or "",
                    columns=[],
                    is_hidden=bool(row.get("[IsHidden]", row.get("IsHidden", False))),
                    row_count=None,
                )
            )

        # Columns
        col_rows = self._execute_dax_query("EVALUATE INFO.VIEW.COLUMNS()")
        for row in col_rows:
            # INFO.VIEW.COLUMNS() liefert [Table] und [DataType] (nicht
            # "Table Name"/"Data Type").
            table_name = (
                row.get("[Table]", row.get("Table", ""))
                or row.get("[Table Name]", row.get("Table Name", ""))
            )
            col_name = row.get("[Name]", row.get("Name", ""))
            data_type = _resolve_data_type(
                row.get("[DataType]", row.get("DataType", "")),
                row.get("[Data Type]", row.get("Data Type", "")),
            )
            col = ColumnInfo(
                name=col_name,
                table_name=table_name,
                data_type=data_type,
                description=row.get("[Description]", row.get("Description", "")) or "",
                is_hidden=bool(row.get("[IsHidden]", row.get("IsHidden", False))),
                is_calculated=False,
                expression=None,
                synonyms=[],
                is_key=False,
            )
            all_columns.append(col)
            for t in tables:
                if t.name == table_name:
                    t.columns.append(col)
                    break

        # Measures & Relationships über die korrigierten INFO.VIEW-Helper
        # (richtige Spaltennamen [Table] bzw. [FromTable]/[ToTable]/…).
        measures = self._extract_measures_via_view()
        relationships = self._extract_relationships_via_view()

        logger.info(
            "INFO.VIEW Fallback: %d Tabellen, %d Spalten, %d Measures, %d Beziehungen",
            len(tables), len(all_columns), len(measures), len(relationships),
        )
        return tables, all_columns, measures, relationships

    def _extract_tables_and_columns(
        self,
    ) -> tuple[list[TableInfo], list[ColumnInfo]]:
        """Extract tables and columns via DAX INFO functions.

        Tries multiple approaches to get table metadata:
        1. INFO.TABLES() with SELECTCOLUMNS
        2. INFO.VIEW.TABLES() (friendlier, more broadly supported)
        3. Simple DAX query to enumerate model tables
        """
        table_rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.TABLES(), "
            "\"ID\", [ID], \"Name\", [Name], "
            "\"Description\", [Description], \"IsHidden\", [IsHidden])"
        )

        # Fallback 1: Try INFO.VIEW.TABLES() if INFO.TABLES() failed
        if not table_rows:
            logger.info("INFO.TABLES() leer – versuche INFO.VIEW.TABLES()…")
            table_rows = self._execute_dax_query("EVALUATE INFO.VIEW.TABLES()")

        # Fallback 2: Use a simple DAX query that lists all tables
        if not table_rows:
            logger.info("INFO.VIEW.TABLES() leer – versuche UNION-basierte Tabellenabfrage…")
            # This DAX query works on any model and returns table names
            table_rows = self._execute_dax_query(
                'EVALUATE '
                'SELECTCOLUMNS('
                'FILTER(INFO.STORAGETABLECOLUMNS(), NOT ISBLANK([COLUMN_ID])), '
                '"Name", [TABLE_ID])'
            )

        col_rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.COLUMNS(), "
            "\"TableID\", [TableID], \"ExplicitName\", [ExplicitName], "
            "\"InferredName\", [InferredName], "
            "\"ExplicitDataType\", [ExplicitDataType], "
            "\"InferredDataType\", [InferredDataType], "
            "\"Description\", [Description], "
            "\"IsHidden\", [IsHidden], \"Type\", [Type], \"Expression\", [Expression], "
            "\"IsKey\", [IsKey])"
        )

        # Fallback for columns: try INFO.VIEW.COLUMNS() which includes table names directly
        if not col_rows:
            logger.info("INFO.COLUMNS() leer – versuche INFO.VIEW.COLUMNS()…")
            col_rows = self._execute_dax_query("EVALUATE INFO.VIEW.COLUMNS()")

        # Build a table-id → name map using actual IDs
        table_id_map: dict[int, str] = {}
        table_name_set: set[str] = set()
        tables: list[TableInfo] = []
        for i, row in enumerate(table_rows):
            name = self._row_get(row, "Name", "") or self._row_get(row, "Table Name", f"Table_{i}")
            if not name:
                name = f"Table_{i}"
            desc = self._row_get(row, "Description", "") or ""
            hidden = self._row_get(row, "IsHidden", False)
            table_id = self._row_get(row, "ID", None)
            if table_id is not None:
                table_id_map[int(table_id)] = name
            table_name_set.add(name)
            tables.append(
                TableInfo(
                    name=name,
                    description=desc,
                    columns=[],
                    is_hidden=bool(hidden),
                    row_count=None,
                )
            )

        # If we got columns via INFO.COLUMNS() but no table_id_map (table_rows
        # didn't provide IDs), build the map by querying INFO.TABLES() explicitly.
        if col_rows and not table_id_map:
            logger.info("table_id_map leer – versuche ID-Zuordnung über INFO.TABLES()…")
            id_rows = self._execute_dax_query(
                "EVALUATE SELECTCOLUMNS(INFO.TABLES(), \"ID\", [ID], \"Name\", [Name])"
            )
            for row in id_rows:
                tid = self._row_get(row, "ID", None)
                tname = self._row_get(row, "Name", "")
                if tid is not None and tname:
                    table_id_map[int(tid)] = tname

        # Extract column synonyms (best-effort, graceful degradation)
        synonyms_map = self._extract_column_synonyms()

        all_columns: list[ColumnInfo] = []
        for row in col_rows:
            table_id = self._row_get(row, "TableID", None)
            # Try ID-based lookup first, then name-based
            table_name = ""
            if table_id is not None and table_id_map:
                try:
                    table_name = table_id_map.get(int(table_id), "")
                except (ValueError, TypeError):
                    table_name = ""
            # For INFO.VIEW.COLUMNS() the table name is directly available
            # INFO.VIEW.COLUMNS() liefert den Tabellennamen als [Table]
            if not table_name:
                table_name = self._row_get(row, "Table", "")
            # Ältere/andere Varianten: "Table Name" bzw. "TableName"
            if not table_name:
                table_name = self._row_get(row, "Table Name", "")
            if not table_name:
                table_name = self._row_get(row, "TableName", "")

            col_name = (
                self._row_get(row, "ExplicitName", "")
                or self._row_get(row, "InferredName", "")
                or self._row_get(row, "Name", "")
            )

            # Skip columns without a name (invalid data)
            if not col_name:
                continue
            # Datentyp auflösen: ExplicitDataType (SSAS-Enum) bevorzugt, sonst
            # InferredDataType (bei importierten Spalten gesetzt), sonst der
            # lesbare Typ aus INFO.VIEW.COLUMNS() ([DataType], je nach API-Variante
            # auch "Data Type"). Numerische Enum-Werte werden in Namen übersetzt.
            data_type = _resolve_data_type(
                self._row_get(row, "ExplicitDataType", ""),
                self._row_get(row, "InferredDataType", ""),
                self._row_get(row, "DataType", ""),
                self._row_get(row, "Data Type", ""),
            )
            desc = self._row_get(row, "Description", "") or ""
            hidden = self._row_get(row, "IsHidden", False)
            col_type = self._row_get(row, "Type", 0)
            expression = self._row_get(row, "Expression", None)
            # Treat empty expression as None (API may return "" for non-calculated columns)
            if expression is not None and not str(expression).strip():
                expression = None
            is_calculated = col_type == 2 or expression is not None

            # Extract isKey from INFO.COLUMNS() result
            is_key = bool(self._row_get(row, "IsKey", False))

            # Look up synonyms from linguistic metadata
            synonyms = synonyms_map.get((table_name, col_name), [])

            col = ColumnInfo(
                name=col_name,
                table_name=table_name,
                data_type=data_type,
                description=desc,
                is_hidden=bool(hidden),
                is_calculated=is_calculated,
                expression=expression,
                synonyms=synonyms,
                is_key=is_key,
            )
            all_columns.append(col)
            # Attach to matching table
            for t in tables:
                if t.name == table_name:
                    t.columns.append(col)
                    break

        # Fallback: If columns have no table_name, try INFO.VIEW.COLUMNS()
        # which returns table names directly as "Table Name"
        unresolved_count = sum(1 for c in all_columns if not c.table_name)
        if unresolved_count > 0 and all_columns:
            logger.info(
                "%d/%d Spalten ohne Tabellenzuordnung – versuche INFO.VIEW.COLUMNS() für Zuordnung…",
                unresolved_count, len(all_columns),
            )
            view_col_rows = self._execute_dax_query("EVALUATE INFO.VIEW.COLUMNS()")
            if view_col_rows:
                # Build a lookup: column_name → table_name from INFO.VIEW.COLUMNS()
                col_table_map: dict[str, str] = {}
                for vrow in view_col_rows:
                    vname = vrow.get("[Name]", vrow.get("Name", ""))
                    vtable = (
                        vrow.get("[Table]", vrow.get("Table", ""))
                        or vrow.get("[Table Name]", vrow.get("Table Name", ""))
                    )
                    if vname and vtable:
                        col_table_map[vname] = vtable

                # Re-assign table names to unresolved columns
                for col in all_columns:
                    if not col.table_name and col.name in col_table_map:
                        col.table_name = col_table_map[col.name]
                        # Attach to matching table
                        for t in tables:
                            if t.name == col.table_name:
                                t.columns.append(col)
                                break

                resolved = unresolved_count - sum(1 for c in all_columns if not c.table_name)
                logger.info("Nachträgliche Zuordnung: %d Spalten zugeordnet.", resolved)

        # If we got columns but no tables, derive tables from column table_names
        if not tables and all_columns:
            seen: dict[str, TableInfo] = {}
            for col in all_columns:
                if col.table_name and col.table_name not in seen:
                    seen[col.table_name] = TableInfo(
                        name=col.table_name, description="", columns=[],
                        is_hidden=False, row_count=None,
                    )
                if col.table_name:
                    seen[col.table_name].columns.append(col)
            tables = list(seen.values())

        logger.info(
            "Tabellen/Spalten-Extraktion: %d Tabellen, %d Spalten",
            len(tables), len(all_columns),
        )
        return tables, all_columns

    def _extract_column_synonyms(self) -> dict[tuple[str, str], list[str]]:
        """Extract column synonyms from linguistic metadata via DAX.

        Attempts to query linguistic metadata using INFO.LINGUISTICSCHEMA()
        or similar DAX functions. Since the Fabric/Service API may not expose
        linguistic metadata via DAX in all environments, this method uses
        graceful degradation: if the query fails, it returns an empty map.

        Returns a mapping of (table_name, column_name) → list of synonyms.
        """
        synonyms_map: dict[tuple[str, str], list[str]] = {}

        # Attempt 1: Try INFO.LINGUISTICSCHEMA() which may contain XML/JSON
        rows = self._execute_dax_query(
            "EVALUATE INFO.LINGUISTICSCHEMA()"
        )

        if rows:
            for row in rows:
                content = self._row_get(row, "Content", "") or self._row_get(row, "Value", "")
                if not content:
                    for val in row.values():
                        if isinstance(val, str) and len(val) > 50:
                            content = val
                            break
                if not content:
                    continue
                try:
                    synonyms_map.update(self._parse_linguistic_xml(content))
                except Exception:
                    try:
                        synonyms_map.update(self._parse_linguistic_json(content))
                    except Exception as exc:
                        logger.debug(
                            "Linguistic Schema konnte nicht geparst werden: %s", exc
                        )

        if synonyms_map:
            logger.info(
                "Synonyme aus Linguistic Schema extrahiert: %d Spalten mit Synonymen",
                len(synonyms_map),
            )
            return synonyms_map

        # Attempt 2: Try a direct EVALUATE on linguistic metadata table
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.LINGUISTICMETADATA(), "
            "\"Content\", [Content])"
        )

        if rows:
            for row in rows:
                content = self._row_get(row, "Content", "")
                if not content:
                    continue
                try:
                    synonyms_map.update(self._parse_linguistic_xml(content))
                except Exception:
                    try:
                        synonyms_map.update(self._parse_linguistic_json(content))
                    except Exception as exc:
                        logger.debug(
                            "Linguistic Metadata konnte nicht geparst werden: %s", exc
                        )

        if synonyms_map:
            logger.info(
                "Synonyme aus Linguistic Metadata extrahiert: %d Spalten mit Synonymen",
                len(synonyms_map),
            )
        else:
            logger.info(
                "Keine Synonyme über DAX-Abfragen verfügbar – "
                "Linguistic Metadata wird in dieser Umgebung möglicherweise nicht unterstützt."
            )

        return synonyms_map

    @staticmethod
    def _parse_linguistic_xml(content: str) -> dict[tuple[str, str], list[str]]:
        """Parse XML-format linguistic metadata to extract column synonyms."""
        import xml.etree.ElementTree as ET

        result: dict[tuple[str, str], list[str]] = {}

        root = ET.fromstring(content)
        ns = ""
        if root.tag.startswith("{"):
            ns = root.tag.split("}")[0] + "}"

        for entity in root.iter(f"{ns}Entity"):
            table_name = entity.get("Name", "") or ""
            if not table_name:
                continue

            for prop in entity.iter(f"{ns}EntityProperty"):
                col_name = prop.get("Name", "") or ""
                if not col_name:
                    continue

                synonyms: list[str] = []
                for word in prop.iter(f"{ns}Word"):
                    synonym = word.text or ""
                    synonym = synonym.strip()
                    if synonym and synonym != col_name:
                        synonyms.append(synonym)

                if synonyms:
                    result[(table_name, col_name)] = synonyms

        return result

    @staticmethod
    def _parse_linguistic_json(content: str) -> dict[tuple[str, str], list[str]]:
        """Parse JSON-format linguistic metadata to extract column synonyms."""
        import json

        result: dict[tuple[str, str], list[str]] = {}

        data = json.loads(content)

        entities = data.get("Language", {}).get("Entities", {}).get("Entity", [])
        if isinstance(entities, dict):
            entities = [entities]

        for entity in entities:
            table_name = entity.get("Name", "") or ""
            if not table_name:
                continue

            properties = entity.get("Properties", {}).get("Property", [])
            if isinstance(properties, dict):
                properties = [properties]

            for prop in properties:
                col_name = prop.get("Name", "") or ""
                if not col_name:
                    continue

                synonyms: list[str] = []
                words = prop.get("Synonyms", {}).get("Synonym", [])
                if isinstance(words, str):
                    words = [words]
                elif isinstance(words, dict):
                    words = [words.get("Word", "")]

                for word in words:
                    if isinstance(word, dict):
                        word = word.get("Word", "") or word.get("#text", "")
                    synonym = str(word).strip()
                    if synonym and synonym != col_name:
                        synonyms.append(synonym)

                if synonyms:
                    result[(table_name, col_name)] = synonyms

        return result

    def _extract_measures(self) -> list[MeasureInfo]:
        """Extract measures via DAX INFO function."""
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.MEASURES(), "
            "\"TableID\", [TableID], \"Name\", [Name], "
            "\"Expression\", [Expression], \"Description\", [Description], "
            "\"FormatString\", [FormatString])"
        )
        # Resolve TableID → table name
        table_id_map = self._get_table_id_map()
        measures: list[MeasureInfo] = []
        for row in rows:
            table_id = self._row_get(row, "TableID", 0)
            table_name = table_id_map.get(int(table_id), str(table_id)) if table_id else ""
            measures.append(
                MeasureInfo(
                    name=self._row_get(row, "Name", ""),
                    table_name=table_name,
                    expression=self._row_get(row, "Expression", "") or "",
                    description=self._row_get(row, "Description", "") or "",
                    format_string=self._row_get(row, "FormatString", None),
                )
            )
        return measures

    def _extract_relationships(self) -> list[RelationshipInfo]:
        """Extract relationships via DAX INFO function."""
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.RELATIONSHIPS(), "
            "\"FromTableID\", [FromTableID], \"FromColumnID\", [FromColumnID], "
            "\"ToTableID\", [ToTableID], \"ToColumnID\", [ToColumnID], "
            "\"CrossFilteringBehavior\", [CrossFilteringBehavior], "
            "\"IsActive\", [IsActive], "
            "\"FromCardinality\", [FromCardinality], \"ToCardinality\", [ToCardinality])"
        )
        # Tabellen- und Spalten-ID-Maps für die Namensauflösung
        table_id_map = self._get_table_id_map()
        column_id_map = self._get_column_id_map()

        relationships: list[RelationshipInfo] = []
        for row in rows:
            from_tid = self._row_get(row, "FromTableID", 0)
            from_cid = self._row_get(row, "FromColumnID", 0)
            to_tid = self._row_get(row, "ToTableID", 0)
            to_cid = self._row_get(row, "ToColumnID", 0)

            from_table = table_id_map.get(int(from_tid), str(from_tid))
            to_table = table_id_map.get(int(to_tid), str(to_tid))
            from_column = column_id_map.get((int(from_tid), int(from_cid)), str(from_cid))
            to_column = column_id_map.get((int(to_tid), int(to_cid)), str(to_cid))

            # Kardinalität aus From/To-Cardinality ableiten
            from_card = self._row_get(row, "FromCardinality", "")
            to_card = self._row_get(row, "ToCardinality", "")
            cardinality = self._derive_cardinality(from_card, to_card)

            cross_filter = str(self._row_get(row, "CrossFilteringBehavior", "Single"))
            # CrossFilteringBehavior: 1=Single, 2=Both
            if cross_filter == "1":
                cross_filter = "Single"
            elif cross_filter == "2":
                cross_filter = "Both"

            relationships.append(
                RelationshipInfo(
                    from_table=from_table,
                    from_column=from_column,
                    to_table=to_table,
                    to_column=to_column,
                    cardinality=cardinality,
                    cross_filter=cross_filter,
                    is_active=bool(self._row_get(row, "IsActive", True)),
                )
            )
        return relationships

    def _get_column_id_map(self) -> dict[tuple[int, int], str]:
        """Build a (TableID, ColumnID) → column name map via DAX INFO function."""
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.COLUMNS(), "
            "\"TableID\", [TableID], \"ID\", [ID], \"ExplicitName\", [ExplicitName])"
        )
        col_map: dict[tuple[int, int], str] = {}
        for row in rows:
            tid = self._row_get(row, "TableID", 0)
            cid = self._row_get(row, "ID", 0)
            name = self._row_get(row, "ExplicitName", "")
            col_map[(int(tid), int(cid))] = name
        return col_map

    @staticmethod
    def _derive_cardinality(from_card: Any, to_card: Any) -> str:
        """Leitet die Kardinalitäts-Bezeichnung aus den DMV-Werten ab."""
        # DMV gibt numerische Werte: 1=One, 2=Many
        card_map = {1: "One", 2: "Many", "1": "One", "2": "Many",
                    "One": "One", "Many": "Many"}
        fc = card_map.get(from_card, "Many")
        tc = card_map.get(to_card, "One")
        return f"{fc}To{tc}"

    def _extract_measures_via_view(self) -> list[MeasureInfo]:
        """Fallback über ``INFO.VIEW.MEASURES()``.

        Wird verwendet, wenn ``INFO.MEASURES()`` in der Umgebung blockiert ist
        (HTTP 400 / DatasetExecuteQueriesError). INFO.VIEW liefert die
        Tabellennamen direkt als ``[Table]`` (keine ID-Auflösung nötig).

        Hinweis: ``[Expression]`` (DAX-Formel) wird laut Microsoft nur für
        Nutzer mit Schreibrechten auf dem Semantic Model befüllt.
        """
        rows = self._execute_dax_query("EVALUATE INFO.VIEW.MEASURES()")
        measures: list[MeasureInfo] = []
        for row in rows:
            name = self._row_get(row, "Name", "")
            if not name:
                continue
            measures.append(
                MeasureInfo(
                    name=name,
                    table_name=self._row_get(row, "Table", ""),
                    expression=self._row_get(row, "Expression", "") or "",
                    description=self._row_get(row, "Description", "") or "",
                    format_string=self._row_get(row, "FormatString", None),
                )
            )
        return measures

    def _extract_relationships_via_view(self) -> list[RelationshipInfo]:
        """Fallback über ``INFO.VIEW.RELATIONSHIPS()``.

        Wird verwendet, wenn ``INFO.RELATIONSHIPS()`` blockiert ist. INFO.VIEW
        liefert Tabellen-/Spaltennamen und Kardinalitäten direkt als lesbare
        Strings ([FromTable], [FromColumn], [FromCardinality], [ToTable],
        [ToColumn], [ToCardinality], [CrossFilteringBehavior], [IsActive]).
        """
        rows = self._execute_dax_query("EVALUATE INFO.VIEW.RELATIONSHIPS()")
        relationships: list[RelationshipInfo] = []
        for row in rows:
            from_table = self._row_get(row, "FromTable", "")
            to_table = self._row_get(row, "ToTable", "")
            if not from_table or not to_table:
                continue
            cardinality = self._derive_cardinality(
                self._row_get(row, "FromCardinality", ""),
                self._row_get(row, "ToCardinality", ""),
            )
            cross_filter = str(self._row_get(row, "CrossFilteringBehavior", "Single"))
            if cross_filter == "1":
                cross_filter = "Single"
            elif cross_filter == "2":
                cross_filter = "Both"
            relationships.append(
                RelationshipInfo(
                    from_table=from_table,
                    from_column=self._row_get(row, "FromColumn", ""),
                    to_table=to_table,
                    to_column=self._row_get(row, "ToColumn", ""),
                    cardinality=cardinality,
                    cross_filter=cross_filter,
                    is_active=bool(self._row_get(row, "IsActive", True)),
                )
            )
        return relationships

    def _extract_rls_roles(self) -> list[RLSRoleInfo]:
        """Extract RLS roles via DAX INFO function, including role members."""
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.ROLES(), "
            "\"ID\", [ID], \"Name\", [Name], \"Description\", [Description])"
        )
        roles: list[RLSRoleInfo] = []
        role_by_id: dict[int, RLSRoleInfo] = {}
        for row in rows:
            role = RLSRoleInfo(
                name=self._row_get(row, "Name", ""),
                description=self._row_get(row, "Description", "") or "",
            )
            roles.append(role)
            rid = self._row_get(row, "ID", None)
            if rid is not None:
                try:
                    role_by_id[int(rid)] = role
                except (ValueError, TypeError):
                    pass

        # Rollen-Mitglieder ergänzen (best effort, via DAX INFO.ROLEMEMBERSHIPS)
        try:
            self._attach_rls_members(role_by_id)
        except Exception as exc:  # noqa: BLE001
            logger.info("RLS-Rollen-Mitglieder konnten nicht ermittelt werden: %s", exc)

        return roles

    def _attach_rls_members(self, role_by_id: dict[int, RLSRoleInfo]) -> None:
        """Ergänzt die Mitglieder je RLS-Rolle über INFO.ROLEMEMBERSHIPS()."""
        if not role_by_id:
            return
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.ROLEMEMBERSHIPS(), "
            "\"RoleID\", [RoleID], \"MemberName\", [MemberName])"
        )
        for row in rows:
            rid = self._row_get(row, "RoleID", None)
            member = self._row_get(row, "MemberName", "")
            if rid is None or not member:
                continue
            try:
                role = role_by_id.get(int(rid))
            except (ValueError, TypeError):
                role = None
            if role is not None:
                role.members.append(str(member))

    def _extract_power_queries(self) -> list[PowerQueryInfo]:
        """Extract Power Query M expressions from partitions via DAX INFO function.

        Falls back to a simpler query without QueryDefinition if the full
        query returns no results (some environments restrict this field).
        """
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.PARTITIONS(), "
            "\"TableID\", [TableID], \"Name\", [Name], "
            "\"QueryDefinition\", [QueryDefinition])"
        )

        # Fallback: If no rows returned, try without QueryDefinition
        # (some environments block this field)
        if not rows:
            logger.info(
                "INFO.PARTITIONS() mit QueryDefinition lieferte keine Daten – "
                "versuche ohne QueryDefinition…"
            )
            rows = self._execute_dax_query(
                "EVALUATE SELECTCOLUMNS(INFO.PARTITIONS(), "
                "\"TableID\", [TableID], \"Name\", [Name], "
                "\"Type\", [Type], \"DataSourceID\", [DataSourceID])"
            )

        # Tabellen-ID → Name Zuordnung für die Auflösung
        table_id_map = self._get_table_id_map()

        pq_list: list[PowerQueryInfo] = []
        for row in rows:
            m_expr = self._row_get(row, "QueryDefinition", "") or ""
            table_id = self._row_get(row, "TableID", None)
            table_name = ""
            if table_id is not None and table_id_map:
                try:
                    table_name = table_id_map.get(int(table_id), "")
                except (ValueError, TypeError):
                    table_name = ""
            # Fallback: use partition name as table name if no mapping found
            partition_name = self._row_get(row, "Name", "")
            if not table_name:
                table_name = partition_name

            if m_expr:
                # Full M expression available
                source_name = self._extract_m_source(m_expr) or "partition"
                pq_list.append(
                    PowerQueryInfo(
                        name=partition_name,
                        source=source_name,
                        target_table=table_name,
                        m_expression=m_expr,
                        description="",
                    )
                )
            elif table_name:
                # No M expression but we know the target table –
                # still record the partition so Chapter 6 isn't empty
                pq_list.append(
                    PowerQueryInfo(
                        name=partition_name or table_name,
                        source="Fabric Service",
                        target_table=table_name,
                        m_expression="",
                        description="",
                    )
                )

        logger.info("Power-Query-Extraktion: %d Partitionen gefunden", len(pq_list))
        return pq_list

    def _get_table_id_map(self) -> dict[int, str]:
        """Build a TableID → table name map via DAX INFO function."""
        rows = self._execute_dax_query(
            "EVALUATE SELECTCOLUMNS(INFO.TABLES(), \"ID\", [ID], \"Name\", [Name])"
        )
        id_map: dict[int, str] = {}
        for row in rows:
            tid = self._row_get(row, "ID", None)
            name = self._row_get(row, "Name", "")
            if tid is not None and name:
                id_map[int(tid)] = name

        # Fallback: If INFO.TABLES() returned no IDs, try INFO.VIEW.TABLES()
        # and build a synthetic map using row index as ID
        if not id_map:
            logger.info("_get_table_id_map: INFO.TABLES() leer – versuche INFO.VIEW.TABLES()…")
            view_rows = self._execute_dax_query("EVALUATE INFO.VIEW.TABLES()")
            for i, row in enumerate(view_rows):
                name = row.get("[Name]", row.get("Name", ""))
                if name:
                    # Use index as synthetic ID – matches order from INFO.PARTITIONS()
                    id_map[i + 1] = name

        return id_map

    @staticmethod
    def _extract_m_source(m_expr: str) -> str | None:
        """Extrahiert die Ursprungsquelle aus einer M-Abfrage."""
        import re

        # SQL Server
        m = re.search(r'Sql\.Database\(\s*"([^"]+)"\s*,\s*"([^"]+)"', m_expr)
        if m:
            return f"SQL: {m.group(1)} / {m.group(2)}"
        # OData
        m = re.search(r'OData\.Feed\(\s*"([^"]+)"', m_expr)
        if m:
            return f"OData: {m.group(1)}"
        # SharePoint
        m = re.search(r'SharePoint\.\w+\(\s*"([^"]+)"', m_expr)
        if m:
            return f"SharePoint: {m.group(1)}"
        # Excel
        m = re.search(r'Excel\.Workbook\([^,]*"([^"]+)"', m_expr)
        if m:
            return f"Excel: {m.group(1)}"
        # Web
        m = re.search(r'Web\.Contents\(\s*"([^"]+)"', m_expr)
        if m:
            return f"Web: {m.group(1)}"
        # Analysis Services
        m = re.search(r'AnalysisServices\.Database\(\s*"([^"]+)"', m_expr)
        if m:
            return f"SSAS: {m.group(1)}"
        # Generischer Fallback
        m = re.search(r'(\w+\.\w+)\(', m_expr)
        if m:
            return m.group(1)
        return None

    def _extract_data_sources(self) -> list[DataSourceInfo]:
        """Extract data sources via REST API."""
        url = self._dataset_url("/datasources")
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Datenquellen konnten nicht extrahiert werden: %s", exc)
            return []
        if not data:
            return []
        sources: list[DataSourceInfo] = []
        seen_names: set[str] = set()
        for item in data.get("value", []):
            conn_details = item.get("connectionDetails", {})
            ds_type = item.get("datasourceType", "Unknown")
            server = conn_details.get("server", "")
            database = conn_details.get("database", None)
            path = conn_details.get("path", "")
            url_val = conn_details.get("url", "")

            # Lesbaren Namen ableiten statt kryptische datasourceId
            name = self._derive_datasource_name(ds_type, server, database, path, url_val)

            # Duplikate vermeiden
            if name in seen_names:
                name = f"{name} ({item.get('datasourceId', '')[:8]})"
            seen_names.add(name)

            # Connection-String zusammenbauen
            connection_string = server
            if not connection_string:
                connection_string = url_val or path

            sources.append(
                DataSourceInfo(
                    name=name,
                    source_type=ds_type,
                    connection_string=connection_string,
                    schema=database,
                )
            )
        return sources

    @staticmethod
    def _derive_datasource_name(
        ds_type: str, server: str, database: str | None, path: str, url: str
    ) -> str:
        """Leitet einen lesbaren Namen für eine Datenquelle ab."""
        if ds_type == "Sql" and server and database:
            return f"SQL: {server} / {database}"
        if ds_type == "AnalysisServices" and server and database:
            return f"SSAS: {server} / {database}"
        if "SharePoint" in ds_type and url:
            return f"SharePoint: {url}"
        if "OData" in ds_type and url:
            return f"OData: {url}"
        if "Web" in ds_type and url:
            return f"Web: {url}"
        if "Excel" in ds_type and path:
            return f"Excel: {path}"
        if server and database:
            return f"{ds_type}: {server} / {database}"
        if server:
            return f"{ds_type}: {server}"
        if url:
            return f"{ds_type}: {url}"
        if path:
            return f"{ds_type}: {path}"
        return f"{ds_type} (unbekannt)"

    @staticmethod
    def _assign_tables_to_sources(
        data_sources: list[DataSourceInfo],
        power_queries: list[PowerQueryInfo],
        tables: list[TableInfo],
    ) -> None:
        """Assign table names to data sources based on power query source matching."""
        # Build a map: source keyword → data source index
        source_map: dict[int, list[str]] = {i: [] for i in range(len(data_sources))}
        table_names = [t.name for t in tables if not t.is_hidden]

        for pq in power_queries:
            if not pq.source or not pq.target_table:
                continue
            # Try to match the power query source to a data source
            matched_idx = 0  # Default to first source
            for idx, ds in enumerate(data_sources):
                # Match by source type or connection string overlap
                if ds.name and pq.source and (
                    ds.name.lower() in pq.source.lower()
                    or pq.source.lower() in ds.name.lower()
                    or ds.source_type.lower() in pq.source.lower()
                ):
                    matched_idx = idx
                    break
            if pq.target_table not in source_map[matched_idx]:
                source_map[matched_idx].append(pq.target_table)

        # Assign tables to sources
        assigned_tables: set[str] = set()
        for idx, table_list in source_map.items():
            if table_list:
                data_sources[idx].tables = table_list
                assigned_tables.update(table_list)

        # Assign remaining unassigned tables to the first source
        unassigned = [t for t in table_names if t not in assigned_tables]
        if unassigned and data_sources:
            if data_sources[0].tables:
                data_sources[0].tables.extend(unassigned)
            else:
                data_sources[0].tables = unassigned

    def _extract_refresh_schedule(self) -> RefreshScheduleInfo | None:
        """Extract refresh schedule via REST API."""
        url = self._dataset_url("/refreshes")
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Refresh-Zeitplan konnte nicht extrahiert werden: %s", exc)
            return None
        if not data:
            return None
        refreshes = data.get("value", [])
        if not refreshes:
            return None
        # Derive schedule info from the most recent refresh entries
        latest = refreshes[0] if refreshes else {}
        return RefreshScheduleInfo(
            frequency="OnDemand",
            times=[latest.get("startTime", "")] if latest.get("startTime") else [],
            timezone="UTC",
        )

    def _extract_visuals(self) -> list[VisualInfo]:
        """Extract visuals via the Power BI Report REST API.

        Uses:
        - GET /groups/{workspace_id}/reports/{report_id}/pages
        - GET /groups/{workspace_id}/reports/{report_id}/pages/{page_name}/visuals

        If no report_id was explicitly provided, attempts to find a report
        that uses this dataset in the same workspace.
        """
        if not self._report_id:
            self._report_id = self._find_report_for_dataset()
            if not self._report_id:
                logger.info(
                    "Keine report_id verfügbar – Visual-Extraktion übersprungen."
                )
                return []

        report_url = f"{_BASE_URL}/groups/{self._workspace_id}/reports/{self._report_id}"

        # 1) Seiten abrufen
        try:
            pages_data = self._api_get(f"{report_url}/pages")
        except Exception as exc:
            logger.warning("Berichtsseiten konnten nicht extrahiert werden: %s", exc)
            return []

        if not pages_data:
            return []

        pages = pages_data.get("value", [])
        if not pages:
            return []

        visuals: list[VisualInfo] = []
        api_visual_count = 0

        # 2) Für jede Seite die Visuals abrufen.
        # Hinweis: Die öffentliche Power BI REST API besitzt KEINEN Endpoint,
        # um die einzelnen Visuals einer Seite aufzulisten – der Aufruf
        # .../pages/{id}/visuals liefert 404. Wir versuchen ihn best-effort und
        # dokumentieren andernfalls zumindest die Seite selbst, damit die
        # Berichtsstruktur (Seiten) im Dokument erscheint.
        for page in pages:
            page_name = page.get("displayName", page.get("name", ""))
            page_id = page.get("name", "")  # API nutzt den internen Namen als ID

            if not page_id:
                continue

            page_visuals: list[VisualInfo] = []
            try:
                visuals_data = self._api_get(f"{report_url}/pages/{page_id}/visuals")
                for vis in (visuals_data or {}).get("value", []):
                    page_visuals.append(
                        VisualInfo(
                            page_name=page_name,
                            visual_type=vis.get("type", "unknown"),
                            title=vis.get("name", None),       # Technische ID
                            display_name=vis.get("title", ""),  # Anzeigename
                            fields=[],  # REST API liefert keine Feld-Bindungen
                        )
                    )
            except Exception as exc:
                logger.info(
                    "Visual-Detail-Endpoint für Seite '%s' nicht verfügbar (%s) – "
                    "dokumentiere nur die Seite.",
                    page_name, exc,
                )

            if page_visuals:
                visuals.extend(page_visuals)
                api_visual_count += len(page_visuals)
            else:
                # Seite als Eintrag festhalten (Visual-Details via REST-API
                # nicht verfügbar).
                visuals.append(
                    VisualInfo(
                        page_name=page_name,
                        visual_type="page",
                        title=page_id,
                        display_name=page_name,
                        fields=[],
                    )
                )

        logger.info(
            "Visual-Extraktion abgeschlossen: %d Visual-Details via API, "
            "%d Seiten dokumentiert.",
            api_visual_count, len(pages),
        )
        return visuals

    def _find_report_for_dataset(self) -> str:
        """Find the first report in the workspace that uses this dataset."""
        url = f"{_BASE_URL}/groups/{self._workspace_id}/reports"
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Berichte konnten nicht abgerufen werden: %s", exc)
            return ""
        if not data:
            return ""
        for item in data.get("value", []):
            if item.get("datasetId") == self._dataset_id:
                report_id = item.get("id", "")
                if report_id:
                    logger.info(
                        "Report '%s' automatisch für Visual-Extraktion gefunden.",
                        item.get("name", report_id),
                    )
                    return report_id
        return ""

    def _extract_dependent_reports(self) -> list[DependentReportInfo]:
        """Extract reports in the workspace that use this semantic model (dataset).

        Calls GET /groups/{workspace_id}/reports and filters by datasetId.
        """
        url = f"{_BASE_URL}/groups/{self._workspace_id}/reports"
        try:
            data = self._api_get(url)
        except Exception as exc:
            logger.warning("Abhängige Berichte konnten nicht extrahiert werden: %s", exc)
            return []
        if not data:
            return []

        reports: list[DependentReportInfo] = []
        for item in data.get("value", []):
            if item.get("datasetId") == self._dataset_id:
                reports.append(
                    DependentReportInfo(
                        name=item.get("name", ""),
                        report_id=item.get("id", ""),
                        web_url=item.get("webUrl", ""),
                    )
                )
        return reports
