# Microsoft Fabric PowerShell Module installeren
Install-Module -Name MicrosoftPowerBIMgmt -Force
Install-Module -Name MicrosoftPowerBIMgmt.Admin -Force
Install-Module -Name MicrosoftPowerBIMgmt.Workspaces -Force

# Aanmelden bij Power BI (Fabric)
Connect-PowerBIServiceAccount

# Fabric Workspace aanmaken
New-PowerBIWorkspace -Name "Fabric Analytics Platform" `
    -Description "Productie workspace voor enterprise analytics" `
    -Type "Premium" `
    -CapacityId "your-capacity-id"

# OneLake bestanden beheren via OneLake file explorer
# OneLake werkt als een normale filesystem maar dan in de cloud
# Voorbeeld: abfss://onelake@fabric.dfs.core.windows.net/

# Fabric Items maken met REST API
$headers = @{
    "Authorization" = "Bearer $accessToken"
    "Content-Type" = "application/json"
}

# Data Warehouse aanmaken
$body = @{
    displayName = "Enterprise Data Warehouse"
    description = "Centraal data warehouse voor alle business units"
    type = "Warehouse"
} | ConvertTo-Json

Invoke-RestMethod -Method Post `
    -Uri "https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items" `
    -Headers $headers `
    -Body $body

# Data Pipeline aanmaken
$pipelineBody = @{
    displayName = "Daily Sales ETL"
    description = "Dagelijkse ETL voor sales data van SAP en CRM"
    type = "Pipeline"
    properties = @{
        activities = @(
            @{
                name = "Copy SAP Data"
                type = "Copy"
                inputs = @(@{referenceName = "SAPSource"})
                outputs = @(@{referenceName = "RawSales"})
            },
            @{
                name = "Transform Data"
                type = "Spark"
                inputs = @(@{referenceName = "RawSales"})
                outputs = @(@{referenceName = "CleanedSales"})
            }
        )
    }
} | ConvertTo-Json -Depth 5

# Capaciteit (CU's) monitoren
Get-PowerBICapacity -Scope Organization |
    Where-Object {$_.State -eq "Active"} |
    Select-Object DisplayName, Sku, State, Admins, Users |
    Format-Table -AutoSize

# Fabric licentie rapport genereren
$report = Get-PowerBICapacity -Scope Organization |
    ForEach-Object {
        [PSCustomObject]@{
            Workspace = $_.DisplayName
            SKU = $_.Sku
            Status = $_.State
            AssignedTo = $_.Admins -join ", "
            MonthlyCost = if($_.Sku -eq "P1") {"€4,995"} 
                         elseif($_.Sku -eq "F64") {"€5,002.88"}
                         else {"Onbekend"}
        }
    }

$report | Export-Csv -Path "fabric_capacity_report.csv" -NoTypeInformation
Write-Host "✅ Fabric capaciteit rapport gegenereerd" -ForegroundColor Green
        

# Microsoft Fabric Notebook (Python)
# OneLake is standaard geconfigureerd als Spark default storage

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from delta.tables import *
import pandas as pd
import json

# Spark sessie in Fabric (geen configuratie nodig)
spark = SparkSession.builder.getOrCreate()
print(f"Spark versie: {spark.version}")
print(f"Fabric Workspace: {spark.conf.get('spark.fabric.workspace.name')}")

