Python表格自动化工具：10个实用工具详解

importos
fromopenpyxlimportload_workbook
fromPILimportImage

defextract_images_from_excel(excel_path, output_folder):
    """从Excel文件中提取所有图片"""
    wb = load_workbook(excel_path)
    ws = wb.active
    
    # 创建输出文件夹
    os.makedirs(output_folder, exist_ok=True)
    
    # 提取图片
    forimageinws._images:
        # 获取图片数据
        img_data = image._data()
        
        # 保存图片
        img = Image.open(io.BytesIO(img_data))
        img_filename = f"image_{image.anchor._from.row}_{image.anchor._from.col}.png"
        img.save(os.path.join(output_folder, img_filename))
    
    print(f"已从 {excel_path} 提取图片到 {output_folder}")

# 使用示例
extract_images_from_excel("产品目录.xlsx", "提取的图片")

importopenpyxl
importre

defbatch_replace_in_excel(excel_path, rules_file, output_path):
    """根据规则文件批量替换Excel内容"""
    # 加载替换规则
    rules = {}
    withopen(rules_file, 'r', encoding='utf-8') asf:
        forlineinf:
            if'→'inline:
                old, new = line.strip().split('→')
                rules[old.strip()] = new.strip()
    
    # 加载Excel文件
    wb = openpyxl.load_workbook(excel_path)
    ws = wb.active
    
    # 执行替换
    forrowinws.iter_rows():
        forcellinrow:
            ifcell.valueandisinstance(cell.value, str):
                forold_text, new_textinrules.items():
                    ifre.search(old_text, cell.value):
                        cell.value = re.sub(old_text, new_text, cell.value)
    
    # 保存结果
    wb.save(output_path)
    print(f"替换完成，结果已保存到 {output_path}")

# 使用示例
batch_replace_in_excel("原始数据.xlsx", "替换规则.txt", "清洗后数据.xlsx")

importpandasaspd
importsqlite3

defexcel_to_sql(excel_path, table_name, db_path="output.db"):
    """将Excel数据转换为SQLite数据库"""
    # 读取Excel数据
    df = pd.read_excel(excel_path)
    
    # 创建数据库连接
    conn = sqlite3.connect(db_path)
    
    # 将数据写入数据库
    df.to_sql(table_name, conn, if_exists='replace', index=False)
    
    # 生成SQL文件
    sql_content = f"CREATE TABLE {table_name} (\n"
    forcolindf.columns:
        sql_content += f"    {col} TEXT,\n"
    sql_content = sql_content.rstrip(',\n') +"\n);\n\n"
    
    # 生成INSERT语句
    for_, rowindf.iterrows():
        values = ", ".join([f"'{str(val).replace(\"'\", \"''\")}'"forvalinrow])
        sql_content += f"INSERT INTO {table_name} VALUES ({values});\n"
    
    # 保存SQL文件
    withopen(f"{table_name}.sql", "w", encoding="utf-8") asf:
        f.write(sql_content)
    
    conn.close()
    print(f"转换完成，SQL文件已保存为 {table_name}.sql")

# 使用示例
excel_to_sql("员工数据.xlsx", "employees")

importpandasaspd
importnumpyasnp

defcompare_excel_files(file1, file2, key_columns=None, output_file="差异报告.xlsx"):
    """比较两个Excel文件的差异"""
    # 读取数据
    df1 = pd.read_excel(file1)
    df2 = pd.read_excel(file2)
    
    # 设置比对键
    ifkey_columns:
        df1 = df1.set_index(key_columns)
        df2 = df2.set_index(key_columns)
    
    # 找出差异
    comparison = df1.compare(df2)
    
    ifnotcomparison.empty:
        # 保存差异报告
        withpd.ExcelWriter(output_file) aswriter:
            comparison.to_excel(writer, sheet_name='差异详情')
            
            # 统计差异
            stats = pd.DataFrame({
                '统计项': ['总行数差异', '不同值数量', '文件1独有行', '文件2独有行'],
                '数量': [
                    len(df1) -len(df2),
                    comparison.shape[0],
                    len(df1[~df1.index.isin(df2.index)]),
                    len(df2[~df2.index.isin(df1.index)])
                ]
            })
            stats.to_excel(writer, sheet_name='差异统计', index=False)
        
        print(f"发现差异，报告已保存到 {output_file}")
        returncomparison
    else:
        print("两个文件内容完全相同")
        returnNone

# 使用示例
compare_excel_files("版本1.xlsx", "版本2.xlsx", key_columns=["员工ID"])

