Python脚本更新Excel表后Power Query及外部连接失效求助

我用Python脚本从Shopify拉取订单并追加到Excel指定表格，脚本功能正常，但运行后打开Excel文件会弹出提示：

我们发现‘Analytics.xlsx’中的部分内容存在问题。是否要尝试尽可能恢复？如果信任此工作簿，请点击‘是’

点击“是”后，所有Power Query和外部连接都会被删除。我的Power Query并未关联脚本更新的表格，预先删除Power Query和外部连接则无报错，求排查原因。

import os
import requests
import pandas as pd
import sys
from time import sleep
from openpyxl import load_workbook
from openpyxl.utils.dataframe import dataframe_to_rows
from openpyxl.worksheet.table import Table
from openpyxl.utils import get_column_letter

# Shopify API credentials and initial URL
SHOP_NAME = 'xyz'
ACCESS_TOKEN = 'token'
BASE_URL = f"https://{SHOP_NAME}.myshopify.com/admin/api/2024-04/orders.json"

headers = {
    "X-Shopify-Access-Token": ACCESS_TOKEN,
    "Content-Type": "application/json"
}

def fetch_orders_since(last_date):
    orders = []
    url = BASE_URL + f"?status=any&limit=250&created_at_min={last_date}"

    while url:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        data = response.json()
        orders.extend(data['orders'])
        link_header = response.headers.get('link')

        next_url = None
        if link_header:
            links = link_header.split(',')
            for link in links:
                if 'rel="next"' in link:
                    next_url = link.split(';')[0].strip()[1:-1]
                    break

        url = next_url

    return orders

def fetch_metafields(order_id):
    metafields_url = f"https://{SHOP_NAME}.myshopify.com/admin/api/2024-04/orders/{order_id}/metafields.json"
    response = requests.get(metafields_url, headers=headers)
    response.raise_for_status()
    return response.json().get('metafields', [])

def add_metafields_to_orders(orders):
    for order in orders:
        order_id = order['id']
        metafields = fetch_metafields(order_id)
        order['metafields'] = metafields
        sleep(0.5)
    return orders

def convert_type(value, target_type):
    if target_type == 'str':
        return str(value)
    elif target_type == 'int':
        try:
            return int(value)
        except ValueError:
            return 0
    elif target_type == 'float':
        try:
            return float(value)
        except ValueError:
            return 0.0
    elif target_type == 'bool':
        return bool(value)
    elif target_type == 'datetime':
        return pd.to_datetime(value, errors='coerce')
    else:
        return value

def get_column_types(ws, table):
    column_types = []
    table_range = table.ref
    start_cell, end_cell = table_range.split(':')
    start_row = ws[start_cell].row

    headers = [cell.value for cell in ws[start_row]]
    for column_name in headers:
        column_letter = ws[start_row][headers.index(column_name)].column_letter
        cell_value = ws[f"{column_letter}{start_row + 1}"].value  # Assuming data starts from the row below headers
        column_types.append(type(cell_value).__name__)
    return column_types

def append_orders_to_excel(orders, file_path, sheet_name='Shopify Orders', table_name='orders'):
    df = pd.json_normalize(orders)
    wb = load_workbook(file_path)
    ws = wb[sheet_name]

    table = ws.tables.get(table_name)

    if table is None:
        raise ValueError(f"Table {table_name} not found in sheet {sheet_name}")

    column_types = get_column_types(ws, table)

    if not df.empty:
        for r in dataframe_to_rows(df, index=False, header=False):
            converted_row = [convert_type(value, column_types[idx]) for idx, value in enumerate(r)]
            ws.append(converted_row)

    # Update the table range to include new rows
    start_cell = table.ref.split(':')[0]
    end_cell = f"{get_column_letter(ws.max_column)}{ws.max_row}"
    table.ref = f"{start_cell}:{end_cell}"

    wb.save(file_path)

if __name__ == "__main__":
    last_date = sys.argv[1] if len(sys.argv) > 1 else '2024-05-26T04:00:00'
    file_path = r'C:\path\to\Analytics.xlsx'

    orders = fetch_orders_since(last_date)
    orders_with_metafields = add_metafields_to_orders(orders)
    append_orders_to_excel(orders_with_metafields, file_path)
    print("Orders have been successfully appended to the Excel file")

