Script to export and synchronize verified answers from a Power BI workspace

fetch_verified_answers.py

"""
Script to export and synchronize verified answers from Power BI Dev environment.
* Works with Large Semantic Model format.
* You must have a "default" report tied to the semantic model. 
* Script expects that report saved locally as PBIP file

This script:
1. Authenticates with the Fabric API using Microsoft authentication
2. Exports the report as a ZIP file from Power BI
3. Extracts and processes the verified answers definitions
4. Converts JSON files from UTF-16LE to UTF-8 encoding
5. Updates the local verified answers definitions directory

"""

# %% Authorize with fabric api
from pathlib import Path
import msal
import requests
import shutil
import tempfile
import zipfile
import json

WORKSPACE_ID = "<<GUID_HERE>>"
REPORT_ID = "<<GUID_HERE>>"
MODEL_NAME = "<<NAME WILL BECOME NAME.SEMANTICMODEL IMMEDIATELY BELOW HERE>>"

LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH = (
    Path(__file__).parent
    / f"{MODEL_NAME}.SemanticModel"
    / "VerifiedAnswers"
    / "definitions"
)

AUTH_CLIENT_ID = "5814bfb4-2705-4994-b8d6-39aabeb5eaeb"
AUTH_TENANT_ID = "be1b4fd9-5fce-4f73-b3dd-3005f9e18ec7"

# Grab a token
BASE_URL = "https://api.fabric.microsoft.com/v1.0"
app = msal.PublicClientApplication(
    client_id=AUTH_CLIENT_ID,
    authority=f"https://login.microsoftonline.com/{AUTH_TENANT_ID}",
    # pip install "msal[broker]>=1.20,<2
    enable_broker_on_windows=True,
)
result = app.acquire_token_interactive(
    [
        "https://api.fabric.microsoft.com/Item.Read.All",
        "https://api.fabric.microsoft.com/Workspace.Read.All",
    ],
    parent_window_handle=msal.PublicClientApplication.CONSOLE_WINDOW_HANDLE,
)

if "access_token" not in result:
    raise RuntimeError(
        f"Authentication failed: {result.get('error_description', result)}"
    )

token = result["access_token"]
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

# %% Export the report as a pbix file
url = f"https://api.powerbi.com/v1.0/myorg/groups/{WORKSPACE_ID}/reports/{REPORT_ID}/Export?downloadType=LiveConnect"
resp = requests.get(url, headers=headers)
resp.raise_for_status()

export_path = Path(tempfile.mkdtemp())
exported_file = export_path / "export.zip"

with open(exported_file, "wb") as f:
    f.write(resp.content)
print(f"saved {len(resp.content)} bytes to {exported_file}")

# %% Extract and sync verified answers
try:
    with zipfile.ZipFile(exported_file, "r") as zip_ref:
        zip_ref.extractall(export_path)

    # remove everything in LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH including subfolders
    shutil.rmtree(LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH, ignore_errors=True)

    # move everything from export_path/VerifiedAnswers/definitions to LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH
    shutil.move(
        export_path / "VerifiedAnswers" / "definitions",
        LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH,
    )

    # now convert every json file in LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH from UTF-16LE to UTF-8
    for file in Path(LOCAL_VERIFIED_ANSWERS_DEFINITIONS_PATH).glob("**/*.json"):
        with open(file, "r", encoding="utf-16le") as f:
            data = json.load(f)
        with open(file, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False)
            f.write("\n")
finally:
    # cleanup temp folder
    shutil.rmtree(export_path, ignore_errors=True)