Power BI Service用户活动日志查询400错误排查求助

问题说明 #

我需要获取Power BI Service中的用户活动信息，包括指定时间范围内每个用户查看的报表、查看时间、查看时长、导出的可视化内容等。目前在调用/admin/activityevents接口获取报表查看记录时遇到400错误，已成功获取Access Token，但接口返回错误提示startDateTime和endDateTime参数需实现IConvertible。

错误详情 #

Access Token JSON: {'token_type': 'Bearer', 'expires_in': 3599, 'ext_expires_in': 3599, 'access_token': 'eyJ0eXAi.........'}
Fetching logs from 2024-06-11T00:00:00.000Z to 2024-06-11T23:59:59.999Z
Requesting logs with URL: https://api.powerbi.com/v1.0/myorg/admin/activityevents?startDateTime=2024-06-11T00%3A00%3A00.000Z&endDateTime=2024-06-11T23%3A59%3A59.999Z
Response Status Code: 400
HTTP error occurred: 400 Client Error: Bad Request for url: https://api.powerbi.com/v1.0/myorg/admin/activityevents?startDateTime=2024-06-11T00%3A00%3A00.000Z&endDateTime=2024-06-11T23%3A59%3A59.999Z
Response text: {"error":{"code":"BadRequest","message":"Bad Request","details":[{"message":"Object must implement IConvertible.","target":"startDateTime"},{"message":"Object must implement IConvertible.","target":"endDateTime"}]}}

问题原因 #

错误根源在于手动拼接URL并使用urlencode处理参数的方式，导致参数传递格式不符合Power BI API的解析要求。虽然时间字符串格式符合ISO 8601，但手动编码后的URL参数可能触发API内部的类型转换异常。

解决方案 #

使用requests库的params参数自动处理URL参数编码，而非手动拼接URL。这种方式能确保参数以API预期的格式传递，避免类型转换错误。

修改后的代码 #

import requests
import datetime
import pandas as pd
from urllib.parse import urlencode, unquote

# Constants (Replace these with your actual values)
CLIENT_ID = '-------------------'
TENANT_ID = '-------------------'
CLIENT_SECRET = '-----------------'

# Get access token using Microsoft identity platform (v2.0; MSAL)
def get_access_token():
    url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    body = {
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'scope': 'https://analysis.windows.net/powerbi/api/.default'
    }
    response = requests.post(url, headers=headers, data=body)
    response.raise_for_status()
    token_json = response.json()
    print("Access Token JSON:", token_json)  # Debugging output
    return token_json.get('access_token')

# Fetch activity logs
def fetch_activity_logs(access_token, start_datetime, end_datetime, continuation_token=None):
    headers = {
        'Authorization': f'Bearer {access_token}'
    }
    params = {
        'startDateTime': start_datetime,
        'endDateTime': end_datetime
    }
    if continuation_token:
        params['continuationToken'] = unquote(continuation_token)
    
    url = "https://api.powerbi.com/v1.0/myorg/admin/activityevents"
    
    # 保留调试输出，显示最终请求URL
    print(f"Requesting logs with URL: {url}?{urlencode(params)}")
    # 使用requests的params参数自动处理编码
    response = requests.get(url, headers=headers, params=params)
    print(f"Response Status Code: {response.status_code}")
    response.raise_for_status()
    return response.json()

# Main script
if __name__ == '__main__':
    try:
        access_token = get_access_token()
        # 设置单日（昨日）的时间范围，符合API要求格式
        yesterday = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1)
        start_datetime = yesterday.strftime("%Y-%m-%dT00:00:00.000Z")
        end_datetime = yesterday.strftime("%Y-%m-%dT23:59:59.999Z")
        print(f"Fetching logs from {start_datetime} to {end_datetime}")  # Debugging output
        # 初始化存储所有日志的列表
        all_logs = []
        continuation_token = None
        # 循环处理分页
        while True:
            logs = fetch_activity_logs(access_token, start_datetime, end_datetime, continuation_token)
            all_logs.extend(logs.get('activityEventEntities', []))
            
            # 检查是否有续传令牌
            continuation_token = logs.get('continuationToken')
            if continuation_token:
                print(f"Continuation Token: {continuation_token}")  # Debugging output
            else:
                break  # 无续传令牌则退出循环
        # 打印原始日志以了解结构
        print("Raw logs:", all_logs)
        # 提取相关数据并创建DataFrame
        data = []
        for log in all_logs:
            if log.get('Activity') == 'ViewReport':  # 过滤查看报表的活动类型
                user = log.get('UserId')
                report = log.get('ItemName')
                start_time = log.get('CreationTime')
                data.append({'User': user, 'Report': report, 'StartTime': start_time})
        # 创建DataFrame
        df = pd.DataFrame(data)
        print(df)
        # 可选：保存到CSV文件
        df.to_csv('powerbi_viewreport_logs_today_filtered.csv', index=False)
    except requests.exceptions.HTTPError as err:
        print(f"HTTP error occurred: {err}")
        print(f"Response text: {err.response.text}")
    except Exception as err:
        print(f"An error occurred: {err}")

额外注意事项 #

• 确保服务主体拥有Power BI租户管理员或工作区管理员权限，否则即使参数正确也会返回权限错误。
• 时间范围不能超过API允许的最大跨度（目前Power BI API允许的最大时间范围为30天）。

内容的提问来源于stack exchange，提问作者Aaron Ramsey