核心原因 #

openpyxl在处理包含Power Query/外部连接的Excel文件时，会丢失这类依赖的XML元数据：

• Excel的Power Query和外部连接信息存储在工作簿的xl/connections.xml、xl/workbook.xml及xl/queries/等隐藏文件中
• openpyxl加载并保存文件时，仅聚焦于单元格、表格等核心数据，不会完整保留这些非内容类元数据，导致Excel检测到文件结构损坏，触发修复流程并删除无法识别的Power Query/连接信息

解决方案 #

方案1：改用pandas ExcelWriter追加（保留元数据） #

修改append_orders_to_excel函数，通过pandas的ExcelWriter以追加模式写入，避免破坏原有元数据：

def append_orders_to_excel(orders, file_path, sheet_name='Shopify Orders', table_name='orders'):
    df = pd.json_normalize(orders)
    if df.empty:
        return

    # 获取现有表格列类型并转换数据
    wb = load_workbook(file_path)
    ws = wb[sheet_name]
    table = ws.tables.get(table_name)
    if table is None:
        raise ValueError(f"Table {table_name} not found in sheet {sheet_name}")
    
    column_types = get_column_types(ws, table)
    for idx, col in enumerate(df.columns):
        df[col] = df[col].apply(lambda x: convert_type(x, column_types[idx]))
    
    # 用ExcelWriter追加数据，保留元数据
    with pd.ExcelWriter(file_path, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer:
        # 计算数据起始行（表格最后一行的下一行）
        start_row = int(table.ref.split(':')[0][1:]) + len(ws[table.ref])
        df.to_excel(writer, sheet_name=sheet_name, startrow=start_row, index=False, header=False)
    
    # 再次加载文件更新表格范围
    wb = load_workbook(file_path)
    ws = wb[sheet_name]
    table = ws.tables.get(table_name)
    start_cell = table.ref.split(':')[0]
    end_cell = f"{get_column_letter(ws.max_column)}{ws.max_row}"
    table.ref = f"{start_cell}:{end_cell}"
    wb.save(file_path)

方案2：使用win32com直接操作Excel（完全保留所有特性） #

如果运行环境是Windows，可通过pywin32调用本地Excel应用，完整保留所有Excel特性：

import win32com.client as win32

def append_orders_to_excel_win32(orders, file_path, sheet_name='Shopify Orders', table_name='orders'):
    df = pd.json_normalize(orders)
    if df.empty:
        return

    # 转换数据类型
    wb = load_workbook(file_path, read_only=True)
    ws = wb[sheet_name]
    table = ws.tables.get(table_name)
    column_types = get_column_types(ws, table)
    for idx, col in enumerate(df.columns):
        df[col] = df[col].apply(lambda x: convert_type(x, column_types[idx]))
    
    # 调用Excel应用
    excel = win32.gencache.EnsureDispatch('Excel.Application')
    excel.Visible = False
    wb = excel.Workbooks.Open(file_path)
    ws = wb.Sheets(sheet_name)
    
    # 获取表格最后一行，写入数据
    table_obj = ws.ListObjects(table_name)
    last_row = table_obj.Range.Rows.Count + table_obj.HeaderRowRange.Row
    data = df.values.tolist()
    ws.Range(ws.Cells(last_row, 1), ws.Cells(last_row + len(data) - 1, len(df.columns))).Value = data
    # 刷新表格范围
    table_obj.Resize(table_obj.Range.Resize(last_row + len(data) - 1 - table_obj.HeaderRowRange.Row + 1))
    
    wb.Save()
    wb.Close()
    excel.Quit()

方案3：临时备份恢复元数据（繁琐不推荐） #

1. 将Excel文件改为.zip格式，解压后提取xl/connections.xml、xl/workbook.xml和xl/queries/文件夹
2. 用openpyxl修改文件后保存
3. 将提取的元文件替换回修改后的zip包，再改回.xlsx格式

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