MicroStratey Extract

microstratey_extract.py

"""
MicroStrategy metadata lineage crawler.

Produces:
    output/
        objects.jsonl
        definitions/
        lineage.csv
        errors.csv

Target lineage:
    report/cube/datamart
        -> metric
        -> nested metric
        -> fact (+ folder path)
        -> nested fact (+ folder path)
        -> column
        -> table

Requirements:
    pip install requests

Environment variables:
    MSTR_BASE_URL
    MSTR_USERNAME
    MSTR_PASSWORD
    MSTR_PROJECT_ID

Example:
    set MSTR_BASE_URL=https://server/MicroStrategyLibrary
    set MSTR_USERNAME=my_user
    set MSTR_PASSWORD=my_password
    set MSTR_PROJECT_ID=PROJECT_GUID

    python extract_mstr_lineage.py
"""

from __future__ import annotations

import csv
import json
import logging
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

BASE_URL = os.environ["MSTR_BASE_URL"].rstrip("/")
USERNAME = os.environ["MSTR_USERNAME"]
PASSWORD = os.environ["MSTR_PASSWORD"]
PROJECT_ID = os.environ["MSTR_PROJECT_ID"]

OUTPUT_DIR = Path("output")
DEFINITION_DIR = OUTPUT_DIR / "definitions"

REQUEST_TIMEOUT_SECONDS = 120
REQUEST_DELAY_SECONDS = 0.05

# MicroStrategy object type 3 is Report. Cubes and datamarts are report
# subtypes, so they are commonly returned by the report object search.
REPORT_OBJECT_TYPE = 3

# MicroStrategy object type for Fact (EnumDSSXMLObjectTypes.DssTypeFact).
# Used to resolve the fact's folder path via /api/objects/{id}.
# ponytail: hard-coded object type. If a future MSTR release changes the
# numeric type, folder-path resolution degrades to None (row still emitted)
# rather than failing the crawl.
FACT_OBJECT_TYPE = 13


# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class ObjectRef:
    object_id: str
    name: str | None
    subtype: str | None
    object_type: str | None = None


@dataclass(frozen=True)
class Dependency:
    object_id: str
    name: str | None
    subtype: str | None
    path: str


@dataclass(frozen=True)
class FolderInfo:
    folder_id: str | None
    folder_name: str | None
    folder_path: str | None


# ---------------------------------------------------------------------------
# REST client
# ---------------------------------------------------------------------------

