Dim conn As ADODB.ConnectionSet conn = New ADODB.Connection' 连接字符串配置conn.ConnectionString = "Provider=Microsoft.ACE.OLEDB.12.0;" & _                       "Data Source=C:\Data\销售数据库.accdb;" & _                       "User ID=Admin;Password=;"conn.Open

Sub ImportExcelToAccess()    Dim conn As ADODB.Connection    Dim rs As ADODB.Recordset    Dim ws As Worksheet    Dim lastRow As Long, i As Long    Set conn = New ADODB.Connection    conn.Open "Provider=Microsoft.ACE.OLEDB.12.0;Data Source=C:\Data\销售数据库.accdb;"    Set ws = ThisWorkbook.Sheets("销售数据")    lastRow = ws.Cells(ws.Rows.Count, 1).End(xlUp).Row    ' 创建记录集并添加数据    Set rs = New ADODB.Recordset    rs.Open "销售记录", conn, adOpenDynamic, adLockOptimistic    For i = 2 To lastRow  ' 跳过标题行        rs.AddNew        rs("日期") = ws.Cells(i, 1).Value        rs("产品编号") = ws.Cells(i, 2).Value        rs("销售数量") = ws.Cells(i, 3).Value        rs("销售额") = ws.Cells(i, 4).Value        rs.Update    Next i    rs.Close    conn.Close    MsgBox "成功导入 " & (lastRow - 1) & " 条记录到Access数据库"End Sub

Sub QueryAccessToExcel()    Dim conn As ADODB.Connection    Dim rs As ADODB.Recordset    Dim ws As Worksheet    Dim i As Integer    Set conn = New ADODB.Connection    conn.Open "Provider=Microsoft.ACE.OLEDB.12.0;Data Source=C:\Data\销售数据库.accdb;"    Set rs = New ADODB.Recordset    ' 执行复杂查询：按产品分组统计销售额    rs.Open "SELECT 产品编号, SUM(销售数量) AS 总数量, SUM(销售额) AS 总销售额 " & _           "FROM 销售记录 " & _           "WHERE 日期 BETWEEN #2024-01-01# AND #2024-12-31# " & _           "GROUP BY 产品编号 " & _           "ORDER BY 总销售额 DESC", conn    Set ws = ThisWorkbook.Sheets.Add    ws.Name = "销售汇总"    ' 写入标题行    For i = 0 To rs.Fields.Count - 1        ws.Cells(1, i + 1).Value = rs.Fields(i).Name    Next i    ' 使用CopyFromRecordset快速导入数据（性能优化）    ws.Range("A2").CopyFromRecordset rs    ' 设置格式    ws.Range("C:D").NumberFormat = "#,##0"    ws.Columns.AutoFit    rs.Close    conn.Close    MsgBox "查询完成，共返回 " & (ws.Cells(ws.Rows.Count, 1).End(xlUp).Row - 1) & " 条记录"End Sub

Sub SafeDatabaseOperation()    On Error GoTo ErrorHandler    Dim conn As ADODB.Connection    Set conn = New ADODB.Connection    ' 启用事务    conn.BeginTrans    conn.Open "Provider=Microsoft.ACE.OLEDB.12.0;Data Source=C:\Data\销售数据库.accdb;"    ' 执行多个数据库操作    conn.Execute "UPDATE 库存 SET 数量 = 数量 - 10 WHERE 产品ID = 'P001'"    conn.Execute "INSERT INTO 销售记录 (产品ID, 数量, 日期) VALUES ('P001', 10, Date)"    ' 提交事务    conn.CommitTrans    MsgBox "操作成功完成"    GoTo CleanUpErrorHandler:    ' 回滚事务    If Not conn Is Nothing Then conn.RollbackTrans    MsgBox "错误 " & Err.Number & ": " & Err.Description & vbCrLf & _           "操作已回滚", vbCriticalCleanUp:    If Not conn Is Nothing Then        If conn.State = adStateOpen Then conn.Close        Set conn = Nothing    End IfEnd Sub

pip install pyodbc pandas sqlalchemy

