SAS Enterprise Guide 8.2：如何自动下载更新ZIP文件并提取执行内容

以下是一套可以实现自动检测网站ZIP文件更新、下载解压并执行内部文件的Python代码，核心逻辑是通过HTTP头信息判断文件是否更新，定时执行同步任务：

核心依赖 #

先安装需要的库：

pip install requests

完整代码 #

import os
import time
import requests
import zipfile
from subprocess import run

# 配置参数，请根据实际情况修改
TARGET_ZIP_URL = "https://example.com/your-target-file.zip"  # 替换为目标ZIP文件的URL
LOCAL_STORAGE_PATH = "./sync_data"  # 本地存储解压后文件的路径
UPDATE_RECORD_FILE = "./update_record.txt"  # 记录上次更新状态的文件
EXECUTABLE_FILE_NAME = "run.exe"  # ZIP内需要执行的文件名（根据实际修改，比如脚本的话是script.py）
CHECK_INTERVAL = 3600  # 检查更新的间隔时间（秒，这里是1小时）

def get_remote_file_etag(url):
    """获取远程文件的ETag（用于判断文件是否更新）"""
    try:
        response = requests.head(url, allow_redirects=True)
        response.raise_for_status()
        return response.headers.get("ETag")
    except requests.exceptions.RequestException as e:
        print(f"获取远程文件状态失败: {str(e)}")
        return None

def load_last_update_record():
    """加载上次更新的记录（ETag）"""
    if os.path.exists(UPDATE_RECORD_FILE):
        with open(UPDATE_RECORD_FILE, "r") as f:
            return f.read().strip()
    return ""

def save_update_record(etag):
    """保存最新的更新记录（ETag）"""
    with open(UPDATE_RECORD_FILE, "w") as f:
        f.write(etag)

def download_zip_file(url, save_path):
    """下载ZIP文件到本地"""
    try:
        response = requests.get(url, stream=True)
        response.raise_for_status()
        with open(save_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        print("ZIP文件下载完成")
        return True
    except requests.exceptions.RequestException as e:
        print(f"下载ZIP文件失败: {str(e)}")
        return False

def extract_zip_file(zip_path, extract_to):
    """解压ZIP文件到指定路径"""
    try:
        os.makedirs(extract_to, exist_ok=True)
        with zipfile.ZipFile(zip_path, "r") as zip_ref:
            zip_ref.extractall(extract_to)
        print("ZIP文件解压完成")
        # 删除临时ZIP文件
        os.remove(zip_path)
        return True
    except zipfile.BadZipFile:
        print("下载的ZIP文件损坏")
        os.remove(zip_path)
        return False
    except Exception as e:
        print(f"解压ZIP文件失败: {str(e)}")
        return False

def execute_target_file(file_path):
    """执行目标文件"""
    if not os.path.exists(file_path):
        print(f"要执行的文件不存在: {file_path}")
        return
    try:
        # 根据文件类型选择执行方式，这里以exe为例，若为Python脚本则用python命令
        if file_path.endswith(".exe"):
            run(file_path, check=True)
        elif file_path.endswith(".py"):
            run(["python", file_path], check=True)
        # 可根据需要添加其他类型文件的执行逻辑
        print("目标文件执行成功")
    except Exception as e:
        print(f"执行目标文件失败: {str(e)}")

def main_sync_task():
    """主同步任务"""
    # 获取远程文件的ETag
    remote_etag = get_remote_file_etag(TARGET_ZIP_URL)
    if not remote_etag:
        return
    
    # 获取上次更新的ETag
    last_etag = load_last_update_record()
    
    # 判断是否需要更新
    if remote_etag != last_etag:
        print("检测到文件更新，开始同步...")
        # 临时ZIP文件路径
        temp_zip_path = "./temp_download.zip"
        
        # 下载ZIP
        if download_zip_file(TARGET_ZIP_URL, temp_zip_path):
            # 解压ZIP
            if extract_zip_file(temp_zip_path, LOCAL_STORAGE_PATH):
                # 执行目标文件
                target_exec_path = os.path.join(LOCAL_STORAGE_PATH, EXECUTABLE_FILE_NAME)
                execute_target_file(target_exec_path)
                # 保存更新记录
                save_update_record(remote_etag)
    else:
        print("文件未更新，无需同步")

if __name__ == "__main__":
    print("启动自动同步任务...")
    while True:
        main_sync_task()
        print(f"等待{CHECK_INTERVAL}秒后再次检查更新...")
        time.sleep(CHECK_INTERVAL)

使用说明 #

1. 修改代码开头的配置参数，替换为你的目标URL、本地路径、要执行的文件名等
2. 若目标文件不是exe或Python脚本，需要在execute_target_file函数中添加对应执行逻辑
3. 运行代码后，程序会按照设定的间隔时间自动检查更新，有更新时自动执行同步流程

注意事项 #

• 请确保执行的文件是可信的，避免运行未知来源的可执行文件带来安全风险
• 可根据需求调整检查间隔时间，比如缩短为30分钟（1800秒）
• 若网站不返回ETag，可改用Last-Modified头信息判断更新，或者下载文件后计算哈希值对比（哈希对比效率较低，适合小文件）

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