Python自动化Excel办公:从入门到实战
2026/7/26 23:50:58 网站建设 项目流程

1. Python自动化办公Excel实用指南:从入门到精通

作为一名长期与Excel和Python打交道的开发者,我深知重复性表格处理工作有多折磨人。曾经为了合并12个分公司的月度报表,我不得不熬夜到凌晨3点手动复制粘贴。直到掌握了Python自动化处理Excel的技巧,同样工作现在只需3分钟就能完成。这篇指南将分享我多年积累的实战经验,带你用Python彻底解放双手。

Python在Excel自动化领域的优势非常明显:它能处理百万行级数据而不会卡顿,可以完美复现所有手动操作步骤,还能实现Excel本身难以完成的复杂逻辑。无论是财务对账、销售分析还是人事管理,掌握这些技能至少能提升10倍工作效率。下面我会从环境搭建开始,逐步深入到实战案例,最后分享几个提升稳定性的独家技巧。

2. 环境准备与基础配置

2.1 Python环境搭建

推荐使用Python 3.8+版本,这个版本在第三方库兼容性和性能之间取得了很好的平衡。新手可以直接安装Anaconda发行版,它预装了数据分析所需的绝大多数工具包:

conda create -n excel_auto python=3.8 conda activate excel_auto

对于有经验的开发者,使用原生Python配合virtualenv会更轻量:

python -m venv excel_venv source excel_venv/bin/activate # Linux/Mac excel_venv\Scripts\activate.bat # Windows

注意:无论哪种方式,请确保将Python添加到系统PATH环境变量。验证方法是在命令行执行python --version能正确显示版本号。

2.2 核心库安装

处理Excel主要依赖以下几个库:

pip install openpyxl pandas xlrd xlwt xlsxwriter

各库的作用如下:

  • openpyxl:处理.xlsx格式文件(Excel 2007+)
  • pandas:数据分析和处理的核心工具
  • xlrd/xlwt:读写.xls格式文件(Excel 2003)
  • xlsxwriter:生成带复杂格式的Excel文件

我强烈建议同时安装pyxlsb以支持二进制.xlsb格式:

pip install pyxlsb

2.3 开发工具选择

VSCode是最适合Excel自动化的IDE之一,需要安装以下扩展:

  1. Python官方扩展
  2. Excel Viewer
  3. Jupyter

配置关键设置(settings.json):

{ "python.linting.enabled": true, "python.formatting.provider": "black", "files.autoSave": "afterDelay" }

3. Excel基础操作全解析

3.1 文件读写操作

使用openpyxl读取工作簿:

from openpyxl import load_workbook # 读取文件 wb = load_workbook('财务报表.xlsx', data_only=True) # data_only获取计算后的值 sheet = wb['2023年度'] # 按名称获取工作表 # 获取单元格值 cell_value = sheet['B2'].value # 获取B2单元格 row_data = [cell.value for cell in sheet[2]] # 获取第2行所有数据 # 写入数据 sheet['C5'] = "=SUM(C2:C4)" # 写入公式 sheet.cell(row=8, column=3, value=42) # 行列号方式写入 # 保存文件 wb.save('财务报表_修改后.xlsx')

避坑指南:打开大文件时添加read_only=True参数可显著提升性能,但此时只能读取不能修改。另存时使用不同文件名可防止原始文件损坏。

3.2 使用Pandas高效处理数据

Pandas的DataFrame是处理表格数据的利器:

import pandas as pd # 读取Excel df = pd.read_excel('销售数据.xlsx', sheet_name='Q1', header=2) # 从第3行开始读 # 数据清洗 df.dropna(subset=['客户ID'], inplace=True) # 删除客户ID为空的行 df['销售额'] = df['单价'] * df['数量'] # 新增计算列 # 分组统计 summary = df.groupby('产品类别')['销售额'].agg(['sum', 'mean', 'count']) # 写入新文件 with pd.ExcelWriter('销售报告.xlsx') as writer: df.to_excel(writer, sheet_name='明细数据') summary.to_excel(writer, sheet_name='汇总统计') # 添加格式 workbook = writer.book fmt = workbook.add_format({'num_format': '#,##0.00'}) writer.sheets['汇总统计'].set_column('B:D', 15, fmt)

3.3 高级格式设置

通过xlsxwriter设置专业级报表格式:

import xlsxwriter workbook = xlsxwriter.Workbook('格式化报告.xlsx') worksheet = workbook.add_worksheet() # 定义格式 title_fmt = workbook.add_format({ 'bold': True, 'font_color': 'white', 'bg_color': '#4F81BD', 'align': 'center', 'valign': 'vcenter' }) currency_fmt = workbook.add_format({'num_format': '$#,##0.00'}) # 应用格式 worksheet.write('A1', '季度销售报告', title_fmt) worksheet.set_column('B:E', 12, currency_fmt) # 添加条件格式 worksheet.conditional_format('B2:B10', { 'type': 'data_bar', 'bar_color': '#63C384' }) workbook.close()

4. 实战案例精讲

4.1 案例一:多文件数据合并

场景:每月需要合并30个分店的销售报表

from pathlib import Path import pandas as pd def merge_excel_files(folder_path, output_file): all_data = [] # 遍历文件夹 for file in Path(folder_path).glob('*.xlsx'): if file.name.startswith('~$'): # 跳过临时文件 continue # 读取每个文件 df = pd.read_excel(file, sheet_name=None) # 读取所有sheet for sheet_name, data in df.items(): data['来源文件'] = file.name data['来源sheet'] = sheet_name all_data.append(data) # 合并保存 merged = pd.concat(all_data, ignore_index=True) merged.to_excel(output_file, index=False) print(f'成功合并{len(all_data)}个表格,总行数:{len(merged)}') # 使用示例 merge_excel_files('分店报表/', '合并报表.xlsx')

经验技巧:添加来源文件来源sheet字段便于后续追踪数据来源。处理前先用glob检查文件格式可避免意外错误。

4.2 案例二:自动化数据校验

自动检测常见数据问题:

def validate_excel(file_path): report = { '空值检查': [], '格式问题': [], '逻辑错误': [] } wb = load_workbook(file_path) for sheet in wb: for row in sheet.iter_rows(values_only=True): # 检查空值 if any(cell is None for cell in row[:3]): # 前3列不能为空 report['空值检查'].append(f"{sheet.title}-行{row[0].row}") # 检查日期格式 if isinstance(row[1], str) and not row[1].count('/') == 2: report['格式问题'].append(f"日期格式错误: {sheet.title}-{row[0]}") # 生成报告 with pd.ExcelWriter('校验报告.xlsx') as writer: for issue_type, details in report.items(): pd.DataFrame({issue_type: details}).to_excel( writer, sheet_name=issue_type[:31], index=False) return report

4.3 案例三:动态报表生成

结合数据库生成动态报表:

import sqlite3 from datetime import datetime def generate_sales_report(start_date, end_date): # 连接数据库 conn = sqlite3.connect('sales.db') # 执行SQL查询 query = f""" SELECT 产品名称, SUM(数量) as 总销量, SUM(金额) as 总销售额 FROM 销售记录 WHERE 日期 BETWEEN '{start_date}' AND '{end_date}' GROUP BY 产品名称 ORDER BY 总销售额 DESC """ df = pd.read_sql(query, conn) # 生成Excel with pd.ExcelWriter(f"销售报告_{datetime.now().strftime('%Y%m%d')}.xlsx") as writer: df.to_excel(writer, sheet_name='销售汇总', index=False) # 添加图表 workbook = writer.book worksheet = writer.sheets['销售汇总'] chart = workbook.add_chart({'type': 'column'}) chart.add_series({ 'name': '总销售额', 'categories': '=销售汇总!$A$2:$A$20', 'values': '=销售汇总!$C$2:$C$20' }) worksheet.insert_chart('E2', chart) print(f"报告已生成:时间段{start_date}至{end_date}")

5. 高级技巧与性能优化

5.1 处理百万行数据

当数据量超过50万行时,需要特殊处理:

# 分块读取 chunk_size = 100000 reader = pd.read_excel('大数据文件.xlsx', chunksize=chunk_size) for i, chunk in enumerate(reader): process(chunk) # 处理每个分块 print(f"已处理{(i+1)*chunk_size}行") # 使用dtype优化内存 dtype_map = { '客户ID': 'str', '金额': 'float32', '数量': 'int16' } pd.read_excel('大数据文件.xlsx', dtype=dtype_map)

5.2 自动化任务调度

结合Windows任务计划或Linux cron实现定时运行:

# 创建run.py import schedule import time def daily_report(): generate_sales_report('2023-01-01', '2023-12-31') schedule.every().day.at("08:00").do(daily_report) while True: schedule.run_pending() time.sleep(60)

然后设置系统定时任务:

# Linux crontab 0 8 * * * /path/to/python /path/to/run.py

5.3 异常处理与日志

