"""Lokaler Bericht Konnektor – verbindet sich mit Power BI Desktop über pythonnet/ADOMD.NET."""
from __future__ import annotations

import logging
import os
from pathlib import Path
from typing import Any

from core.connectors.base import IConnector
from core.models import (
    ColumnInfo,
    ConnectionConfig,
    DataSourceInfo,
    MeasureInfo,
    PBIConnectionError,
    PowerQueryInfo,
    RelationshipInfo,
    ReportMetadata,
    RLSRoleInfo,
    TableInfo,
)

logger = logging.getLogger(__name__)

# Prüfe ob pythonnet verfügbar ist
_ADOMD_AVAILABLE = False
_ADOMD_ERROR = ""


def _init_pythonnet():
    """Initialisiert pythonnet robust für das gefrorene EXE-Bundle.

    Hintergrund
    -----------
    ``import clr`` ruft intern ``pythonnet.load()`` auf. Dieses berechnet den
    Pfad zur .NET-Assembly als ``Path(pythonnet.__file__).parent / "runtime" /
    "Python.Runtime.dll"`` und lässt ``clr_loader`` die Methode
    ``Python.Runtime.Loader.Initialize`` daraus auflösen.

    Zwei bekannte Probleme führen zu
    ``RuntimeError: Failed to resolve Python.Runtime.Loader.Initialize``:

    1. **Leerzeichen im Pfad**: liegt die EXE in einem Ordner mit Leerzeichen
       (z.B. 'PowerBI-AutoDoku 7'), kann ``clr_loader`` die DLL nicht als
       .NET-Assembly laden.
    2. **Beschädigter Runtime-Cache**: Wird die Runtime (früher nach ``%TEMP%``)
       kopiert, kann sie durch Temp-Bereinigung (Datenträgerbereinigung /
       Storage Sense) oder abgebrochene Kopiervorgänge unvollständig werden.
       Die auf mtime basierende Prüfung reparierte solche Reste nie – der
       Fehler trat dann bei jedem Start erneut auf.

    Lösung
    ------
    Im gefrorenen Bundle wird die Runtime deterministisch in ein
    **leerzeichenfreies, stabiles** Verzeichnis (bevorzugt unter
    ``%LOCALAPPDATA%``, nicht ``%TEMP%``) bereitgestellt. Die Bereitstellung ist
    **integritätsgeprüft** (Dateigrößen), **atomar** (``.part`` + ``os.replace``)
    und **versionsspezifisch** (Hash über Quelldateien), und der Import wird bei
    einem ``RuntimeError`` mit frisch erzwungener Kopie einmal wiederholt
    (Selbstheilung gegen beschädigte Caches).
    """
    import sys

    if not getattr(sys, "frozen", False):
        # Normaler Entwicklungsmodus – einfach importieren
        import clr  # type: ignore[import-untyped]  # noqa: F401
        return

    base_path = sys._MEIPASS
    bundled_runtime = os.path.join(base_path, "pythonnet", "runtime")
    bundled_dll = os.path.join(bundled_runtime, "Python.Runtime.dll")

    # DLL fehlt im Bundle – normaler Import liefert die aussagekräftigste Meldung.
    if not os.path.exists(bundled_dll):
        import clr  # type: ignore[import-untyped]  # noqa: F401
        return

    # Schneller Pfad: keine Leerzeichen -> Direktimport versuchen.
    if " " not in base_path:
        try:
            import clr  # type: ignore[import-untyped]  # noqa: F401
            return
        except RuntimeError as exc:
            # z.B. beschädigte gebündelte DLL -> über gestagte Kopie selbstheilen
            logger.warning("Direkter clr-Import fehlgeschlagen, nutze Staging: %s", exc)

    # Selbstheilende, gestagte Bereitstellung (max. 2 Versuche).
    last_err: Exception | None = None
    for attempt in range(2):
        work_dir = _stage_net_runtime(bundled_runtime, force=(attempt > 0))
        _point_pythonnet_at(work_dir)
        sys.modules.pop("clr", None)
        try:
            import clr  # type: ignore[import-untyped]  # noqa: F401
            return
        except RuntimeError as exc:
            last_err = exc
            logger.warning(
                "clr-Import aus gestagter Runtime fehlgeschlagen "
                "(Versuch %d/2): %s", attempt + 1, exc
            )

    # Beide Versuche gescheitert – ursprünglichen Fehler weiterreichen.
    raise last_err if last_err is not None else RuntimeError(
        "pythonnet konnte nicht initialisiert werden."
    )


