上一篇讲了使用Direct Query Mode来实现数据自动刷新,但是Direct Mode只能适用于Database这种数据源,很多其它的源都不行。对于其它类型的数据源,就只能另想办法了。
PBI刷新可以用以下2种方式:
1,在PBI Desktop中点击刷新,然后刷新完成后,再Publish
2,将报告发布到WorkSpace中,然后在选中Dataset,再刷新
这里介绍的是第二种方式,使用 Selenium + Request 方式刷新Dataset
先说刷新:当我们点击Workspace里的refresh之后,按下F12,点开Network,可以看到刷新PBI,实际上是触发了一个POST请求:https://wabi-west-europe-redirect.analysis.windows.net/powerbi/content/packages/9642769/refresh/,其中红字部分会根据报告的不同而改变,OK,接下来想办法如何正常执行这个POST请求就好了。
POST请求的Header很好拼接,只有authorization这一个变量,代表了登录人的信息,其它都是固定不变的
headers = { 'Host': 'wabi-mc-sha-redirect.analysis.chinacloudapi.cn', 'User-Agent': r'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', 'Authorization': authorization_json, 'Content-Type': 'application/json;charset=UTF-8', 'Origin': 'https://app.powerbi.com' } response = requests.post(url, headers=headers) if response.status_code == 200: log.info('Refresh successfully') else: log.error('Refresh failed')
最后,我们在想办法拿到登录的Cookie就可以了。这里使用Selenium来获取用户的登录信息
很多人使用selenium driver实例化浏览器的时候,其实都是使用的一个临时用户资料,打开一个初始浏览器,所以每次都要输入用户名,密码。如果先模拟登录,再取用户信息,这样就会有很多多余的操作。因此我们使用真实的用户去实例化浏览器。打开chrome://version/,可以找到用户资料的存储路径,如下
接下来使用真实的用户去登录PBI Workspace
from selenium import webdriver
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
capabilities = DesiredCapabilities.CHROME capabilities["goog:loggingPrefs"] = {"performance": "ALL"} option = webdriver.ChromeOptions() user_data = '--user-data-dir=C:/Users/your_name/AppData/Local/Google/Chrome/User Data' option.add_argument(user_data) # 设置成用户自己的数据目录 # chrome_driver = C:\LegacyApp\Python36\chromedriver.exe driver = webdriver.Chrome(executable_path=chrome_driver, chrome_options=option) workplace = 'https://app.powerbi.com/groups/4458f0a9-c43f-4dd5-913e-3fa5a042de64/list'
到此为止,就使用了真实用户登录到power bi online的workspace。
我们再获取一下authorization
authorization_json = '' for entry in logs: if "Bearer" in str(entry["message"]): json_message_data = json.loads(str(entry["message"])) authorization_json = json_message_data['message']['params']['request']['headers']['Authorization'] break # url = switch('Operation_And_Production_System') log.info(f'Authorization is {authorization_json}')
最后,我们再把 authorization,传给第一步里面的headers,就可以正常使用POST请求来刷新PBI了。
可以将程序打包成exe,挂到Task Schedule中运行
标签:数据源,实时,headers,json,刷新,message,PBI,authorization From: https://www.cnblogs.com/kykstyle/p/17968214