def demonstrate_fabric_features():
    """Demonstreer Fabric-specifieke features"""
    
    # 1. Data lezen van OneLake (Delta format)
    print("1. Data lezen van OneLake...")
    
    # OneLake path: abfss://workspace@onelake.dfs.fabric.microsoft.com/
    sales_df = spark.read.format("delta").load(
        "Files/Sales/Transactions"
    )
    
    print(f"   Sales data: {sales_df.count():,} rijen")
    
    # 2. Data transformeren met Spark
    print("2. Data transformeren...")
    
    daily_metrics = sales_df \
        .groupBy("TransactionDate", "StoreId", "ProductCategory") \
        .agg(
            sum("Amount").alias("TotalSales"),
            avg("Quantity").alias("AvgQuantity"),
            count("*").alias("TransactionCount")
        ) \
        .withColumn("Year", year(col("TransactionDate"))) \
        .withColumn("Month", month(col("TransactionDate")))
    
    # 3. Opslaan naar OneLake (Delta Lake)
    print("3. Opslaan naar OneLake als Delta table...")
    
    daily_metrics.write \
        .format("delta") \
        .mode("overwrite") \
        .partitionBy("Year", "Month") \
        .save("Tables/DailySalesMetrics")
    
    # 4. Direct Query met SQL Endpoint
    print("4. Direct SQL query op Warehouse...")
    
    # Maak een SQL Endpoint view
    daily_metrics.createOrReplaceTempView("daily_sales_metrics_view")
    
    # SQL query uitvoeren
    sql_result = spark.sql("""
        SELECT 
            Year,
            Month,
            ProductCategory,
            SUM(TotalSales) as MonthlySales,
            AVG(AvgQuantity) as AvgDailyQuantity
        FROM daily_sales_metrics_view
        WHERE Year = 2024
        GROUP BY Year, Month, ProductCategory
        ORDER BY MonthlySales DESC
        LIMIT 10
    """)
    
    print("   Top 10 product categorieën 2024:")
    sql_result.show(truncate=False)
    
    # 5. Shortcuts gebruiken - virtuele links naar externe data
    print("5. Shortcuts maken naar externe data...")
    
    # Maak een shortcut naar Azure Data Lake Storage
    spark.sql("""
        CREATE SHORTCUT IF NOT EXISTS ExternalCustomerData
        ON 'abfss://customer-data@externaladls.dfs.core.windows.net/raw/'
        WITH (
            TYPE = 'ADLS',
            ACCOUNT_NAME = 'externalstorage'
        )
    """)
    
    # Lees data via shortcut
    external_data = spark.read.parquet(
        "ExternalCustomerData/customers/*.parquet"
    )
    
    print(f"   Externe customer data: {external_data.count():,} rijen")
    
    return daily_metrics, sql_result

def fabric_data_science_example():
    """Data Science met Fabric ML"""
    
    print("\n6. Machine Learning met Fabric ML...")
    
    from synapse.ml import *
    from synapse.ml.lightgbm import *
    
    # Laad data voor ML
    ml_data = spark.read.format("delta").load(
        "Tables/MLReadyData"
    )
    
    # Feature engineering
    feature_cols = [c for c in ml_data.columns if c not in ["Label", "CustomerId"]]
    
    # Train LightGBM model
    lgbm = LightGBMClassifier(
        featuresCol="features",
        labelCol="Label",
        predictionCol="prediction",
        objective="binary",
        numLeaves=31,
        maxDepth=5,
        numIterations=100
    )
    
    # Data splitsen
    train_data, test_data = ml_data.randomSplit([0.8, 0.2])
    
    # Model trainen
    model = lgbm.fit(train_data)
    
    # Voorspellingen maken
    predictions = model.transform(test_data)
    
    # Model opslaan in MLflow (geïntegreerd in Fabric)
    import mlflow
    
    with mlflow.start_run():
        mlflow.log_param("num_leaves", 31)
        mlflow.log_param("max_depth", 5)
        mlflow.spark.log_model(model, "churn_prediction_model")
        
        # Metrics loggen
        from sklearn.metrics import accuracy_score, f1_score
        
        pdf = predictions.select("Label", "prediction").toPandas()
        accuracy = accuracy_score(pdf["Label"], pdf["prediction"])
        f1 = f1_score(pdf["Label"], pdf["prediction"])
        
        mlflow.log_metric("accuracy", accuracy)
        mlflow.log_metric("f1_score", f1)
    
    print(f"   Model accuracy: {accuracy:.2%}")
    print(f"   F1 score: {f1:.2%}")
    
    return model, predictions