def _pick_stage_base() -> str:
    """Wählt ein leerzeichenfreies, möglichst stabiles Basisverzeichnis.

    Reihenfolge: %LOCALAPPDATA% (überlebt Temp-Bereinigung) -> %TEMP% ->
    <SystemDrive>\\ProgramData -> <SystemDrive>\\Temp. Fällt auf ein festes
    leerzeichenfreies Verzeichnis auf dem Systemlaufwerk zurück.
    """
    import tempfile

    drive = os.environ.get("SYSTEMDRIVE", "C:") + os.sep
    candidates = [
        os.environ.get("LOCALAPPDATA"),
        tempfile.gettempdir(),
        os.path.join(drive, "ProgramData"),
        os.path.join(drive, "Temp"),
    ]
    for cand in candidates:
        if cand and " " not in cand:
            try:
                os.makedirs(cand, exist_ok=True)
                return cand
            except OSError:
                continue

    fallback = os.path.join(drive, "PBIAutoDoku")
    os.makedirs(fallback, exist_ok=True)
    return fallback


def _stage_net_runtime(src_runtime: str, force: bool = False) -> str:
    """Stellt die .NET-Runtime integritätsgeprüft und atomar bereit.

    Kopiert alle Dateien aus ``src_runtime`` in ein versionsspezifisches,
    leerzeichenfreies Zielverzeichnis. Vorhandene Dateien mit passender Größe
    werden übersprungen (außer ``force=True``); abweichende/fehlende/beschädigte
    Dateien werden atomar neu geschrieben. Gibt das Arbeitsverzeichnis zurück,
    dessen Unterordner ``runtime`` die Assemblies enthält.
    """
    import hashlib
    import shutil

    entries = sorted(
        (fname, os.path.getsize(os.path.join(src_runtime, fname)))
        for fname in os.listdir(src_runtime)
        if os.path.isfile(os.path.join(src_runtime, fname))
    )

    # Versions-/Inhaltsschlüssel: verhindert Kollisionen zwischen App-Versionen
    # und erzwingt bei geänderten Quelldateien automatisch eine frische Kopie.
    key = hashlib.sha1(repr(entries).encode("utf-8")).hexdigest()[:12]

    base = _pick_stage_base()
    work_dir = os.path.join(base, "pbi_autodoku_net", key)
    dst_runtime = os.path.join(work_dir, "runtime")
    os.makedirs(dst_runtime, exist_ok=True)

    for fname, size in entries:
        src = os.path.join(src_runtime, fname)
        dst = os.path.join(dst_runtime, fname)

        if (not force) and os.path.exists(dst) and os.path.getsize(dst) == size:
            continue

        tmp = dst + ".part"
        shutil.copyfile(src, tmp)
        if os.path.getsize(tmp) != size:
            # Kopie unvollständig -> aufräumen, damit kein Rest zurückbleibt.
            try:
                os.remove(tmp)
            except OSError:
                pass
            raise RuntimeError(f"Kopie der .NET-Runtime unvollständig: {fname}")
        os.replace(tmp, dst)  # atomar

    return work_dir


def _point_pythonnet_at(work_dir: str) -> None:
    """Biegt ``pythonnet.__file__`` auf das gestagte Runtime-Verzeichnis um.

    ``pythonnet.load()`` leitet den DLL-Pfad aus ``pythonnet.__file__`` ab.
    Zusätzlich wird evtl. vorhandener Ladezustand eines fehlgeschlagenen
    Vorversuchs zurückgesetzt, damit der erneute Import wirklich neu lädt.
    """
    import pythonnet

    fake_init = os.path.join(work_dir, "__init__.py")
    if not os.path.exists(fake_init):
        with open(fake_init, "w", encoding="utf-8") as fh:
            fh.write("")
    pythonnet.__file__ = fake_init

    # pythonnet.load() bricht früh ab, wenn bereits (teilweise) geladen wurde.
    for attr in ("_LOADER_ASSEMBLY", "_RUNTIME"):
        if hasattr(pythonnet, attr):
            setattr(pythonnet, attr, None)


try:
    _init_pythonnet()
    _ADOMD_AVAILABLE = True
except ImportError:
    _ADOMD_ERROR = "pythonnet ist nicht installiert. Bitte installieren: pip install pythonnet"
except RuntimeError as e:
    _ADOMD_ERROR = (
        f"pythonnet konnte die .NET Runtime nicht laden: {e}\n"
        "Bitte .NET Framework 4.7.2 oder höher installieren: "
        "https://dotnet.microsoft.com/download/dotnet-framework"
    )
except Exception as e:
    _ADOMD_ERROR = f"Unerwarteter Fehler beim Laden von pythonnet: {e}"

_WORKSPACE_ROOT = (
    Path(os.environ.get("LOCALAPPDATA", ""))
    / "Microsoft"
    / "Power BI Desktop"
    / "AnalysisServicesWorkspaces"
)