class MicroStrategyClient:
    def __init__(
        self,
        base_url: str,
        username: str,
        password: str,
        project_id: str,
    ) -> None:
        self.base_url = base_url
        self.username = username
        self.password = password
        self.project_id = project_id
        self.session = self._create_session()
        self.auth_token: str | None = None

    @staticmethod
    def _create_session() -> requests.Session:
        session = requests.Session()

        retries = Retry(
            total=5,
            connect=5,
            read=5,
            status=5,
            backoff_factor=1,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset({"GET", "POST"}),
            raise_on_status=False,
        )

        adapter = HTTPAdapter(max_retries=retries)
        session.mount("https://", adapter)
        session.mount("http://", adapter)

        return session

    def login(self) -> None:
        response = self.session.post(
            f"{self.base_url}/api/auth/login",
            json={
                "username": self.username,
                "password": self.password,
                "loginMode": 1,
            },
            headers={
                "Accept": "application/json",
                "Content-Type": "application/json",
            },
            timeout=REQUEST_TIMEOUT_SECONDS,
        )
        response.raise_for_status()

        token = (
            response.headers.get("X-MSTR-AuthToken")
            or response.headers.get("x-mstr-authtoken")
        )

        if not token:
            raise RuntimeError(
                "Authentication succeeded, but X-MSTR-AuthToken "
                "was not returned."
            )

        self.auth_token = token

    @property
    def headers(self) -> dict[str, str]:
        if not self.auth_token:
            raise RuntimeError("Call login() before making API requests.")

        return {
            "Accept": "application/json",
            "X-MSTR-AuthToken": self.auth_token,
            "X-MSTR-ProjectID": self.project_id,
        }

    def get_json(
        self,
        endpoint: str,
        params: dict[str, Any] | None = None,
    ) -> dict[str, Any] | list[Any]:
        time.sleep(REQUEST_DELAY_SECONDS)

        response = self.session.get(
            f"{self.base_url}{endpoint}",
            headers=self.headers,
            params=params,
            timeout=REQUEST_TIMEOUT_SECONDS,
        )

        response.raise_for_status()

        if not response.content:
            return {}

        return response.json()

    def search_report_objects(self) -> list[ObjectRef]:
        """
        Retrieves all objects whose top-level object type is Report.

        Cubes and datamarts are report subtypes in MicroStrategy metadata,
        so subtype routing happens after discovery.
        """
        payload = self.get_json(
            "/api/searches/results",
            params={
                "type": REPORT_OBJECT_TYPE,
                "limit": -1,
            },
        )

        records = extract_search_records(payload)

        results: list[ObjectRef] = []

        for record in records:
            object_id = first_string(
                record,
                "id",
                "objectId",
                "objectID",
            )

            if not object_id:
                continue

            results.append(
                ObjectRef(
                    object_id=object_id,
                    name=first_string(record, "name"),
                    subtype=first_string(
                        record,
                        "subType",
                        "subtype",
                    ),
                    object_type=first_string(
                        record,
                        "type",
                        "objectType",
                    ),
                )
            )

        return deduplicate_object_refs(results)

    def get_report_like_definition(
        self,
        object_ref: ObjectRef,
    ) -> tuple[str, dict[str, Any] | list[Any]]:
        """
        Route a report-like object to the correct Modeling API.

        A fallback sequence is included because search results do not always
        expose subtype consistently across MicroStrategy versions.
        """
        subtype = normalize_subtype(object_ref.subtype)

        if is_cube_subtype(subtype):
            candidates = [
                ("cube", f"/api/model/cubes/{object_ref.object_id}"),
            ]
        elif is_datamart_subtype(subtype):
            candidates = [
                ("datamart", f"/api/model/datamarts/{object_ref.object_id}"),
            ]
        else:
            candidates = [
                ("report", f"/api/model/reports/{object_ref.object_id}"),
                ("cube", f"/api/model/cubes/{object_ref.object_id}"),
                ("datamart", f"/api/model/datamarts/{object_ref.object_id}"),
            ]

        last_error: Exception | None = None

        for object_kind, endpoint in candidates:
            try:
                definition = self.get_json(
                    endpoint,
                    params={
                        "showExpressionAs": "tree",
                        "showFilterTokens": "false",
                        "showAdvancedProperties": "true",
                    },
                )
                return object_kind, definition

            except requests.HTTPError as exc:
                last_error = exc

                status = (
                    exc.response.status_code
                    if exc.response is not None
                    else None
                )

                # Continue only for object/endpoint mismatch responses.
                if status not in {400, 404, 405}:
                    raise

        raise RuntimeError(
            f"No supported report-like definition endpoint worked for "
            f"{object_ref.object_id} ({object_ref.name})."
        ) from last_error

    def get_metric_definition(
        self,
        metric_id: str,
    ) -> dict[str, Any] | list[Any]:
        return self.get_json(
            f"/api/model/metrics/{metric_id}",
            params={
                "showExpressionAs": "tree",
                "showAdvancedProperties": "true",
            },
        )

    def get_fact_definition(
        self,
        fact_id: str,
    ) -> dict[str, Any] | list[Any]:
        return self.get_json(
            f"/api/model/facts/{fact_id}",
            params={
                "showExpressionAs": "tree",
                "showAdvancedProperties": "true",
            },
        )

    def get_object_ancestors(
        self,
        object_id: str,
        object_type: int | None,
    ) -> dict[str, Any] | list[Any]:
        """
        Retrieves an object's metadata, including the `ancestors` folder chain,
        used to reconstruct the object's folder path.

        `type` is optional because some MicroStrategy versions omit ancestors
        when the type is wrong; the caller retries without it as a fallback.
        """
        params = {} if object_type is None else {"type": object_type}

        return self.get_json(
            f"/api/objects/{object_id}",
            params=params,
        )


