使用Selenium自动刷新PowerBI - 谷冬

公众号：yunops

Selenium 不仅仅是一个优秀的自动化测试工具，其内置 webdriver 模块结合浏览器，还可以作为一种良好的自动化工具，用途代替人工执行一些需要在浏览器上进行的一些重复繁琐的工作，以下就自动刷新 PowerBI 数据集（更新大盘实时数据）的场景进行一次实际应用说明。

一、场景描述

我们使用了微软的 PowerBI 产品进行一些业务数据的图表化展示，常规化分为两种数据类型：
流式数据：需要 Client 端自行调用 API 上报数据；
实时数据：Server 端通过网关从 Client 抓取；
实际使用场景中，使用实时数据的图表（数据集）有个缺陷，实时数据抓取到 PowerBI Server 端后，图表并不会自动刷新，需要使用内置的定时刷新功能设置定期刷新（根据购买不同规格的服务提供不同的每日刷新次数限制）或者手动在页面上执行刷新操作，定时刷新有次数限制，人为手动刷新又是个极度劳神费力的活，所以就有了自动化的需求；
实现数据集自动化刷新提供以下两种方案：
第一种：自然就是调用 API 进行刷新，本来这是个很推荐的办法，奈何 PowerBI 的 API 接口认证太不友好了，调用 API 需要创建 Azure AD 用于权限验证，创建 Azure AD 需要 Azure 账户，创建 Azure 账户需要绑定双币信用卡，可拉倒吧~~此路不通。
第二种：模拟手动操作，可用来模拟手动操作的办法也不止一种，比如说使用按键精灵啊~使用 selenium 啊~等等，按键精灵坐标定位不太稳妥，还必须按键精灵、浏览器等软件全程打开，总之就很费劲；不如 selenium。

二、方案设计

自动化离不了编程，根据个人技术栈选择编程语言，我们这里选择 Python，综合定型为：在 CentOS 服务器上安装 Chrome，运行一个后台 Python 脚本，通过 Selenium Webdriver 调用 Chrome（无痕、无界面模式），结合 Python 的 schedule 模块定期执行刷新操作的整个流程：登录 —> 选择目标任务 —> 执行刷新操作 —> 注销登录。

三、环境准备

一个有权限执行刷新操作的用户（用户名+密码）
一台 CentOS7 系统的服务器，并且安装 Chrome、浏览器驱动：chromedriver、Python3（以及 schedule 和 selenium 等模块）；详细的安装过程就不细说了，相比大家也不需要这种基础的教程~就其中几个要点说一下：

CentOS7 中使用 yum 包管理器安装 Chrome 时需要添加选项参数禁用 GPG 签名检查（国内网络问题），否则会导致安装失败，命令示例：yum install google-chrome-stable --nogpgcheck -y
chromedriver 下载地址：
- 官方：https://chromedriver.storage.googleapis.com/index.html
- 国内淘宝镜像：http://npm.taobao.org/mirrors/chromedriver/

chromedriver 放置位置：

Windows：Python 安装目录中的 Scripts 目录里，如：D:\Program Files\Python\Python39\Scripts\
Linux（CentOS）：放到 PATH 路径下即可，如：/usr/local/sbin/ 或者 /usr/local/bin/

四、编写脚本：

核心功能就是使用 selenium 实现刷新操作的整个流程，并使用 python 的 schedule 模块实现定时调度，重点就是各个网页元素的定位，这里推荐使用 Selenium IDE 浏览器插件（直接在浏览器应用商店安装即可）自动捕获，然后将捕获到的操作流导出为 Python 格式，改吧改吧（删除冗余代码，更换更合适的网页元素定位方式：CSS_SELECTOR、ID、、XPATH、LINK_TEXT）就行~
由于代码运行速度远高于浏览器渲染速度，所以存在浏览器未渲染完成导致网页元素定位失败的问题，这种情况需要通过设置等待解决，有三种等待方式可选：强制等待（sleep）、隐形等待（implicitly_wait）、显示等待（WebDriverWait），不推荐使用固定等待和隐性等待，推荐使用更智能的显性等待，以下示例代码中用的时强制等待，别问为啥，问就是懒~
关于 schedule 补充使用说明：在实际使用中，不使用并行任务的话，通过 schedule 设置了好几个定时任务，但只有最后一个被执行，使用 sche dule 的 Parallel execution 模式解决。
示例代码：

# -*- coding: utf-8 -*-
###############################################################################
# About：用于定期自动运行 Power BI 数据集刷新任务
# Author：Alan
# Release：2021.11.04
# Usage：python3 -u task-refresh.py
# Attention：
#     1. 依赖Python3运行环境、schedule 和 selenium 等三方模块、Chrome浏览器
#     2. 如有新增、删除任务等需求，直接编辑主函数中的任务列表即可
###############################################################################
import os
import sys
import time
import logging
import threading
import schedule
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    filename=os.path.join(sys.path[0], 'task_refresh.log'),
    filemode='a',
    format=
    "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")


