Sub FetchCustomerLevelFromMySQL()    Dim conn As Object ' ADODB.Connection    Dim rs As Object   ' ADODB.Recordset    Dim connStr As String    Dim sqlStr As String    Dim ws As Worksheet    Dim lastRow As Long, i As Long    Dim customerIdCol As Integer, vipLevelCol As Integer    Dim startTime As Double, endTime As Double    Dim customerId As String    On Error GoTo ErrorHandler    startTime = Timer ' 记录开始时间    ' 1. 定义连接字符串 (请替换为实际的数据库信息)    ' 格式: "Driver={MySQL ODBC 8.0 Unicode Driver};Server=服务器地址;Port=端口;Database=数据库名;Uid=用户名;Pwd=密码;"    connStr = "Driver={MySQL ODBC 8.0 Unicode Driver};Server=192.168.1.100;Port=3306;Database=customer_db;Uid=app_user;Pwd=YourSecurePassword123;"    ' 注意：实际应用中，密码不应硬编码。可考虑从单元格、输入框或加密文件中读取。    ' 2. 定义SQL查询语句    ' 假设我们需要VIP级别为1、2、3的客户    sqlStr = "SELECT customer_id, vip_level, customer_name FROM customers " & _             "WHERE vip_level IN ('VIP1', 'VIP2', 'VIP3') AND status = 'active'"    ' 3. 创建并打开连接    Set conn = CreateObject("ADODB.Connection")    Set rs = CreateObject("ADODB.Recordset")    conn.Open connStr    ' 4. 执行查询    rs.Open sqlStr, conn, 1, 1 ' 1,1 对应 adOpenKeyset, adLockReadOnly    ' 5. 准备Excel工作表    Set ws = ThisWorkbook.Sheets("外呼名单")    ' 假设A列是客户ID，B列是客户名，我们将VIP级别填在C列    customerIdCol = 1 ' A列    vipLevelCol = 3  ' C列    ' 设置表头    ws.Cells(1, vipLevelCol).Value = "VIP等级（实时）"    ' 6. 构建内存字典以便快速查找（优化关键：避免在Excel中逐行查找）    Dim vipDict As Object    Set vipDict = CreateObject("Scripting.Dictionary")    vipDict.CompareMode = vbTextCompare    ' 将查询结果加载到字典中，键为customer_id，值为vip_level    Do Until rs.EOF        customerId = CStr(rs.Fields("customer_id").Value)        ' 使用字典存储，customer_id 作为键，值可以是一个数组存储多个字段        vipDict(customerId) = Array(rs.Fields("vip_level").Value, rs.Fields("customer_name").Value)        rs.MoveNext    Loop    ' 7. 遍历Excel中的客户ID，从字典中匹配并填写VIP等级    lastRow = ws.Cells(ws.Rows.Count, customerIdCol).End(xlUp).Row    Dim matchCount As Long, updateCount As Long    matchCount = 0    updateCount = 0    For i = 2 To lastRow ' 假设第1行是标题        customerId = Trim(CStr(ws.Cells(i, customerIdCol).Value))        If Len(customerId) > 0 Then            If vipDict.Exists(customerId) Then                ' 匹配成功，写入VIP等级                Dim customerInfo                customerInfo = vipDict(customerId)                ws.Cells(i, vipLevelCol).Value = customerInfo(0) ' vip_level                ' 可选：也更新客户名（如果数据库中有更准确的）                ' ws.Cells(i, customerIdCol + 1).Value = customerInfo(1)                matchCount = matchCount + 1            Else                ' 未在数据库中找到，可以标记或清空                ws.Cells(i, vipLevelCol).Value = "非VIP"            End If        End If    Next i    ' 8. 清理与统计    rs.Close    conn.Close    Set rs = Nothing    Set conn = Nothing    endTime = Timer    Dim timeTaken As Double    timeTaken = endTime - startTime    ' 9. 输出结果报告    Dim msg As String    msg = "数据获取完成！" & vbNewLine & vbNewLine    msg = msg & "数据库查询记录数：" & vipDict.Count & vbNewLine    msg = msg & "Excel中匹配成功数：" & matchCount & vbNewLine    msg = msg & "总耗时：" & Format(timeTaken, "0.00") & " 秒"    MsgBox msg, vbInformation, "MySQL数据导入报告"    Exit SubErrorHandler:    MsgBox "发生错误：" & Err.Description & vbNewLine & _           "请检查数据库连接信息和SQL语句。", vbCritical    If Not rs Is Nothing Then        If rs.State = 1 Then rs.Close    End If    If Not conn Is Nothing Then        If conn.State = 1 Then conn.Close    End If    Set rs = Nothing    Set conn = NothingEnd Sub

