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

import base64
import binascii
import json
import logging
import re
import time
from typing import Any

import msal
import requests

from core.connectors.base import IConnector
from core.models import (
    AuthenticationError,
    ColumnInfo,
    ConnectionConfig,
    CopyJobActivityInfo,
    CopyJobColumnMappingInfo,
    CopyJobInfo,
    DashboardInfo,
    DashboardTileInfo,
    DataflowAttributeInfo,
    DataflowEntityInfo,
    DataflowInfo,
    DataPipelineActivityInfo,
    DataPipelineInfo,
    DataPipelineParameterInfo,
    DataSourceInfo,
    DependentReportInfo,
    EnvironmentInfo,
    EnvironmentLibraryInfo,
    EventhouseDatabaseInfo,
    EventhouseInfo,
    EventstreamInfo,
    EventstreamNodeInfo,
    GraphQlApiInfo,
    GraphQlDataSourceInfo,
    KqlDatabaseInfo,
    KqlQuerysetInfo,
    KqlQueryInfo,
    LakehouseInfo,
    LakehouseTableInfo,
    MeasureInfo,
    MetadataExtractionError,
    MirroredDatabaseInfo,
    MirroredTableInfo,
    NotebookCellInfo,
    NotebookInfo,
    PBIConnectionError,
    PowerQueryInfo,
    RefreshScheduleInfo,
    ReflexInfo,
    ReflexObjectInfo,
    ReflexRuleInfo,
    RelationshipInfo,
    ReportMetadata,
    RLSRoleInfo,
    SparkJobDefinitionInfo,
    SqlDatabaseInfo,
    TableInfo,
    VisualInfo,
    WarehouseInfo,
    WorkspaceItemInfo,
    WorkspaceMetadata,
    WorkspaceUserInfo,
)

logger = logging.getLogger(__name__)

_BASE_URL = "https://api.powerbi.com/v1.0/myorg"
_FABRIC_BASE_URL = "https://api.fabric.microsoft.com/v1"
_SCOPE = ["https://analysis.windows.net/powerbi/api/.default"]
# Fabric-Audience-Scopes für die Item-Definition-APIs (getDefinition benötigt
# Item.ReadWrite.All bzw. typenspezifische ReadWrite-Scopes). Der Power-BI-Scope
# oben deckt Lese-/Listen-APIs ab, aber NICHT getDefinition (→ 403).
_FABRIC_SCOPE_APP = ["https://api.fabric.microsoft.com/.default"]
_FABRIC_SCOPE_DELEGATED = ["https://api.fabric.microsoft.com/Item.ReadWrite.All"]
# Well-known Microsoft Power BI first-party Client. Dieser darf KEINE Tokens für
# die Fabric-API anfordern (AADSTS65002 – Preauthorization erforderlich); daher
# wird für ihn kein Fabric-Token beschafft. Nur ein EIGENER Client (Service
# Principal oder eigene Client-ID) kann den Fabric-Scope erhalten.
_WELL_KNOWN_PBI_CLIENT_ID = "23d8f6bd-1eb0-4cc2-a08c-7bf525c67bcd"
_AUTHORITY_TEMPLATE = "https://login.microsoftonline.com/{tenant_id}"
_MAX_RETRIES = 3
_BACKOFF_BASE = 1.0  # seconds
# Long-Running-Operation (getDefinition kann 202 + Location liefern)
_LRO_MAX_WAIT = 120.0  # Sekunden, Gesamtbudget fürs Pollen
_LRO_POLL_DEFAULT = 2.0  # Sekunden, Fallback-Poll-Intervall

# 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 ""


def _is_fabric_sql_endpoint(server: str) -> bool:
    """True, wenn der Server ein Fabric-/OneLake-SQL-Analytics-Endpoint ist.

    Wird genutzt, um DirectLake-Quellen (die auf den SQL-Endpoint eines
    Lakehouse/Warehouse zeigen) von klassischen SQL-Import-/DirectQuery-Quellen
    (z. B. ``*.database.windows.net``) zu unterscheiden.
    """
    s = (server or "").lower()
    return any(marker in s for marker in (
        "datawarehouse.fabric.microsoft.com",
        "datawarehouse.pbidedicated.windows.net",
        "pbidedicated.windows.net",
        "fabric.microsoft.com",
        "onelake",
    ))


# ---------------------------------------------------------------------------
# Dataflow-Parser (Gen1 model.json & Gen2 definition parts)
# ---------------------------------------------------------------------------

_SHARED_RE = re.compile(
    r'shared\s+(#"(?P<q>[^"]+)"|(?P<n>[A-Za-z_][\w.]*))\s*=',
    re.MULTILINE,
)


def _split_m_shared(document: str) -> dict[str, str]:
    """Zerlegt ein M-/Mashup-Dokument in ``shared``-Ausdrücke je Query-Name.

    Best-effort: Findet ``shared <Name> = …;``-Anweisungen und ordnet den
    Ausdruck (ohne abschließendes Semikolon) dem Query-Namen zu.
    """
    if not document:
        return {}
    matches = list(_SHARED_RE.finditer(document))
    result: dict[str, str] = {}
    for idx, match in enumerate(matches):
        name = match.group("q") or match.group("n") or ""
        start = match.end()
        end = matches[idx + 1].start() if idx + 1 < len(matches) else len(document)
        expr = document[start:end].strip()
        expr = expr.rstrip().rstrip(";").rstrip()
        if name:
            result[name] = expr
    return result


def _parse_gen1_model(model: dict[str, Any], base: DataflowInfo) -> DataflowInfo:
    """Baut ein :class:`DataflowInfo` aus einer Gen1 ``model.json``-Definition."""
    base.name = base.name or model.get("name", "") or ""
    base.description = base.description or model.get("description", "") or ""
    base.modified_time = model.get("modifiedTime", "") or base.modified_time
    base.raw_metadata = model

    mashup = model.get("pbi:mashup") or {}
    document = mashup.get("document", "") or ""
    base.document_m = document
    shared = _split_m_shared(document)

    entities: list[DataflowEntityInfo] = []
    for ent in model.get("entities", []) or []:
        name = ent.get("name", "") or ""
        attributes = [
            DataflowAttributeInfo(
                name=attr.get("name", "") or "",
                data_type=str(attr.get("dataType", "") or ""),
                description=attr.get("description", "") or "",
            )
            for attr in (ent.get("attributes") or [])
        ]
        is_computed = bool(
            ent.get("isComputed")
            or str(ent.get("$type", "")).lower().startswith("computed")
            or ent.get("pbi:refreshPolicy")
        )
        entities.append(
            DataflowEntityInfo(
                name=name,
                description=ent.get("description", "") or "",
                attributes=attributes,
                m_expression=shared.get(name, ""),
                is_computed=is_computed,
            )
        )
    base.entities = entities
    return base


def _decode_part_payload(part: dict[str, Any]) -> str:
    """Dekodiert den Payload eines Definition-Parts (InlineBase64 oder Klartext)."""
    payload = part.get("payload", "") or ""
    payload_type = (part.get("payloadType", "") or "").lower()
    if payload_type == "inlinebase64":
        try:
            return base64.b64decode(payload).decode("utf-8", errors="replace")
        except (binascii.Error, ValueError):
            return ""
    return payload