importpandasaspd
importglob
importos

defsummarize_excel_files(folder_path, summary_column, output_file="汇总结果.xlsx"):
    """汇总文件夹中所有Excel文件的指定列"""
    # 获取所有Excel文件
    excel_files = glob.glob(os.path.join(folder_path, "*.xlsx"))
    excel_files.extend(glob.glob(os.path.join(folder_path, "*.xls")))
    
    ifnotexcel_files:
        print("未找到Excel文件")
        return
    
    # 读取并合并数据
    all_data = []
    file_info = []
    
    forfile_pathinexcel_files:
        try:
            df = pd.read_excel(file_path)
            ifsummary_columnindf.columns:
                # 添加文件名列
                df['来源文件'] = os.path.basename(file_path)
                all_data.append(df)
                file_info.append({
                    '文件名': os.path.basename(file_path),
                    '行数': len(df),
                    f'{summary_column}总和': df[summary_column].sum(),
                    f'{summary_column}平均值': df[summary_column].mean()
                })
        exceptExceptionase:
            print(f"读取文件 {file_path} 时出错: {e}")
    
    ifnotall_data:
        print("没有找到包含指定列的数据")
        return
    
    # 合并所有数据
    combined_df = pd.concat(all_data, ignore_index=True)
    
    # 生成汇总统计
    summary_stats = pd.DataFrame(file_info)
    group_summary = combined_df.groupby('来源文件')[summary_column].agg(['sum', 'mean', 'count']).reset_index()
    
    # 保存结果
    withpd.ExcelWriter(output_file) aswriter:
        combined_df.to_excel(writer, sheet_name='合并数据', index=False)
        summary_stats.to_excel(writer, sheet_name='文件统计', index=False)
        group_summary.to_excel(writer, sheet_name='分组汇总', index=False)
    
    print(f"汇总完成，结果已保存到 {output_file}")
    returncombined_df

# 使用示例
summarize_excel_files("月度报表", "销售额")

fromopenpyxlimportload_workbook
fromopenpyxl.stylesimportFont, Alignment, PatternFill, Border, Side
fromopenpyxl.formatting.ruleimportColorScaleRule

defbeautify_excel(input_file, output_file):
    """美化Excel表格格式"""
    wb = load_workbook(input_file)
    ws = wb.active
    
    # 设置标题样式
    title_font = Font(name='微软雅黑', size=14, bold=True, color='FFFFFF')
    title_fill = PatternFill(start_color='366092', end_color='366092', fill_type='solid')
    
    forcellinws[1]:
        cell.font = title_font
        cell.fill = title_fill
        cell.alignment = Alignment(horizontal='center', vertical='center')
    
    # 设置表头样式
    header_font = Font(name='微软雅黑', size=11, bold=True, color='000000')
    header_fill = PatternFill(start_color='D9E1F2', end_color='D9E1F2', fill_type='solid')
    
    forcellinws[2]:
        cell.font = header_font
        cell.fill = header_fill
        cell.alignment = Alignment(horizontal='center', vertical='center')
    
    # 设置数据区域样式
    thin_border = Border(
        left=Side(style='thin'),
        right=Side(style='thin'),
        top=Side(style='thin'),
        bottom=Side(style='thin')
    )
    
    forrowinws.iter_rows(min_row=3):
        forcellinrow:
            cell.border = thin_border
            cell.alignment = Alignment(horizontal='left', vertical='center')
    
    # 自动调整列宽
    forcolumninws.columns:
        max_length = 0
        column_letter = column[0].column_letter
        
        forcellincolumn:
            try:
                iflen(str(cell.value)) >max_length:
                    max_length = len(str(cell.value))
            except:
                pass
        
        adjusted_width = min(max_length+2, 50)
        ws.column_dimensions[column_letter].width = adjusted_width
    
    # 添加条件格式（数值越大颜色越深）
    ifws.max_column>= 3:
        color_scale_rule = ColorScaleRule(
            start_type='min', start_color='FFFFFF',
            end_type='max', end_color='63BE7B'
        )
        ws.conditional_formatting.add(f'C3:C{ws.max_row}', color_scale_rule)
    
    wb.save(output_file)
    print(f"表格美化完成，结果已保存到 {output_file}")

# 使用示例
beautify_excel("原始报表.xlsx", "美化后报表.xlsx")

importpandasaspd
fromopenpyxlimportWorkbook
fromopenpyxl.utils.dataframeimportdataframe_to_rows