健壮的脚本需要完善的错误处理:

import logging from openpyxl.utils.exceptions import InvalidFileException logging.basicConfig( filename='excel_auto.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def safe_process(file): try: wb = load_workbook(file) # 处理逻辑... except InvalidFileException as e: logging.error(f"文件格式错误: {file} - {str(e)}") except Exception as e: logging.critical(f"处理{file}时发生未知错误", exc_info=True) else: logging.info(f"成功处理{file}")

6. 常见问题解决方案

6.1 文件被占用错误

当遇到"文件正在被其他程序使用"错误时:

import os from time import sleep def safe_save(wb, filename, retries=3): for i in range(retries): try: wb.save(filename) return True except PermissionError: sleep(2) # 等待2秒重试 return False

替代方案是使用临时文件:

import tempfile def temp_save(wb, original_path): with tempfile.NamedTemporaryFile(delete=False) as tmp: temp_path = tmp.name + '.xlsx' wb.save(temp_path) os.replace(temp_path, original_path)

6.2 公式计算问题

确保公式计算结果更新:

# 方法1:强制计算 wb = load_workbook('带公式.xlsx', data_only=False) sheet = wb.active sheet['B2'] = "=SUM(A1:A10)" wb.save('公式更新.xlsx') # 方法2:使用win32com(仅Windows) import win32com.client excel = win32com.client.Dispatch("Excel.Application") wb = excel.Workbooks.Open(r'C:\路径\文件.xlsx') wb.Save() excel.Quit()

6.3 中文编码问题

处理包含中文的文件名或内容:

# 设置系统默认编码 import sys import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') # Pandas读取时指定编码 df = pd.read_excel('中文文件.xlsx', engine='openpyxl', encoding='utf-8-sig') # 写入CSV时处理中文 df.to_csv('输出.csv', encoding='utf_8_sig', index=False)

7. 扩展应用场景

7.1 与邮件系统集成

自动发送Excel报表邮件:

import smtplib from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email import encoders def send_email_with_excel(to_email, subject, body, excel_path): msg = MIMEMultipart() msg['From'] = 'auto_report@company.com' msg['To'] = to_email msg['Subject'] = subject # 添加正文 msg.attach(MIMEText(body, 'plain')) # 添加Excel附件 with open(excel_path, 'rb') as f: part = MIMEBase('application', 'octet-stream') part.set_payload(f.read()) encoders.encode_base64(part) part.add_header('Content-Disposition', f'attachment; filename="{os.path.basename(excel_path)}"') msg.attach(part) # 发送邮件 with smtplib.SMTP('smtp.company.com', 587) as server: server.starttls() server.login('user', 'password') server.send_message(msg)

7.2 Web数据抓取与Excel整合

结合爬虫自动更新数据:

import requests from bs4 import BeautifulSoup def scrape_to_excel(url, output_file): response = requests.get(url) soup = BeautifulSoup(response.text, 'html.parser') data = [] for row in soup.select('table.data tr'): cols = [col.get_text(strip=True) for col in row.find_all('td')] if cols: data.append(cols) df = pd.DataFrame(data, columns=['日期', '产品', '销量', '金额']) df.to_excel(output_file, index=False)

7.3 创建Excel插件

用PyXLL将Python函数变成Excel公式:

# 在pyxll.cfg中添加 [PYTHON] pythonpath = /path/to/your/scripts [FUNCTIONS] my_sum = your_module.your_function # Python函数 @xl_func def my_sum(x, y): """在Excel中使用的自定义求和函数""" return float(x) + float(y)

8. 维护与最佳实践

8.1 代码组织建议

推荐的项目结构:

/excel_automation │── /config # 配置文件 │ └── settings.ini │── /data # 输入输出文件 │ ├── /input │ └── /output │── /lib # 公共函数 │ └── excel_utils.py │── /scripts # 主程序 │ ├── daily_report.py │ └── data_clean.py │── requirements.txt └── README.md

8.2 版本控制策略

处理Excel文件时的Git策略:

  1. 将原始Excel文件添加到.gitignore
  2. 只提交处理脚本和生成的样例文件
  3. 使用git-lfs管理大型二进制文件
# .gitignore 示例 *.xlsx *.xls ~$*

8.3 性能监控

添加执行时间记录:

import time from functools import wraps def time_logger(func): @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) end = time.perf_counter() print(f"{func.__name__} 执行时间: {end - start:.2f}秒") return result return wrapper @time_logger def process_large_file(file_path): # 处理逻辑...

