Raporty Power BI z Pythona: eksport, harmonogram, publikacja

Odświeżanie datasetów Power BI

Automatyczne odświeżanie danych w Power BI Service:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import requests
from pathlib import Path

def refresh_power_bi_dataset(workspace_id: str, dataset_id: str, access_token: str):
    """Wywołuje odświeżenie datasetu w Power BI Service."""
    url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/refreshes"

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

    response = requests.post(url, headers=headers)

    if response.status_code == 202:
        print("Odświeżenie datasetu uruchomione pomyślnie")
        return True
    else:
        print(f"Błąd podczas odświeżania: {response.status_code}")
        print(response.text)
        return False

# Użycie wymaga tokenu autoryzacyjnego
refresh_power_bi_dataset(
    workspace_id="your-workspace-id",
    dataset_id="your-dataset-id",
    access_token="your-access-token"
)

Automatyczne generowanie danych

Przygotowanie danych przed odświeżeniem:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import pandas as pd
from datetime import datetime

def update_power_bi_data_source(source_file: str, output_file: str):
    """Aktualizuje źródło danych dla Power BI."""
    # Wczytanie istniejących danych
    df = pd.read_excel(source_file)

    # Dodanie nowych wierszy (symulacja)
    new_data = pd.DataFrame({
        "Data": [datetime.now()],
        "Wartość": [100],
        "Kategoria": ["Nowa"]
    })

    # Łączenie
    updated_df = pd.concat([df, new_data], ignore_index=True)

    # Zapisanie zaktualizowanych danych
    updated_df.to_excel(output_file, index=False)
    print(f"Dane zaktualizowane: {output_file} ({len(updated_df)} wierszy)")

update_power_bi_data_source("dane_stare.xlsx", "dane_aktualne.xlsx")

Eksport raportów do PDF

Eksport raportów Power BI do PDF wymaga użycia API:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import requests
import time

def export_power_bi_report_to_pdf(
    workspace_id: str, 
    report_id: str, 
    access_token: str,
    output_path: str
):
    """Eksportuje raport Power BI do PDF."""
    # Rozpoczęcie eksportu
    export_url = f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}/reports/{report_id}/ExportTo"

    headers = {
        "Authorization": f"Bearer {access_token}"
    }

    params = {
        "format": "PDF"
    }

    response = requests.post(export_url, headers=headers, params=params)

    if response.status_code == 202:
        # Pobranie URL do pobrania pliku
        export_id = response.headers.get("Content-Location")

        # Oczekiwanie na zakończenie eksportu
        time.sleep(10)  # W praktyce sprawdzaj status

        # Pobranie pliku (uproszczone)
        print(f"Eksport rozpoczęty. ID: {export_id}")
        print(f"Plik zostanie zapisany w: {output_path}")
    else:
        print(f"Błąd eksportu: {response.status_code}")

export_power_bi_report_to_pdf(
    workspace_id="workspace-id",
    report_id="report-id",
    access_token="token",
    output_path="raport.pdf"
)

Harmonogram automatycznego odświeżania

Tworzenie harmonogramu zadań:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import schedule
import time
from datetime import datetime

def refresh_power_bi_job():
    """Zadanie harmonogramowe do odświeżania Power BI."""
    print(f"[{datetime.now()}] Rozpoczynam odświeżanie Power BI...")

    # 1. Aktualizacja danych źródłowych
    update_power_bi_data_source("dane.xlsx", "dane_aktualne.xlsx")

    # 2. Odświeżenie datasetu (wymaga konfiguracji API)
    # refresh_power_bi_dataset(...) - wywołanie funkcji odświeżania w harmonogramie

    print(f"[{datetime.now()}] Odświeżanie zakończone")

# Harmonogram codzienny o 8:00
schedule.every().day.at("08:00").do(refresh_power_bi_job)

# Harmonogram tygodniowy
schedule.every().monday.at("09:00").do(refresh_power_bi_job)

# Uruchomienie harmonogramu
print("Harmonogram odświeżania Power BI uruchomiony...")
while True:
    schedule.run_pending()
    time.sleep(60)

Wysyłka raportów e-mailem

Połączenie z eksportem raportów i wysyłką:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email import encoders
from pathlib import Path

def send_power_bi_report_email(
    report_path: str,
    recipient: str,
    subject: str = "Raport Power BI"
):
    """Wysyła raport Power BI e-mailem."""
    msg = MIMEMultipart()
    msg["From"] = "raporty@firma.pl"
    msg["To"] = recipient
    msg["Subject"] = subject

    body = "W załączniku znajduje się aktualny raport Power BI."
    msg.attach(MIMEText(body, "plain"))

    # Załączanie pliku
    with open(report_path, "rb") as f:
        part = MIMEBase("application", "octet-stream")
        part.set_payload(f.read())
        encoders.encode_base64(part)
        part.add_header(
            "Content-Disposition",
            f"attachment; filename= {Path(report_path).name}"
        )
        msg.attach(part)

    # Wysyłka (wymaga konfiguracji SMTP)
    # server = smtplib.SMTP("smtp.gmail.com", 587)
    # server.starttls()
    # server.login("user", "password")
    # server.send_message(msg)
    # server.quit()

    print(f"Raport wysłany do: {recipient}")

send_power_bi_report_email(
    "raport.pdf",
    "uzytkownik@firma.pl",
    "Raport Power BI - Miesięczny"
)

Kompletna automatyzacja pipeline

Połączenie wszystkich elementów:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import schedule
import time
from datetime import datetime

def complete_power_bi_automation():
    """Kompletna automatyzacja: odświeżanie -> eksport -> wysyłka."""
    print(f"[{datetime.now()}] Rozpoczynam automatyzację Power BI...")

    # 1. Aktualizacja danych
    update_power_bi_data_source("dane.xlsx", "dane_aktualne.xlsx")

    # 2. Odświeżenie datasetu
    # refresh_power_bi_dataset(...) - wywołanie funkcji w kompletnym pipeline automatyzacji

    # 3. Eksport do PDF (wymaga czasu na wygenerowanie)
    time.sleep(30)
    # export_power_bi_report_to_pdf(...)

    # 4. Wysyłka e-mailem
    # send_power_bi_report_email("raport.pdf", "uzytkownik@firma.pl")

    print(f"[{datetime.now()}] Automatyzacja zakończona")

# Harmonogram
schedule.every().day.at("08:00").do(complete_power_bi_automation)

# Uruchomienie
while True:
    schedule.run_pending()
    time.sleep(60)

Podsumowanie

Automatyzacja raportów Power BI obejmuje:

• Odświeżanie datasetów przez API
• Generowanie i aktualizację danych
• Eksport raportów do PDF/CSV
• Harmonogramy zadań dla cyklicznego uruchamiania
• Wysyłkę raportów e-mailem

To kompletne rozwiązanie do automatyzacji procesów raportowania w Power BI.

---

➡️ Następny artykuł

Po automatyzacji raportowania, poznaj fundamenty pracy z bazami danych:

Praca z bazą danych w automatyzacji: SQLite/PostgreSQL — wzorce połączeń, transakcje, migracje i tworzenie niezawodnych pipeline'ów ETL w Pythonie.