defcreate_pivot_table(input_file, output_file, 
                       index_cols, values_cols, 
                       aggfunc='sum', filters=None):
    """创建数据透视表"""
    # 读取数据
    df = pd.read_excel(input_file)
    
    # 应用筛选条件
    iffilters:
        forcolumn, valueinfilters.items():
            ifisinstance(value, list):
                df = df[df[column].isin(value)]
            else:
                df = df[df[column] == value]
    
    # 创建数据透视表
    pivot_table = pd.pivot_table(
        df,
        index=index_cols,
        values=values_cols,
        aggfunc=aggfunc,
        fill_value=0,
        margins=True,
        margins_name='总计'
    )
    
    # 计算百分比（如果适用）
    iflen(values_cols) == 1andaggfuncin ['sum', 'count']:
        total = pivot_table.loc['总计', values_cols[0]]
        pivot_table['百分比'] = (pivot_table[values_cols[0]] /total*100).round(2)
    
    # 保存到Excel
    withpd.ExcelWriter(output_file, engine='openpyxl') aswriter:
        df.to_excel(writer, sheet_name='原始数据', index=False)
        pivot_table.to_excel(writer, sheet_name='数据透视表')
        
        # 添加图表（可选）
        workbook = writer.book
        worksheet = writer.sheets['数据透视表']
        
        # 这里可以添加图表生成代码
    
    print(f"数据透视表已生成，保存到 {output_file}")
    returnpivot_table

# 使用示例
create_pivot_table(
    input_file="销售数据.xlsx",
    output_file="销售分析.xlsx",
    index_cols=['地区', '产品类别'],
    values_cols=['销售额', '销售量'],
    aggfunc={'销售额': 'sum', '销售量': 'sum'},
    filters={'年份': [2024]}
)

importpandasaspd
importnumpyasnp
importre
fromdatetimeimportdatetime

defvalidate_and_clean_excel(input_file, output_file, validation_rules):
    """验证并清洗Excel数据"""
    df = pd.read_excel(input_file)
    
    issues = []
    cleaned_df = df.copy()
    
    # 检查空值
    null_counts = df.isnull().sum()
    ifnull_counts.sum() >0:
        issues.append(f"发现 {null_counts.sum()} 个空值")
        # 可选：填充空值
        forcolindf.columns:
            ifdf[col].dtypein ['int64', 'float64']:
                cleaned_df[col] = cleaned_df[col].fillna(0)
            else:
                cleaned_df[col] = cleaned_df[col].fillna('未知')
    
    # 检查重复值
    duplicates = df.duplicated().sum()
    ifduplicates>0:
        issues.append(f"发现 {duplicates} 个重复行")
        cleaned_df = cleaned_df.drop_duplicates()
    
    # 应用自定义验证规则
    forruleinvalidation_rules:
        column = rule.get('column')
        rule_type = rule.get('type')
        condition = rule.get('condition')
        
        ifcolumnindf.columns:
            ifrule_type == 'range':
                min_val, max_val = condition
                out_of_range = df[(df[column] <min_val) | (df[column] >max_val)]
                iflen(out_of_range) >0:
                    issues.append(f"列 '{column}' 有 {len(out_of_range)} 个值超出范围 [{min_val}, {max_val}]")
            
            elifrule_type == 'regex':
                invalid = df[~df[column].astype(str).str.match(condition)]
                iflen(invalid) >0:
                    issues.append(f"列 '{column}' 有 {len(invalid)} 个值不符合格式要求")
    
    # 检查数据类型
    forcolindf.columns:
        ifdf[col].dtype == 'object':
            # 尝试转换为日期
            try:
                pd.to_datetime(df[col], errors='raise')
                issues.append(f"列 '{col}' 应设为日期格式")
            except:
                pass
    
    # 保存结果和问题报告
    withpd.ExcelWriter(output_file) aswriter:
        cleaned_df.to_excel(writer, sheet_name='清洗后数据', index=False)
        
        # 创建问题报告
        ifissues:
            issues_df = pd.DataFrame({'问题描述': issues})
            issues_df.to_excel(writer, sheet_name='问题报告', index=False)
    
    print(f"数据验证完成，发现 {len(issues)} 个问题")
    ifissues:
        forissueinissues:
            print(f"  - {issue}")
    
    returncleaned_df, issues