9. 安全注意事项

9.1 文件安全处理

防止恶意文件攻击:

from openpyxl import Workbook from tempfile import mkstemp import shutil def safe_open(file_path): """验证文件安全性""" if not file_path.endswith(('.xlsx', '.xls')): raise ValueError("仅支持Excel文件") # 创建临时副本 fd, temp_path = mkstemp(suffix='.xlsx') shutil.copy2(file_path, temp_path) try: wb = load_workbook(temp_path) return wb finally: os.close(fd) os.unlink(temp_path)

9.2 敏感数据处理

加密包含敏感信息的Excel:

from io import BytesIO from cryptography.fernet import Fernet def encrypt_excel(df, password): """将DataFrame加密保存""" buffer = BytesIO() with pd.ExcelWriter(buffer) as writer: df.to_excel(writer) cipher_suite = Fernet(password) encrypted = cipher_suite.encrypt(buffer.getvalue()) with open('加密文件.xlsx', 'wb') as f: f.write(encrypted)

9.3 宏病毒防护

检查文件是否包含宏:

import zipfile def check_macros(file_path): """检查Excel是否包含宏""" try: with zipfile.ZipFile(file_path) as z: return 'xl/vbaProject.bin' in z.namelist() except zipfile.BadZipFile: return False

10. 实战经验分享

10.1 处理特殊格式的坑

遇到合并单元格时的处理技巧:

def get_merged_cell_value(sheet, cell): """获取合并单元格的真实值""" for range_ in sheet.merged_cells.ranges: if cell.coordinate in range_: return range_.start_cell.value return cell.value

处理自定义数字格式:

def parse_custom_number(cell): """解析如"1,000.00 USD"这样的自定义格式""" value = cell.value if isinstance(value, str): return float(''.join(c for c in value if c.isdigit() or c == '.')) return value

10.2 跨平台兼容性问题

处理Windows和Mac的路径差异:

from pathlib import Path def get_absolute_path(relative_path): """获取跨平台绝对路径""" base_dir = Path(__file__).parent return (base_dir / relative_path).resolve()

处理不同系统的换行符:

def clean_text(text): """统一换行符""" if text is None: return "" return text.replace('\r\n', '\n').replace('\r', '\n')

10.3 性能优化实战

加速openpyxl读取的秘诀:

def fast_read_excel(file_path): """快速读取大Excel文件""" wb = load_workbook(file_path, read_only=True, data_only=True) data = [] for row in wb.active.iter_rows(values_only=True): data.append(row) wb.close() return data

减少Pandas内存占用的技巧:

def optimize_memory(df): """优化DataFrame内存使用""" # 转换数值类型 for col in df.select_dtypes(include=['int64']): df[col] = pd.to_numeric(df[col], downcast='integer') # 转换字符串类型 for col in df.select_dtypes(include=['object']): df[col] = df[col].astype('category') return df

11. 资源推荐与学习路径

11.1 进阶学习资料

推荐书籍:

  • 《Python for Excel》by Felix Zumstein
  • 《Pandas Cookbook》by Theodore Petrou
  • 《Automate the Boring Stuff with Python》第12章

在线课程:

  • Udemy: "Python for Data Science and Machine Learning"
  • Coursera: "Data Science at Scale with Python"

11.2 实用工具集

开发辅助工具:

  • Tablib:数据格式转换库
  • PyInstaller:打包成exe
  • tqdm:进度条显示

调试工具:

  • Excel Viewer:快速预览文件内容
  • File Watcher:监控文件变化
  • Memory Profiler:内存使用分析

11.3 社区支持

遇到问题时可以求助:

  • Stack Overflow:使用[python][excel]标签
  • GitHub:openpyxl和pandas的issue区
  • Reddit:/r/learnpython和/r/excel

中文社区:

  • 知乎Python话题
  • CSDN Python板块
  • 掘金数据分析专栏

12. 持续集成与自动化部署

12.1 自动化测试框架

为Excel脚本添加单元测试:

import unittest from openpyxl import Workbook class TestExcelFunctions(unittest.TestCase): @classmethod def setUpClass(cls): cls.wb = Workbook() cls.sheet = cls.wb.active cls.sheet['A1'] = 10 cls.sheet['A2'] = 20 def test_sum_formula(self): self.sheet['A3'] = "=SUM(A1:A2)" self.assertEqual(self.sheet['A3'].value, "=SUM(A1:A2)") self.assertEqual(self.sheet['A3'].data_type, 'f') @classmethod def tearDownClass(cls): cls.wb.close() if __name__ == '__main__': unittest.main()