def power_bi_integration():
    """Power BI Direct Lake mode demo"""
    
    print("\n7. Power BI Direct Lake integration...")
    
    # Maak een dataset klaar voor Direct Lake
    # Direct Lake laadt data direct van OneLake zonder import
    
    # 1. Maak een semantic model
    semantic_model_sql = """
    CREATE TABLE SalesSemanticModel (
        SalesAmount decimal(18,2),
        Quantity int,
        ProductName varchar(100),
        Category varchar(50),
        TransactionDate date,
        StoreLocation varchar(200)
    )
    """
    
    # 2. Vul het semantic model
    spark.sql("""
        INSERT INTO SalesSemanticModel
        SELECT 
            Amount as SalesAmount,
            Quantity,
            p.ProductName,
            p.Category,
            t.TransactionDate,
            s.StoreLocation
        FROM Files/Sales/Transactions t
        JOIN Tables/Products p ON t.ProductId = p.ProductId
        JOIN Tables/Stores s ON t.StoreId = s.StoreId
        WHERE t.TransactionDate >= '2024-01-01'
    """)
    
    print("   Semantic model gemaakt voor Power BI Direct Lake")
    print("   Power BI kan nu direct queryen op OneLake zonder data kopiëren!")
    
    # 3. Power BI dataset URL genereren
    workspace_name = spark.conf.get("spark.fabric.workspace.name")
    dataset_id = "sales_semantic_model"
    
    pbix_url = f"https://app.powerbi.com/groups/me/datasets/{dataset_id}"
    print(f"   Power BI dataset URL: {pbix_url}")

if __name__ == "__main__":
    print("=" * 70)
    print("MICROSOFT FABRIC DEMO")
    print("=" * 70)
    
    # Demo basis features
    metrics, sql_result = demonstrate_fabric_features()
    
    # Demo Data Science
    try:
        model, predictions = fabric_data_science_example()
    except Exception as e:
        print(f"   ML demo skipped: {e}")
    
    # Demo Power BI integration
    power_bi_integration()
    
    print("\n" + "=" * 70)
    print("DEMO VOLTOOID")
    print("=" * 70)
    print("\nVolgende stappen:")
    print("1. Open Power BI voor Direct Lake dashboards")
    print("2. Bekijk data lineage in Purview")
    print("3. Configureer data governance policies")
    print("4. Maak data pipelines in Data Factory")
        

# Een shortcut gedraagt zich als een gewone map in je Lakehouse
df = spark.read.format("delta").load("Files/verkoop_extern/orders")

# Wegschrijven als beheerde tabel, zodat het SQL-eindpunt hem ziet
(df.write
   .format("delta")
   .mode("overwrite")
   .option("overwriteSchema", "true")
   .saveAsTable("bronze_orders"))

# Het volledige OneLake-pad werkt ook, handig tussen workspaces
pad = "abfss://Verkoop@onelake.dfs.fabric.microsoft.com/Sales.Lakehouse/Tables/orders"
spark.read.format("delta").load(pad).createOrReplaceTempView("orders_extern")
        

-- Vanuit het Warehouse een Lakehouse-tabel bevragen via drieledige naamgeving
SELECT
    k.klant_id,
    k.naam,
    SUM(o.bedrag) AS omzet
FROM SalesLakehouse.dbo.orders        AS o
JOIN dim_klant                        AS k ON k.klant_id = o.klant_id
WHERE o.besteldatum >= '2026-01-01'
GROUP BY k.klant_id, k.naam;

-- In het Warehouse mag je wel schrijven
CREATE TABLE fact_omzet AS
SELECT besteldatum, klant_id, SUM(bedrag) AS omzet
FROM SalesLakehouse.dbo.orders
GROUP BY besteldatum, klant_id;