import pyodbcimport pandas as pdfrom sqlalchemy import create_enginedef create_access_connection(db_path):    """创建Access数据库连接"""    connection_string = (        r"DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};"        r"DBQ=" + db_path + ";"    )    try:        conn = pyodbc.connect(connection_string)        print("数据库连接成功")        return conn    except pyodbc.Error as e:        print(f"连接失败: {e}")        return None# 使用SQLAlchemy创建引擎（用于pandas）def create_sqlalchemy_engine(db_path):    """创建SQLAlchemy引擎"""    connection_url = f"access+pyodbc:///?odbc_connect={pyodbc.connect(db_path).connection_string}"    engine = create_engine(connection_url)    return engine3.2 使用pandas进行数据导入导出pandas提供了高级数据操作接口，极大简化了数据库交互：class AccessDataManager:    """Access数据库管理器"""    def __init__(self, db_path):        self.db_path = db_path        self.engine = create_sqlalchemy_engine(db_path)    def excel_to_access(self, excel_path, sheet_name, table_name):        """将Excel工作表导入Access数据库"""        try:            # 读取Excel数据            df = pd.read_excel(excel_path, sheet_name=sheet_name)            # 数据清洗与预处理            df_cleaned = self.clean_data(df)            # 导入到Access            df_cleaned.to_sql(table_name, self.engine,                             if_exists='replace', index=False)            print(f"成功导入 {len(df_cleaned)} 条记录到表 {table_name}")            return True        except Exception as e:            print(f"导入失败: {e}")            return False    def clean_data(self, df):        """数据清洗处理"""        # 处理空值        df_cleaned = df.dropna(subset=['关键字段'])        # 数据类型转换        df_cleaned['日期字段'] = pd.to_datetime(df_cleaned['日期字段'], errors='coerce')        df_cleaned['数值字段'] = pd.to_numeric(df_cleaned['数值字段'], errors='coerce')        # 去除重复记录        df_cleaned = df_cleaned.drop_duplicates()        return df_cleaned    def query_to_dataframe(self, sql_query):        """执行查询返回DataFrame"""        try:            df = pd.read_sql(sql_query, self.engine)            print(f"查询返回 {len(df)} 条记录")            return df        except Exception as e:            print(f"查询失败: {e}")            return pd.DataFrame()  # 返回空DataFrame    def dataframe_to_excel(self, df, excel_path, sheet_name):        """将DataFrame导出到Excel"""        try:            with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:                df.to_excel(writer, sheet_name=sheet_name, index=False)            print(f"数据已导出到 {excel_path}")            return True        except Exception as e:            print(f"导出失败: {e}")            return False# 使用示例if __name__ == "__main__":    manager = AccessDataManager(r"C:\Data\销售数据库.accdb")    # 从Excel导入数据    manager.excel_to_access("销售数据.xlsx", "月度数据", "销售记录")    # 执行复杂查询    query = """    SELECT 产品编号,            SUM(销售数量) AS 总数量,           SUM(销售额) AS 总销售额,           AVG(销售额) AS 平均销售额    FROM 销售记录     WHERE 日期 BETWEEN '2024-01-01' AND '2024-12-31'    GROUP BY 产品编号    HAVING SUM(销售额) > 10000    ORDER BY 总销售额 DESC    """    result_df = manager.query_to_dataframe(query)    # 导出到Excel    if not result_df.empty:        manager.dataframe_to_excel(result_df, "销售分析报告.xlsx", "汇总数据")

def advanced_data_analysis(db_manager):    """高级数据分析示例"""    # 执行基础查询    sales_df = db_manager.query_to_dataframe("SELECT * FROM 销售记录")    if sales_df.empty:        print("无数据可分析")        return    # 使用pandas进行复杂分析    analysis_results = {}    # 1. 销售趋势分析    sales_df['日期'] = pd.to_datetime(sales_df['日期'])    monthly_sales = sales_df.groupby(sales_df['日期'].dt.to_period('M')).agg({        '销售额': ['sum', 'mean', 'count'],        '销售数量': 'sum'    }).round(2)    analysis_results['月度趋势'] = monthly_sales    # 2. 产品表现分析    product_performance = sales_df.groupby('产品编号').agg({        '销售额': ['sum', 'mean', 'std'],        '销售数量': 'sum'    }).round(2)    # 添加排名    product_performance['销售额排名'] = product_performance[('销售额', 'sum')].rank(ascending=False)    analysis_results['产品表现'] = product_performance    # 3. 高级统计计算    from scipy import stats    sales_values = sales_df['销售额'].dropna()    if len(sales_values) > 0:        analysis_results['相关系数'] = sales_df[['销售额', '销售数量']].corr()        analysis_results['销售额统计'] = {            '偏度': stats.skew(sales_values),            '峰度': stats.kurtosis(sales_values),            '变异系数': sales_values.std() / sales_values.mean()        }    return analysis_results# 生成分析报告def generate_analysis_report(analysis_results, output_path):    """生成详细分析报告"""    with pd.ExcelWriter(output_path, engine='openpyxl') as writer:        for sheet_name, data in analysis_results.items():            if isinstance(data, pd.DataFrame):                data.to_excel(writer, sheet_name=sheet_name)            else:                # 处理字典类型结果                pd.DataFrame.from_dict(data, orient='index').to_excel(                    writer, sheet_name=sheet_name)    print(f"分析报告已生成: {output_path}")