12.2 CI/CD集成

GitHub Actions自动化示例:

name: Excel Automation CI on: [push, pull_request] jobs: test: runs-on: windows-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: '3.8' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - name: Run tests run: | python -m unittest discover

12.3 容器化部署

Docker镜像构建:

FROM python:3.8-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "daily_report.py"]

构建命令:

docker build -t excel-automation . docker run -v $(pwd)/data:/app/data excel-automation

13. 扩展思路:与其他工具集成

13.1 与数据库联动

从SQL数据库导出到Excel:

import pyodbc import pandas as pd def export_sql_to_excel(connection_str, query, output_file): conn = pyodbc.connect(connection_str) df = pd.read_sql(query, conn) # 添加Excel格式 writer = pd.ExcelWriter(output_file, engine='xlsxwriter') df.to_excel(writer, index=False) # 自动调整列宽 for column in df: col_idx = df.columns.get_loc(column) writer.sheets['Sheet1'].set_column(col_idx, col_idx, max(df[column].astype(str).map(len).max(), len(column)) + 2) writer.save()

13.2 与Power BI/Tableau集成

生成供BI工具使用的数据模型:

def create_bi_ready_excel(data_frames, output_file): with pd.ExcelWriter(output_file) as writer: for name, df in data_frames.items(): df.to_excel(writer, sheet_name=name[:31], index=False) # 添加为Excel表格对象(方便Power BI识别) worksheet = writer.sheets[name[:31]] (max_row, max_col) = df.shape worksheet.add_table(0, 0, max_row, max_col-1, { 'columns': [{'header': col} for col in df.columns], 'style': 'Table Style Medium 2', 'name': name })

13.3 与云存储结合

自动上传到Google Drive:

from pydrive.auth import GoogleAuth from pydrive.drive import GoogleDrive def upload_to_drive(file_path, folder_id): gauth = GoogleAuth() gauth.LocalWebserverAuth() # 首次使用需要授权 drive = GoogleDrive(gauth) file = drive.CreateFile({'title': os.path.basename(file_path), 'parents': [{'id': folder_id}]}) file.SetContentFile(file_path) file.Upload() return file['id']

14. 可视化增强技巧

14.1 条件格式高级应用

创建热力图效果:

def add_heatmap(writer, sheet_name, range_, min_color, max_color): workbook = writer.book worksheet = writer.sheets[sheet_name] worksheet.conditional_format(range_, { 'type': '2_color_scale', 'min_color': min_color, 'max_color': max_color, 'min_type': 'percentile', 'min_value': 10, 'max_type': 'percentile', 'max_value': 90 })

14.2 动态图表生成

根据数据自动创建图表:

def create_dynamic_chart(writer, sheet_name, data_range, title): workbook = writer.book worksheet = writer.sheets[sheet_name] chart = workbook.add_chart({'type': 'line'}) chart.add_series({ 'values': f'={sheet_name}!{data_range}', 'categories': f'={sheet_name}!$A$2:$A${len(data)+1}', 'name': '实际值' }) chart.set_title({'name': title}) chart.set_x_axis({'name': '日期'}) chart.set_y_axis({'name': '数值'}) worksheet.insert_chart('E2', chart)

14.3 交互式控件

添加下拉菜单和数据验证:

def add_data_validation(writer, sheet_name, cell, options): workbook = writer.book worksheet = writer.sheets[sheet_name] # 创建选项列表 option_str = ",".join(options) # 添加数据验证 worksheet.data_validation(cell, { 'validate': 'list', 'source': option_str, 'input_title': '请选择:', 'error_title': '无效输入', 'error_message': '请从列表中选择' })

15. 项目架构设计模式

15.1 面向对象封装

将常用功能封装成类:

class ExcelAutomator: def __init__(self, template_path): self.template = template_path self.wb = None def __enter__(self): self.wb = load_workbook(self.template) return self def __exit__(self, exc_type, exc_val, exc_tb): if self.wb: self.wb.close() def fill_data(self, sheet_name, data_dict): """根据字典填充数据""" sheet = self.wb[sheet_name] for cell, value in data_dict.items(): sheet[cell] = value def save_as(self, output_path): self.wb.save(output_path) # 使用示例 data = {'B2': '季度报告', 'C5': 42, 'D8': datetime.now()} with ExcelAutomator('模板.xlsx') as excel: excel.fill_data('Sheet1', data) excel.save_as('输出报告.xlsx')