# ---------------------------------------------------------------------------
# Generic JSON utilities
# ---------------------------------------------------------------------------

def walk_json(
    value: Any,
    path: str = "$",
) -> Iterator[tuple[str, Any]]:
    """
    Recursively yields every node with a JSONPath-like location.
    """
    yield path, value

    if isinstance(value, dict):
        for key, child in value.items():
            yield from walk_json(child, f"{path}.{key}")

    elif isinstance(value, list):
        for index, child in enumerate(value):
            yield from walk_json(child, f"{path}[{index}]")


def first_string(
    value: dict[str, Any],
    *keys: str,
) -> str | None:
    for key in keys:
        item = value.get(key)

        if isinstance(item, str) and item.strip():
            return item.strip()

        if isinstance(item, int):
            return str(item)

    return None


def normalize_subtype(value: str | None) -> str:
    return (value or "").strip().lower().replace("-", "_").replace(" ", "_")


def is_cube_subtype(subtype: str) -> bool:
    return subtype == "report_cube" or "cube" in subtype


def is_datamart_subtype(subtype: str) -> bool:
    return subtype == "report_datamart" or "datamart" in subtype


def extract_search_records(
    payload: dict[str, Any] | list[Any],
) -> list[dict[str, Any]]:
    """
    Handles the most common search response containers.
    """
    if isinstance(payload, list):
        return [item for item in payload if isinstance(item, dict)]

    if not isinstance(payload, dict):
        return []

    for key in (
        "result",
        "results",
        "objects",
        "searchResults",
        "items",
    ):
        candidate = payload.get(key)

        if isinstance(candidate, list):
            return [
                item
                for item in candidate
                if isinstance(item, dict)
            ]

    # Last-resort scan for the first list of object-like dictionaries.
    for _, node in walk_json(payload):
        if (
            isinstance(node, list)
            and node
            and all(isinstance(item, dict) for item in node)
        ):
            if any(
                first_string(item, "id", "objectId", "objectID")
                for item in node
            ):
                return node

    return []


def deduplicate_object_refs(
    objects: list[ObjectRef],
) -> list[ObjectRef]:
    by_id: dict[str, ObjectRef] = {}

    for obj in objects:
        existing = by_id.get(obj.object_id)

        if existing is None:
            by_id[obj.object_id] = obj
            continue

        by_id[obj.object_id] = ObjectRef(
            object_id=obj.object_id,
            name=obj.name or existing.name,
            subtype=obj.subtype or existing.subtype,
            object_type=obj.object_type or existing.object_type,
        )

    return list(by_id.values())


# ---------------------------------------------------------------------------
# Dependency extraction
# ---------------------------------------------------------------------------

REFERENCE_ID_KEYS = (
    "id",
    "objectId",
    "objectID",
)

REFERENCE_SUBTYPE_KEYS = (
    "subType",
    "subtype",
)

REFERENCE_TYPE_KEYS = (
    "type",
    "objectType",
)

REFERENCE_NAME_KEYS = (
    "name",
    "objectName",
)

# EnumDSSXMLObjectTypes base object types. MicroStrategy expresses references
# either as strings ("metric", "fact") or numeric type/subType codes.
OBJECT_TYPE_NAMES = {
    4: "metric",
    12: "attribute",
    13: "fact",
    15: "table",
}


def numeric_type_name(
    node: dict[str, Any],
    keys: tuple[str, ...],
) -> str | None:
    """
    Maps numeric MicroStrategy type/subType codes to a reference kind.

    A subType encodes its base object type in the high byte, so the base type
    is recovered with `code >> 8` (e.g. metric subType 1024 >> 8 == 4).
    """
    for key in keys:
        value = node.get(key)

        if isinstance(value, bool):
            continue

        if isinstance(value, int):
            code = value
        elif isinstance(value, str) and value.strip().isdigit():
            code = int(value.strip())
        else:
            continue

        for candidate in (code, code >> 8):
            if candidate in OBJECT_TYPE_NAMES:
                return OBJECT_TYPE_NAMES[candidate]

    return None