# 使用示例
rules = [
    {'column': '年龄', 'type': 'range', 'condition': (18, 65)},
    {'column': '邮箱', 'type': 'regex', 'condition': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'}
]

validate_and_clean_excel("原始数据.xlsx", "清洗后数据.xlsx", rules)

importpandasaspd
importmatplotlib.pyplotasplt
fromopenpyxlimportload_workbook
fromopenpyxl.drawing.imageimportImage
fromdatetimeimportdatetime
importos

defgenerate_automated_report(data_file, template_file, output_file, report_date=None):
    """生成自动化报表"""
    ifreport_dateisNone:
        report_date = datetime.now().strftime('%Y-%m-%d')
    
    # 读取数据
    df = pd.read_excel(data_file)
    
    # 计算关键指标
    metrics = {
        '报告日期': report_date,
        '总记录数': len(df),
        '总销售额': df['销售额'].sum(),
        '平均销售额': df['销售额'].mean(),
        '最大销售额': df['销售额'].max(),
        '最小销售额': df['销售额'].min(),
        '客户数': df['客户ID'].nunique(),
        '产品数': df['产品ID'].nunique()
    }
    
    # 按类别分组统计
    category_stats = df.groupby('产品类别').agg({
        '销售额': ['sum', 'mean', 'count'],
        '利润': 'sum'
    }).round(2)
    
    # 生成图表
    plt.figure(figsize=(10, 6))
    
    # 销售额柱状图
    sales_by_category = df.groupby('产品类别')['销售额'].sum().sort_values(ascending=False)
    plt.subplot(1, 2, 1)
    sales_by_category.plot(kind='bar', color='skyblue')
    plt.title('各产品类别销售额')
    plt.xlabel('产品类别')
    plt.ylabel('销售额')
    plt.xticks(rotation=45)
    
    # 销售额占比饼图
    plt.subplot(1, 2, 2)
    sales_by_category.plot(kind='pie', autopct='%1.1f%%')
    plt.title('销售额占比')
    plt.ylabel('')
    
    # 保存图表
    chart_file = 'temp_chart.png'
    plt.tight_layout()
    plt.savefig(chart_file, dpi=300, bbox_inches='tight')
    plt.close()
    
    # 使用模板生成报表
    wb = load_workbook(template_file)
    ws = wb.active
    
    # 填充数据
    ws['B2'] = f"销售报告 - {report_date}"
    
    # 填充指标
    metric_cells = {
        'B5': metrics['总记录数'],
        'B6': f"{metrics['总销售额']:,.2f}",
        'B7': f"{metrics['平均销售额']:,.2f}",
        'B8': metrics['客户数'],
        'B9': metrics['产品数']
    }
    
    forcell, valueinmetric_cells.items():
        ws[cell] = value
    
    # 填充分类统计
    start_row = 12
    fori, (category, stats) inenumerate(category_stats.iterrows()):
        row = start_row+i
        ws[f'A{row}'] = category
        ws[f'B{row}'] = stats[('销售额', 'sum')]
        ws[f'C{row}'] = stats[('销售额', 'mean')]
        ws[f'D{row}'] = stats[('销售额', 'count')]
        ws[f'E{row}'] = stats[('利润', 'sum')]
    
    # 插入图表
    img = Image(chart_file)
    ws.add_image(img, 'G5')
    
    # 保存报表
    wb.save(output_file)
    
    # 清理临时文件
    ifos.path.exists(chart_file):
        os.remove(chart_file)
    
    print(f"报表已生成: {output_file}")
    returnmetrics

# 使用示例
generate_automated_report(
    data_file="销售数据.xlsx",
    template_file="报告模板.xlsx",
    output_file=f"销售报告_{datetime.now().strftime('%Y%m%d')}.xlsx"
)

importpandasaspd
importschedule
importtime
importsmtplib
fromemail.mime.textimportMIMEText
fromemail.mime.multipartimportMIMEMultipart
fromdatetimeimportdatetime
importjson

classExcelDataMonitor:
    def__init__(self, config_file):
        """初始化数据监控器"""
        withopen(config_file, 'r', encoding='utf-8') asf:
            self.config = json.load(f)
        
        self.history_data = {}
    
    defcheck_thresholds(self, df):
        """检查阈值条件"""
        alerts = []
        
        forruleinself.config.get('threshold_rules', []):
            column = rule['column']
            condition = rule['condition']
            threshold = rule['threshold']
            message = rule['message']
            
            ifcolumnindf.columns:
                ifcondition == 'greater_than':
                    violations = df[df[column] >threshold]
                elifcondition == 'less_than':
                    violations = df[df[column] <threshold]
                elifcondition == 'equals':
                    violations = df[df[column] == threshold]
                elifcondition == 'not_equals':
                    violations = df[df[column] != threshold]
                else:
                    continue
                
                iflen(violations) >0:
                    alert_msg = f"{message}\n违反数量: {len(violations)}\n"
                    alert_msg += "违反记录:\n"
                    for_, rowinviolations.head(5).iterrows():
                        alert_msg += f"  - {row.to_dict()}\n"
                    alerts.append(alert_msg)
        
        returnalerts
    
    defcheck_data_changes(self, current_df, previous_df):
        """检查数据变化"""
        changes = []
        
        ifprevious_dfisnotNone:
            # 检查行数变化
            row_change = len(current_df) -len(previous_df)
            ifabs(row_change) >self.config.get('max_row_change', 10):
                changes.append(f"行数变化较大: {row_change} 行")
            
            # 检查关键指标变化
            formetricinself.config.get('key_metrics', []):
                ifmetricincurrent_df.columnsandmetricinprevious_df.columns:
                    current_sum = current_df[metric].sum()
                    previous_sum = previous_df[metric].sum()
                    
                    ifprevious_sum!= 0:
                        change_pct = (current_sum-previous_sum) /previous_sum*100
                        ifabs(change_pct) >self.config.get('max_change_pct', 20):
                            changes.append(f"{metric} 变化: {change_pct:.1f}%")
        
        returnchanges
    
    defsend_alert(self, alerts, changes):
        """发送告警通知"""
        ifnotalertsandnotchanges:
            return
        
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        subject = f"数据监控告警 - {timestamp}"
        
        # 构建邮件内容
        body = f"数据监控告警报告\n"
        body += f"时间: {timestamp}\n"
        body += f"监控文件: {self.config['monitor_file']}\n\n"
        
        ifalerts:
            body += "=== 阈值告警 ===\n"
            foralertinalerts:
                body += f"{alert}\n"
        
        ifchanges:
            body += "\n=== 数据变化告警 ===\n"
            forchangeinchanges:
                body += f"{change}\n"
        
        # 发送邮件（示例）
        ifself.config.get('email_alerts', False):
            self.send_email(subject, body)
        
        # 记录日志
        withopen('monitor_log.txt', 'a', encoding='utf-8') asf:
            f.write(f"\n[{timestamp}]\n{body}\n")
        
        print(f"发现告警，已发送通知")
    
    defsend_email(self, subject, body):
        """发送邮件（简化示例）"""
        # 实际使用时需要配置SMTP服务器
        print(f"发送邮件告警:\n主题: {subject}\n内容: {body[:200]}...")
        # 实际邮件发送代码需要根据SMTP服务器配置
    
    defmonitor_task(self):
        """监控任务"""
        try:
            # 读取当前数据
            current_df = pd.read_excel(self.config['monitor_file'])
            
            # 检查阈值
            alerts = self.check_thresholds(current_df)
            
            # 检查数据变化
            file_key = self.config['monitor_file']
            previous_df = self.history_data.get(file_key)
            changes = self.check_data_changes(current_df, previous_df)
            
            # 发送告警
            ifalertsorchanges:
                self.send_alert(alerts, changes)
            
            # 更新历史数据
            self.history_data[file_key] = current_df
            
            print(f"监控完成: {datetime.now().strftime('%H:%M:%S')}")
            
        exceptExceptionase:
            print(f"监控任务出错: {e}")
    
    defstart_monitoring(self):
        """启动监控"""
        interval = self.config.get('check_interval_minutes', 5)
        
        print(f"开始监控 {self.config['monitor_file']}，检查间隔: {interval}分钟")
        
        # 安排定时任务
        schedule.every(interval).minutes.do(self.monitor_task)
        
        # 立即执行一次
        self.monitor_task()
        
        # 保持运行
        whileTrue:
            schedule.run_pending()
            time.sleep(1)

# 配置示例
config = {
    "monitor_file": "实时销售数据.xlsx",
    "check_interval_minutes": 5,
    "email_alerts": True,
    "max_row_change": 50,
    "max_change_pct": 30,
    "threshold_rules": [
        {
            "column": "库存量",
            "condition": "less_than",
            "threshold": 10,
            "message": "库存量低于安全阈值"
        },
        {
            "column": "错误率",
            "condition": "greater_than",
            "threshold": 0.05,
            "message": "错误率超过5%"
        }
    ],
    "key_metrics": ["销售额", "订单量", "客户数"]
}

# 保存配置
withopen('monitor_config.json', 'w', encoding='utf-8') asf:
    json.dump(config, f, ensure_ascii=False, indent=2)

# 启动监控（在实际使用中可能需要作为后台服务运行）
# monitor = ExcelDataMonitor('monitor_config.json')
# monitor.start_monitoring()