import pymysqlimport pandas as pdfrom sqlalchemy import create_engineimport osfrom datetime import datetimeimport warningswarnings.filterwarnings('ignore')def fetch_customer_level_with_pandas(mysql_config, excel_input_path, excel_output_path=None):    """    从MySQL获取客户等级，并与Excel外呼名单合并。    参数:    mysql_config: 字典，包含数据库连接信息。    excel_input_path: 输入Excel文件路径。    excel_output_path: 输出Excel文件路径。如果为None，则覆盖原文件。    """    start_time = datetime.now()    if excel_output_path is None:        excel_output_path = excel_input_path.replace('.xlsx', '_with_vip.xlsx')    # 1. 建立数据库连接 (两种方式任选)    # 方式A: 使用pymysql + pandas.read_sql    connection = None    try:        # 读取Excel外呼名单        print(f"正在读取Excel文件: {excel_input_path}")        df_excel = pd.read_excel(excel_input_path, dtype={'customer_id': str})  # 确保ID为字符串        print(f"Excel记录数: {len(df_excel)}")        # 构建SQL查询，动态使用Excel中的ID列表        # 注意：如果ID列表很长，需要分批次查询或使用其他优化        id_list = df_excel['customer_id'].dropna().unique().tolist()        if not id_list:            print("警告：Excel中没有有效的customer_id。")            return        # 将ID列表格式化为SQL IN语句的字符串        # 为了防止SQL注入，我们使用参数化查询，但pymysql的IN语句需要特殊处理        # 这里使用安全的方式构建查询        placeholders = ','.join(['%s'] * len(id_list))        sql = f"""            SELECT customer_id, vip_level, customer_name, last_purchase_date, total_amount            FROM customers            WHERE customer_id IN ({placeholders})              AND status = 'active'              AND vip_level IS NOT NULL        """        print("正在连接数据库并执行查询...")        connection = pymysql.connect(**mysql_config)        # 执行查询，结果直接读入DataFrame        df_db = pd.read_sql(sql, connection, params=id_list)        print(f"数据库查询记录数: {len(df_db)}")        # 2. 数据合并（类似SQL的LEFT JOIN）        # 使用how='left'保留所有Excel中的记录，即使数据库中没有匹配        df_merged = pd.merge(df_excel, df_db, on='customer_id', how='left', suffixes=('', '_db'))        # 3. 数据清理与重命名        # 如果数据库中有同名字段，可以选择用数据库的值覆盖Excel的值        # 这里假设我们需要用数据库的vip_level更新        if 'vip_level_db' in df_merged.columns:            df_merged['vip_level'] = df_merged['vip_level_db']            df_merged.drop(columns=['vip_level_db'], inplace=True)        # 标记未匹配到的记录        df_merged['vip_level'] = df_merged['vip_level'].fillna('非VIP')        # 4. 保存到Excel        print(f"正在保存结果到: {excel_output_path}")        with pd.ExcelWriter(excel_output_path, engine='openpyxl') as writer:            df_merged.to_excel(writer, sheet_name='外呼名单_VIP', index=False)            # 可选：添加一个统计摘要sheet            summary_data = {                '统计项': ['处理时间', '输入记录数', '匹配VIP数', '非VIP数', '数据更新时间'],                '数值': [                    start_time.strftime('%Y-%m-%d %H:%M:%S'),                    len(df_excel),                    (df_merged['vip_level'] != '非VIP').sum(),                    (df_merged['vip_level'] == '非VIP').sum(),                    datetime.now().strftime('%Y-%m-%d %H:%M:%S')                ]            }            df_summary = pd.DataFrame(summary_data)            df_summary.to_excel(writer, sheet_name='执行摘要', index=False)        # 5. 打印统计信息        end_time = datetime.now()        time_taken = (end_time - start_time).total_seconds()        print("\n" + "="*50)        print("数据合并完成报告")        print("="*50)        print(f"输入文件: {excel_input_path}")        print(f"输出文件: {excel_output_path}")        print(f"总记录数: {len(df_merged)}")        print(f"VIP客户数: {(df_merged['vip_level'] != '非VIP').sum()}")        print(f"处理耗时: {time_taken:.2f} 秒")    except Exception as e:        print(f"处理过程中发生错误: {e}")        import traceback        traceback.print_exc()    finally:        if connection:            connection.close()def fetch_customer_level_with_sqlalchemy(connection_str, excel_input_path, sql_query=None):    """    使用sqlalchemy连接数据库，适合更复杂的查询和操作。    """    try:        # 创建数据库引擎        engine = create_engine(connection_str)        # 读取Excel        df_excel = pd.read_excel(excel_input_path, dtype={'customer_id': str})        # 如果没有提供SQL，则构建一个基于Excel中ID的查询        if sql_query is None:            id_list = df_excel['customer_id'].dropna().unique().tolist()            if not id_list:                print("警告：Excel中没有有效的customer_id。")                return            # 使用sqlalchemy的参数化查询            from sqlalchemy import text            sql = text("""                SELECT customer_id, vip_level, customer_name                 FROM customers                 WHERE customer_id IN :ids AND status = 'active'            """)            # 注意：这里需要将列表转换为元组            with engine.connect() as conn:                result = conn.execute(sql, {'ids': tuple(id_list)})                df_db = pd.DataFrame(result.fetchall(), columns=result.keys())        else:            # 执行自定义SQL            df_db = pd.read_sql(sql_query, engine)        # 后续合并逻辑与上面相同        # ...    except Exception as e:        print(f"错误: {e}")# 主程序示例if __name__ == "__main__":    # 数据库配置（从环境变量读取更安全）    mysql_config = {        'host': 'localhost',        'port': 3306,        'user': 'your_username',        'password': 'your_password',  # 建议从环境变量读取        'database': 'customer_db',        'charset': 'utf8mb4',        'cursorclass': pymysql.cursors.DictCursor    }    # 使用sqlalchemy连接字符串的格式    # connection_str = "mysql+pymysql://username:password@host:port/database"    # Excel文件路径    input_excel = "data/外呼客户名单.xlsx"    output_excel = "data/外呼客户名单_带VIP等级.xlsx"    # 调用函数    fetch_customer_level_with_pandas(mysql_config, input_excel, output_excel)    # 更安全的配置方式：从配置文件或环境变量读取    # import os    # mysql_config = {    #     'host': os.getenv('DB_HOST', 'localhost'),    #     'port': int(os.getenv('DB_PORT', 3306)),    #     'user': os.getenv('DB_USER'),    #     'password': os.getenv('DB_PASSWORD'),    #     'database': os.getenv('DB_NAME'),    # }