def classify_reference(node: dict[str, Any]) -> str | None:
    """
    Classifies an object reference based on type and subtype fields.
    """
    numeric = numeric_type_name(
        node,
        REFERENCE_TYPE_KEYS + REFERENCE_SUBTYPE_KEYS,
    )
    if numeric:
        return numeric

    subtype = normalize_subtype(
        first_string(node, *REFERENCE_SUBTYPE_KEYS)
    )
    object_type = normalize_subtype(
        first_string(node, *REFERENCE_TYPE_KEYS)
    )

    combined = f"{subtype} {object_type}"

    if "metric" in combined:
        return "metric"

    if "fact" in combined:
        return "fact"

    if "attribute" in combined:
        return "attribute"

    if "table" in combined:
        return "table"

    if "column" in combined:
        return "column"

    return None


def extract_dependencies(
    payload: dict[str, Any] | list[Any],
    dependency_kind: str,
) -> list[Dependency]:
    """
    Finds typed object references anywhere in the response.

    This deliberately scans the entire definition rather than assuming one
    exact response shape. MicroStrategy places references in templates,
    expressions, filters, conditions, transformations, and nested trees.
    """
    dependencies: dict[str, Dependency] = {}

    for path, node in walk_json(payload):
        if not isinstance(node, dict):
            continue

        if classify_reference(node) != dependency_kind:
            continue

        object_id = first_string(node, *REFERENCE_ID_KEYS)

        if not object_id:
            continue

        dependencies.setdefault(
            object_id,
            Dependency(
                object_id=object_id,
                name=first_string(node, *REFERENCE_NAME_KEYS),
                subtype=first_string(node, *REFERENCE_SUBTYPE_KEYS),
                path=path,
            ),
        )

    return list(dependencies.values())


# ---------------------------------------------------------------------------
# Physical table/column extraction
# ---------------------------------------------------------------------------

COLUMN_NAME_KEYS = {
    "columnname",
    "column_name",
    "column",
    "warehousecolumn",
    "physicalcolumn",
}

TABLE_NAME_KEYS = {
    "tablename",
    "table_name",
    "table",
    "warehousetable",
    "physicaltable",
}

TABLE_ID_KEYS = {
    "tableid",
    "table_id",
    "physicaltableid",
}

COLUMN_ID_KEYS = {
    "columnid",
    "column_id",
    "physicalcolumnid",
}


def normalized_key(value: str) -> str:
    return value.strip().lower().replace("-", "_").replace(" ", "_")


def extract_physical_mappings(
    payload: dict[str, Any] | list[Any],
) -> list[dict[str, str | None]]:
    """
    Extracts column and table values from fact definitions.

    The exact field arrangement varies by server version and fact expression:
    - table and column may be siblings;
    - column may be nested beneath a table object;
    - the expression may contain typed object references;
    - only IDs may be supplied.

    The function preserves JSON paths so unmapped variants can be diagnosed.
    """
    mappings: list[dict[str, str | None]] = []

    for path, node in walk_json(payload):
        if not isinstance(node, dict):
            continue

        normalized = {
            normalized_key(str(key)): value
            for key, value in node.items()
        }

        column_name = get_scalar_from_key_set(
            normalized,
            COLUMN_NAME_KEYS,
        )
        table_name = get_scalar_from_key_set(
            normalized,
            TABLE_NAME_KEYS,
        )
        column_id = get_scalar_from_key_set(
            normalized,
            COLUMN_ID_KEYS,
        )
        table_id = get_scalar_from_key_set(
            normalized,
            TABLE_ID_KEYS,
        )

        if not any([column_name, column_id, table_name, table_id]):
            continue

        mappings.append(
            {
                "column_name": column_name,
                "column_id": column_id,
                "table_name": table_name,
                "table_id": table_id,
                "source_path": path,
            }
        )

    return deduplicate_mappings(mappings)


def get_scalar_from_key_set(
    node: dict[str, Any],
    keys: set[str],
) -> str | None:
    for key in keys:
        value = node.get(key)

        if isinstance(value, (str, int, float)) and str(value).strip():
            return str(value).strip()

        if isinstance(value, dict):
            nested_value = first_string(
                value,
                "name",
                "id",
                "objectId",
            )
            if nested_value:
                return nested_value

    return None