-- COPY INTO laadt rechtstreeks vanuit opslag
COPY INTO dim_product
FROM 'https://account.blob.core.windows.net/data/producten.parquet'
WITH (FILE_TYPE = 'PARQUET');
        

from pyspark.sql import functions as F
from pyspark.sql.window import Window

bronze = spark.read.table("bronze_orders")

venster = Window.partitionBy("order_id").orderBy(F.col("geladen_op").desc())

silver = (bronze
    .withColumn("rang", F.row_number().over(venster))
    .filter(F.col("rang") == 1)                    # alleen de laatste versie
    .drop("rang")
    .filter(F.col("order_id").isNotNull())
    .withColumn("bedrag", F.col("bedrag").cast("decimal(12,2)"))
    .withColumn("besteldatum", F.to_date("besteldatum", "dd-MM-yyyy")))

(silver.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("silver_orders"))
        

from pyspark.sql import functions as F
from delta.tables import DeltaTable

# Het watermerk van de vorige run ophalen
try:
    watermerk = (spark.read.table("silver_orders")
                      .agg(F.max("gewijzigd_op")).collect()[0][0])
except Exception:
    watermerk = "1900-01-01"

nieuw = (spark.read.table("bronze_orders")
              .filter(F.col("gewijzigd_op") > F.lit(watermerk)))

print(f"{nieuw.count()} gewijzigde rijen sinds {watermerk}")

# MERGE: bestaande rijen bijwerken, nieuwe toevoegen
doel = DeltaTable.forName(spark, "silver_orders")
(doel.alias("d")
     .merge(nieuw.alias("n"), "d.order_id = n.order_id")
     .whenMatchedUpdateAll()
     .whenNotMatchedInsertAll()
     .execute())
        

import datetime

def log_run(stap, status, rijen=0, melding=""):
    regel = [(stap, status, rijen, melding, datetime.datetime.now())]
    kolommen = ["stap", "status", "rijen", "melding", "tijdstip"]
    (spark.createDataFrame(regel, kolommen)
          .write.format("delta").mode("append").saveAsTable("pipeline_log"))

try:
    aantal = verwerk_orders()          # je eigen functie
    log_run("silver_orders", "gelukt", aantal)
except Exception as fout:
    log_run("silver_orders", "mislukt", 0, str(fout)[:500])
    raise                               # opnieuw opgooien, anders denkt de pipeline dat het goed ging
        

-- Expliciete maatstaven, geen kolommen die je in een visual sleept
Omzet = SUM(fact_omzet[bedrag])

Omzet vorig jaar =
CALCULATE([Omzet], SAMEPERIODLASTYEAR(dim_datum[datum]))

Groei % =
VAR Huidig = [Omzet]
VAR Vorig  = [Omzet vorig jaar]
RETURN DIVIDE(Huidig - Vorig, Vorig)      -- DIVIDE vangt deling door nul af

-- Lopend totaal binnen het jaar
Omzet YTD =
TOTALYTD([Omzet], dim_datum[datum])
        

-- Rijniveau: iedere accountmanager ziet alleen zijn eigen regio
CREATE SCHEMA beveiliging;
GO

CREATE FUNCTION beveiliging.regio_filter(@regio AS VARCHAR(50))
RETURNS TABLE WITH SCHEMABINDING
AS
RETURN SELECT 1 AS toegestaan
       WHERE @regio = USER_NAME()
          OR USER_NAME() = 'datateam@bedrijf.nl';
GO

CREATE SECURITY POLICY beveiliging.regio_beleid
ADD FILTER PREDICATE beveiliging.regio_filter(regio) ON dbo.fact_omzet
WITH (STATE = ON);
GO

-- Kolomniveau: niet iedereen hoeft salarissen te zien
GRANT SELECT ON dbo.dim_medewerker(medewerker_id, naam, afdeling) TO analist_rol;