def _parse_gen2_definition(
    definition: dict[str, Any], base: DataflowInfo
) -> DataflowInfo:
    """Baut ein :class:`DataflowInfo` aus einer Gen2-Definition (getDefinition)."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition

    document = ""
    query_meta: dict[str, Any] = {}
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("mashup.pq"):
            document = _decode_part_payload(part)
        elif path.endswith("querymetadata.json"):
            raw = _decode_part_payload(part)
            try:
                query_meta = json.loads(raw) if raw else {}
            except (json.JSONDecodeError, ValueError):
                query_meta = {}

    base.document_m = document
    shared = _split_m_shared(document)

    # Query-Namen bevorzugt aus queryMetadata, sonst aus dem Mashup-Dokument.
    names: list[str] = []
    meta_queries = (
        query_meta.get("queriesMetadata")
        or query_meta.get("queries")
        or {}
    )
    if isinstance(meta_queries, dict):
        names = list(meta_queries.keys())
    elif isinstance(meta_queries, list):
        names = [
            q.get("queryName", q.get("name", ""))
            for q in meta_queries
            if isinstance(q, dict)
        ]
    if not names:
        names = list(shared.keys())

    entities: list[DataflowEntityInfo] = []
    for name in names:
        if not name:
            continue
        meta_entry = (
            meta_queries.get(name, {})
            if isinstance(meta_queries, dict)
            else {}
        )
        is_computed = bool(
            isinstance(meta_entry, dict)
            and (meta_entry.get("isComputed") or meta_entry.get("computed"))
        )
        entities.append(
            DataflowEntityInfo(
                name=name,
                description=(
                    meta_entry.get("description", "")
                    if isinstance(meta_entry, dict)
                    else ""
                ),
                attributes=[],
                m_expression=shared.get(name, ""),
                is_computed=is_computed,
            )
        )
    base.entities = entities
    return base


# ---------------------------------------------------------------------------
# Notebook-Parser (Fabric getDefinition: notebook-content.ipynb / .py)
# ---------------------------------------------------------------------------

# Magic-Befehle am Zellanfang, die die Sprache einer Code-Zelle festlegen.
_CELL_MAGIC_RE = re.compile(r"^\s*%%(\w+)")

# Sprache aus dem `# CELL`-Marker des Python-Quellformats (z. B. "# CELL ... language=python").
_PY_LANGUAGE_RE = re.compile(r'language\s*[:=]\s*"?(?P<lang>[\w]+)"?', re.IGNORECASE)


def _normalize_language(language: str) -> str:
    """Normalisiert Kernel-/Sprachbezeichnungen auf ein kurzes Kürzel."""
    lang = (language or "").strip().lower()
    mapping = {
        "synapse_pyspark": "pyspark",
        "pyspark": "pyspark",
        "python": "python",
        "ipython": "python",
        "spark": "scala",
        "synapse_spark": "scala",
        "scala": "scala",
        "sql": "sql",
        "sparksql": "sql",
        "sparkr": "r",
        "r": "r",
        "csharp": "csharp",
        "html": "html",
    }
    return mapping.get(lang, lang)


def _cell_language(source: str, default_language: str) -> str:
    """Ermittelt die Sprache einer Code-Zelle (Magic-Befehl schlägt Default)."""
    for line in (source or "").splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        match = _CELL_MAGIC_RE.match(stripped)
        if match:
            return _normalize_language(match.group(1))
        break
    return _normalize_language(default_language)


def _extract_default_lakehouse(metadata: dict[str, Any]) -> str:
    """Liest das angebundene Standard-Lakehouse aus den Notebook-Metadaten."""
    deps = metadata.get("dependencies") or {}
    lakehouse = deps.get("lakehouse") or {}
    if not isinstance(lakehouse, dict):
        return ""
    name = lakehouse.get("default_lakehouse_name")
    if name:
        return str(name)
    known = lakehouse.get("known_lakehouses") or []
    if isinstance(known, list) and known and isinstance(known[0], dict):
        return str(known[0].get("id", "") or "")
    return ""


def _extract_default_language(metadata: dict[str, Any]) -> str:
    """Liest die Standardsprache/den Kernel aus den Notebook-Metadaten."""
    lang_info = metadata.get("language_info") or {}
    if isinstance(lang_info, dict) and lang_info.get("name"):
        return str(lang_info.get("name"))
    kernelspec = metadata.get("kernelspec") or {}
    if isinstance(kernelspec, dict):
        return str(
            kernelspec.get("language") or kernelspec.get("display_name") or ""
        )
    return ""


def _parse_notebook_ipynb(nb_json: dict[str, Any], base: NotebookInfo) -> NotebookInfo:
    """Baut ein :class:`NotebookInfo` aus einer ipynb-JSON-Definition."""
    base.raw_metadata = nb_json
    metadata = nb_json.get("metadata") or {}
    default_language = _normalize_language(_extract_default_language(metadata))
    base.default_language = default_language or base.default_language
    base.default_lakehouse = (
        base.default_lakehouse or _extract_default_lakehouse(metadata)
    )

    cells: list[NotebookCellInfo] = []
    source_parts: list[str] = []
    for idx, cell in enumerate(nb_json.get("cells", []) or [], start=1):
        cell_type = "markdown" if cell.get("cell_type") == "markdown" else "code"
        raw_source = cell.get("source", "")
        if isinstance(raw_source, list):
            source = "".join(raw_source)
        else:
            source = str(raw_source or "")
        cell_lang = (
            "" if cell_type == "markdown"
            else _cell_language(source, default_language)
        )
        cells.append(
            NotebookCellInfo(
                index=idx,
                cell_type=cell_type,
                source=source,
                language=cell_lang,
            )
        )
        source_parts.append(source)
    base.cells = cells
    base.document_source = "\n\n".join(source_parts)
    return base


def _parse_notebook_pysource(source: str, base: NotebookInfo) -> NotebookInfo:
    """Baut ein :class:`NotebookInfo` aus dem Python-Quellformat (# CELL/# MARKDOWN)."""
    base.document_source = source or ""
    lines = (source or "").splitlines()

    cells: list[NotebookCellInfo] = []
    current_type: str | None = None
    current_lang = ""
    buffer: list[str] = []
    index = 0

    def _flush() -> None:
        nonlocal index
        if current_type is None:
            return
        if current_type == "markdown":
            content = "\n".join(
                ln[2:] if ln.startswith("# ") else ln.lstrip("#").lstrip()
                for ln in buffer
            ).strip()
        else:
            content = "\n".join(buffer).strip()
        if content:
            index += 1
            if current_type == "markdown":
                lang = ""
            else:
                lang = current_lang or _cell_language(content, base.default_language)
            cells.append(
                NotebookCellInfo(
                    index=index,
                    cell_type=current_type,
                    source=content,
                    language=lang,
                )
            )

    for line in lines:
        stripped = line.strip()
        if stripped.startswith("# CELL"):
            _flush()
            current_type = "code"
            match = _PY_LANGUAGE_RE.search(stripped)
            current_lang = _normalize_language(match.group("lang")) if match else ""
            buffer = []
            continue
        if stripped.startswith("# MARKDOWN"):
            _flush()
            current_type = "markdown"
            current_lang = ""
            buffer = []
            continue
        if stripped.startswith("# META") or stripped.startswith("# METADATA"):
            # Metadaten-Zeilen des Quellformats überspringen
            continue
        if current_type is not None:
            buffer.append(line)
    _flush()

    base.cells = cells
    return base


def _parse_notebook_definition(
    definition: dict[str, Any], base: NotebookInfo
) -> NotebookInfo:
    """Baut ein :class:`NotebookInfo` aus einer Fabric getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition

    ipynb_payload = ""
    py_payload = ""
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith(".ipynb"):
            ipynb_payload = _decode_part_payload(part)
        elif path.endswith("notebook-content.py") or path.endswith(".py"):
            py_payload = _decode_part_payload(part)

    if ipynb_payload:
        try:
            nb_json = json.loads(ipynb_payload)
            return _parse_notebook_ipynb(nb_json, base)
        except (json.JSONDecodeError, ValueError):
            pass
    if py_payload:
        return _parse_notebook_pysource(py_payload, base)
    return base


# ---------------------------------------------------------------------------
# Copyjob-Parser (Fabric getDefinition: copyjob-content.json)
# ---------------------------------------------------------------------------


def _summarize_connection(conn: dict[str, Any]) -> tuple[str, str]:
    """Ermittelt (Typ, lesbare Details) für eine Copyjob-Quelle/-Ziel."""
    if not isinstance(conn, dict):
        return "", ""
    settings = conn.get("connectionSettings") or {}
    conn_type = (
        settings.get("type") if isinstance(settings, dict) else ""
    ) or conn.get("type", "") or ""

    props = settings.get("typeProperties") or {} if isinstance(settings, dict) else {}
    ext = settings.get("externalReferences") or {} if isinstance(settings, dict) else {}

    details: list[str] = []
    if isinstance(props, dict):
        if props.get("database"):
            details.append(f"DB={props['database']}")
        if props.get("schema"):
            details.append(f"Schema={props['schema']}")
        if props.get("artifactId"):
            details.append(f"ArtifactId={props['artifactId']}")
        if props.get("workspaceId"):
            details.append(f"WorkspaceId={props['workspaceId']}")
        if props.get("rootFolder"):
            details.append(f"RootFolder={props['rootFolder']}")
        if props.get("endpoint"):
            details.append(f"Endpoint={props['endpoint']}")
    if isinstance(ext, dict) and ext.get("connection"):
        details.append(f"Connection={ext['connection']}")

    return str(conn_type), ", ".join(details)


def _object_label(dataset_settings: dict[str, Any]) -> str:
    """Baut eine lesbare Objektbezeichnung aus datasetSettings (Tabelle/Datei)."""
    if not isinstance(dataset_settings, dict):
        return ""
    # Tabelle (SQL/Lakehouse): schema.table oder table
    table = dataset_settings.get("table") or dataset_settings.get("Table")
    if table:
        schema = dataset_settings.get("schema") or dataset_settings.get("Schema")
        return f"{schema}.{table}" if schema else str(table)
    # Datei-basiert: Location
    loc = dataset_settings.get("location") or dataset_settings.get("Location") or {}
    if isinstance(loc, dict):
        parts = [
            loc.get("container"),
            loc.get("fileSystem"),
            loc.get("bucketName"),
            loc.get("folderPath"),
            loc.get("fileName"),
        ]
        path = "/".join(str(p).strip("/") for p in parts if p)
        if path:
            return path
    return ""


def _parse_copyjob_activity(
    activity: dict[str, Any], index: int
) -> CopyJobActivityInfo:
    """Baut ein :class:`CopyJobActivityInfo` aus einer Copyjob-Aktivität."""
    props = activity.get("properties") or {}
    source = props.get("source") or {}
    destination = props.get("destination") or {}

    source_object = _object_label(source.get("datasetSettings") or {})
    destination_object = _object_label(destination.get("datasetSettings") or {})
    write_behavior = (
        destination.get("writeBehavior")
        or destination.get("writeBehaviour")
        or ""
    )
    table_option = destination.get("tableOption") or ""
    change_settings = source.get("changeDataSettings") or {}
    read_method = (
        change_settings.get("readMethod")
        or change_settings.get("ReadMethod")
        or ""
    )

    mappings: list[CopyJobColumnMappingInfo] = []
    translator = props.get("translator") or {}
    for mapping in translator.get("mappings") or []:
        if not isinstance(mapping, dict):
            continue
        src = mapping.get("source") or {}
        dst = mapping.get("destination") or {}
        mappings.append(
            CopyJobColumnMappingInfo(
                source_name=str(src.get("name", "") or ""),
                source_type=str(
                    src.get("type", "") or src.get("physicalType", "") or ""
                ),
                destination_name=str(dst.get("name", "") or ""),
                destination_type=str(
                    dst.get("type", "") or dst.get("physicalType", "") or ""
                ),
            )
        )

    return CopyJobActivityInfo(
        index=index,
        source_object=source_object,
        destination_object=destination_object,
        write_behavior=str(write_behavior or ""),
        table_option=str(table_option or ""),
        read_method=str(read_method or ""),
        mappings=mappings,
    )


def _parse_copyjob_content(content: dict[str, Any], base: CopyJobInfo) -> CopyJobInfo:
    """Baut ein :class:`CopyJobInfo` aus dem ``copyjob-content.json``-Inhalt."""
    base.raw_metadata = content
    properties = content.get("properties") or {}
    base.job_mode = str(properties.get("jobMode", "") or "")

    src_type, src_details = _summarize_connection(properties.get("source") or {})
    dst_type, dst_details = _summarize_connection(
        properties.get("destination") or {}
    )
    base.source_type = src_type
    base.source_details = src_details
    base.destination_type = dst_type
    base.destination_details = dst_details

    policy = properties.get("policy") or {}
    if isinstance(policy, dict):
        base.timeout = str(policy.get("timeout", "") or "")
        if policy.get("retry") is not None:
            base.retry = str(policy.get("retry"))

    activities: list[CopyJobActivityInfo] = []
    for idx, activity in enumerate(content.get("activities") or [], start=1):
        if isinstance(activity, dict):
            activities.append(_parse_copyjob_activity(activity, idx))
    base.activities = activities
    return base


def _parse_copyjob_definition(
    definition: dict[str, Any], base: CopyJobInfo
) -> CopyJobInfo:
    """Baut ein :class:`CopyJobInfo` aus einer Fabric getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition

    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("copyjob-content.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict) and (
                "properties" in content or "activities" in content
            ):
                return _parse_copyjob_content(content, base)
    return base


# ---------------------------------------------------------------------------
# Lakehouse-Parser (Fabric GET properties + /tables)
# ---------------------------------------------------------------------------


def _parse_lakehouse_properties(item: dict[str, Any], base: LakehouseInfo) -> LakehouseInfo:
    """Übernimmt die Item-Properties (SQL-Endpoint, OneLake-Pfade) in das Modell."""
    base.raw_metadata = item
    base.name = base.name or item.get("displayName", item.get("name", "")) or ""
    base.description = base.description or item.get("description", "") or ""

    props = item.get("properties") or {}
    if isinstance(props, dict):
        base.default_schema = props.get("defaultSchema", "") or ""
        base.onelake_tables_path = props.get("oneLakeTablesPath", "") or ""
        base.onelake_files_path = props.get("oneLakeFilesPath", "") or ""
        sql = props.get("sqlEndpointProperties") or {}
        if isinstance(sql, dict):
            base.sql_connection_string = sql.get("connectionString", "") or ""
            base.sql_endpoint_id = sql.get("id", "") or ""
            base.sql_provisioning_status = sql.get("provisioningStatus", "") or ""
    return base


def _parse_lakehouse_tables(entries: list[dict[str, Any]]) -> list[LakehouseTableInfo]:
    """Baut die Tabellenliste aus den ``data``-Einträgen der Tables-API."""
    tables: list[LakehouseTableInfo] = []
    for entry in entries or []:
        if not isinstance(entry, dict):
            continue
        tables.append(
            LakehouseTableInfo(
                name=str(entry.get("name", "") or ""),
                table_type=str(entry.get("type", "") or ""),
                table_format=str(entry.get("format", "") or ""),
                location=str(entry.get("location", "") or ""),
            )
        )
    return tables


# ---------------------------------------------------------------------------
# Warehouse-/Eventhouse-Parser (Fabric GET properties)
# ---------------------------------------------------------------------------


def _parse_warehouse_properties(item: dict[str, Any], base: WarehouseInfo) -> WarehouseInfo:
    """Übernimmt die Warehouse-Properties (SQL-Verbindung, Collation) ins Modell."""
    base.raw_metadata = item
    base.name = base.name or item.get("displayName", item.get("name", "")) or ""
    base.description = base.description or item.get("description", "") or ""
    props = item.get("properties") or {}
    if isinstance(props, dict):
        base.connection_string = props.get("connectionString", "") or ""
        base.collation_type = props.get("collationType", "") or ""
        base.created_date = props.get("createdDate", "") or ""
        base.last_updated_time = props.get("lastUpdatedTime", "") or ""
    return base


def _parse_eventhouse_properties(
    item: dict[str, Any],
    base: EventhouseInfo,
    db_name_map: dict[str, str] | None = None,
) -> EventhouseInfo:
    """Übernimmt die Eventhouse-Properties (URIs, KQL-DBs) ins Modell.

    ``db_name_map`` ordnet KQL-Datenbank-IDs optional lesbare Namen zu.
    """
    base.raw_metadata = item
    base.name = base.name or item.get("displayName", item.get("name", "")) or ""
    base.description = base.description or item.get("description", "") or ""
    props = item.get("properties") or {}
    databases: list[EventhouseDatabaseInfo] = []
    if isinstance(props, dict):
        base.query_service_uri = props.get("queryServiceUri", "") or ""
        base.ingestion_service_uri = props.get("ingestionServiceUri", "") or ""
        if props.get("minimumConsumptionUnits") is not None:
            base.minimum_consumption_units = str(props.get("minimumConsumptionUnits"))
        name_map = db_name_map or {}
        for db_id in props.get("databasesItemIds", []) or []:
            databases.append(
                EventhouseDatabaseInfo(
                    id=str(db_id), name=name_map.get(str(db_id), "")
                )
            )
    base.databases = databases
    return base


# ---------------------------------------------------------------------------
# KQL-Datenbank-Parser (Fabric GET properties)
# ---------------------------------------------------------------------------


def _parse_kqldatabase_properties(
    item: dict[str, Any], base: KqlDatabaseInfo
) -> KqlDatabaseInfo:
    """Übernimmt die KQL-Datenbank-Properties (URIs, Typ, Eventhouse) ins Modell."""
    base.raw_metadata = item
    base.name = base.name or item.get("displayName", item.get("name", "")) or ""
    base.description = base.description or item.get("description", "") or ""
    props = item.get("properties") or {}
    if isinstance(props, dict):
        base.parent_eventhouse_id = props.get("parentEventhouseItemId", "") or ""
        base.query_service_uri = props.get("queryServiceUri", "") or ""
        base.ingestion_service_uri = props.get("ingestionServiceUri", "") or ""
        base.database_type = props.get("databaseType", "") or ""
        if props.get("oneLakeCachingPeriod") is not None:
            base.onelake_caching_period = str(props.get("oneLakeCachingPeriod"))
        if props.get("oneLakeStandardStoragePeriod") is not None:
            base.onelake_standard_storage_period = str(
                props.get("oneLakeStandardStoragePeriod")
            )
    return base


# ---------------------------------------------------------------------------
# SQL-Database-Parser (Fabric GET properties)
# ---------------------------------------------------------------------------


def _parse_sqldatabase_properties(
    item: dict[str, Any], base: SqlDatabaseInfo
) -> SqlDatabaseInfo:
    """Übernimmt die SQL-Database-Properties (Verbindung, FQDN, Collation) ins Modell."""
    base.raw_metadata = item
    base.name = base.name or item.get("displayName", item.get("name", "")) or ""
    base.description = base.description or item.get("description", "") or ""
    props = item.get("properties") or {}
    if isinstance(props, dict):
        base.connection_string = props.get("connectionString", "") or ""
        base.database_name = props.get("databaseName", "") or ""
        base.server_fqdn = props.get("serverFqdn", "") or ""
        base.collation = props.get("collation", "") or ""
        base.earliest_restore_point = props.get("earliestRestorePoint", "") or ""
        base.latest_restore_point = props.get("latestRestorePoint", "") or ""
        if props.get("backupRetentionDays") is not None:
            base.backup_retention_days = str(props.get("backupRetentionDays"))
    return base


# ---------------------------------------------------------------------------
# Environment-Parser (Fabric GET properties + libraries + sparkcompute)
# ---------------------------------------------------------------------------


def _parse_environment_libraries(data: dict[str, Any]) -> list[EnvironmentLibraryInfo]:
    """Baut die Bibliotheksliste aus einer ``libraries``-Antwort."""
    libraries: list[EnvironmentLibraryInfo] = []
    for entry in (data or {}).get("libraries", []) or []:
        if not isinstance(entry, dict):
            continue
        libraries.append(
            EnvironmentLibraryInfo(
                name=str(entry.get("name", "") or ""),
                library_type=str(entry.get("libraryType", "") or ""),
                version=str(entry.get("version", "") or ""),
            )
        )
    return libraries


def _parse_environment_sparkcompute(
    data: dict[str, Any], base: EnvironmentInfo
) -> EnvironmentInfo:
    """Übernimmt die Spark-Compute-/Runtime-Einstellungen in das Modell."""
    if not isinstance(data, dict):
        return base
    base.runtime_version = str(data.get("runtimeVersion", "") or "")
    pool = data.get("instancePool") or {}
    if isinstance(pool, dict):
        base.instance_pool_name = str(pool.get("name", "") or "")
        base.instance_pool_type = str(pool.get("type", "") or "")
    if data.get("driverCores") is not None:
        base.driver_cores = str(data.get("driverCores"))
    base.driver_memory = str(data.get("driverMemory", "") or "")
    if data.get("executorCores") is not None:
        base.executor_cores = str(data.get("executorCores"))
    base.executor_memory = str(data.get("executorMemory", "") or "")
    dyn = data.get("dynamicExecutorAllocation") or {}
    if isinstance(dyn, dict) and dyn.get("enabled") is not None:
        if dyn.get("enabled"):
            base.dynamic_allocation = (
                f"aktiv ({dyn.get('minExecutors', '?')}–{dyn.get('maxExecutors', '?')})"
            )
        else:
            base.dynamic_allocation = "deaktiviert"
    return base


# ---------------------------------------------------------------------------
# Mirrored-Database-Parser (Fabric getDefinition: mirroring.json)
# ---------------------------------------------------------------------------


def _parse_mirroreddb_content(
    content: dict[str, Any], base: MirroredDatabaseInfo
) -> MirroredDatabaseInfo:
    """Baut ein :class:`MirroredDatabaseInfo` aus dem ``mirroring.json``-Inhalt."""
    base.raw_metadata = content
    props = content.get("properties") if isinstance(content.get("properties"), dict) else content
    props = props or {}

    source = props.get("source") or {}
    if isinstance(source, dict):
        base.source_type = str(source.get("type", "") or "")
        src_type_props = source.get("typeProperties") or {}
        if isinstance(src_type_props, dict):
            base.source_database = str(src_type_props.get("database", "") or "")

    target = props.get("target") or {}
    if isinstance(target, dict):
        base.target_type = str(target.get("type", "") or "")
        tgt_type_props = target.get("typeProperties") or {}
        if isinstance(tgt_type_props, dict):
            base.default_schema = str(tgt_type_props.get("defaultSchema", "") or "")
            base.target_format = str(tgt_type_props.get("format", "") or "")
            if tgt_type_props.get("retentionInDays") is not None:
                base.retention_days = str(tgt_type_props.get("retentionInDays"))

    tables: list[MirroredTableInfo] = []
    for entry in props.get("mountedTables", []) or []:
        if not isinstance(entry, dict):
            continue
        src = entry.get("source") or {}
        type_props = src.get("typeProperties") if isinstance(src, dict) else {}
        type_props = type_props or {}
        tables.append(
            MirroredTableInfo(
                schema_name=str(type_props.get("schemaName", "") or ""),
                table_name=str(type_props.get("tableName", "") or ""),
            )
        )
    base.mirrored_tables = tables
    return base


def _parse_mirroreddb_definition(
    definition: dict[str, Any], base: MirroredDatabaseInfo
) -> MirroredDatabaseInfo:
    """Baut ein :class:`MirroredDatabaseInfo` aus einer getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("mirroring.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                return _parse_mirroreddb_content(content, base)
    return base


# ---------------------------------------------------------------------------
# Eventstream-Parser (Fabric getDefinition: eventstream.json)
# ---------------------------------------------------------------------------


def _parse_eventstream_nodes(entries: list[Any]) -> list[EventstreamNodeInfo]:
    """Baut eine Knotenliste (Name/Typ) aus einer sources/destinations/operators-Liste."""
    nodes: list[EventstreamNodeInfo] = []
    for entry in entries or []:
        if not isinstance(entry, dict):
            continue
        nodes.append(
            EventstreamNodeInfo(
                name=str(entry.get("name", "") or ""),
                node_type=str(entry.get("type", "") or entry.get("operatorType", "") or ""),
            )
        )
    return nodes


def _parse_eventstream_content(
    content: dict[str, Any], base: EventstreamInfo
) -> EventstreamInfo:
    """Baut ein :class:`EventstreamInfo` aus dem ``eventstream.json``-Inhalt."""
    base.raw_metadata = content
    props = content.get("properties") if isinstance(content.get("properties"), dict) else content
    props = props or {}
    base.sources = _parse_eventstream_nodes(props.get("sources", []))
    base.destinations = _parse_eventstream_nodes(props.get("destinations", []))
    operators = props.get("operators")
    if operators is None:
        operators = props.get("streams")
    base.operators = _parse_eventstream_nodes(operators or [])
    return base


def _parse_eventstream_definition(
    definition: dict[str, Any], base: EventstreamInfo
) -> EventstreamInfo:
    """Baut ein :class:`EventstreamInfo` aus einer getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("eventstream.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                return _parse_eventstream_content(content, base)
    return base


# ---------------------------------------------------------------------------
# Spark-Job-Definition-Parser (Fabric getDefinition: SparkJobDefinitionV1.json)
# ---------------------------------------------------------------------------


def _parse_sparkjob_content(
    content: dict[str, Any], base: SparkJobDefinitionInfo
) -> SparkJobDefinitionInfo:
    """Baut ein :class:`SparkJobDefinitionInfo` aus dem ``SparkJobDefinitionV1.json``."""
    base.raw_metadata = content
    props = content
    for key in ("properties", "jobProperties"):
        if isinstance(content.get(key), dict):
            props = content[key]
            break
    base.language = str(props.get("language", "") or "")
    base.executable_file = str(props.get("executableFile", "") or "")
    base.main_class = str(props.get("mainClass", "") or "")
    base.default_lakehouse = str(props.get("defaultLakehouseArtifactId", "") or "")
    args = props.get("commandLineArguments", "")
    if isinstance(args, list):
        base.command_line_arguments = " ".join(str(a) for a in args)
    else:
        base.command_line_arguments = str(args or "")
    libs = props.get("additionalLibraryUris") or []
    base.additional_libraries = [str(u) for u in libs if u] if isinstance(libs, list) else []
    return base


def _parse_sparkjob_definition(
    definition: dict[str, Any], base: SparkJobDefinitionInfo
) -> SparkJobDefinitionInfo:
    """Baut ein :class:`SparkJobDefinitionInfo` aus einer getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("sparkjobdefinitionv1.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                return _parse_sparkjob_content(content, base)
    return base


# ---------------------------------------------------------------------------
# KQL-Queryset-Parser (Fabric getDefinition: RealTimeQueryset.json)
# ---------------------------------------------------------------------------


def _parse_kqlqueryset_content(
    content: dict[str, Any], base: KqlQuerysetInfo
) -> KqlQuerysetInfo:
    """Baut ein :class:`KqlQuerysetInfo` aus dem ``RealTimeQueryset.json``-Inhalt."""
    base.raw_metadata = content
    props = content.get("properties") if isinstance(content.get("properties"), dict) else content
    props = props or {}
    entries = props.get("queries")
    if not entries:
        entries = props.get("tabs")
    queries: list[KqlQueryInfo] = []
    for entry in entries or []:
        if not isinstance(entry, dict):
            continue
        text = (
            entry.get("queryText")
            or entry.get("text")
            or entry.get("content")
            or entry.get("query")
            or ""
        )
        queries.append(
            KqlQueryInfo(
                name=str(entry.get("name", "") or entry.get("title", "") or ""),
                query_text=str(text or ""),
            )
        )
    base.queries = queries
    return base


def _parse_kqlqueryset_definition(
    definition: dict[str, Any], base: KqlQuerysetInfo
) -> KqlQuerysetInfo:
    """Baut ein :class:`KqlQuerysetInfo` aus einer getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("realtimequeryset.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                return _parse_kqlqueryset_content(content, base)
    return base


# ---------------------------------------------------------------------------
# Reflex-Parser (Fabric getDefinition: ReflexEntities.json)
# ---------------------------------------------------------------------------


def _parse_reflex_content(
    content: dict[str, Any], base: ReflexInfo
) -> ReflexInfo:
    """Baut ein :class:`ReflexInfo` aus dem ``ReflexEntities.json``-Inhalt."""
    base.raw_metadata = content
    props = content.get("properties") if isinstance(content.get("properties"), dict) else content
    props = props or {}

    obj_entries = props.get("objects")
    if not obj_entries:
        obj_entries = props.get("entities")
    objects: list[ReflexObjectInfo] = []
    for entry in obj_entries or []:
        if not isinstance(entry, dict):
            continue
        objects.append(
            ReflexObjectInfo(
                name=str(entry.get("name", "") or entry.get("displayName", "") or ""),
                object_type=str(entry.get("type", "") or entry.get("kind", "") or ""),
            )
        )
    base.objects = objects

    rule_entries = props.get("rules")
    if not rule_entries:
        rule_entries = props.get("triggers")
    rules: list[ReflexRuleInfo] = []
    for entry in rule_entries or []:
        if not isinstance(entry, dict):
            continue
        rules.append(
            ReflexRuleInfo(
                name=str(entry.get("name", "") or entry.get("displayName", "") or ""),
                rule_type=str(entry.get("type", "") or entry.get("kind", "") or ""),
            )
        )
    base.rules = rules
    return base


def _parse_reflex_definition(
    definition: dict[str, Any], base: ReflexInfo
) -> ReflexInfo:
    """Baut ein :class:`ReflexInfo` aus einer getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("reflexentities.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                return _parse_reflex_content(content, base)
    return base


# ---------------------------------------------------------------------------
# GraphQL-API-Parser (Fabric getDefinition)
# ---------------------------------------------------------------------------


def _parse_graphqlapi_content(
    content: dict[str, Any], base: GraphQlApiInfo
) -> GraphQlApiInfo:
    """Baut ein :class:`GraphQlApiInfo` aus dem Definition-Inhalt (best-effort)."""
    base.raw_metadata = content
    props = content.get("properties") if isinstance(content.get("properties"), dict) else content
    props = props or {}
    entries = props.get("dataSources")
    if not entries:
        entries = props.get("datasources")
    data_sources: list[GraphQlDataSourceInfo] = []
    for entry in entries or []:
        if not isinstance(entry, dict):
            continue
        data_sources.append(
            GraphQlDataSourceInfo(
                name=str(
                    entry.get("name", "")
                    or entry.get("displayName", "")
                    or entry.get("id", "")
                    or ""
                ),
                source_type=str(entry.get("type", "") or entry.get("kind", "") or ""),
            )
        )
    base.data_sources = data_sources
    schema = props.get("schema") or props.get("graphqlSchema") or ""
    base.schema = str(schema or "")
    return base


def _parse_graphqlapi_definition(
    definition: dict[str, Any], base: GraphQlApiInfo
) -> GraphQlApiInfo:
    """Baut ein :class:`GraphQlApiInfo` aus einer getDefinition-Antwort.

    Das GraphQL-Schema wird bevorzugt aus einem ``*.graphql``-Part gelesen; die
    Datenquellen aus einem JSON-Part.
    """
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        raw = _decode_part_payload(part)
        if not raw:
            continue
        if path.endswith(".graphql") or path.endswith(".gql"):
            base.schema = raw
            continue
        if path.endswith(".json"):
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict):
                schema = base.schema
                _parse_graphqlapi_content(content, base)
                if not base.schema and schema:
                    base.schema = schema
    return base


# ---------------------------------------------------------------------------
# Data Pipeline-Parser (Fabric getDefinition: pipeline-content.json)
# ---------------------------------------------------------------------------

# Container-Aktivitäten mit verschachtelten Unteraktivitäten und deren Feldnamen.
_PIPELINE_NESTED_KEYS = (
    "activities",
    "ifTrueActivities",
    "ifFalseActivities",
    "defaultActivities",
)


def _parse_pipeline_depends_on(activity: dict[str, Any]) -> list[str]:
    """Liest die Namen der Vorgänger-Aktivitäten aus ``dependsOn``."""
    result: list[str] = []
    for dep in activity.get("dependsOn", []) or []:
        if isinstance(dep, dict) and dep.get("activity"):
            result.append(str(dep["activity"]))
        elif isinstance(dep, str):
            result.append(dep)
    return result


def _collect_pipeline_activities(
    activities: list[Any],
    container: str,
    out: list[DataPipelineActivityInfo],
) -> None:
    """Sammelt Aktivitäten rekursiv (inkl. verschachtelter Container-Aktivitäten)."""
    for activity in activities or []:
        if not isinstance(activity, dict):
            continue
        name = str(activity.get("name", "") or "")
        act_type = str(activity.get("type", "") or "")
        policy = activity.get("policy") or {}
        timeout = str(policy.get("timeout", "") or "") if isinstance(policy, dict) else ""
        retry = (
            str(policy.get("retry"))
            if isinstance(policy, dict) and policy.get("retry") is not None
            else ""
        )
        out.append(
            DataPipelineActivityInfo(
                name=name,
                activity_type=act_type,
                depends_on=_parse_pipeline_depends_on(activity),
                container=container,
                timeout=timeout,
                retry=retry,
            )
        )
        # Verschachtelte Aktivitäten (ForEach/If/Until/Switch …)
        type_props = activity.get("typeProperties") or {}
        for key in _PIPELINE_NESTED_KEYS:
            nested = type_props.get(key)
            if isinstance(nested, list):
                _collect_pipeline_activities(nested, name, out)
        # Switch-Cases
        for case in type_props.get("cases", []) or []:
            if isinstance(case, dict) and isinstance(case.get("activities"), list):
                _collect_pipeline_activities(case["activities"], name, out)


def _parse_pipeline_parameters(
    raw: dict[str, Any],
) -> list[DataPipelineParameterInfo]:
    """Baut Parameter-/Variablen-Infos aus einem ``{name: {type, defaultValue}}``-Dict."""
    result: list[DataPipelineParameterInfo] = []
    if not isinstance(raw, dict):
        return result
    for pname, pdef in raw.items():
        pdef = pdef if isinstance(pdef, dict) else {}
        default = pdef.get("defaultValue")
        result.append(
            DataPipelineParameterInfo(
                name=str(pname),
                param_type=str(pdef.get("type", "") or ""),
                default_value="" if default is None else str(default),
            )
        )
    return result


def _parse_pipeline_content(content: dict[str, Any], base: DataPipelineInfo) -> DataPipelineInfo:
    """Baut ein :class:`DataPipelineInfo` aus dem ``pipeline-content.json``-Inhalt."""
    base.raw_metadata = content
    properties = content.get("properties") or {}
    base.description = base.description or str(properties.get("description", "") or "")

    activities: list[DataPipelineActivityInfo] = []
    _collect_pipeline_activities(properties.get("activities", []), "", activities)
    base.activities = activities
    base.parameters = _parse_pipeline_parameters(properties.get("parameters") or {})
    base.variables = _parse_pipeline_parameters(properties.get("variables") or {})
    return base


def _parse_pipeline_definition(
    definition: dict[str, Any], base: DataPipelineInfo
) -> DataPipelineInfo:
    """Baut ein :class:`DataPipelineInfo` aus einer Fabric getDefinition-Antwort."""
    parts = definition.get("definition", {}).get("parts", []) or []
    base.raw_metadata = definition
    for part in parts:
        path = (part.get("path", "") or "").lower()
        if path.endswith("pipeline-content.json") or path.endswith(".json"):
            raw = _decode_part_payload(part)
            if not raw:
                continue
            try:
                content = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(content, dict) and "properties" in content:
                return _parse_pipeline_content(content, base)
    return base


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

    def __init__(self) -> None:
        self._access_token: str | None = None
        self._fabric_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."
                    )
                pbi_token = self._authenticate_service_principal(
                    client_id, client_secret, authority
                )
                # Eigene App → Fabric-Token (Item.ReadWrite.All via .default) für
                # die Definition-APIs best-effort beschaffen.
                self._fabric_access_token = self._fabric_token_service_principal(
                    client_id, client_secret, authority
                )
                return pbi_token
            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()

        # DirectLake: Die REST-API /datasources liefert für DirectLake-Modelle nur
        # generische Typen (z. B. "OneLakeStorage"/"FabricService"), nicht das
        # konkret genutzte Lakehouse/Warehouse. Über die strukturierte Datenquelle
        # des Modells (INFO.EXPRESSIONS) lässt sich der SQL-Endpoint/OneLake-Pfad
        # ermitteln und gegen die Workspace-Items zum echten Namen auflösen.
        try:
            directlake_sources = self._resolve_directlake_data_sources()
            if directlake_sources:
                data_sources = self._merge_directlake_sources(
                    data_sources, directlake_sources
                )
        except Exception as exc:  # pragma: no cover - defensiver Schutz
            logger.warning("DirectLake-Quellenauflösung fehlgeschlagen: %s", exc)

        # 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 (Gen1 – Power BI REST API)
        seen_dataflow_ids: set[str] = set()
        try:
            data = self._api_get(
                f"{_BASE_URL}/groups/{self._workspace_id}/dataflows"
            )
            if data:
                for item in data.get("value", []):
                    df_id = item.get("objectId", item.get("id", ""))
                    seen_dataflow_ids.add(df_id)
                    # Manche Tenants liefern bereits Gen2-Dataflows über diese
                    # API mit einem generation-Feld (2) – dann entsprechend taggen.
                    gen_raw = item.get("generation")
                    generation = "Gen2" if str(gen_raw) == "2" else "Gen1"
                    items.append(
                        WorkspaceItemInfo(
                            id=df_id,
                            name=item.get("name", ""),
                            item_type="Dataflow",
                            description=item.get("description", "") or "",
                            configured_by=item.get("configuredBy", "") or "",
                            subtype=generation,
                        )
                    )
        except Exception as exc:
            logger.info("Dataflows (Gen1) nicht verfügbar: %s", exc)
        # Dataflows (Gen2 – Fabric REST API); nur ergänzen, was nicht bereits
        # über die Power BI API erfasst wurde (Dedupe über die ID).
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/dataflows"
            )
            if data:
                for item in data.get("value", []):
                    df_id = item.get("id", "")
                    if not df_id or df_id in seen_dataflow_ids:
                        continue
                    seen_dataflow_ids.add(df_id)
                    items.append(
                        WorkspaceItemInfo(
                            id=df_id,
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Dataflow",
                            description=item.get("description", "") or "",
                            subtype="Gen2",
                        )
                    )
        except Exception as exc:
            logger.info("Dataflows (Gen2) nicht verfügbar: %s", exc)
        # Notebooks (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/notebooks"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Notebook",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Notebooks nicht verfügbar: %s", exc)
        # Copyjobs (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/copyJobs"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="CopyJob",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Copyjobs nicht verfügbar: %s", exc)
        # Lakehouses (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/lakehouses"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Lakehouse",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Lakehouses nicht verfügbar: %s", exc)
        # Warehouses (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/warehouses"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Warehouse",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Warehouses nicht verfügbar: %s", exc)
        # Eventhouses (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/eventhouses"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Eventhouse",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Eventhouses nicht verfügbar: %s", exc)
        # Data Pipelines (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/dataPipelines"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="DataPipeline",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Data Pipelines nicht verfügbar: %s", exc)
        # KQL-Datenbanken (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/kqlDatabases"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="KQLDatabase",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("KQL-Datenbanken nicht verfügbar: %s", exc)
        # SQL Databases (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/sqlDatabases"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="SQLDatabase",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("SQL Databases nicht verfügbar: %s", exc)
        # Environments (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/environments"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Environment",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Environments nicht verfügbar: %s", exc)
        # Mirrored Databases (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/mirroredDatabases"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="MirroredDatabase",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Mirrored Databases nicht verfügbar: %s", exc)
        # Eventstreams (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/eventstreams"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Eventstream",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Eventstreams nicht verfügbar: %s", exc)
        # Spark Job Definitions (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/sparkJobDefinitions"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="SparkJobDefinition",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Spark Job Definitions nicht verfügbar: %s", exc)
        # KQL Querysets (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/kqlQuerysets"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="KQLQueryset",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("KQL Querysets nicht verfügbar: %s", exc)
        # Reflexe (Data Activator, Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/reflexes"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="Reflex",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("Reflexe nicht verfügbar: %s", exc)
        # GraphQL APIs (Fabric REST API)
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/GraphQLApis"
            )
            if data:
                for item in data.get("value", []):
                    items.append(
                        WorkspaceItemInfo(
                            id=item.get("id", ""),
                            name=item.get("displayName", item.get("name", "")),
                            item_type="GraphQLApi",
                            description=item.get("description", "") or "",
                        )
                    )
        except Exception as exc:
            logger.info("GraphQL APIs 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

    # ------------------------------------------------------------------
    # Dataflow extraction (Gen1 & Gen2)
    # ------------------------------------------------------------------

    def extract_dataflow(
        self, dataflow_id: str, generation: str = "Gen1", *, name: str = "",
        description: str = "", web_url: str = "",
    ) -> DataflowInfo:
        """Extrahiert die Detail-Metadaten eines Dataflows (Gen1 oder Gen2).

        Gen1 werden über die Power BI REST API (``/dataflows/{id}`` → CDM
        ``model.json``) gelesen, Gen2 über die Fabric REST API
        (``getDefinition`` → ``mashup.pq`` + ``queryMetadata.json``). Fehlt der
        Zugriff, wird ein minimales :class:`DataflowInfo` mit den bereits
        bekannten Basisdaten zurückgegeben (graceful degradation).

        Args:
            dataflow_id: Objekt-/Item-ID des Dataflows.
            generation: "Gen1" oder "Gen2".
            name/description/web_url: Aus der Enumeration bekannte Basiswerte.
        """
        base = DataflowInfo(
            id=dataflow_id,
            name=name,
            generation=generation or "Gen1",
            description=description,
            web_url=web_url,
        )
        try:
            if (generation or "").lower() == "gen2":
                return self._extract_dataflow_gen2(dataflow_id, base)
            return self._extract_dataflow_gen1(dataflow_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Dataflow-Details nicht verfügbar (%s, %s): %s",
                dataflow_id,
                generation,
                exc,
            )
            return base

    def _extract_dataflow_gen1(
        self, dataflow_id: str, base: DataflowInfo
    ) -> DataflowInfo:
        """Liest die ``model.json``-Definition eines Gen1-Dataflows."""
        url = f"{_BASE_URL}/groups/{self._workspace_id}/dataflows/{dataflow_id}"
        model = self._api_get(url)
        if not model:
            return base
        return _parse_gen1_model(model, base)

    def _extract_dataflow_gen2(
        self, dataflow_id: str, base: DataflowInfo
    ) -> DataflowInfo:
        """Liest die Definition (mashup.pq / queryMetadata) eines Gen2-Dataflows."""
        # Item-Basisdaten (Name/Beschreibung), falls noch nicht bekannt.
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"dataflows/{dataflow_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Gen2 Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_gen2_definition(definition, base)

    # ------------------------------------------------------------------
    # Notebook extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_notebook(
        self, notebook_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> NotebookInfo:
        """Extrahiert die Detail-Metadaten eines Fabric-Notebooks.

        Der Quelltext wird über die Fabric REST API (``getDefinition`` →
        ``notebook-content.ipynb`` bzw. ``notebook-content.py``) gelesen. Fehlt
        der Zugriff, wird ein minimales :class:`NotebookInfo` mit den bereits
        bekannten Basisdaten zurückgegeben (graceful degradation).

        Args:
            notebook_id: Item-ID des Notebooks.
            name/description/web_url: Aus der Enumeration bekannte Basiswerte.
        """
        base = NotebookInfo(
            id=notebook_id,
            name=name,
            description=description,
            web_url=web_url,
        )
        try:
            return self._extract_notebook(notebook_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Notebook-Details nicht verfügbar (%s): %s", notebook_id, exc
            )
            return base

    def _extract_notebook(
        self, notebook_id: str, base: NotebookInfo
    ) -> NotebookInfo:
        """Liest die Definition (notebook-content) eines Fabric-Notebooks."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"notebooks/{notebook_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Notebook Item-Metadaten nicht verfügbar: %s", exc)

        # ipynb-Format anfordern (deterministisch parsbar); fällt der Service
        # auf das Python-Quellformat zurück, wird dieses ebenfalls verarbeitet.
        def_url = f"{meta_url}/getDefinition?format=ipynb"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_notebook_definition(definition, base)

    # ------------------------------------------------------------------
    # Copyjob extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_copyjob(
        self, copyjob_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> CopyJobInfo:
        """Extrahiert die Detail-Metadaten eines Fabric Copy Jobs.

        Die Kopierlogik wird über die Fabric REST API (``getDefinition`` →
        ``copyjob-content.json``) gelesen. Fehlt der Zugriff, wird ein minimales
        :class:`CopyJobInfo` mit den bereits bekannten Basisdaten zurückgegeben
        (graceful degradation).

        Args:
            copyjob_id: Item-ID des Copyjobs.
            name/description/web_url: Aus der Enumeration bekannte Basiswerte.
        """
        base = CopyJobInfo(
            id=copyjob_id,
            name=name,
            description=description,
            web_url=web_url,
        )
        try:
            return self._extract_copyjob(copyjob_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Copyjob-Details nicht verfügbar (%s): %s", copyjob_id, exc
            )
            return base

    def _extract_copyjob(
        self, copyjob_id: str, base: CopyJobInfo
    ) -> CopyJobInfo:
        """Liest die Definition (copyjob-content.json) eines Fabric Copy Jobs."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"copyJobs/{copyjob_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Copyjob Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_copyjob_definition(definition, base)

    # ------------------------------------------------------------------
    # Data Pipeline extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_datapipeline(
        self, pipeline_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> DataPipelineInfo:
        """Extrahiert die Detail-Metadaten einer Fabric Data Pipeline.

        Aktivitäten/Parameter/Variablen werden über die Fabric REST API
        (``getDefinition`` → ``pipeline-content.json``) gelesen. Fehlt der
        Zugriff, wird ein minimales :class:`DataPipelineInfo` zurückgegeben
        (graceful degradation).
        """
        base = DataPipelineInfo(
            id=pipeline_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_datapipeline(pipeline_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Pipeline-Details nicht verfügbar (%s): %s", pipeline_id, exc
            )
            return base

    def _extract_datapipeline(
        self, pipeline_id: str, base: DataPipelineInfo
    ) -> DataPipelineInfo:
        """Liest die Definition (pipeline-content.json) einer Data Pipeline."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"dataPipelines/{pipeline_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Pipeline Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_pipeline_definition(definition, base)

    # ------------------------------------------------------------------
    # Dashboard extraction (Power BI / Fabric)
    # ------------------------------------------------------------------

    def extract_dashboard(
        self, dashboard_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> DashboardInfo:
        """Extrahiert die Metadaten eines Power BI / Fabric Dashboards.

        Dashboards besitzen kein ``getDefinition``; die Doku basiert auf den
        Item-Properties (über die Dashboards-Liste) und der Kachelliste
        (``/tiles``) – beides über die Power BI REST API (nur Lese-Rechte).
        Fehlt der Zugriff, wird ein minimales :class:`DashboardInfo`
        zurückgegeben (graceful degradation).
        """
        base = DashboardInfo(
            id=dashboard_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_dashboard(dashboard_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Dashboard-Details nicht verfügbar (%s): %s", dashboard_id, exc
            )
            return base

    def _extract_dashboard(
        self, dashboard_id: str, base: DashboardInfo
    ) -> DashboardInfo:
        """Liest Properties (über die Dashboards-Liste) und Kacheln eines Dashboards."""
        base_url = f"{_BASE_URL}/groups/{self._workspace_id}/dashboards"

        # Metadaten: bevorzugt Einzelabruf, sonst über die Liste suchen.
        meta: dict[str, Any] | None = None
        try:
            meta = self._api_get(f"{base_url}/{dashboard_id}")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Dashboard-Einzelabruf nicht verfügbar: %s", exc)
        if not meta:
            try:
                data = self._api_get(base_url)
                for item in (data or {}).get("value", []):
                    if item.get("id") == dashboard_id:
                        meta = item
                        break
            except Exception as exc:  # noqa: BLE001
                logger.debug("Dashboard-Liste nicht verfügbar: %s", exc)

        if meta:
            base.raw_metadata = meta
            base.name = base.name or meta.get(
                "displayName", meta.get("name", "")
            ) or ""
            base.web_url = base.web_url or meta.get("webUrl", "") or ""
            base.is_read_only = bool(meta.get("isReadOnly", False))
            base.data_classification = meta.get("dataClassification", "") or ""

        # Kacheln (Tiles)
        tiles: list[DashboardTileInfo] = []
        try:
            tdata = self._api_get(f"{base_url}/{dashboard_id}/tiles")
            for tile in (tdata or {}).get("value", []):
                tiles.append(
                    DashboardTileInfo(
                        id=str(tile.get("id", "") or ""),
                        title=str(tile.get("title", "") or ""),
                        subtitle=str(tile.get("subTitle", "") or ""),
                        report_id=str(tile.get("reportId", "") or ""),
                        dataset_id=str(tile.get("datasetId", "") or ""),
                    )
                )
        except Exception as exc:  # noqa: BLE001 – Kacheln best-effort
            logger.info("Dashboard-Kacheln nicht verfügbar (%s): %s", dashboard_id, exc)
        base.tiles = tiles
        return base

    # ------------------------------------------------------------------
    # Lakehouse extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_lakehouse(
        self, lakehouse_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> LakehouseInfo:
        """Extrahiert die Metadaten eines Fabric-Lakehouses.

        Liest die Item-Properties (SQL-Analytics-Endpoint, OneLake-Pfade) sowie
        die Tabellenliste (paginiert über ``/tables``). Ein Lakehouse besitzt
        keine ``getDefinition`` und benötigt nur **Lese**-Rechte. Fehlt der
        Zugriff, wird ein minimales :class:`LakehouseInfo` zurückgegeben
        (graceful degradation).

        Args:
            lakehouse_id: Item-ID des Lakehouses.
            name/description/web_url: Aus der Enumeration bekannte Basiswerte.
        """
        base = LakehouseInfo(
            id=lakehouse_id,
            name=name,
            description=description,
            web_url=web_url,
        )
        try:
            return self._extract_lakehouse(lakehouse_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Lakehouse-Details nicht verfügbar (%s): %s", lakehouse_id, exc
            )
            return base

    def _extract_lakehouse(
        self, lakehouse_id: str, base: LakehouseInfo
    ) -> LakehouseInfo:
        """Liest Properties und Tabellenliste eines Fabric-Lakehouses."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"lakehouses/{lakehouse_id}"
        )
        item = self._api_get(meta_url)
        if item:
            _parse_lakehouse_properties(item, base)

        # Tabellen (paginiert über continuationUri/continuationToken).
        tables: list[LakehouseTableInfo] = []
        try:
            url: str | None = f"{meta_url}/tables"
            guard = 0
            while url and guard < 100:
                guard += 1
                data = self._api_get(url)
                if not data:
                    break
                tables.extend(_parse_lakehouse_tables(data.get("data", [])))
                url = data.get("continuationUri") or None
        except Exception as exc:  # noqa: BLE001 – Tabellen best-effort
            logger.info("Lakehouse-Tabellen nicht verfügbar (%s): %s", lakehouse_id, exc)
        base.tables = tables
        return base

    # ------------------------------------------------------------------
    # Warehouse extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_warehouse(
        self, warehouse_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> WarehouseInfo:
        """Extrahiert die Metadaten eines Fabric-Warehouses (nur Lese-Rechte).

        Fehlt der Zugriff, wird ein minimales :class:`WarehouseInfo` mit den
        Basisdaten zurückgegeben (graceful degradation).
        """
        base = WarehouseInfo(
            id=warehouse_id, name=name, description=description, web_url=web_url
        )
        try:
            url = (
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
                f"warehouses/{warehouse_id}"
            )
            item = self._api_get(url)
            if item:
                _parse_warehouse_properties(item, base)
            return base
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Warehouse-Details nicht verfügbar (%s): %s", warehouse_id, exc
            )
            return base

    # ------------------------------------------------------------------
    # Eventhouse extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_eventhouse(
        self, eventhouse_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> EventhouseInfo:
        """Extrahiert die Metadaten eines Fabric-Eventhouses (nur Lese-Rechte).

        Löst die enthaltenen KQL-Datenbank-IDs best-effort auf lesbare Namen auf.
        Fehlt der Zugriff, wird ein minimales :class:`EventhouseInfo`
        zurückgegeben (graceful degradation).
        """
        base = EventhouseInfo(
            id=eventhouse_id, name=name, description=description, web_url=web_url
        )
        try:
            url = (
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
                f"eventhouses/{eventhouse_id}"
            )
            item = self._api_get(url)
            if not item:
                return base
            db_name_map = self._kql_database_name_map()
            _parse_eventhouse_properties(item, base, db_name_map)
            return base
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Eventhouse-Details nicht verfügbar (%s): %s", eventhouse_id, exc
            )
            return base

    def _kql_database_name_map(self) -> dict[str, str]:
        """Best-effort: Ordnet KQL-Datenbank-IDs im Workspace ihre Namen zu."""
        mapping: dict[str, str] = {}
        try:
            data = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/kqlDatabases"
            )
            if data:
                for item in data.get("value", []):
                    db_id = item.get("id", "")
                    if db_id:
                        mapping[db_id] = item.get(
                            "displayName", item.get("name", "")
                        ) or ""
        except Exception as exc:  # noqa: BLE001 – nicht fatal
            logger.debug("KQL-Datenbanken nicht auflösbar: %s", exc)
        return mapping

    # ------------------------------------------------------------------
    # KQL database extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_kqldatabase(
        self, kqldatabase_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> KqlDatabaseInfo:
        """Extrahiert die Metadaten einer Fabric KQL-Datenbank (nur Lese-Rechte).

        KQL-Datenbanken besitzen keine ``getDefinition``; die Doku basiert auf
        den Item-Properties (Query-/Ingestion-URI, übergeordnetes Eventhouse,
        Datenbanktyp). Fehlt der Zugriff, wird ein minimales
        :class:`KqlDatabaseInfo` zurückgegeben (graceful degradation).
        """
        base = KqlDatabaseInfo(
            id=kqldatabase_id, name=name, description=description, web_url=web_url
        )
        try:
            url = (
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
                f"kqlDatabases/{kqldatabase_id}"
            )
            item = self._api_get(url)
            if item:
                _parse_kqldatabase_properties(item, base)
            return base
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "KQL-Datenbank-Details nicht verfügbar (%s): %s",
                kqldatabase_id,
                exc,
            )
            return base

    # ------------------------------------------------------------------
    # SQL database extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_sqldatabase(
        self, sqldatabase_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> SqlDatabaseInfo:
        """Extrahiert die Metadaten einer Fabric SQL Database (nur Lese-Rechte).

        SQL Databases besitzen keine ``getDefinition``; die Doku basiert auf den
        Item-Properties (SQL-Verbindung, Server-FQDN, Collation). Fehlt der
        Zugriff, wird ein minimales :class:`SqlDatabaseInfo` zurückgegeben
        (graceful degradation).
        """
        base = SqlDatabaseInfo(
            id=sqldatabase_id, name=name, description=description, web_url=web_url
        )
        try:
            url = (
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
                f"sqlDatabases/{sqldatabase_id}"
            )
            item = self._api_get(url)
            if item:
                _parse_sqldatabase_properties(item, base)
            return base
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "SQL-Database-Details nicht verfügbar (%s): %s",
                sqldatabase_id,
                exc,
            )
            return base

    # ------------------------------------------------------------------
    # Environment extraction (Fabric)
    # ------------------------------------------------------------------

    def extract_environment(
        self, environment_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> EnvironmentInfo:
        """Extrahiert die Metadaten eines Fabric-Environments (nur Lese-Rechte).

        Liest die Item-Properties, die installierten Bibliotheken und die
        Spark-Compute-/Runtime-Einstellungen (best-effort über separate
        Endpunkte). Fehlt der Zugriff, wird ein minimales
        :class:`EnvironmentInfo` zurückgegeben (graceful degradation).
        """
        base = EnvironmentInfo(
            id=environment_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_environment(environment_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Environment-Details nicht verfügbar (%s): %s", environment_id, exc
            )
            return base

    def _extract_environment(
        self, environment_id: str, base: EnvironmentInfo
    ) -> EnvironmentInfo:
        """Liest Properties, Bibliotheken und Spark-Compute eines Environments."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"environments/{environment_id}"
        )
        item = self._api_get(meta_url)
        if item:
            base.raw_metadata = item
            base.name = base.name or item.get(
                "displayName", item.get("name", "")
            ) or ""
            base.description = base.description or item.get("description", "") or ""

        # Installierte Bibliotheken (paginiert, best-effort).
        libraries: list[EnvironmentLibraryInfo] = []
        try:
            url: str | None = f"{meta_url}/libraries?beta=false"
            guard = 0
            while url and guard < 100:
                guard += 1
                data = self._api_get(url)
                if not data:
                    break
                libraries.extend(_parse_environment_libraries(data))
                url = data.get("continuationUri") or None
        except Exception as exc:  # noqa: BLE001 – Bibliotheken best-effort
            logger.info(
                "Environment-Bibliotheken nicht verfügbar (%s): %s",
                environment_id,
                exc,
            )
        base.libraries = libraries

        # Spark-Compute-/Runtime-Einstellungen (best-effort).
        try:
            compute = self._api_get(f"{meta_url}/sparkcompute?beta=false")
            if compute:
                _parse_environment_sparkcompute(compute, base)
        except Exception as exc:  # noqa: BLE001 – Compute best-effort
            logger.info(
                "Environment-Spark-Compute nicht verfügbar (%s): %s",
                environment_id,
                exc,
            )
        return base

    # ------------------------------------------------------------------
    # Mirrored Database extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_mirroreddb(
        self, mirroreddb_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> MirroredDatabaseInfo:
        """Extrahiert die Metadaten einer Fabric Mirrored Database.

        Quelle/Ziel und gespiegelte Tabellen werden über die Fabric REST API
        (``getDefinition`` → ``mirroring.json``) gelesen. Fehlt der Zugriff,
        wird ein minimales :class:`MirroredDatabaseInfo` zurückgegeben
        (graceful degradation).
        """
        base = MirroredDatabaseInfo(
            id=mirroreddb_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_mirroreddb(mirroreddb_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Mirrored-Database-Details nicht verfügbar (%s): %s",
                mirroreddb_id,
                exc,
            )
            return base

    def _extract_mirroreddb(
        self, mirroreddb_id: str, base: MirroredDatabaseInfo
    ) -> MirroredDatabaseInfo:
        """Liest die Definition (mirroring.json) einer Mirrored Database."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"mirroredDatabases/{mirroreddb_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Mirrored-Database Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_mirroreddb_definition(definition, base)

    # ------------------------------------------------------------------
    # Eventstream extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_eventstream(
        self, eventstream_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> EventstreamInfo:
        """Extrahiert die Metadaten eines Fabric Eventstreams.

        Quellen/Ziele/Operatoren werden über die Fabric REST API
        (``getDefinition`` → ``eventstream.json``) gelesen. Fehlt der Zugriff,
        wird ein minimales :class:`EventstreamInfo` zurückgegeben (graceful
        degradation).
        """
        base = EventstreamInfo(
            id=eventstream_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_eventstream(eventstream_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Eventstream-Details nicht verfügbar (%s): %s", eventstream_id, exc
            )
            return base

    def _extract_eventstream(
        self, eventstream_id: str, base: EventstreamInfo
    ) -> EventstreamInfo:
        """Liest die Definition (eventstream.json) eines Eventstreams."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"eventstreams/{eventstream_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Eventstream Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_eventstream_definition(definition, base)

    # ------------------------------------------------------------------
    # Spark Job Definition extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_sparkjob(
        self, sparkjob_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> SparkJobDefinitionInfo:
        """Extrahiert die Metadaten einer Fabric Spark Job Definition.

        Konfiguration (Datei/Sprache/Argumente/Bibliotheken) wird über die
        Fabric REST API (``getDefinition`` → ``SparkJobDefinitionV1.json``)
        gelesen. Fehlt der Zugriff, wird ein minimales
        :class:`SparkJobDefinitionInfo` zurückgegeben (graceful degradation).
        """
        base = SparkJobDefinitionInfo(
            id=sparkjob_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_sparkjob(sparkjob_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Spark-Job-Definition-Details nicht verfügbar (%s): %s",
                sparkjob_id,
                exc,
            )
            return base

    def _extract_sparkjob(
        self, sparkjob_id: str, base: SparkJobDefinitionInfo
    ) -> SparkJobDefinitionInfo:
        """Liest die Definition (SparkJobDefinitionV1.json) einer Spark Job Definition."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"sparkJobDefinitions/{sparkjob_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Spark-Job Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_sparkjob_definition(definition, base)

    # ------------------------------------------------------------------
    # KQL Queryset extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_kqlqueryset(
        self, queryset_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> KqlQuerysetInfo:
        """Extrahiert die Metadaten eines Fabric KQL-Querysets.

        Die gespeicherten Abfragen werden über die Fabric REST API
        (``getDefinition`` → ``RealTimeQueryset.json``) gelesen. Fehlt der
        Zugriff, wird ein minimales :class:`KqlQuerysetInfo` zurückgegeben
        (graceful degradation).
        """
        base = KqlQuerysetInfo(
            id=queryset_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_kqlqueryset(queryset_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "KQL-Queryset-Details nicht verfügbar (%s): %s", queryset_id, exc
            )
            return base

    def _extract_kqlqueryset(
        self, queryset_id: str, base: KqlQuerysetInfo
    ) -> KqlQuerysetInfo:
        """Liest die Definition (RealTimeQueryset.json) eines KQL-Querysets."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"kqlQuerysets/{queryset_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("KQL-Queryset Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_kqlqueryset_definition(definition, base)

    # ------------------------------------------------------------------
    # Reflex extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_reflex(
        self, reflex_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> ReflexInfo:
        """Extrahiert die Metadaten eines Fabric Reflex-Items (Data Activator).

        Objekte/Entitäten und Regeln/Trigger werden über die Fabric REST API
        (``getDefinition`` → ``ReflexEntities.json``) gelesen. Fehlt der Zugriff,
        wird ein minimales :class:`ReflexInfo` zurückgegeben (graceful
        degradation).
        """
        base = ReflexInfo(
            id=reflex_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_reflex(reflex_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "Reflex-Details nicht verfügbar (%s): %s", reflex_id, exc
            )
            return base

    def _extract_reflex(self, reflex_id: str, base: ReflexInfo) -> ReflexInfo:
        """Liest die Definition (ReflexEntities.json) eines Reflex-Items."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"reflexes/{reflex_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("Reflex Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_reflex_definition(definition, base)

    # ------------------------------------------------------------------
    # GraphQL API extraction (Fabric, definition-based)
    # ------------------------------------------------------------------

    def extract_graphqlapi(
        self, graphqlapi_id: str, *, name: str = "", description: str = "",
        web_url: str = "",
    ) -> GraphQlApiInfo:
        """Extrahiert die Metadaten eines Fabric „API for GraphQL"-Items.

        Datenquellen und Schema werden über die Fabric REST API
        (``getDefinition``) gelesen. Fehlt der Zugriff, wird ein minimales
        :class:`GraphQlApiInfo` zurückgegeben (graceful degradation).
        """
        base = GraphQlApiInfo(
            id=graphqlapi_id, name=name, description=description, web_url=web_url
        )
        try:
            return self._extract_graphqlapi(graphqlapi_id, base)
        except Exception as exc:  # noqa: BLE001 – graceful degradation
            logger.info(
                "GraphQL-API-Details nicht verfügbar (%s): %s", graphqlapi_id, exc
            )
            return base

    def _extract_graphqlapi(
        self, graphqlapi_id: str, base: GraphQlApiInfo
    ) -> GraphQlApiInfo:
        """Liest die Definition (Schema + Datenquellen) einer GraphQL-API."""
        meta_url = (
            f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}/"
            f"GraphQLApis/{graphqlapi_id}"
        )
        try:
            meta = self._api_get(meta_url)
            if meta:
                base.name = base.name or meta.get(
                    "displayName", meta.get("name", "")
                )
                base.description = base.description or meta.get("description", "")
        except Exception as exc:  # noqa: BLE001
            logger.debug("GraphQL-API Item-Metadaten nicht verfügbar: %s", exc)

        def_url = f"{meta_url}/getDefinition"
        definition = self._api_post(def_url, {})
        if not definition:
            return base
        return _parse_graphqlapi_definition(definition, base)

    # ------------------------------------------------------------------
    # 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 _fabric_token_service_principal(
        client_id: str, client_secret: str, authority: str
    ) -> str | None:
        """Best-effort Fabric-Token (Item-Definition-APIs) für einen SP.

        Erfordert, dass die eigene App-Registrierung die Fabric-Berechtigung
        ``Item.ReadWrite.All`` (o. ä.) mit Admin-Consent besitzt. Schlägt die
        Beschaffung fehl, wird ``None`` zurückgegeben (Fallback auf den
        Power-BI-Token). Ein Service Principal nutzt die EIGENE App und ist
        daher nicht von der First-Party-Preauthorization (AADSTS65002) betroffen.
        """
        try:
            app = msal.ConfidentialClientApplication(
                client_id,
                authority=authority,
                client_credential=client_secret,
            )
            result = app.acquire_token_for_client(scopes=_FABRIC_SCOPE_APP)
            if result and "access_token" in result:
                return result["access_token"]
            logger.info(
                "Fabric-Token (Service-Principal) nicht verfügbar: %s",
                result.get("error_description", result.get("error", "")),
            )
        except Exception as exc:  # noqa: BLE001 – nicht fatal
            logger.info("Fabric-Token (Service-Principal) fehlgeschlagen: %s", exc)
        return None

    @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}"
        )

    def _authenticate_device_code(self, 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.

        Wird eine **eigene** Client-ID angegeben (nicht der well-known Power-BI-
        Client), wird zusätzlich best-effort ein Fabric-Token (delegiert,
        ``Item.ReadWrite.All``) für die Item-Definition-APIs beschafft. Für den
        well-known Microsoft-Client wird das bewusst NICHT versucht, da dieser
        keine Fabric-Tokens anfordern darf (AADSTS65002).
        """
        # Ist eine eigene (Nicht-Microsoft-)Client-ID gesetzt?
        is_custom_client = bool(client_id) and client_id != _WELL_KNOWN_PBI_CLIENT_ID

        # Well-known Microsoft Power BI first-party client ID als Default.
        if not client_id:
            client_id = _WELL_KNOWN_PBI_CLIENT_ID

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

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

        if not pbi_token:
            # Interactive Browser Flow — öffnet Login direkt
            result = app.acquire_token_interactive(
                scopes=_SCOPE,
                prompt="select_account",
            )
            if "access_token" not in result:
                error_desc = result.get(
                    "error_description", result.get("error", "Unbekannter Fehler")
                )
                raise AuthenticationError(
                    f"SSO-Authentifizierung fehlgeschlagen: {error_desc}"
                )
            pbi_token = result["access_token"]

        # Fabric-Token nur für EIGENE Clients (own app registration) beschaffen.
        if is_custom_client:
            self._fabric_access_token = self._fabric_token_public(app)
        else:
            self._fabric_access_token = None

        return pbi_token

    @staticmethod
    def _fabric_token_public(app) -> str | None:  # type: ignore[no-untyped-def]
        """Best-effort Fabric-Token (delegated) über eine eigene PublicClient-App.

        Nur für EIGENE Client-IDs aufrufen (nicht den well-known Microsoft-
        Client). Versucht zuerst einen stillen Token, sonst interaktiven Consent
        für ``Item.ReadWrite.All``. Jeder Fehler ist nicht fatal – dann wird der
        Power-BI-Token als Fallback verwendet.
        """
        accounts = app.get_accounts()
        account = accounts[0] if accounts else None
        try:
            if account is not None:
                result = app.acquire_token_silent(
                    scopes=_FABRIC_SCOPE_DELEGATED, account=account
                )
                if result and "access_token" in result:
                    return result["access_token"]
        except Exception as exc:  # noqa: BLE001
            logger.debug("Fabric-Token (still) fehlgeschlagen: %s", exc)
        try:
            result = app.acquire_token_interactive(scopes=_FABRIC_SCOPE_DELEGATED)
            if result and "access_token" in result:
                return result["access_token"]
            logger.info(
                "Fabric-Token (interaktiv) nicht erhalten: %s",
                result.get("error_description", result.get("error", "")),
            )
        except Exception as exc:  # noqa: BLE001
            logger.info("Fabric-Token (interaktiv) fehlgeschlagen: %s", exc)
        return None

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

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

    def _headers_for(self, url: str) -> dict[str, str]:
        """Wählt den passenden Token je API-Audience.

        Fabric-APIs (``api.fabric.microsoft.com``) nutzen – falls vorhanden –
        den Fabric-Token (mit ``Item.ReadWrite.All`` für ``getDefinition``);
        andernfalls und für Power-BI-APIs den Standard-Token.
        """
        if url.startswith(_FABRIC_BASE_URL) and self._fabric_access_token:
            return {"Authorization": f"Bearer {self._fabric_access_token}"}
        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_for(url), 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 _await_lro(self, initial_resp) -> dict[str, Any] | None:  # type: ignore[no-untyped-def]
        """Verarbeitet eine Long-Running-Operation (HTTP 202) der Fabric-API.

        Pollt die im ``Location``-Header angegebene Operation, bis sie
        abgeschlossen ist, und liefert das Ergebnis (``/result``). Gibt ``None``
        zurück, wenn kein Ergebnis ermittelt werden kann (z. B. Timeout).
        """
        headers = initial_resp.headers or {}
        location = headers.get("Location") or headers.get("Operation-Location")
        if not location:
            op_id = headers.get("x-ms-operation-id")
            if op_id:
                location = f"{_FABRIC_BASE_URL}/operations/{op_id}"
        if not location:
            return None
        try:
            retry_after = float(headers.get("Retry-After", _LRO_POLL_DEFAULT))
        except (TypeError, ValueError):
            retry_after = _LRO_POLL_DEFAULT

        deadline = time.monotonic() + _LRO_MAX_WAIT
        while time.monotonic() < deadline:
            time.sleep(max(0.5, min(retry_after, 10.0)))
            status_resp = requests.get(
                location, headers=self._headers_for(location), timeout=30
            )
            status_resp.raise_for_status()
            payload = status_resp.json() if status_resp.content else {}
            status = str(payload.get("status", "")).lower()
            if status in ("succeeded", "completed"):
                result_url = location.rstrip("/") + "/result"
                result_resp = requests.get(
                    result_url, headers=self._headers_for(result_url), timeout=30
                )
                result_resp.raise_for_status()
                return result_resp.json() if result_resp.content else None
            if status in ("failed", "canceled", "cancelled"):
                raise MetadataExtractionError(
                    f"Fabric-Operation fehlgeschlagen: "
                    f"{payload.get('error', status or 'unbekannt')}"
                )
            try:
                retry_after = float(status_resp.headers.get("Retry-After", retry_after))
            except (TypeError, ValueError):
                pass
        logger.info("Fabric-Operation Zeitüberschreitung beim Pollen: %s", location)
        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.

        Behandelt zusätzlich Long-Running-Operations: Antwortet die Fabric-API
        mit ``202 Accepted``, wird die Operation über den ``Location``-Header
        gepollt und das Ergebnis geliefert (siehe :meth:`_await_lro`).
        """
        last_exc: Exception | None = None
        for attempt in range(retries):
            try:
                resp = requests.post(
                    url, headers=self._headers_for(url), json=json_body, timeout=30
                )
                if resp.status_code == 202:
                    return self._await_lro(resp)
                resp.raise_for_status()
                return resp.json() if resp.content else None
            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

    # ------------------------------------------------------------------
    # DirectLake-Datenquellen-Auflösung
    # ------------------------------------------------------------------

    def _resolve_directlake_data_sources(self) -> list[DataSourceInfo]:
        """Löst DirectLake-Quellen zum konkreten Lakehouse/Warehouse auf.

        DirectLake-Modelle referenzieren ihr Lakehouse/Warehouse über eine
        strukturierte Datenquelle (M-Ausdruck, ``INFO.EXPRESSIONS()``), die den
        SQL-Analytics-Endpoint bzw. OneLake-Pfad enthält. Dieser wird gegen die
        Lakehouses/Warehouses des Workspace aufgelöst, um den lesbaren Namen zu
        erhalten.

        Returns:
            Liste aufgelöster :class:`DataSourceInfo`; leer, wenn keine
            DirectLake-Quelle erkannt bzw. aufgelöst werden konnte.
        """
        expressions = self._extract_expressions()
        if not expressions:
            return []
        # Nur wenn überhaupt eine DirectLake-typische Quelle vorliegt, die
        # (potenziell teuren) Item-Abfragen durchführen.
        has_directlake = any(
            self._parse_directlake_expression(e.get("expression", ""))
            for e in expressions
        )
        if not has_directlake:
            return []
        targets = self._list_directlake_targets()
        return self._match_directlake_sources(expressions, targets)

    def _extract_expressions(self) -> list[dict[str, str]]:
        """Liest die geteilten M-Ausdrücke des Modells via ``INFO.EXPRESSIONS()``.

        Für DirectLake enthält typischerweise ein Ausdruck (z. B.
        ``DatabaseQuery``) die strukturierte Datenquelle mit SQL-Endpoint bzw.
        OneLake-Pfad.
        """
        rows = self._execute_dax_query(
            'EVALUATE SELECTCOLUMNS(INFO.EXPRESSIONS(), '
            '"Name", [Name], "Expression", [Expression])'
        )
        expressions: list[dict[str, str]] = []
        for row in rows:
            name = self._row_get(row, "Name", "")
            expr = self._row_get(row, "Expression", "")
            if expr:
                expressions.append({"name": str(name), "expression": str(expr)})
        return expressions

    def _list_directlake_targets(self) -> list[dict[str, str]]:
        """Listet Lakehouses und Warehouses des Workspace inkl. SQL-Endpoint.

        Jeder Eintrag enthält ``name``, ``type`` ("Lakehouse"/"Warehouse"),
        ``id``, ``server`` (SQL-Connection-String) und ``sql_endpoint_id``.
        """
        targets: list[dict[str, str]] = []
        ws = self._workspace_id

        # Lakehouses – SQL-Endpoint steckt in properties.sqlEndpointProperties
        try:
            data = self._api_get(f"{_FABRIC_BASE_URL}/workspaces/{ws}/lakehouses")
            for item in (data or {}).get("value", []):
                item_id = item.get("id", "") or ""
                name = item.get("displayName", item.get("name", "")) or ""
                props = item.get("properties") or {}
                sql = (
                    props.get("sqlEndpointProperties") or {}
                ) if isinstance(props, dict) else {}
                server = sql.get("connectionString", "") or ""
                sql_id = sql.get("id", "") or ""
                # Falls die Liste keine Properties liefert, Item einzeln nachladen.
                if not server and item_id:
                    server, sql_id = self._fetch_lakehouse_sql_endpoint(item_id)
                targets.append({
                    "name": name, "type": "Lakehouse", "id": item_id,
                    "server": server, "sql_endpoint_id": sql_id,
                })
        except Exception as exc:
            logger.info("Lakehouses für DirectLake-Auflösung nicht verfügbar: %s", exc)

        # Warehouses – SQL-Connection-String steckt in properties.connectionString
        try:
            data = self._api_get(f"{_FABRIC_BASE_URL}/workspaces/{ws}/warehouses")
            for item in (data or {}).get("value", []):
                item_id = item.get("id", "") or ""
                name = item.get("displayName", item.get("name", "")) or ""
                props = item.get("properties") or {}
                server = (
                    props.get("connectionString", "") if isinstance(props, dict) else ""
                )
                if not server and item_id:
                    server = self._fetch_warehouse_connection_string(item_id)
                targets.append({
                    "name": name, "type": "Warehouse", "id": item_id,
                    "server": server or "", "sql_endpoint_id": "",
                })
        except Exception as exc:
            logger.info("Warehouses für DirectLake-Auflösung nicht verfügbar: %s", exc)

        return targets

    def _fetch_lakehouse_sql_endpoint(self, lakehouse_id: str) -> tuple[str, str]:
        """Lädt ``(server, sql_endpoint_id)`` eines Lakehouse nach (best-effort)."""
        try:
            item = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}"
                f"/lakehouses/{lakehouse_id}"
            )
            props = (item or {}).get("properties") or {}
            sql = (
                props.get("sqlEndpointProperties") or {}
            ) if isinstance(props, dict) else {}
            return sql.get("connectionString", "") or "", sql.get("id", "") or ""
        except Exception as exc:
            logger.info("Lakehouse-SQL-Endpoint %s nicht abrufbar: %s", lakehouse_id, exc)
            return "", ""

    def _fetch_warehouse_connection_string(self, warehouse_id: str) -> str:
        """Lädt den ``connectionString`` eines Warehouse nach (best-effort)."""
        try:
            item = self._api_get(
                f"{_FABRIC_BASE_URL}/workspaces/{self._workspace_id}"
                f"/warehouses/{warehouse_id}"
            )
            props = (item or {}).get("properties") or {}
            return props.get("connectionString", "") if isinstance(props, dict) else ""
        except Exception as exc:
            logger.info("Warehouse-Connection %s nicht abrufbar: %s", warehouse_id, exc)
            return ""

    @staticmethod
    def _parse_directlake_expression(m_expr: str) -> dict[str, str] | None:
        """Extrahiert die DirectLake-Zielbindung aus einem M-Ausdruck.

        Erkennt zwei DirectLake-Varianten:
          * DirectLake-over-SQL: ``Sql.Database("<endpoint>", "<db-guid>")`` →
            ``{"kind": "sql", "server": ..., "database": ...}`` (nur für
            Fabric-/OneLake-SQL-Endpoints).
          * DirectLake-on-OneLake: OneLake-Pfad mit ``<name>.Lakehouse`` bzw.
            GUIDs → ``{"kind": "onelake", "path": ..., "item": ...}``.

        Returns:
            Dict mit der Zielbindung oder ``None``, wenn keine DirectLake-Quelle
            erkennbar ist (klassische SQL-/Import-Quellen werden bewusst ignoriert,
            da diese bereits über ``/datasources`` erfasst werden).
        """
        import re

        if not m_expr:
            return None

        # DirectLake-over-SQL: Sql.Database("server", "db")
        m = re.search(r'Sql\.Database\(\s*"([^"]+)"\s*,\s*"([^"]+)"', m_expr)
        if m:
            server = m.group(1)
            database = m.group(2)
            if _is_fabric_sql_endpoint(server):
                return {"kind": "sql", "server": server, "database": database}

        # DirectLake-on-OneLake / Lakehouse-Konnektor.
        # Beispiele:
        #   AzureStorage.DataLake(
        #       "https://onelake.dfs.fabric.microsoft.com/<ws>/<name>.Lakehouse/Tables")
        #   ... "https://onelake.dfs.fabric.microsoft.com/<ws-guid>/<item-guid>" ...
        low = m_expr.lower()
        if "onelake" in low or ".lakehouse" in low or ".warehouse" in low:
            url_m = re.search(
                r'"(https?://[^"]*onelake[^"]*)"', m_expr, re.IGNORECASE
            )
            path = url_m.group(1) if url_m else ""
            item_m = re.search(r'/([^/"]+)\.(?:Lakehouse|Warehouse)\b', m_expr)
            item = item_m.group(1) if item_m else ""
            if path or item:
                return {"kind": "onelake", "path": path, "item": item}

        return None

    @staticmethod
    def _match_directlake_sources(
        expressions: list[dict[str, str]],
        targets: list[dict[str, str]],
    ) -> list[DataSourceInfo]:
        """Ordnet DirectLake-Ausdrücke den Lakehouses/Warehouses zu.

        Args:
            expressions: Liste von ``{"name", "expression"}`` (INFO.EXPRESSIONS).
            targets: Liste von ``{"name","type","id","server","sql_endpoint_id"}``.

        Returns:
            Aufgelöste :class:`DataSourceInfo` mit lesbarem Lakehouse-/
            Warehouse-Namen. Nicht auflösbare DirectLake-Quellen werden mit dem
            rohen Endpoint zurückgegeben, damit die Doku nicht erneut nur
            "OneLakeStorage" anzeigt.
        """
        results: list[DataSourceInfo] = []
        seen: set[str] = set()

        for expr in expressions:
            parsed = FabricConnector._parse_directlake_expression(
                expr.get("expression", "")
            )
            if not parsed:
                continue

            matched: dict[str, str] | None = None
            if parsed["kind"] == "sql":
                server = (parsed.get("server") or "").strip().lower()
                for t in targets:
                    tserver = (t.get("server") or "").strip().lower()
                    if tserver and server and tserver == server:
                        matched = t
                        break
            else:  # onelake
                item = (parsed.get("item") or "").strip().lower()
                path = (parsed.get("path") or "").lower()
                for t in targets:
                    tname = (t.get("name") or "").strip().lower()
                    tid = (t.get("id") or "").strip().lower()
                    if item and tname and item == tname:
                        matched = t
                        break
                    if tid and tid in path:
                        matched = t
                        break

            if matched:
                key = f"{matched['type']}:{matched['name']}"
                if key in seen:
                    continue
                seen.add(key)
                results.append(DataSourceInfo(
                    name=matched["name"],
                    source_type=f"DirectLake ({matched['type']})",
                    connection_string=matched.get("server", "") or parsed.get("path", ""),
                    schema=parsed.get("database") or None,
                ))
            else:
                raw = (
                    parsed.get("server")
                    or parsed.get("path")
                    or parsed.get("item")
                    or ""
                )
                if not raw:
                    continue
                key = f"DirectLake:{raw}"
                if key in seen:
                    continue
                seen.add(key)
                results.append(DataSourceInfo(
                    name=f"DirectLake: {raw}",
                    source_type="DirectLake",
                    connection_string=raw,
                    schema=parsed.get("database") or None,
                ))

        return results

    @staticmethod
    def _merge_directlake_sources(
        existing: list[DataSourceInfo],
        directlake: list[DataSourceInfo],
    ) -> list[DataSourceInfo]:
        """Ersetzt generische DirectLake-Platzhalter durch aufgelöste Quellen.

        Aus ``/datasources`` stammende Einträge mit generischem DirectLake-Typ
        (z. B. "OneLakeStorage"/"FabricService"/"Extension") werden entfernt und
        durch die konkret aufgelösten Lakehouse-/Warehouse-Quellen ersetzt. Alle
        übrigen (echten) Quellen bleiben erhalten.
        """
        generic_markers = {
            "onelakestorage", "fabricservice", "extension",
            "onelake", "fabric", "unknown", "",
        }
        kept = [
            ds for ds in existing
            if (ds.source_type or "").strip().lower() not in generic_markers
        ]
        existing_names = {(ds.name or "").strip().lower() for ds in kept}
        merged = list(kept)
        for ds in directlake:
            if (ds.name or "").strip().lower() in existing_names:
                continue
            merged.append(ds)
        return merged

    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