# Store-Version von Power BI Desktop
_WORKSPACE_ROOT_STORE = (
    Path(os.environ.get("USERPROFILE", ""))
    / "Microsoft"
    / "Power BI Desktop Store App"
    / "AnalysisServicesWorkspaces"
)


class LocalConnector(IConnector):
    """Konnektor für lokal in Power BI Desktop geöffnete Berichte."""

    def __init__(self) -> None:
        self._connection: Any | None = None
        self._port: str = ""
        self._workspace_name: str = ""

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

    def connect(self, config: ConnectionConfig) -> None:
        """Connect to a locally running Power BI Desktop instance.

        ``config.params`` may contain:
        - ``port``: explicit SSAS port to connect to
        - ``name``: report name to select from discovered instances
        """
        if not _ADOMD_AVAILABLE:
            raise PBIConnectionError(_ADOMD_ERROR)

        params = config.params
        port = params.get("port", "")

        if not port:
            instances = self.discover_instances()
            if not instances:
                raise PBIConnectionError(
                    "Kein geöffneter Power BI Desktop Bericht gefunden. "
                    "Bitte öffnen Sie einen Bericht in Power BI Desktop."
                )

            report_name = params.get("name", "") or params.get("report_name", "")
            if report_name:
                matched = [
                    inst for inst in instances
                    if inst.get("name") == report_name
                    or inst["workspace_name"] == report_name
                ]
                if not matched:
                    raise PBIConnectionError(
                        f"Kein Power BI Desktop Bericht mit dem Namen "
                        f"'{report_name}' gefunden."
                    )
                port = matched[0]["port"]
                self._workspace_name = matched[0].get("name", matched[0]["workspace_name"])
            elif len(instances) == 1:
                port = instances[0]["port"]
                self._workspace_name = instances[0].get("name", instances[0]["workspace_name"])
            else:
                raise PBIConnectionError(
                    "Mehrere Power BI Desktop Berichte geöffnet. "
                    "Bitte wählen Sie einen Bericht aus."
                )

        self._port = port
        conn_str = (
            f"Provider=MSOLAP;Data Source=127.0.0.1:{self._port};"
        )

        try:
            self._connection = self._create_adomd_connection(conn_str)
        except Exception as exc:
            self._connection = None
            raise PBIConnectionError(
                f"Verbindung zu localhost:{self._port} fehlgeschlagen: {exc}"
            ) from exc

    @staticmethod
    def _create_adomd_connection(conn_str: str) -> Any:
        """Erstellt eine ADOMD.NET-Verbindung über pythonnet."""
        import clr  # type: ignore[import-untyped]

        # ADOMD.NET Assembly laden
        try:
            clr.AddReference("Microsoft.AnalysisServices.AdomdClient")
        except Exception:
            # Versuche den GAC-Pfad oder den lokalen Pfad
            import glob
            adomd_paths = glob.glob(
                os.path.join(
                    os.environ.get("ProgramFiles(x86)", ""),
                    "Microsoft.NET", "**", "Microsoft.AnalysisServices.AdomdClient.dll"
                ),
                recursive=True,
            )
            if not adomd_paths:
                # Versuche im Power BI Desktop Ordner
                pbi_paths = glob.glob(
                    os.path.join(
                        os.environ.get("ProgramFiles", ""),
                        "Microsoft Power BI Desktop", "**",
                        "Microsoft.AnalysisServices.AdomdClient.dll"
                    ),
                    recursive=True,
                )
                adomd_paths.extend(pbi_paths)
                # Store App Pfad
                store_paths = glob.glob(
                    os.path.join(
                        os.environ.get("ProgramFiles", r"C:\Program Files"),
                        "WindowsApps", "**",
                        "Microsoft.AnalysisServices.AdomdClient.dll"
                    ),
                    recursive=True,
                )
                adomd_paths.extend(store_paths)

            if adomd_paths:
                clr.AddReference(adomd_paths[0])
            else:
                raise PBIConnectionError(
                    "ADOMD.NET Client-Bibliothek nicht gefunden. "
                    "Bitte installieren Sie die 'Microsoft Analysis Services "
                    "Client Libraries' (MSOLAP Provider)."
                )

        from Microsoft.AnalysisServices.AdomdClient import AdomdConnection  # type: ignore

        conn = AdomdConnection(conn_str)
        conn.Open()
        return conn

    def extract_metadata(self) -> ReportMetadata:
        """Extract metadata via DMV queries over the local SSAS connection."""
        if self._connection is None:
            raise PBIConnectionError(
                "Nicht verbunden – bitte zuerst connect() aufrufen."
            )

        # Echten Berichtsnamen ermitteln:
        # 1) Fenstertitel (vom Benutzer gewählter Bericht)
        # 2) Katalogname (kann GUID sein bei PBI Desktop)
        # 3) Fallback: workspace_name oder localhost:port
        report_name = self._workspace_name or f"localhost:{self._port}"

        # Fenstertitel ist am zuverlässigsten
        # _workspace_name enthält bereits den Fenstertitel wenn verfügbar
        # Prüfe ob es eine GUID ist und ersetze ggf. durch Katalognamen
        import re
        is_guid = bool(re.match(
            r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
            report_name, re.IGNORECASE,
        ))
        is_workspace_id = report_name.startswith("AnalysisServicesWorkspace_")

        if is_guid or is_workspace_id or report_name.startswith("localhost:"):
            # Versuche den echten Namen aus dem Fenstertitel zu holen
            titles = self._get_pbi_window_titles()
            if titles:
                report_name = titles[0]
            else:
                # Fallback: Katalogname (kann auch GUID sein)
                db_name = self._get_database_name()
                if db_name and not bool(re.match(
                    r"^[0-9a-f]{8}-", db_name, re.IGNORECASE
                )):
                    report_name = db_name

        tables = self._extract_tables()

        # Table-ID → Name Mapping aufbauen
        table_id_map: dict[int, str] = {}
        for i, t in enumerate(tables):
            tid = getattr(t, "_table_id", i)
            table_id_map[tid] = t.name

        self._extract_columns(tables)
        measures = self._extract_measures_with_names(table_id_map)
        relationships = self._extract_relationships_with_names(table_id_map)
        rls_roles = self._extract_rls_roles_with_filters(table_id_map)
        data_sources = self._extract_data_sources()
        power_queries = self._extract_power_queries_with_names(table_id_map)

        # Interne PBI-Tabellen filtern
        internal_names = {
            t.name for t in tables
            if t.name.startswith("LocalDateTable_")
            or t.name.startswith("DateTableTemplate_")
        }
        tables = [t for t in tables if t.name not in internal_names]
        all_columns = [c for c in [col for t in tables for col in t.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
        ]
        data_sources = [
            ds for ds in data_sources
            if ds.name not in internal_names
        ]

        # Tabellenklassifikation
        from core.connectors.pbip_connector import PBIPConnector
        PBIPConnector._classify_tables(tables, relationships, measures)

        return ReportMetadata(
            report_name=report_name,
            tables=tables,
            columns=all_columns,
            measures=measures,
            relationships=relationships,
            data_sources=data_sources,
            power_queries=power_queries,
            rls_roles=rls_roles,
            visuals=[],
            refresh_schedule=None,
            raw_metadata={},
        )

    def _get_database_name(self) -> str:
        """Ermittelt den Datenbanknamen (= Berichtsnamen) über DMV."""
        try:
            rows = self._execute_dmv(
                "SELECT [CATALOG_NAME] FROM $SYSTEM.DBSCHEMA_CATALOGS"
            )
            if rows:
                return rows[0].get("CATALOG_NAME", "") or ""
        except Exception:
            pass
        return ""

    def disconnect(self) -> None:
        """Close the ADOMD.NET connection."""
        if self._connection is not None:
            try:
                self._connection.Close()
                self._connection.Dispose()
            except Exception as exc:
                logger.warning("Fehler beim Schließen der Verbindung: %s", exc)
            finally:
                self._connection = None

    def test_connection(self) -> bool:
        """Try a simple DMV query to verify the connection works."""
        if self._connection is None:
            return False
        try:
            rows = self._execute_dmv(
                "SELECT TOP 1 [Name] FROM $SYSTEM.TMSCHEMA_TABLES"
            )
            return rows is not None
        except Exception:
            return False

    # ------------------------------------------------------------------
    # Instance discovery
    # ------------------------------------------------------------------

    @staticmethod
    def discover_instances() -> list[dict[str, str]]:
        """Find all open Power BI Desktop instances.

        Scans both the classic and Store App paths for
        ``msmdsrv.port.txt`` files, prüft ob der Port erreichbar ist,
        und ordnet Berichtsnamen über PBI Desktop Fenstertitel zu.

        Returns a list of dicts with keys ``port``, ``workspace_name``,
        and ``name`` (the human-readable report name).
        """
        import socket

        candidates: list[dict[str, str]] = []

        for root in (_WORKSPACE_ROOT, _WORKSPACE_ROOT_STORE):
            if not root.exists():
                continue

            for workspace_dir in root.iterdir():
                if not workspace_dir.is_dir():
                    continue
                port_file = workspace_dir / "Data" / "msmdsrv.port.txt"
                if not port_file.exists():
                    port_file = workspace_dir / "msmdsrv.port.txt"
                if port_file.exists():
                    try:
                        raw = port_file.read_bytes()
                        try:
                            port = raw.decode("utf-16-le").strip().strip("\ufeff")
                        except UnicodeDecodeError:
                            port = raw.decode("utf-8").strip()
                        port = port.replace("\x00", "")
                        if port:
                            candidates.append({
                                "port": port,
                                "workspace_name": workspace_dir.name,
                                "name": workspace_dir.name,
                            })
                    except Exception as exc:
                        logger.warning(
                            "Port-Datei %s konnte nicht gelesen werden: %s",
                            port_file, exc,
                        )

        # Nur Instanzen behalten, deren Port tatsächlich erreichbar ist
        instances: list[dict[str, str]] = []
        for inst in candidates:
            try:
                s = socket.create_connection(
                    ("127.0.0.1", int(inst["port"])), timeout=1
                )
                s.close()
                instances.append(inst)
            except Exception:
                logger.debug(
                    "Port %s nicht erreichbar – überspringe %s",
                    inst["port"], inst["workspace_name"],
                )

        # Berichtsnamen aus PBI Desktop Fenstertiteln zuordnen
        report_names = LocalConnector._get_pbi_window_titles()
        if report_names and instances:
            if len(report_names) <= len(instances):
                for i, name in enumerate(report_names):
                    if i < len(instances):
                        instances[i]["name"] = name

        return instances

    @staticmethod
    def _get_pbi_window_titles() -> list[str]:
        """Ermittelt die Fenstertitel aller PBI Desktop Prozesse.

        Nutzt Win32 API über ctypes — funktioniert auch in
        --windowed PyInstaller-Builds ohne Konsole.
        """
        titles: list[str] = []
        try:
            import ctypes
            import ctypes.wintypes

            user32 = ctypes.windll.user32

            # Alle Fenster durchgehen
            EnumWindows = user32.EnumWindows
            GetWindowTextW = user32.GetWindowTextW
            GetWindowTextLengthW = user32.GetWindowTextLengthW
            GetWindowThreadProcessId = user32.GetWindowThreadProcessId
            IsWindowVisible = user32.IsWindowVisible

            # PBI Desktop PIDs ermitteln
            import subprocess
            try:
                # Versuche über tasklist (funktioniert immer, auch ohne Konsole)
                result = subprocess.run(
                    ["tasklist", "/FI", "IMAGENAME eq PBIDesktop.exe", "/FO", "CSV", "/NH"],
                    capture_output=True, text=True, timeout=5,
                    creationflags=0x08000000,  # CREATE_NO_WINDOW
                )
                pbi_pids: set[int] = set()
                if result.returncode == 0:
                    for line in result.stdout.strip().split("\n"):
                        parts = line.strip().strip('"').split('","')
                        if len(parts) >= 2 and "PBIDesktop" in parts[0]:
                            try:
                                pbi_pids.add(int(parts[1].strip('"')))
                            except ValueError:
                                pass
            except Exception:
                pbi_pids = set()

            if not pbi_pids:
                return titles

            # Callback für EnumWindows
            WNDENUMPROC = ctypes.WINFUNCTYPE(
                ctypes.wintypes.BOOL,
                ctypes.wintypes.HWND,
                ctypes.wintypes.LPARAM,
            )

            def enum_callback(hwnd, _lparam):
                if not IsWindowVisible(hwnd):
                    return True
                pid = ctypes.wintypes.DWORD()
                GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
                if pid.value in pbi_pids:
                    length = GetWindowTextLengthW(hwnd)
                    if length > 0:
                        buf = ctypes.create_unicode_buffer(length + 1)
                        GetWindowTextW(hwnd, buf, length + 1)
                        title = buf.value.strip()
                        # PBI Desktop Titel: "ReportName - Power BI Desktop"
                        if title and "Power BI" not in title.split(" - ")[-1:][0] if " - " not in title else True:
                            # Entferne " - Power BI Desktop" Suffix
                            if " - Power BI Desktop" in title:
                                title = title.rsplit(" - Power BI Desktop", 1)[0]
                            elif " – Power BI Desktop" in title:
                                title = title.rsplit(" – Power BI Desktop", 1)[0]
                            if title:
                                titles.append(title)
                return True

            EnumWindows(WNDENUMPROC(enum_callback), 0)

        except Exception as exc:
            logger.debug("Fenstertitel konnten nicht ermittelt werden: %s", exc)

        return titles

    # ------------------------------------------------------------------
    # DMV query helpers
    # ------------------------------------------------------------------

    def _execute_dmv(self, query: str) -> list[dict[str, Any]]:
        """Execute a DMV query via ADOMD.NET and return rows as list of dicts."""
        if self._connection is None:
            raise PBIConnectionError("Keine aktive Verbindung.")

        from Microsoft.AnalysisServices.AdomdClient import AdomdCommand  # type: ignore

        cmd = AdomdCommand(query, self._connection)
        try:
            reader = cmd.ExecuteReader()
            columns = [reader.GetName(i) for i in range(reader.FieldCount)]
            rows: list[dict[str, Any]] = []
            while reader.Read():
                row = {}
                for i, col in enumerate(columns):
                    val = reader.GetValue(i)
                    # .NET null → Python None
                    if val is not None and str(type(val)) != "<class 'System.DBNull'>":
                        row[col] = val
                    else:
                        row[col] = None
                rows.append(row)
            reader.Close()
            return rows
        except Exception as exc:
            raise PBIConnectionError(f"DMV-Abfrage fehlgeschlagen: {exc}") from exc

    def _safe_dmv(self, query: str, label: str) -> list[dict[str, Any]]:
        """Execute a DMV query, logging warnings on failure instead of raising."""
        try:
            return self._execute_dmv(query)
        except Exception as exc:
            logger.warning("%s-Abfrage fehlgeschlagen: %s", label, exc)
            return []

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

    # SSAS Partition Mode Enum → lesbare Namen
    _PARTITION_MODE_MAP: dict[int, str] = {
        0: "Import",
        1: "DirectQuery",
        2: "Dual",
        3: "Push",
    }

    # SSAS Partition SourceType → Modus-Hinweis
    _PARTITION_SOURCE_MAP: dict[int, str] = {
        1: "Import",       # Query partition (M)
        2: "Calculated",   # Calculated partition
        4: "Import",       # M partition
    }

    def _extract_tables(self) -> list[TableInfo]:
        """Extract tables via TMSCHEMA_TABLES with storage mode from partitions."""
        rows = self._safe_dmv(
            "SELECT [ID], [Name], [Description], [IsHidden] FROM $SYSTEM.TMSCHEMA_TABLES",
            "Tabellen",
        )

        # Speichermodus aus Partitionen holen
        part_rows = self._safe_dmv(
            "SELECT [TableID], [Mode], [SourceType] FROM $SYSTEM.TMSCHEMA_PARTITIONS",
            "Partitionen-Modus",
        )
        table_mode: dict[int, str] = {}
        for pr in part_rows:
            tid = pr.get("TableID", 0)
            if tid in table_mode:
                continue  # Erste Partition gewinnt
            mode_val = pr.get("Mode")
            source_type = pr.get("SourceType")
            if isinstance(mode_val, int) and mode_val in self._PARTITION_MODE_MAP:
                table_mode[tid] = self._PARTITION_MODE_MAP[mode_val]
            elif isinstance(source_type, int) and source_type in self._PARTITION_SOURCE_MAP:
                table_mode[tid] = self._PARTITION_SOURCE_MAP[source_type]

        tables: list[TableInfo] = []
        for row in rows:
            tid = row.get("ID", 0)
            t = TableInfo(
                name=row.get("Name", ""),
                description=row.get("Description", "") or "",
                columns=[],
                is_hidden=bool(row.get("IsHidden", False)),
                row_count=None,
                mode=table_mode.get(tid, "Import"),
            )
            t._table_id = tid  # type: ignore[attr-defined]
            tables.append(t)
        return tables

    # SSAS DataType Enum → lesbare Namen
    _DATA_TYPE_MAP: dict[int, str] = {
        2: "string",
        6: "int64",
        8: "double",
        9: "dateTime",
        10: "decimal",
        11: "boolean",
        17: "binary",
    }

    def _extract_columns(
        self, tables: list[TableInfo]
    ) -> list[ColumnInfo]:
        """Extract columns via TMSCHEMA_COLUMNS and attach to tables."""
        rows = self._safe_dmv(
            "SELECT [ID], [TableID], [ExplicitName], [ExplicitDataType], "
            "[Description], [IsHidden], [Type], [Expression], [IsKey] "
            "FROM $SYSTEM.TMSCHEMA_COLUMNS",
            "Spalten",
        )

        # Table ID → TableInfo Mapping
        table_by_id: dict[int, TableInfo] = {}
        for t in tables:
            tid = getattr(t, "_table_id", None)
            if tid is not None:
                table_by_id[tid] = t

        # Extract synonyms from linguistic metadata
        synonyms_map = self._extract_column_synonyms()

        all_columns: list[ColumnInfo] = []
        for row in rows:
            col_name = row.get("ExplicitName", "") or ""

            # Interne Systemspalten überspringen
            if col_name.startswith("RowNumber"):
                continue

            tid = row.get("TableID", 0)
            table = table_by_id.get(tid)
            table_name = table.name if table else str(tid)

            expression = row.get("Expression") or None
            col_type = row.get("Type", 0)
            is_calculated = col_type == 2 or expression is not None

            # Datentyp-ID in lesbaren Namen umwandeln
            raw_dt = row.get("ExplicitDataType", "")
            if isinstance(raw_dt, int):
                data_type = self._DATA_TYPE_MAP.get(raw_dt, str(raw_dt))
            else:
                data_type = str(raw_dt) if raw_dt else ""

            # isKey aus DMV-Abfrage extrahieren
            is_key = bool(row.get("IsKey", False))

            # Synonyme aus Linguistic Metadata zuordnen
            synonyms = synonyms_map.get((table_name, col_name), [])

            col = ColumnInfo(
                name=col_name,
                table_name=table_name,
                data_type=data_type,
                description=row.get("Description", "") or "",
                is_hidden=bool(row.get("IsHidden", False)),
                is_calculated=is_calculated,
                expression=expression,
                synonyms=synonyms,
                is_key=is_key,
            )
            all_columns.append(col)

            if table:
                table.columns.append(col)

        return all_columns

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

        The linguistic metadata contains an XML document with synonym
        definitions for tables and columns. This method parses that content
        and returns a mapping of (table_name, column_name) → list of synonyms.
        """
        rows = self._safe_dmv(
            "SELECT [Content] FROM $SYSTEM.TMSCHEMA_LINGUISTIC_METADATA",
            "Linguistic Metadata",
        )

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

        for row in rows:
            content = row.get("Content", "") or ""
            if not content:
                continue

            # The linguistic metadata content can be XML or JSON format.
            # Try XML parsing first (most common for PBI Desktop local models).
            try:
                synonyms_map.update(self._parse_linguistic_xml(content))
            except Exception:
                # Try JSON format as fallback
                try:
                    synonyms_map.update(self._parse_linguistic_json(content))
                except Exception as exc:
                    logger.debug(
                        "Linguistic Metadata konnte nicht geparst werden: %s", exc
                    )

        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)
        # Namespace handling – the XML may use a default namespace
        ns = ""
        if root.tag.startswith("{"):
            ns = root.tag.split("}")[0] + "}"

        # Look for Entity elements (tables) containing EntityProperty (columns)
        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] = []
                # Synonyms can be in Word elements within the property
                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)

        # Navigate the JSON structure for entities/columns
        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 TMSCHEMA_MEASURES (legacy, IDs not resolved)."""
        return self._extract_measures_with_names({})

    def _extract_measures_with_names(
        self, table_id_map: dict[int, str]
    ) -> list[MeasureInfo]:
        """Extract measures via TMSCHEMA_MEASURES with table name resolution."""
        rows = self._safe_dmv(
            "SELECT [Name], [TableID], [Expression], [Description], "
            "[FormatString] FROM $SYSTEM.TMSCHEMA_MEASURES",
            "Measures",
        )
        measures: list[MeasureInfo] = []
        for row in rows:
            tid = row.get("TableID", 0)
            table_name = table_id_map.get(tid, str(tid)) if isinstance(tid, int) else str(tid)
            measures.append(MeasureInfo(
                name=row.get("Name", ""),
                table_name=table_name,
                expression=row.get("Expression", "") or "",
                description=row.get("Description", "") or "",
                format_string=row.get("FormatString") or None,
            ))
        return measures

    def _extract_relationships(self) -> list[RelationshipInfo]:
        """Extract relationships (legacy, IDs not resolved)."""
        return self._extract_relationships_with_names({})

    def _extract_relationships_with_names(
        self, table_id_map: dict[int, str]
    ) -> list[RelationshipInfo]:
        """Extract relationships via TMSCHEMA_RELATIONSHIPS with name resolution."""
        rows = self._safe_dmv(
            "SELECT [FromTableID], [FromColumnID], [ToTableID], [ToColumnID], "
            "[FromCardinality], [ToCardinality], [CrossFilteringBehavior], "
            "[IsActive] FROM $SYSTEM.TMSCHEMA_RELATIONSHIPS",
            "Beziehungen",
        )

        # Column-ID → Name Mapping aufbauen
        col_rows = self._safe_dmv(
            "SELECT [ID], [TableID], [ExplicitName] FROM $SYSTEM.TMSCHEMA_COLUMNS",
            "Spalten-IDs",
        )
        col_id_map: dict[tuple[int, int], str] = {}
        for cr in col_rows:
            tid = cr.get("TableID", 0)
            cid = cr.get("ID", 0)
            cname = cr.get("ExplicitName", "") or ""
            if isinstance(tid, int) and isinstance(cid, int):
                col_id_map[(tid, cid)] = cname

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

            from_table = table_id_map.get(from_tid, str(from_tid)) if isinstance(from_tid, int) else str(from_tid)
            to_table = table_id_map.get(to_tid, str(to_tid)) if isinstance(to_tid, int) else str(to_tid)
            from_col = col_id_map.get((from_tid, from_cid), str(from_cid))
            to_col = col_id_map.get((to_tid, to_cid), str(to_cid))

            # Kardinalität aus From/To Cardinality ableiten
            from_card = row.get("FromCardinality", 2)  # 2 = Many
            to_card = row.get("ToCardinality", 1)  # 1 = One
            if from_card == 2 and to_card == 1:
                cardinality = "ManyToOne"
            elif from_card == 1 and to_card == 2:
                cardinality = "OneToMany"
            elif from_card == 1 and to_card == 1:
                cardinality = "OneToOne"
            else:
                cardinality = "ManyToMany"

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

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

    def _extract_rls_roles(self) -> list[RLSRoleInfo]:
        """Extract RLS roles (legacy, without filters)."""
        return self._extract_rls_roles_with_filters({})

    def _extract_rls_roles_with_filters(
        self, table_id_map: dict[int, str]
    ) -> list[RLSRoleInfo]:
        """Extract RLS roles with filter expressions via DMV."""
        role_rows = self._safe_dmv(
            "SELECT [ID], [Name], [Description] FROM $SYSTEM.TMSCHEMA_ROLES",
            "RLS-Rollen",
        )

        # Role-ID → Name Mapping
        role_id_map: dict[int, str] = {}
        roles_by_name: dict[str, RLSRoleInfo] = {}
        for row in role_rows:
            rid = row.get("ID", 0)
            name = row.get("Name", "")
            role_id_map[rid] = name
            roles_by_name[name] = RLSRoleInfo(
                name=name,
                description=row.get("Description", "") or "",
                filter_expressions={},
            )

        # Table Permissions mit Filter-Expressions
        perm_rows = self._safe_dmv(
            "SELECT [RoleID], [TableID], [FilterExpression] "
            "FROM $SYSTEM.TMSCHEMA_TABLE_PERMISSIONS",
            "Tabellenberechtigungen",
        )
        for row in perm_rows:
            rid = row.get("RoleID", 0)
            tid = row.get("TableID", 0)
            expr = row.get("FilterExpression", "") or ""
            role_name = role_id_map.get(rid, "")
            table_name = table_id_map.get(tid, str(tid)) if isinstance(tid, int) else str(tid)
            if role_name and role_name in roles_by_name and expr:
                roles_by_name[role_name].filter_expressions[table_name] = expr

        return list(roles_by_name.values())

    def _extract_data_sources(self) -> list[DataSourceInfo]:
        """Extract data sources via TMSCHEMA_DATA_SOURCES."""
        rows = self._safe_dmv(
            "SELECT * FROM $SYSTEM.TMSCHEMA_DATA_SOURCES", "Datenquellen"
        )
        sources: list[DataSourceInfo] = []
        for row in rows:
            sources.append(DataSourceInfo(
                name=row.get("Name", ""),
                source_type=str(row.get("Type", "Unknown")),
                connection_string=row.get("ConnectionString", "") or "",
                schema=None,
            ))
        return sources

    def _extract_power_queries(self) -> list[PowerQueryInfo]:
        """Extract Power Query M expressions (legacy)."""
        return self._extract_power_queries_with_names({})

    def _extract_power_queries_with_names(
        self, table_id_map: dict[int, str]
    ) -> list[PowerQueryInfo]:
        """Extract Power Query M expressions via TMSCHEMA_PARTITIONS."""
        rows = self._safe_dmv(
            "SELECT [TableID], [Name], [QueryDefinition] "
            "FROM $SYSTEM.TMSCHEMA_PARTITIONS",
            "Power Query",
        )
        from core.connectors.pbip_connector import PBIPConnector

        pq_list: list[PowerQueryInfo] = []
        for row in rows:
            m_expr = row.get("QueryDefinition", "") or ""
            if m_expr:
                tid = row.get("TableID", 0)
                table_name = table_id_map.get(tid, str(tid)) if isinstance(tid, int) else str(tid)
                source_name = PBIPConnector._extract_m_source(m_expr) or "M"
                pq_list.append(PowerQueryInfo(
                    name=row.get("Name", ""),
                    source=source_name,
                    target_table=table_name,
                    m_expression=m_expr,
                    description="",
                ))
        return pq_list