def deduplicate_mappings(
    mappings: list[dict[str, str | None]],
) -> list[dict[str, str | None]]:
    output: list[dict[str, str | None]] = []
    seen: set[tuple[str | None, ...]] = set()

    for mapping in mappings:
        key = (
            mapping["column_name"],
            mapping["column_id"],
            mapping["table_name"],
            mapping["table_id"],
            mapping["source_path"],
        )

        if key not in seen:
            seen.add(key)
            output.append(mapping)

    return output


# ---------------------------------------------------------------------------
# Folder path resolution
# ---------------------------------------------------------------------------

def build_folder_info(
    payload: dict[str, Any] | list[Any],
) -> FolderInfo:
    """
    Reconstructs an object's containing folder and full folder path from its
    `ancestors` chain.

    MicroStrategy returns ancestors ordered from the project root down to the
    immediate parent folder, so the last element is the containing folder.
    """
    empty = FolderInfo(folder_id=None, folder_name=None, folder_path=None)

    if not isinstance(payload, dict):
        return empty

    ancestors = payload.get("ancestors")

    if not isinstance(ancestors, list):
        return empty

    folders = [item for item in ancestors if isinstance(item, dict)]

    if not folders:
        return empty

    # MicroStrategy's `level` decreases toward the object, so the containing
    # folder has the smallest level. Sorting by descending level yields a
    # stable root->parent order regardless of the array's original order.
    if all(isinstance(item.get("level"), int) for item in folders):
        folders = sorted(
            folders,
            key=lambda item: item["level"],
            reverse=True,
        )

    names = [
        item["name"].strip()
        for item in folders
        if isinstance(item.get("name"), str) and item["name"].strip()
    ]

    parent = folders[-1]

    return FolderInfo(
        folder_id=first_string(parent, "id", "objectId", "objectID"),
        folder_name=first_string(parent, "name"),
        folder_path="/".join(names) if names else None,
    )


def folder_fields(info: FolderInfo) -> dict[str, str | None]:
    return {
        "fact_folder_id": info.folder_id,
        "fact_folder_name": info.folder_name,
        "fact_folder_path": info.folder_path,
    }


# ---------------------------------------------------------------------------
# Persistence
# ---------------------------------------------------------------------------

def safe_filename(value: str) -> str:
    return "".join(
        character
        if character.isalnum() or character in {"-", "_", "."}
        else "_"
        for character in value
    )


def save_definition(
    object_kind: str,
    object_id: str,
    payload: Any,
) -> None:
    directory = DEFINITION_DIR / object_kind
    directory.mkdir(parents=True, exist_ok=True)

    path = directory / f"{safe_filename(object_id)}.json"

    with path.open("w", encoding="utf-8") as handle:
        json.dump(
            payload,
            handle,
            ensure_ascii=False,
            indent=2,
        )


def append_jsonl(path: Path, value: dict[str, Any]) -> None:
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(value, ensure_ascii=False))
        handle.write("\n")


# ---------------------------------------------------------------------------
# Lineage crawler
# ---------------------------------------------------------------------------

