描述

certd证书下载脚本，用于certd证书自动续期部署工具的证书的下载。

代码

import json
from playwright.sync_api import sync_playwright
import requests
import logging
import tqdm

# 顶层函数定义
def load_config():
    with open('ssl_down.json') as f:
        return json.load(f)
config = load_config()
# 新增独立的下载函数
# 在文件顶部添加日志模块

# 初始化日志配置
logging.basicConfig(
    filename='ssl_download.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    encoding='utf-8'  # 新增编码设置

)
print("[步骤] 日志初始化完成")
logging.info("--日志初始化完成--")
def download_file(page, row_key, save_name):
    try:
        logging.info(f"开始下载 {save_name}")
        print(f"[步骤] 正在处理ID: {row_key} 的下载...")
        # 保持页面操作在主线程序列化执行

        # 原选择器
        # btn_selector = f'//tr[@data-row-key="{row_key}"]//button[@order="3"]'

        # 新选择器（新增title属性和order值变更）
        btn_selector = f'//tr[@data-row-key="{row_key}"]//button[@order="4" and @title="下载证书"]'

        page.wait_for_selector(btn_selector)
        page.click(btn_selector)

        # 弹窗处理
        download_link = page.wait_for_selector('.ant-modal-content a[href*="download"]').get_attribute('href')
        # 修改下载链接拼接方式

        download_url = f'{config["domain"]}{download_link}'

        # 下载逻辑（添加进度条）
        response = requests.get(
            download_url,
            headers={"User-Agent": "Mozilla/5.0"},
            cookies={cookie['name']: cookie['value'] for cookie in page.context.cookies()},
            stream=True  # 启用流式下载
        )

        # 获取文件总大小
        total_size = int(response.headers.get('content-length', 0))
        block_size = 1024  # 1KB
        progress_bar = tqdm.tqdm(
            total=total_size, 
            unit='iB', 
            unit_scale=True,
            desc=f"[下载] {save_name}"
        )

        with open(save_name, 'wb') as f:
            for data in response.iter_content(block_size):
                progress_bar.update(len(data))
                f.write(data)
        progress_bar.close()

        # 确保进度条完成
        if total_size != 0 and progress_bar.n != total_size:
            print("下载中断，文件可能不完整")

        logging.info(f"文件 {save_name} 下载成功")
        print(f"[完成] {save_name} 已保存")

        # 加强弹窗关闭逻辑
        def close_modal():
            print("[弹窗] 正在关闭下载提示...")
            # 修改后的选择器
            know_btn = page.wait_for_selector(
                'button.ant-btn-primary:has-text("关 闭")',
                timeout=40000
            )

            # 保持原有验证逻辑
            page.wait_for_function('''(btn) => {
                return btn.offsetWidth > 0 && btn.offsetHeight > 0;
            }''', arg=know_btn)

            # 点击方式保持不变
            know_btn.hover()
            page.mouse.down()
            page.mouse.up()

            # 验证弹窗关闭
            page.wait_for_selector('.ant-modal-content', state='hidden', timeout=15000)
            print("[弹窗] 弹窗关闭确认完成")

        close_modal()

    except Exception as e:
        logging.error(f"下载失败: {str(e)}", exc_info=True)
        print(f"[错误] 下载过程中发生异常: {str(e)}")
        raise  # 重新抛出异常供外层捕获

# 在main函数的导航操作后添加全局弹窗检测
def main():
    try:
        logging.info("====== 程序启动 ======")
        print("[系统] 正在初始化配置...")

        with sync_playwright() as p:
            print("[浏览器] 正在启动Chromium...")
            browser = p.chromium.launch(headless=config['headless'])
            context = browser.new_context()
            page = context.new_page()

            # 在此处添加页面设置
            page.set_default_timeout(45000)
            page.set_default_navigation_timeout(60000)

            try:  # 内部try只包裹可能失败的操作
                # 访问登录页面
                page.goto(f'{config["domain"]}/#/login')

                # 等待登录表单加载
                page.wait_for_selector('#custom-validation_username', state='visible')  # 改为使用实际ID

                # 输入用户名密码（根据实际ID修改选择器）
                page.fill('#custom-validation_username', config['username'])  # 原选择器是 input[name="username"]
                page.fill('#custom-validation_password', config['password'])   # 原选择器是 input[name="password"]

                # 点击登录按钮（更精确的选择器）
                page.click('button.login-button:has-text("登 录")')  # 原选择器是 button:has-text("登录")

                # 等待登录成功（替换固定等待为条件等待）

                page.wait_for_selector('#app', state='visible')  # 根据实际成功页元素添加
                print("登录成功！")

                # 新增点击菜单操作
                # 等待菜单加载

                # 通过组合选择器定位证书自动化流水线菜单项
                # 修改前的定位方式
                # menu_xpath = '//div[contains(@class, "menu-item-title")]/span[contains(., "证书自动化流水线")]'

                # 修改后的定位方式
                menu_xpath = '//li[contains(@class, "vben-menu-item")]//div[@class="vben-menu-item__content"]/span[text()="证书自动化流水线"]'

                # 添加显式等待和增强点击稳定性
                page.wait_for_selector(menu_xpath, state='visible', timeout=30000)
                page.hover(menu_xpath)
                page.wait_for_timeout(1000)  # 确保元素可交互
                page.click(menu_xpath)

                # 添加页面加载验证（根据实际情况选择一种）
                # 方式1：等待新URL（推荐用于页面跳转）
                page.wait_for_url(f'{config["domain"]}/#/certd/pipeline*', timeout=30000)  # 使用通配符匹配路径变化

                # 方式2：等待新页面关键元素（推荐用于单页应用）

                page.wait_for_selector('#app', state='visible')  # 根据实际成功页元素添加

                print("已成功进入证书流水线页面")

                # 从配置读取下载任务
                for task in config['download_tasks']:
                    download_file(
                        page,
                        task['row_key'],
                        task['save_name']
                    )

                # 删除原来的硬编码下载调用
                # download_file(page, '4', 'applic.zip')
                # download_file(page, '2', 'api_cn.zip')

            except Exception as e:
                logging.error(f"页面操作失败: {str(e)}", exc_info=True)
                raise  # 重新抛出异常到外层
            finally:
                # 将浏览器关闭移到最外层
                if hasattr(browser, 'is_connected') and browser.is_connected():
                    browser.close()
                    print("[系统] 浏览器已正常关闭")

    except Exception as e:
        logging.critical(f"程序异常终止: {str(e)}", exc_info=True)
        print(f"[严重错误] 程序运行失败: {str(e)}")
    finally:
        # 删除此处无效的页面设置代码
        logging.info("====== 程序结束 ======\n")

        # 设置全局等待策略

# 在main函数的导航操作后添加全局弹窗检测

if __name__ == "__main__":
    main()

JSON代码

{
    "domain": "http://ssl.api.com",
    "username": "admin",
    "password": "你的密码",
    "headless": true,
    "download_tasks": [   
        {
            "row_key": "4",
            "save_name": "applic.zip"
        },
        {
            "row_key": "2", 
            "save_name": "api.zip"
        }
    ]
}

配置说明

使用时删除这个注释行16行||提示： row_key 是流水线页面的ID号 ，save_name 是保存的文件名。{"row_key": "2", "save_name": "api_cn.zip"} 可以添加多个任务配置。

脚本下载

下载地址:点击下载