class PowerBIRefresh():
    def __init__(self, platform_name, task_name) -> None:
        self.opts = Options()
        # 无痕模式
        self.opts.add_argument("--incognito")
        # 设置为无界面模式 - 在无桌面环境的 Linux 中运行时必开此选项
        self.opts.add_argument('--headless')
        # 不加载图片, 提升速度
        self.opts.add_argument('--blink-settings=imagesEnabled=false')
        # 窗口最大化
        self.opts.add_argument("--start-maximized")
        # 解决DevToolsActivePort文件不存在的报错
        self.opts.add_argument('--no-sandbox')
        # 自定义变量，指定登录 power bi 的账户
        self.platform_name = platform_name
        self.task_name = task_name
        self.base_url = "<your_login_url>"
        self.username = "<your_username>"
        self.password = "<your_password>"

    # 判断当前页面指定元素是否存在
    def is_element_present(self, by, value):
        try:
            self.driver.find_element(by, value)
        except NoSuchElementException:
            return False
        return True

    # 执行刷新操作
    def task_refresh(self):
        start_time = time.time()
        logging.info("开始执行刷新任务, 【任务信息】平台: %s - 任务名称: %s", self.platform_name,
                     self.task_name)
        try:
            self.driver = webdriver.Chrome(chrome_options=self.opts)
            # 登录
            self.driver.get(self.base_url)
            self.driver.find_element(By.ID, "i0116").send_keys(self.username)
            time.sleep(0.5)
            self.driver.find_element(By.ID, "i0118").send_keys(self.password)
            time.sleep(2)
            self.driver.find_element(By.ID, "idSIButton9").click()
            time.sleep(1)
            self.driver.find_element(By.ID, "idSIButton9").click()
            time.sleep(0.5)
            self.driver.find_element(By.ID, "idSIButton9").click()
            time.sleep(1)
            # 打开工作区侧边栏
            self.driver.find_element(By.CSS_SELECTOR,
                                     ".workspacesPaneExpander").click()
            time.sleep(0.5)
            # 选择工作区
            self.driver.find_element(
                By.XPATH,
                "//button[@title='%s']" % self.platform_name).click()

            time.sleep(0.5)
            # 切换到：“数据集 + 数据流”一栏
            self.driver.find_element(By.CSS_SELECTOR,
                                     ".mat-tab-link:nth-child(3)").click()
            time.sleep(0.5)
            # 选择数据集
            self.driver.find_element(By.LINK_TEXT, self.task_name).click()
            time.sleep(1)
            # 展开刷线操作选项
            self.driver.find_element(
                By.CSS_SELECTOR,
                ".ng-star-inserted:nth-child(2) > .mat-menu-trigger > .mat-icon"
            ).click()
            time.sleep(0.5)

            # 点击刷新
            self.driver.find_element(
                By.CSS_SELECTOR,
                ".mat-menu-panel > div > button:nth-child(1)").click()
            time.sleep(5)

            # 检测是否正在刷新
            switch = self.is_element_present(By.CSS_SELECTOR, ".flat-button")
            if switch is True:
                logging.info("当前数据集刷新任务正在执行中, 中断本次执行...")
                self.driver.find_element(By.CSS_SELECTOR,
                                         ".flat-button").click()
            time.sleep(1)

            # 退出登录【可选】 - 在 CentOS 下注销按钮无法正常定位，会截获异常并退出，但也不影响主要功能；
            self.driver.find_element(By.CSS_SELECTOR,
                                     ".pbi-glyph-user").click()
            time.sleep(1)
            self.driver.find_element(By.LINK_TEXT, "注销").click()
            time.sleep(0.5)

            # 统计耗时
            end_time = time.time()
            excute_time = end_time - start_time
            print("数据集: %s刷新任务执行完成，耗时：%s" % (self.task_name, excute_time))
            logging.info("平台: %s - 任务名称: %s数据集刷新任务完成，本次执行耗时: %s",
                         self.platform_name, self.task_name, excute_time)
            self.driver.close()
            self.driver.quit()
        except Exception as e:
            dingding_alert()
            self.driver.close()
            self.driver.quit()
            logging.error(e)
            sys.exit(1)


# 空白函数 - 用于扩充告警功能
def dingding_alert():
    pass


# 用于并行执行任务
def threading_job(platform_name, task_name):
    job = PowerBIRefresh(platform_name, task_name)
    job_thread = threading.Thread(target=job.task_refresh)
    job_thread.start()


# 主函数
if __name__ == '__main__':
    # 任务列表
    # 并行 - 计划任务调度
    # 任务1：001-实时数据明细
    schedule.every(3).minutes.do(threading_job,
                                 platform_name="工作区001",
                                 task_name="001-实时数据明细").tag('demo_tag')
    # 任务2: 002-实时数据明细
    schedule.every(10).minutes.do(threading_job,
                                  platform_name="工作区002",
                                  task_name="002-实时数据明细").tag('demo_tag')

    # 开始计划任务调度
    while True:
        schedule.run_pending()
        time.sleep(5)

    # 取消任务
    # schedule.clear('demo_tag')

    # # 手动调度
    # do = PowerBIRefresh("工作区001", "001-实时数据明细")
    # do.task_refresh()

五、进程托管

可以使用 nohup 进行简单的后台托管，也可以托管为 systemd 系统任务
这里使用托管为 systemd 系统任务的方法，好处是支持故障自启动、开机自启动，在 /etc/systemd/system/ 目录中新增birefresh.service文件，添加如下内容（Python 可执行文件位置、脚本路径等信息自行更换）：

[Unit]
Description=refresh
After=multi-user.target

[Service]
ExecStart=/usr/bin/python3 -u <your_script_path>
Restart=on-failure
RestartSec=1min
StartLimitInterval=3600
StartLimitBurst=5

[Install]
WantedBy=multi-user.target

执行重载配置命令：systemctl daemon-reload
设置开机自启动：systemctl enable birefresh
愉快的使用 systemctl 进行生命周期管理：systemctl start/stop/restart/status birefresh
在 Power BI 控制台验证成果吧~

浏览器自动化工具 - selenium

一、场景描述

二、方案设计

三、环境准备

四、编写脚本：

五、进程托管