class LineageCrawler:
    def __init__(self, client: MicroStrategyClient) -> None:
        self.client = client

        self.metric_cache: dict[str, dict[str, Any] | list[Any]] = {}
        self.fact_cache: dict[str, dict[str, Any] | list[Any]] = {}
        self.fact_folder_cache: dict[str, FolderInfo] = {}

        self.failed_metric_ids: set[str] = set()
        self.failed_fact_ids: set[str] = set()

        self.errors: list[dict[str, str]] = []
        self.rows: list[dict[str, str | None]] = []

    # -- traversal ----------------------------------------------------------

    def crawl_all(self) -> None:
        report_objects = self.client.search_report_objects()

        logging.info(
            "Discovered %d report-like objects.",
            len(report_objects),
        )

        for index, report_ref in enumerate(report_objects, start=1):
            logging.info(
                "[%d/%d] %s — %s",
                index,
                len(report_objects),
                report_ref.object_id,
                report_ref.name,
            )

            try:
                self.crawl_report_object(report_ref)
            except Exception as exc:
                self.record_error(
                    stage="report-like-definition",
                    object_id=report_ref.object_id,
                    object_name=report_ref.name,
                    error=exc,
                )

    def crawl_report_object(self, report_ref: ObjectRef) -> None:
        object_kind, definition = (
            self.client.get_report_like_definition(report_ref)
        )

        save_definition(
            object_kind,
            report_ref.object_id,
            definition,
        )

        information = extract_information(definition)

        root_name = (
            first_string(information, "name")
            or report_ref.name
            or report_ref.object_id
        )

        root_subtype = (
            first_string(information, "subType", "subtype")
            or report_ref.subtype
        )

        append_jsonl(
            OUTPUT_DIR / "objects.jsonl",
            {
                "object_id": report_ref.object_id,
                "object_name": root_name,
                "object_kind": object_kind,
                "object_subtype": root_subtype,
            },
        )

        direct_metrics = extract_dependencies(
            definition,
            "metric",
        )
        direct_facts = extract_dependencies(
            definition,
            "fact",
        )

        # Some reports contain direct fact references.
        for fact in direct_facts:
            self.crawl_fact_tree(
                root_id=report_ref.object_id,
                root_name=root_name,
                root_kind=object_kind,
                root_subtype=root_subtype,
                metric_chain=[],
                fact_dependency=fact,
                fact_chain=[],
            )

        for metric in direct_metrics:
            self.crawl_metric_tree(
                root_id=report_ref.object_id,
                root_name=root_name,
                root_kind=object_kind,
                root_subtype=root_subtype,
                metric_dependency=metric,
                metric_chain=[],
                visited_metrics=set(),
            )

    def crawl_metric_tree(
        self,
        *,
        root_id: str,
        root_name: str,
        root_kind: str,
        root_subtype: str | None,
        metric_dependency: Dependency,
        metric_chain: list[Dependency],
        visited_metrics: set[str],
    ) -> None:
        metric_id = metric_dependency.object_id

        if metric_id in visited_metrics:
            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | {
                    "metric_id": metric_id,
                    "metric_name": metric_dependency.name,
                    "metric_chain": format_chain(
                        metric_chain + [metric_dependency]
                    ),
                    "status": "METRIC_CYCLE_DETECTED",
                }
            )
            return

        current_chain = metric_chain + [metric_dependency]
        current_visited = visited_metrics | {metric_id}

        try:
            definition = self.get_metric_definition(metric_id)
        except Exception as exc:
            self.record_error(
                stage="metric-definition",
                object_id=metric_id,
                object_name=metric_dependency.name,
                error=exc,
            )

            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | {
                    "metric_id": metric_id,
                    "metric_name": metric_dependency.name,
                    "metric_chain": format_chain(current_chain),
                    "status": "METRIC_DEFINITION_FAILED",
                }
            )
            return

        nested_metrics = extract_dependencies(
            definition,
            "metric",
        )
        facts = extract_dependencies(
            definition,
            "fact",
        )

        # Ignore self-references returned in the metric's information block.
        nested_metrics = [
            dependency
            for dependency in nested_metrics
            if dependency.object_id != metric_id
        ]

        for fact in facts:
            self.crawl_fact_tree(
                root_id=root_id,
                root_name=root_name,
                root_kind=root_kind,
                root_subtype=root_subtype,
                metric_chain=current_chain,
                fact_dependency=fact,
                fact_chain=[],
            )

        for nested_metric in nested_metrics:
            self.crawl_metric_tree(
                root_id=root_id,
                root_name=root_name,
                root_kind=root_kind,
                root_subtype=root_subtype,
                metric_dependency=nested_metric,
                metric_chain=current_chain,
                visited_metrics=current_visited,
            )

        if not nested_metrics and not facts:
            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | {
                    "metric_id": metric_id,
                    "metric_name": metric_dependency.name,
                    "metric_chain": format_chain(current_chain),
                    "status": "METRIC_HAS_NO_FACT_REFERENCE",
                }
            )

    def crawl_fact_tree(
        self,
        *,
        root_id: str,
        root_name: str,
        root_kind: str,
        root_subtype: str | None,
        metric_chain: list[Dependency],
        fact_dependency: Dependency,
        fact_chain: list[Dependency],
    ) -> None:
        fact_id = fact_dependency.object_id
        folder_info = self.get_fact_folder_info(fact_id)

        if fact_id in {item.object_id for item in fact_chain}:
            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | metric_fields(metric_chain)
                | folder_fields(folder_info)
                | {
                    "fact_id": fact_id,
                    "fact_name": fact_dependency.name,
                    "fact_chain": format_chain(
                        fact_chain + [fact_dependency]
                    ),
                    "status": "FACT_CYCLE_DETECTED",
                }
            )
            return

        current_chain = fact_chain + [fact_dependency]

        try:
            definition = self.get_fact_definition(fact_id)
        except Exception as exc:
            self.record_error(
                stage="fact-definition",
                object_id=fact_id,
                object_name=fact_dependency.name,
                error=exc,
            )

            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | metric_fields(metric_chain)
                | folder_fields(folder_info)
                | {
                    "fact_id": fact_id,
                    "fact_name": fact_dependency.name,
                    "fact_chain": format_chain(current_chain),
                    "status": "FACT_DEFINITION_FAILED",
                }
            )
            return

        nested_facts = [
            dependency
            for dependency in extract_dependencies(definition, "fact")
            if dependency.object_id != fact_id
        ]

        mappings = extract_physical_mappings(definition)

        if mappings:
            for mapping in mappings:
                has_column = bool(
                    mapping["column_name"] or mapping["column_id"]
                )
                has_table = bool(
                    mapping["table_name"] or mapping["table_id"]
                )
                status = (
                    "OK"
                    if has_column and has_table
                    else "PARTIAL_PHYSICAL_MAPPING"
                )

                self.rows.append(
                    self.base_row(
                        root_id,
                        root_name,
                        root_kind,
                        root_subtype,
                    )
                    | metric_fields(metric_chain)
                    | folder_fields(folder_info)
                    | {
                        "fact_id": fact_id,
                        "fact_name": fact_dependency.name,
                        "fact_chain": format_chain(current_chain),
                        "column_id": mapping["column_id"],
                        "column_name": mapping["column_name"],
                        "table_id": mapping["table_id"],
                        "table_name": mapping["table_name"],
                        "definition_path": mapping["source_path"],
                        "status": status,
                    }
                )

        for nested_fact in nested_facts:
            self.crawl_fact_tree(
                root_id=root_id,
                root_name=root_name,
                root_kind=root_kind,
                root_subtype=root_subtype,
                metric_chain=metric_chain,
                fact_dependency=nested_fact,
                fact_chain=current_chain,
            )

        if not mappings and not nested_facts:
            self.rows.append(
                self.base_row(
                    root_id,
                    root_name,
                    root_kind,
                    root_subtype,
                )
                | metric_fields(metric_chain)
                | folder_fields(folder_info)
                | {
                    "fact_id": fact_id,
                    "fact_name": fact_dependency.name,
                    "fact_chain": format_chain(current_chain),
                    "status": "FACT_HAS_NO_PHYSICAL_MAPPING",
                }
            )

    # -- cached fetch helpers ----------------------------------------------

    def get_metric_definition(
        self,
        metric_id: str,
    ) -> dict[str, Any] | list[Any]:
        if metric_id in self.metric_cache:
            return self.metric_cache[metric_id]

        definition = self.client.get_metric_definition(metric_id)
        self.metric_cache[metric_id] = definition
        return definition

    def get_fact_definition(
        self,
        fact_id: str,
    ) -> dict[str, Any] | list[Any]:
        if fact_id in self.fact_cache:
            return self.fact_cache[fact_id]

        definition = self.client.get_fact_definition(fact_id)
        self.fact_cache[fact_id] = definition
        return definition

    def get_fact_folder_info(self, fact_id: str) -> FolderInfo:
        if fact_id in self.fact_folder_cache:
            return self.fact_folder_cache[fact_id]

        info = FolderInfo(folder_id=None, folder_name=None, folder_path=None)

        for object_type in (FACT_OBJECT_TYPE, None):
            try:
                payload = self.client.get_object_ancestors(fact_id, object_type)
            except Exception as exc:
                self.record_error(
                    stage="fact-folder",
                    object_id=fact_id,
                    object_name=None,
                    error=exc,
                )
                continue

            candidate = build_folder_info(payload)

            if candidate.folder_id or candidate.folder_name or candidate.folder_path:
                info = candidate
                break

        self.fact_folder_cache[fact_id] = info
        return info

    # -- row/error helpers --------------------------------------------------

    def base_row(
        self,
        root_id: str,
        root_name: str,
        root_kind: str,
        root_subtype: str | None,
    ) -> dict[str, str | None]:
        return {
            "root_id": root_id,
            "root_name": root_name,
            "root_kind": root_kind,
            "root_subtype": root_subtype,
        }

    def record_error(
        self,
        *,
        stage: str,
        object_id: str,
        object_name: str | None,
        error: Exception,
    ) -> None:
        logging.warning(
            "Error [%s] for %s (%s): %s",
            stage,
            object_id,
            object_name,
            error,
        )

        self.errors.append(
            {
                "stage": stage,
                "object_id": object_id,
                "object_name": object_name or "",
                "error_type": type(error).__name__,
                "error": str(error),
            }
        )

    # -- output -------------------------------------------------------------

    def write_outputs(self) -> None:
        write_csv(
            OUTPUT_DIR / "lineage.csv",
            self.rows,
            fieldnames=[
                "root_id",
                "root_name",
                "root_kind",
                "root_subtype",
                "metric_id",
                "metric_name",
                "metric_chain",
                "fact_id",
                "fact_name",
                "fact_folder_id",
                "fact_folder_name",
                "fact_folder_path",
                "fact_chain",
                "column_id",
                "column_name",
                "table_id",
                "table_name",
                "definition_path",
                "status",
            ],
        )

        write_csv(
            OUTPUT_DIR / "errors.csv",
            self.errors,
            fieldnames=[
                "stage",
                "object_id",
                "object_name",
                "error_type",
                "error",
            ],
        )