def batch_processing_optimized(db_manager, large_excel_path, chunk_size=10000):    """批量处理大文件（内存优化）"""    # 分块读取大型Excel文件    chunk_reader = pd.read_excel(large_excel_path, chunksize=chunk_size)    total_processed = 0    for i, chunk in enumerate(chunk_reader):        # 处理每个数据块        processed_chunk = db_manager.clean_data(chunk)        # 分块写入数据库        if_exists = 'replace' if i == 0 else 'append'        processed_chunk.to_sql('大数据表', db_manager.engine,                             if_exists=if_exists, index=False)        total_processed += len(processed_chunk)        print(f"已处理第 {i+1} 块数据，累计 {total_processed} 条记录")        # 释放内存        del processed_chunk    print(f"批量处理完成，共处理 {total_processed} 条记录")def parallel_processing(db_manager, excel_files):    """多文件并行处理"""    from concurrent.futures import ThreadPoolExecutor    import os    def process_single_file(file_path):        """处理单个文件"""        try:            file_name = os.path.basename(file_path)            sheet_name = os.path.splitext(file_name)[0]            table_name = sheet_name            db_manager.excel_to_access(file_path, sheet_name, table_name)            return f"成功处理: {file_name}"        except Exception as e:            return f"处理失败 {file_name}: {e}"    # 使用线程池并行处理    with ThreadPoolExecutor(max_workers=3) as executor:        results = list(executor.map(process_single_file, excel_files))    for result in results:        print(result)

# Python可轻松集成其他数据源和高级功能def multi_database_integration():    """多数据库集成示例"""    # 同时连接多个数据源    sources = {        'access': 'Access连接字符串',        'mysql': 'MySQL连接字符串',         'api': 'Web API端点'    }    # 统一数据处理流程    combined_data = pd.DataFrame()    for name, conn_str in sources.items():        data = pd.read_sql("SELECT * FROM 主要表", conn_str)        data['数据源'] = name        combined_data = pd.concat([combined_data, data])    return combined_data

class SalesAnalysisSystem:    """销售数据分析系统（混合架构）"""    def __init__(self, access_db_path, template_folder):        self.db_manager = AccessDataManager(access_db_path)        self.template_folder = template_folder    def daily_automation_pipeline(self):        """每日自动化流水线"""        try:            # 1. 收集各门店Excel数据            store_files = self.collect_store_reports()            # 2. 并行导入数据库            self.batch_import_data(store_files)            # 3. 数据质量检查            quality_report = self.data_quality_check()            # 4. 生成分析报告            analysis_results = self.comprehensive_analysis()            # 5. 生成最终报告            self.generate_final_report(analysis_results, quality_report)            # 6. 异常警报            self.send_alerts(analysis_results)            return "每日处理流程完成"        except Exception as e:            error_msg = f"自动化流程失败: {e}"            self.send_alerts({'错误信息': error_msg})            return error_msg    def generate_vba_interface(self):        """生成VBA前端接口供Excel调用"""        vba_code = """        Sub 运行销售分析()            ' 调用Python后端处理            Shell "python sales_analysis.py --daily"            ' 刷新Excel数据连接            ThisWorkbook.RefreshAll            MsgBox "销售分析完成，数据已更新"        End Sub        """        with open(os.path.join(self.template_folder, "自动化接口.bas"), "w") as f:            f.write(vba_code)

' 禁用屏幕更新和自动计算提升性能Sub OptimizedVBAOperation()    Application.ScreenUpdating = False    Application.Calculation = xlCalculationManual    Application.EnableEvents = False    ' 执行数据操作...    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    Application.EnableEvents = TrueEnd Sub

def memory_efficient_processing(large_dataset):    """内存高效处理"""    # 使用生成器避免内存溢出    for chunk in pd.read_sql_query(large_dataset, chunksize=10000):        processed_chunk = process_data_chunk(chunk)        yield processed_chunk# 使用索引提升查询性能def create_query_indexes(db_manager):    """创建查询索引"""    index_queries = [        "CREATE INDEX idx_sales_date ON 销售记录(日期)",        "CREATE INDEX idx_product_id ON 销售记录(产品编号)",        "CREATE INDEX idx_date_product ON 销售记录(日期, 产品编号)"    ]    for query in index_queries:        try:            db_manager.engine.execute(query)            print("索引创建成功")        except Exception as e:            print(f"索引创建失败: {e}")

import loggingfrom datetime import datetimedef setup_logging():    """配置日志系统"""    logging.basicConfig(        level=logging.INFO,        format='%(asctime)s - %(levelname)s - %(message)s',        handlers=[            logging.FileHandler(f"database_log_{datetime.now().strftime('%Y%m%d')}.log"),            logging.StreamHandler()        ]    )    return logging.getLogger(__name__)class RobustDataManager:    """健壮的数据管理器"""    def __init__(self, db_path):        self.logger = setup_logging()        self.db_path = db_path    def safe_database_operation(self, operation_func, *args):        """安全的数据库操作封装"""        try:            result = operation_func(*args)            self.logger.info(f"操作成功: {operation_func.__name__}")            return result        except pyodbc.Error as e:            self.logger.error(f"数据库错误: {e}")            # 重试逻辑            return self.retry_operation(operation_func, *args)        except Exception as e:            self.logger.error(f"未知错误: {e}")            raise