"""PBIP-Projekt Konnektor – liest Metadaten aus einer PBIP-Projektordnerstruktur."""
from __future__ import annotations

import json
import logging
import re
from pathlib import Path

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

logger = logging.getLogger(__name__)


class PBIPConnector(IConnector):
    """Konnektor für PBIP-Projektordner (dateisystembasiert)."""

    def __init__(self) -> None:
        self._folder_path: Path | None = None
        self._pbip_file: Path | None = None
        self._report_folder: Path | None = None
        self._semantic_model_folder: Path | None = None
        self._definition_folder: Path | None = None
        self._report_name: str = ""

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

    def connect(self, config: ConnectionConfig) -> None:
        """Verbindung herstellen – Ordnerstruktur validieren."""
        # Unterstütze sowohl camelCase (Frontend) als auch snake_case (Tests/Legacy)
        raw_path = (
            config.params.get("folder_path", "")
            or config.params.get("folderPath", "")
        )
        if not raw_path:
            raise PBIPStructureError(
                "Kein Ordnerpfad angegeben."
            )
        folder_path = Path(raw_path)
        if not folder_path.is_dir():
            raise PBIPStructureError(
                f"Der Ordner existiert nicht: {folder_path}"
            )

        # Find .pbip file – suche im gewählten Ordner, dann im Elternordner,
        # dann rekursiv eine Ebene tief (falls ein Unterordner gewählt wurde)
        pbip_files = list(folder_path.glob("*.pbip"))
        if not pbip_files:
            # Prüfe Elternordner (falls User einen Unterordner wie .Report gewählt hat)
            parent = folder_path.parent
            parent_pbip = list(parent.glob("*.pbip"))
            if parent_pbip:
                # Verwende den Elternordner als Projektordner
                folder_path = parent
                pbip_files = parent_pbip
            else:
                # Suche eine Ebene tief (falls User einen übergeordneten Ordner gewählt hat)
                pbip_files = list(folder_path.glob("*/*.pbip"))
                if pbip_files:
                    # Verwende den Ordner der gefundenen .pbip-Datei
                    folder_path = pbip_files[0].parent

        if not pbip_files:
            raise PBIPStructureError(
                f"Keine .pbip-Datei im Ordner gefunden: {folder_path}"
            )
        self._pbip_file = pbip_files[0]
        self._folder_path = folder_path
        self._report_name = self._pbip_file.stem

        # Find Report folder
        report_folders = [
            d for d in folder_path.iterdir()
            if d.is_dir() and d.name.endswith(".Report")
        ]
        self._report_folder = report_folders[0] if report_folders else None

        # Find SemanticModel/definition folder
        sm_folders = [
            d for d in folder_path.iterdir()
            if d.is_dir() and d.name.endswith(".SemanticModel")
        ]
        if sm_folders:
            self._semantic_model_folder = sm_folders[0]
            definition = sm_folders[0] / "definition"
            self._definition_folder = definition if definition.is_dir() else None

        # At least one of report or semantic model must exist
        if self._report_folder is None and self._definition_folder is None:
            raise PBIPStructureError(
                "Keine gültige PBIP-Projektstruktur gefunden: "
                "Weder .Report/ noch .SemanticModel/definition/ vorhanden."
            )

    def extract_metadata(self) -> ReportMetadata:
        """Extrahiert Metadaten aus TMDL-Dateien und report.json."""
        tables: list[TableInfo] = []
        columns: list[ColumnInfo] = []
        measures: list[MeasureInfo] = []
        relationships: list[RelationshipInfo] = []
        rls_roles: list[RLSRoleInfo] = []
        data_sources: list[DataSourceInfo] = []
        power_queries: list[PowerQueryInfo] = []
        visuals: list[VisualInfo] = []

        if self._definition_folder and self._definition_folder.is_dir():
            tables, columns, measures, power_queries, data_sources = (
                self._parse_tables()
            )
            relationships = self._parse_relationships()
            rls_roles = self._parse_roles()

            # Interne PBI-Tabellen herausfiltern
            internal_names = {
                t.name for t in tables if self._is_internal_table(t.name)
            }
            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
            ]
            data_sources = [
                ds for ds in data_sources
                if ds.name not in internal_names
            ]

            # Tabellen als Fakt/Dimension klassifizieren
            self._classify_tables(tables, relationships, measures)

        if self._report_folder and self._report_folder.is_dir():
            visuals = self._parse_report_visuals()

        return ReportMetadata(
            report_name=self._report_name,
            tables=tables,
            columns=columns,
            measures=measures,
            relationships=relationships,
            data_sources=data_sources,
            power_queries=power_queries,
            rls_roles=rls_roles,
            visuals=visuals,
            refresh_schedule=None,
            raw_metadata={},
        )

    @staticmethod
    def _is_internal_table(name: str) -> bool:
        """Prüft ob eine Tabelle eine interne PBI-Systemtabelle ist."""
        return (
            name.startswith("LocalDateTable_")
            or name.startswith("DateTableTemplate_")
        )

    def disconnect(self) -> None:
        """No-op – dateisystembasiert."""

    def test_connection(self) -> bool:
        """Prüft ob der Ordner existiert und eine .pbip-Datei enthält."""
        if self._folder_path is None:
            return False
        if not self._folder_path.is_dir():
            return False
        return bool(list(self._folder_path.glob("*.pbip")))


    # ------------------------------------------------------------------
    # TMDL table parsing
    # ------------------------------------------------------------------

    def _parse_tables(
        self,
    ) -> tuple[
        list[TableInfo],
        list[ColumnInfo],
        list[MeasureInfo],
        list[PowerQueryInfo],
        list[DataSourceInfo],
    ]:
        """Parse alle Tabellen-TMDL-Dateien."""
        tables: list[TableInfo] = []
        all_columns: list[ColumnInfo] = []
        all_measures: list[MeasureInfo] = []
        all_pq: list[PowerQueryInfo] = []
        all_ds: list[DataSourceInfo] = []

        assert self._definition_folder is not None
        tables_dir = self._definition_folder / "tables"
        if not tables_dir.is_dir():
            # Try model.tmdl as fallback
            model_file = self._definition_folder / "model.tmdl"
            if model_file.is_file():
                content = self._safe_read(model_file)
                if content is not None:
                    t, c, m, pq, ds = self._parse_table_tmdl(
                        content, model_file.stem
                    )
                    tables.extend(t)
                    all_columns.extend(c)
                    all_measures.extend(m)
                    all_pq.extend(pq)
                    all_ds.extend(ds)
            return tables, all_columns, all_measures, all_pq, all_ds

        for tmdl_file in sorted(tables_dir.glob("*.tmdl")):
            content = self._safe_read(tmdl_file)
            if content is None:
                continue
            t, c, m, pq, ds = self._parse_table_tmdl(
                content, tmdl_file.stem
            )
            tables.extend(t)
            all_columns.extend(c)
            all_measures.extend(m)
            all_pq.extend(pq)
            all_ds.extend(ds)

        return tables, all_columns, all_measures, all_pq, all_ds

    def _parse_table_tmdl(
        self, content: str, default_name: str
    ) -> tuple[
        list[TableInfo],
        list[ColumnInfo],
        list[MeasureInfo],
        list[PowerQueryInfo],
        list[DataSourceInfo],
    ]:
        """Parse a single TMDL file for table info."""
        # Extract table name
        table_name = default_name
        table_match = re.search(
            r"^table\s+'([^']+)'", content, re.MULTILINE
        )
        if table_match:
            table_name = table_match.group(1)

        # Extract table-level properties (single-tab indented, before any
        # column/measure/partition blocks which are also single-tab indented)
        table_props = self._extract_table_level_props(content)

        # Check hidden at table level only
        is_hidden = bool(
            re.search(r"^\tisHidden\s*$", table_props, re.MULTILINE)
        ) or bool(
            re.search(r"^\tisHidden\s*:\s*true", table_props, re.MULTILINE | re.IGNORECASE)
        )

        # Extract description at table level
        description = ""
        desc_match = re.search(
            r"^\tdescription\s*:\s*(.+)$", table_props, re.MULTILINE
        )
        if not desc_match:
            desc_match = re.search(
                r"^\tdescription\s*$\s+'''(.+?)'''",
                table_props,
                re.MULTILINE | re.DOTALL,
            )
        if desc_match:
            description = desc_match.group(1).strip()

        columns = self._extract_columns(content, table_name)
        measures = self._extract_measures(content, table_name)
        pq, ds = self._extract_partitions(content, table_name)

        # Speichermodus aus Partition extrahieren
        mode = self._extract_storage_mode(content)

        table = TableInfo(
            name=table_name,
            description=description,
            columns=columns,
            is_hidden=is_hidden,
            row_count=None,
            mode=mode,
        )
        return [table], columns, measures, pq, ds


    # ------------------------------------------------------------------
    # Column extraction
    # ------------------------------------------------------------------

    def _extract_columns(
        self, content: str, table_name: str
    ) -> list[ColumnInfo]:
        """Extract columns from TMDL content."""
        columns: list[ColumnInfo] = []
        # Match column blocks: column 'Name' or column Name
        col_pattern = re.compile(
            r"^[ \t]+column\s+'?([^'\n]+?)'?\s*$",
            re.MULTILINE,
        )
        for match in col_pattern.finditer(content):
            col_name = match.group(1).strip()
            # Extract the block after this column declaration
            block = self._extract_block(content, match.start())

            data_type = ""
            dt_match = re.search(r"dataType\s*:\s*(\S+)", block)
            if dt_match:
                data_type = dt_match.group(1).strip()

            is_hidden = bool(
                re.search(r"^\s*isHidden\s*$", block, re.MULTILINE)
            ) or bool(
                re.search(r"^\s*isHidden\s*:\s*true", block, re.MULTILINE | re.IGNORECASE)
            )

            expression = None
            expr_match = re.search(
                r"expression\s*:\s*(.+?)$", block, re.MULTILINE
            )
            if not expr_match:
                expr_match = re.search(
                    r"expression\s*$\s+'''(.+?)'''",
                    block,
                    re.MULTILINE | re.DOTALL,
                )
            if expr_match:
                expression = expr_match.group(1).strip()

            is_calculated = expression is not None

            col_desc = ""
            cd_match = re.search(
                r"description\s*:\s*(.+?)$", block, re.MULTILINE
            )
            if cd_match:
                col_desc = cd_match.group(1).strip()

            # Extract synonyms sub-block (one synonym per line, quoted)
            synonyms: list[str] = []
            syn_match = re.search(
                r"^\s+synonyms\s*$", block, re.MULTILINE
            )
            if syn_match:
                syn_block = self._extract_block(block, syn_match.start())
                for syn_line in syn_block.split("\n")[1:]:
                    stripped = syn_line.strip()
                    if stripped:
                        # Remove surrounding quotes if present
                        synonym = stripped.strip("'\"")
                        if synonym:
                            synonyms.append(synonym)

            # Extract isKey (TMDL convention: presence means true)
            is_key = bool(
                re.search(r"^\s*isKey\s*$", block, re.MULTILINE)
            ) or bool(
                re.search(r"^\s*isKey\s*:\s*true", block, re.MULTILINE | re.IGNORECASE)
            )

            columns.append(
                ColumnInfo(
                    name=col_name,
                    table_name=table_name,
                    data_type=data_type,
                    description=col_desc,
                    is_hidden=is_hidden,
                    is_calculated=is_calculated,
                    expression=expression,
                    synonyms=synonyms,
                    is_key=is_key,
                )
            )
        return columns

    # ------------------------------------------------------------------
    # Measure extraction
    # ------------------------------------------------------------------

    def _extract_measures(
        self, content: str, table_name: str
    ) -> list[MeasureInfo]:
        """Extract measures from TMDL content."""
        measures: list[MeasureInfo] = []
        # Measures: measure 'Name' = <expression>
        # or multi-line with ```
        measure_pattern = re.compile(
            r"^[ \t]+measure\s+'([^']+)'\s*=\s*(.+?)$",
            re.MULTILINE,
        )
        for match in measure_pattern.finditer(content):
            m_name = match.group(1).strip()
            expr_start = match.group(2).strip()

            block = self._extract_block(content, match.start())

            # Build full expression – may span multiple lines with ```
            if expr_start.startswith("```"):
                # Multi-line expression
                expr_block = re.search(
                    r"```\s*\n(.*?)```", block, re.DOTALL
                )
                expression = (
                    expr_block.group(1).strip() if expr_block else expr_start
                )
            else:
                expression = expr_start

            fmt_string = None
            fmt_match = re.search(
                r"formatString\s*:\s*(.+?)$", block, re.MULTILINE
            )
            if fmt_match:
                fmt_string = fmt_match.group(1).strip()

            m_desc = ""
            md_match = re.search(
                r"description\s*:\s*(.+?)$", block, re.MULTILINE
            )
            if not md_match:
                md_match = re.search(
                    r"description\s*$\s+'''(.+?)'''",
                    block,
                    re.MULTILINE | re.DOTALL,
                )
            if md_match:
                m_desc = md_match.group(1).strip()

            measures.append(
                MeasureInfo(
                    name=m_name,
                    table_name=table_name,
                    expression=expression,
                    description=m_desc,
                    format_string=fmt_string,
                )
            )
        return measures


    # ------------------------------------------------------------------
    # Partition / Power Query / Data Source extraction
    # ------------------------------------------------------------------

    def _extract_partitions(
        self, content: str, table_name: str
    ) -> tuple[list[PowerQueryInfo], list[DataSourceInfo]]:
        """Extract partition blocks for Power Query M expressions."""
        pq_list: list[PowerQueryInfo] = []
        ds_list: list[DataSourceInfo] = []

        # partition blocks — unterstützt Tab- und Space-Einrückung
        part_pattern = re.compile(
            r"^[ \t]+partition\s+'?([^'\n]+?)'?\s*=\s*(\S+)",
            re.MULTILINE,
        )
        for match in part_pattern.finditer(content):
            part_name = match.group(1).strip()
            source_type = match.group(2).strip()
            block = self._extract_block(content, match.start())

            # Extract M expression — mehrere Formate unterstützen
            m_expr = self._extract_m_expression(block)

            if m_expr:
                # Ursprungsquelle aus der M-Abfrage extrahieren
                source_name = self._extract_m_source(m_expr) or source_type

                pq_list.append(
                    PowerQueryInfo(
                        name=part_name,
                        source=source_name,
                        target_table=table_name,
                        m_expression=m_expr,
                        description="",
                    )
                )

                # Try to derive a data source from the M expression
                ds = self._derive_data_source(part_name, m_expr)
                if ds:
                    ds_list.append(ds)

        return pq_list, ds_list

    @staticmethod
    def _extract_m_expression(block: str) -> str:
        """Extrahiert den M-Ausdruck aus einem Partition-Block.

        Unterstützt mehrere TMDL-Formate:
        1. expression\\n'''...'''  (älteres Format)
        2. expression: ...         (einzeilig)
        3. source =\\n  let...     (neueres PBIP-Format)
        """
        # Format 1: expression mit triple-quotes
        m = re.search(
            r"expression\s*$\s+'''(.*?)'''",
            block, re.MULTILINE | re.DOTALL,
        )
        if m:
            return m.group(1).strip()

        # Format 2: expression einzeilig
        m = re.search(r"expression\s*:\s*(.+?)$", block, re.MULTILINE)
        if m:
            return m.group(1).strip()

        # Format 3: source = \n <eingerückter M-Code> (PBIP-Format)
        m = re.search(r"source\s*=\s*$", block, re.MULTILINE)
        if m:
            # Alles nach "source =" bis zum Ende des eingerückten Blocks
            after_source = block[m.end():]
            lines = after_source.split("\n")
            m_lines: list[str] = []
            for line in lines:
                stripped = line.strip()
                if not stripped:
                    if m_lines:
                        # Leere Zeile nach M-Code = Ende
                        break
                    continue
                # Eingerückte Zeilen gehören zum M-Code
                if line.startswith((" ", "\t")):
                    m_lines.append(stripped)
                elif m_lines:
                    break
            if m_lines:
                return "\n".join(m_lines)

        return ""

    @staticmethod
    def _extract_storage_mode(content: str) -> str:
        """Extrahiert den Speichermodus aus einer TMDL-Datei.

        Sucht nach 'mode: import', 'mode: directQuery', 'mode: dual'
        in Partition-Blöcken. Falls 'calculated' als Partitionstyp
        vorkommt, ist es eine berechnete Tabelle.
        """
        # Prüfe auf calculated partition
        if re.search(r"partition\s+.+?=\s*calculated", content, re.MULTILINE):
            return "Calculated"

        # Suche nach mode: in Partition-Blöcken
        mode_match = re.search(
            r"mode\s*:\s*(import|directQuery|dual|default)",
            content, re.IGNORECASE,
        )
        if mode_match:
            mode_raw = mode_match.group(1).lower()
            mode_map = {
                "import": "Import",
                "directquery": "DirectQuery",
                "dual": "Dual",
                "default": "Import",
            }
            return mode_map.get(mode_raw, mode_raw)

        # Fallback: wenn partition = m vorhanden → Import
        if re.search(r"partition\s+.+?=\s*m\b", content, re.MULTILINE):
            return "Import"

        return ""

    @staticmethod
    def _classify_tables(
        tables: list[TableInfo],
        relationships: list[RelationshipInfo],
        measures: list[MeasureInfo],
    ) -> None:
        """Klassifiziert Tabellen als Fakt-, Dimensions- oder Hilfstabelle.

        Heuristik:
        - Tabelle mit 'fact' oder 'fakt' im Namen → Faktentabelle
        - Tabelle mit 'dim' im Namen → Dimensionstabelle
        - Tabelle die auf der Many-Seite von Beziehungen steht → Faktentabelle
        - Tabelle die auf der One-Seite steht → Dimensionstabelle
        - Tabelle mit Measures aber ohne Datenpartition → Hilfstabelle
        - Versteckte Tabellen → Hilfstabelle
        - LocalDateTable / DateTableTemplate → Hilfstabelle
        """
        # Sammle Beziehungsinformationen
        many_side_tables: set[str] = set()
        one_side_tables: set[str] = set()
        for rel in relationships:
            if "Many" in rel.cardinality or "many" in rel.cardinality.lower():
                # fromTable ist die Many-Seite bei ManyToOne
                if "ManyToOne" in rel.cardinality:
                    many_side_tables.add(rel.from_table)
                    one_side_tables.add(rel.to_table)
                elif "OneToMany" in rel.cardinality:
                    many_side_tables.add(rel.to_table)
                    one_side_tables.add(rel.from_table)

        # Tabellen mit Measures
        measure_tables = {m.table_name for m in measures}

        for table in tables:
            name_lower = table.name.lower()

            # Hilfstabellen erkennen
            if (table.is_hidden
                or "LocalDateTable" in table.name
                or "DateTableTemplate" in table.name
                or table.name.startswith("_")):
                table.table_type = "Hilfstabelle"
                continue

            # Namensbasierte Erkennung
            if "fact" in name_lower or "fakt" in name_lower:
                table.table_type = "Faktentabelle"
                continue
            if "dim" in name_lower:
                table.table_type = "Dimensionstabelle"
                continue

            # Reine Measure-Tabelle (keine Spalten mit sourceColumn)
            if (table.name in measure_tables
                and not any(not c.is_calculated for c in table.columns)):
                table.table_type = "Hilfstabelle (Measures)"
                continue

            # Beziehungsbasierte Erkennung
            if table.name in many_side_tables and table.name not in one_side_tables:
                table.table_type = "Faktentabelle"
                continue
            if table.name in one_side_tables and table.name not in many_side_tables:
                table.table_type = "Dimensionstabelle"
                continue

            # Fallback
            if table.name in many_side_tables:
                table.table_type = "Faktentabelle"
            elif table.name in one_side_tables:
                table.table_type = "Dimensionstabelle"
            else:
                table.table_type = "Tabelle"

    @staticmethod
    def _derive_data_source(name: str, m_expr: str) -> DataSourceInfo | None:
        """Try to derive a DataSourceInfo from an M expression."""
        # Common patterns in M expressions
        source_type = "Unknown"
        connection_string = ""

        if "Sql.Database" in m_expr:
            source_type = "SQL"
            cs_match = re.search(
                r'Sql\.Database\(\s*"([^"]+)"', m_expr
            )
            if cs_match:
                connection_string = cs_match.group(1)
        elif "OData.Feed" in m_expr:
            source_type = "OData"
            cs_match = re.search(r'OData\.Feed\(\s*"([^"]+)"', m_expr)
            if cs_match:
                connection_string = cs_match.group(1)
        elif "SharePoint" in m_expr:
            source_type = "SharePoint"
            cs_match = re.search(r'"(https?://[^"]+)"', m_expr)
            if cs_match:
                connection_string = cs_match.group(1)
        elif "Excel.Workbook" in m_expr:
            source_type = "Excel"
            cs_match = re.search(r'"([^"]+\.xlsx?)"', m_expr)
            if cs_match:
                connection_string = cs_match.group(1)
        elif "Csv.Document" in m_expr:
            source_type = "CSV"
        elif "Web.Contents" in m_expr:
            source_type = "Web"
            cs_match = re.search(r'"(https?://[^"]+)"', m_expr)
            if cs_match:
                connection_string = cs_match.group(1)
        else:
            # No recognisable source – skip
            return None

        return DataSourceInfo(
            name=name,
            source_type=source_type,
            connection_string=connection_string,
        )

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

        Sucht nach dem ersten Funktionsaufruf (Source-Schritt) in der
        M-Abfrage und gibt eine lesbare Beschreibung zurück.

        Beispiele:
          Sql.Database("server", "db") → "SQL: server / db"
          OData.Feed("https://...") → "OData: https://..."
          SharePoint.Tables("https://...") → "SharePoint: https://..."
          Excel.Workbook(...) → "Excel: datei.xlsx"
        """
        # 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)}"

        # CSV
        if "Csv.Document" in m_expr:
            return "CSV"

        # Oracle
        m = re.search(
            r'Oracle\.Database\(\s*"([^"]+)"', m_expr
        )
        if m:
            return f"Oracle: {m.group(1)}"

        # ODBC
        m = re.search(r'Odbc\.DataSource\(\s*"([^"]+)"', m_expr)
        if m:
            return f"ODBC: {m.group(1)}"

        # Analysis Services
        m = re.search(
            r'AnalysisServices\.Database\(\s*"([^"]+)"', m_expr
        )
        if m:
            return f"SSAS: {m.group(1)}"

        # Dataverse / CDS
        m = re.search(r'CommonDataService\.Database\(\s*"([^"]+)"', m_expr)
        if m:
            return f"Dataverse: {m.group(1)}"

        # Generischer Fallback: ersten Funktionsaufruf mit Punkt extrahieren
        m = re.search(r'(\w+\.\w+)\(', m_expr)
        if m:
            return m.group(1)

        return None

    # ------------------------------------------------------------------
    # Relationship parsing
    # ------------------------------------------------------------------

    def _parse_relationships(self) -> list[RelationshipInfo]:
        """Parse relationships.tmdl."""
        assert self._definition_folder is not None
        rel_file = self._definition_folder / "relationships.tmdl"
        if not rel_file.is_file():
            return []

        content = self._safe_read(rel_file)
        if content is None:
            return []

        relationships: list[RelationshipInfo] = []
        rel_pattern = re.compile(
            r"^[ \t]*relationship\s+(\S+)", re.MULTILINE
        )
        for match in rel_pattern.finditer(content):
            block = self._extract_block(content, match.start())

            # Format: fromColumn: Table.Column oder 'Table Name'.Column
            from_table, from_col = self._parse_table_column_ref(
                self._get_field(block, "fromColumn")
            )
            to_table, to_col = self._parse_table_column_ref(
                self._get_field(block, "toColumn")
            )

            # Fallback: separate fromTable/toTable Felder
            if not from_table:
                from_table = self._get_field(block, "fromTable")
            if not to_table:
                to_table = self._get_field(block, "toTable")

            cardinality = self._get_field(block, "cardinality") or "OneToMany"
            cross_filter = self._get_field(block, "crossFilteringBehavior") or "Single"
            is_active = "isActive" not in block.lower() or bool(
                re.search(r"isActive\s*:\s*true", block, re.IGNORECASE)
            )

            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=is_active,
                )
            )
        return relationships

    @staticmethod
    def _parse_table_column_ref(ref: str) -> tuple[str, str]:
        """Parst eine Tabelle.Spalte-Referenz aus TMDL.

        Formate:
          - Table.Column → ("Table", "Column")
          - 'Table Name'.Column → ("Table Name", "Column")
          - 'Table Name'.'Column Name' → ("Table Name", "Column Name")
        """
        if not ref:
            return "", ""

        # Format: 'Table Name'.Column oder 'Table Name'.'Column Name'
        m = re.match(r"'([^']+)'\.(?:'([^']+)'|(\S+))", ref)
        if m:
            table = m.group(1)
            col = m.group(2) or m.group(3) or ""
            return table, col

        # Format: Table.Column (ohne Anführungszeichen)
        if "." in ref:
            parts = ref.split(".", 1)
            return parts[0].strip("'"), parts[1].strip("'")

        return "", ref


    # ------------------------------------------------------------------
    # RLS role parsing
    # ------------------------------------------------------------------

    def _parse_roles(self) -> list[RLSRoleInfo]:
        """Parse roles/*.tmdl files."""
        assert self._definition_folder is not None
        roles_dir = self._definition_folder / "roles"
        if not roles_dir.is_dir():
            return []

        roles: list[RLSRoleInfo] = []
        for tmdl_file in sorted(roles_dir.glob("*.tmdl")):
            content = self._safe_read(tmdl_file)
            if content is None:
                continue

            role_name = tmdl_file.stem
            rn_match = re.search(
                r"^role\s+'([^']+)'", content, re.MULTILINE
            )
            if rn_match:
                role_name = rn_match.group(1)

            description = ""
            desc_match = re.search(
                r"description\s*:\s*(.+?)$", content, re.MULTILINE
            )
            if desc_match:
                description = desc_match.group(1).strip()

            # Extract filter expressions per table
            filters: dict[str, str] = {}

            # Format 1: tablePermission 'TableName' mit filterExpression
            tp_pattern = re.compile(
                r"tablePermission\s+'([^']+)'", re.MULTILINE
            )
            for tp_match in tp_pattern.finditer(content):
                tbl = tp_match.group(1)
                tp_block = self._extract_block(content, tp_match.start())
                fe_match = re.search(
                    r"filterExpression\s*:\s*(.+?)$",
                    tp_block,
                    re.MULTILINE,
                )
                if not fe_match:
                    fe_match = re.search(
                        r"filterExpression\s*$\s+'''(.+?)'''",
                        tp_block,
                        re.MULTILINE | re.DOTALL,
                    )
                if fe_match:
                    filters[tbl] = fe_match.group(1).strip()

            # Format 2: tablePermission TableName = ``` ... ```
            tp_pattern2 = re.compile(
                r"tablePermission\s+([^\s=]+)\s*=\s*```(.*?)```",
                re.DOTALL,
            )
            for tp_match in tp_pattern2.finditer(content):
                tbl = tp_match.group(1).strip().strip("'")
                expr = tp_match.group(2).strip()
                if tbl and expr:
                    filters[tbl] = expr

            roles.append(
                RLSRoleInfo(
                    name=role_name,
                    description=description,
                    filter_expressions=filters,
                )
            )
        return roles

    # ------------------------------------------------------------------
    # Report / visual parsing
    # ------------------------------------------------------------------

    def _parse_report_visuals(self) -> list[VisualInfo]:
        """Parse report.json or PBIR structure for visuals."""
        assert self._report_folder is not None
        visuals: list[VisualInfo] = []

        report_json = self._report_folder / "report.json"
        if report_json.is_file():
            content = self._safe_read(report_json)
            if content is not None:
                visuals = self._parse_report_json(content)
                if visuals:
                    return visuals

        # Fallback: PBIR folder structure (definition/pages/...)
        pages_dir = self._report_folder / "definition" / "pages"
        if pages_dir.is_dir():
            visuals = self._parse_pbir_pages(pages_dir)

        return visuals

    def _parse_report_json(self, content: str) -> list[VisualInfo]:
        """Parse a report.json for visual information."""
        visuals: list[VisualInfo] = []
        try:
            data = json.loads(content)
        except json.JSONDecodeError:
            logger.warning("report.json konnte nicht als JSON geparst werden.")
            return visuals

        # report.json can have different structures; handle common ones
        sections = data.get("sections", data.get("pages", []))
        if isinstance(sections, list):
            for section in sections:
                page_name = section.get("displayName", section.get("name", ""))
                visual_containers = section.get(
                    "visualContainers", section.get("visuals", [])
                )
                if isinstance(visual_containers, list):
                    for vc in visual_containers:
                        vis = self._parse_visual_container(vc, page_name)
                        if vis:
                            visuals.append(vis)
        return visuals

    @staticmethod
    def _parse_visual_container(
        vc: dict, page_name: str
    ) -> VisualInfo | None:
        """Parse a single visual container from report.json."""
        # The config is often a JSON string inside the container
        config_raw = vc.get("config", "{}")
        if isinstance(config_raw, str):
            try:
                config = json.loads(config_raw)
            except json.JSONDecodeError:
                config = {}
        else:
            config = config_raw if isinstance(config_raw, dict) else {}

        single_visual = config.get("singleVisual", config)
        visual_type = single_visual.get(
            "visualType", single_visual.get("type", "unknown")
        )
        title = None
        title_obj = single_visual.get("vcObjects", {}).get("title", [{}])
        if isinstance(title_obj, list) and title_obj:
            props = title_obj[0].get("properties", {})
            text_obj = props.get("text", {})
            if isinstance(text_obj, dict):
                expr = text_obj.get("expr", {})
                title = expr.get("Literal", {}).get("Value", None)
                if title and title.startswith("'") and title.endswith("'"):
                    title = title[1:-1]

        # Extract field names
        fields: list[str] = []
        projections = single_visual.get("projections", {})
        if isinstance(projections, dict):
            for _role, items in projections.items():
                if isinstance(items, list):
                    for item in items:
                        if isinstance(item, dict):
                            qr = item.get("queryRef", "")
                            if qr:
                                fields.append(qr)

        return VisualInfo(
            page_name=page_name,
            visual_type=visual_type,
            title=title,
            fields=fields,
        )

    @staticmethod
    def _extract_visual_title(visual_obj: dict) -> str:
        """Extrahiert den Anzeigenamen eines Visuals.

        Sucht in mehreren Stellen:
        1. vcObjects.title[*].properties.text.expr.Literal.Value
        2. objects.header[*].properties.text.expr.Literal.Value
        3. objects.title[*].properties.text.expr.Literal.Value
        4. visual.title (direktes Feld)
        """
        # Methode 1: vcObjects.title
        for container_key in ("vcObjects", "objects", "visualContainerObjects"):
            container = visual_obj.get(container_key, {})
            for title_key in ("title", "header"):
                title_list = container.get(title_key, [])
                if not isinstance(title_list, list):
                    continue
                for t in title_list:
                    if not isinstance(t, dict):
                        continue
                    props = t.get("properties", {})
                    text = props.get("text", {})
                    if isinstance(text, dict):
                        expr = text.get("expr", {})
                        lit = expr.get("Literal", {})
                        val = lit.get("Value", "")
                        if val:
                            return val.strip("'\"")

        # Methode 2: Direktes title-Feld
        title = visual_obj.get("title")
        if isinstance(title, str) and title:
            return title

        return ""

    @staticmethod
    def _extract_visual_fields(visual_obj: dict) -> list[str]:
        """Extrahiert Feldnamen aus einem PBIR visual-Objekt.

        Sucht in query.queryState.*.projections[*].field für
        Column (Entity.Property) und Measure Referenzen.
        """
        fields: list[str] = []
        query = visual_obj.get("query", {})
        query_state = query.get("queryState", {})

        for _role, role_data in query_state.items():
            if not isinstance(role_data, dict):
                continue
            projections = role_data.get("projections", [])
            if not isinstance(projections, list):
                continue
            for proj in projections:
                if not isinstance(proj, dict):
                    continue
                # queryRef ist der einfachste Weg
                qr = proj.get("queryRef", "")
                if qr:
                    fields.append(qr)
                    continue
                # Fallback: field.Column.Expression.SourceRef.Entity + Property
                field_obj = proj.get("field", {})
                col = field_obj.get("Column", {})
                if col:
                    entity = col.get("Expression", {}).get("SourceRef", {}).get("Entity", "")
                    prop = col.get("Property", "")
                    if entity and prop:
                        fields.append(f"{entity}.{prop}")
                    elif prop:
                        fields.append(prop)
                # Measure
                measure = field_obj.get("Measure", {})
                if measure:
                    entity = measure.get("Expression", {}).get("SourceRef", {}).get("Entity", "")
                    prop = measure.get("Property", "")
                    if entity and prop:
                        fields.append(f"{entity}.{prop}")
                    elif prop:
                        fields.append(prop)

        return fields

    def _parse_pbir_pages(self, pages_dir: Path) -> list[VisualInfo]:
        """Parse PBIR page folders for visuals."""
        visuals: list[VisualInfo] = []
        for page_dir in sorted(pages_dir.iterdir()):
            if not page_dir.is_dir():
                continue
            page_name = page_dir.name
            # Try page.json for display name
            page_json = page_dir / "page.json"
            if page_json.is_file():
                pj_content = self._safe_read(page_json)
                if pj_content:
                    try:
                        pj = json.loads(pj_content)
                        page_name = pj.get("displayName", page_name)
                    except json.JSONDecodeError:
                        pass

            visuals_dir = page_dir / "visuals"
            if not visuals_dir.is_dir():
                continue
            for vis_dir in sorted(visuals_dir.iterdir()):
                if not vis_dir.is_dir():
                    continue
                vis_json = vis_dir / "visual.json"
                if vis_json.is_file():
                    vj_content = self._safe_read(vis_json)
                    if vj_content:
                        try:
                            vj = json.loads(vj_content)
                            visual_obj = vj.get("visual", {})
                            visual_type = visual_obj.get("visualType", "unknown")
                            vis_id = vj.get("name", vis_dir.name)
                            # Anzeigename aus vcObjects.title extrahieren
                            display_name = self._extract_visual_title(visual_obj)
                            fields = self._extract_visual_fields(visual_obj)
                            visuals.append(
                                VisualInfo(
                                    page_name=page_name,
                                    visual_type=visual_type,
                                    title=vis_id,
                                    display_name=display_name,
                                    fields=fields,
                                )
                            )
                        except json.JSONDecodeError:
                            logger.warning(
                                "visual.json nicht lesbar: %s", vis_json
                            )
        return visuals


    # ------------------------------------------------------------------
    # Helpers
    # ------------------------------------------------------------------

    def _safe_read(self, path: Path) -> str | None:
        """Read a file, returning None and logging a warning on failure."""
        try:
            return path.read_text(encoding="utf-8")
        except Exception as exc:
            logger.warning(
                "Datei konnte nicht gelesen werden: %s – %s", path, exc
            )
            # Log via PBIPFileReadError but don't abort
            _ = PBIPFileReadError(f"Datei nicht lesbar: {path} – {exc}")
            return None

    @staticmethod
    def _extract_table_level_props(content: str) -> str:
        """Extract only the table-level properties (before first column/measure/partition)."""
        lines = content.split("\n")
        result: list[str] = []
        for line in lines:
            stripped = line.lstrip()
            # Stop at the first column, measure, or partition block
            if line != line.lstrip() and stripped == line.lstrip():
                keyword = stripped.split(" ")[0].split("'")[0].strip()
                if keyword in ("column", "measure", "partition"):
                    break
            result.append(line)
        return "\n".join(result)

    @staticmethod
    def _extract_block(content: str, start: int) -> str:
        """Extract an indented block starting from *start*.

        Returns text from *start* up to the next line at the same or lower
        indentation level (or end of file).
        Supports both tab and space indentation.
        """
        lines = content[start:].split("\n")
        if not lines:
            return ""
        # First line determines base indent (count leading whitespace)
        first_line = lines[0]
        base_indent = len(first_line) - len(first_line.lstrip())
        block_lines = [first_line]
        for line in lines[1:]:
            if line.strip() == "":
                block_lines.append(line)
                continue
            indent = len(line) - len(line.lstrip())
            if indent <= base_indent:
                break
            block_lines.append(line)
        return "\n".join(block_lines)

    @staticmethod
    def _get_field(block: str, field_name: str) -> str:
        """Extract a simple field value from a TMDL block."""
        match = re.search(
            rf"{field_name}\s*:\s*'?([^'\n]+)'?", block, re.MULTILINE
        )
        return match.group(1).strip() if match else ""