# ---------------------------------------------------------------------------
# Output helpers
# ---------------------------------------------------------------------------

def extract_information(payload: Any) -> dict[str, Any]:
    if isinstance(payload, dict):
        information = payload.get("information")

        if isinstance(information, dict):
            return information

    return {}


def format_chain(items: list[Dependency]) -> str | None:
    if not items:
        return None

    return " > ".join(
        f"{item.name or '[unnamed]'} [{item.object_id}]"
        for item in items
    )


def metric_fields(
    metric_chain: list[Dependency],
) -> dict[str, str | None]:
    if not metric_chain:
        return {
            "metric_id": None,
            "metric_name": None,
            "metric_chain": None,
        }

    leaf = metric_chain[-1]

    return {
        "metric_id": leaf.object_id,
        "metric_name": leaf.name,
        "metric_chain": format_chain(metric_chain),
    }


def write_csv(
    path: Path,
    rows: list[dict[str, Any]],
    fieldnames: list[str],
) -> None:
    with path.open(
        "w",
        encoding="utf-8-sig",
        newline="",
    ) as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=fieldnames,
            extrasaction="ignore",
        )
        writer.writeheader()
        writer.writerows(rows)


# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------

def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    DEFINITION_DIR.mkdir(parents=True, exist_ok=True)

    # objects.jsonl is append-only during the crawl, so clear it up front to
    # avoid mixing stale objects from a previous run into this run's output.
    (OUTPUT_DIR / "objects.jsonl").unlink(missing_ok=True)

    client = MicroStrategyClient(
        base_url=BASE_URL,
        username=USERNAME,
        password=PASSWORD,
        project_id=PROJECT_ID,
    )

    client.login()

    crawler = LineageCrawler(client)
    crawler.crawl_all()
    crawler.write_outputs()

    logging.info(
        "Finished: %d lineage rows, %d errors.",
        len(crawler.rows),
        len(crawler.errors),
    )
    logging.info("Output directory: %s", OUTPUT_DIR.resolve())


if __name__ == "__main__":
    main()