15.2 插件系统设计

支持自定义处理插件:

class ExcelProcessor: def __init__(self): self.plugins = [] def register_plugin(self, plugin): self.plugins.append(plugin) def process_file(self, file_path): wb = load_workbook(file_path) for plugin in self.plugins: plugin.before_process(wb) for sheet in wb: for plugin in self.plugins: plugin.process_sheet(sheet) for plugin in self.plugins: plugin.after_process(wb) return wb # 示例插件 class FormatCleanerPlugin: def process_sheet(self, sheet): for row in sheet.iter_rows(): for cell in row: cell.style = 'Normal'

15.3 配置驱动架构

使用JSON/YAML配置定义处理流程:

# config/process.yaml steps: - name: 数据清洗 actions: - type: remove_empty_rows columns: [客户ID, 产品编号] - type: standardize_dates columns: [订单日期] format: "%Y-%m-%d" - name: 数据计算 actions: - type: add_formula cell: "D2" formula: "=B2*C2" fill_range: "D2:D100"

对应的处理器:

def process_with_config(wb, config_file): with open(config_file) as f: config = yaml.safe_load(f) for step in config['steps']: print(f"正在执行: {step['name']}") for action in step['actions']: execute_action(wb, action)

16. 性能监控与调优

16.1 内存分析

使用memory_profiler检测内存泄漏:

@profile def process_large_excel(file_path): df = pd.read_excel(file_path) # 处理逻辑... return df.groupby('类别').sum() if __name__ == '__main__': process_large_excel('大数据.xlsx')

运行分析:

python -m memory_profiler script.py

16.2 执行时间优化

使用cProfile分析性能瓶颈:

import cProfile def profile_excel_processing(): pr = cProfile.Profile() pr.enable() # 要分析的代码 process_large_file('大数据.xlsx') pr.disable() pr.print_stats(sort='cumtime') if __name__ == '__main__': profile_excel_processing()

16.3 并行处理

使用多进程加速:

from multiprocessing import Pool def process_sheet(sheet_name): df = pd.read_excel('大数据.xlsx', sheet_name=sheet_name) # 处理逻辑... return df if __name__ == '__main__': sheet_names = ['Q1', 'Q2', 'Q3', 'Q4'] with Pool(4) as p: results = p.map(process_sheet, sheet_names) final_df = pd.concat(results)

17. 错误处理与恢复机制

17.1 事务处理模式

实现原子性操作:

import shutil def safe_excel_operation(input_path, output_path, process_func): """确保操作失败时不影响原始文件""" temp_path = output_path + '.tmp' try: # 执行处理 wb = load_workbook(input_path) process_func(wb) wb.save(temp_path) # 验证文件 test_wb = load_workbook(temp_path) test_wb.close() # 替换原文件 shutil.move(temp_path, output_path) except Exception as e: if os.path.exists(temp_path): os.remove(temp_path) raise e

17.2 断点续处理

记录处理进度:

import json class ProgressTracker: def __init__(self, state_file='progress.json'): self.state_file = state_file self.state = self._load_state() def _load_state(self): try: with open(self.state_file) as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {'last_processed': 0} def save_state(self, last_row): self.state['last_processed'] = last_row with open(self.state_file, 'w') as f: json.dump(self.state, f) def process_file(self, file_path): wb = load_workbook(file_path) sheet = wb.active start_row = self.state['last_processed'] + 1 for row in sheet.iter_rows(min_row=start_row, values_only=True): try: process_row(row) self.save_state(row[0]) # 假设第一列是行ID except Exception as e: print(f"处理行{row[0]}失败: {str(e)}") break

17.3 自动重试机制

处理网络或资源问题:

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def download_excel(url): response = requests.get(url, timeout=10) response.raise_for_status() with open('download.xlsx', 'wb') as f: f.write(response.content)

18. 文档与注释规范

18.1 自动化文档生成

使用pydoc生成API文档:

python -m pydoc -w your_module

18.2 Excel元数据注释

在Excel中添加文档说明:

def add_documentation_sheet(wb, content): if '文档说明' in wb.sheetnames: doc_sheet = wb['文档说明'] else: doc_sheet = wb.create_sheet('文档说明', 0) doc_sheet['A1'] = '最后更新时间' doc_sheet['B1'] = datetime.now().strftime('%Y-%m-%d %H:%M') for i, line in enumerate(content.split('\n'), start=3): doc_sheet[f'A{i}'] = line

18.3 代码注释标准

符合PEP

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询