Python实战项目中文教程：从数据爬虫到自动化办公案例详解

一、项目设计与全链路架构

在很多企业日常办公中，员工需要周期性地打开行业网站或竞品监测页面，复制关键指标并填入固定的周报 Excel 表格中，最后排版成 Word 简报发往工作群或抄送邮箱。这种纯重复性的操作耗时且易出错。通过 Python 将网络请求、数据加工与文档对象模型结合，可以构建一套高度可靠的自动化办公流水线。

工程整体划分为四个解耦的层级：网络采集层负责稳健获取源数据并处理网络波动；清洗计算层进行数据校验、空值填补与指标聚合；文档排版层将数据映射到单元格样式及文档段落中；自动化调度层负责定时触发与任务监控。

二、模块一：目标数据采集与健壮爬虫实现

网络爬虫是整个流水线的数据入口。在编写工业级脚本时，不能简单使用一次性的 requests.get()，而必须考虑网络延迟、服务端限流、反爬响应以及 HTML 结构微调带来的异常崩溃问题。

1.1 完整数据采集代码示例

以下脚本演示抓取目标行情/商品列表的核心代码。采用 urllib3 重试适配器（Retry Adapter），当遇到 500、502、503 等服务端临时报错或网络闪断时，底层会自动完成 3 次重试，保障定时流水线的无人值守稳定性。

import time
import logging
import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def create_resilient_session(retries=3, backoff_factor=1.0) -> requests.Session:
    """创建具备自动重试机制与标准请求头的 Session"""
    session = requests.Session()
    session.headers.update({
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/124.0.0.0 Safari/537.36"
        ),
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
    })
    retry_strategy = Retry(
        total=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

def fetch_product_data(target_url: str) -> list:
    """抓取页面关键列表数据，返回规范化字典数组"""
    session = create_resilient_session()
    logging.info(f"开始请求数据源: {target_url}")
    
    try:
        response = session.get(target_url, timeout=(5.0, 15.0))
        response.raise_for_status()
        response.encoding = response.apparent_encoding  # 自动纠正中文编码
    except requests.RequestException as e:
        logging.error(f"抓取页面失败: {e}")
        return []

    soup = BeautifulSoup(response.text, "html.parser")
    records = []
    
    # 模拟抓取包含商品/项目列表的容器节点
    items = soup.select(".item-card, tr.data-row")
    if not items:
        # 当网站改版或选择器不匹配时预警
        logging.warning("未匹配到目标DOM节点，请检查选择器规则。")
        return []

    for item in items:
        title_node = item.select_one(".title, .item-name")
        price_node = item.select_one(".price, .item-val")
        sales_node = item.select_one(".sales, .item-vol")
        category_node = item.select_one(".category, .item-tag")

        if title_node and price_node:
            records.append({
                "item_name": title_node.get_text(strip=True),
                "price_raw": price_node.get_text(strip=True),
                "volume_raw": sales_node.get_text(strip=True) if sales_node else "0",
                "category": category_node.get_text(strip=True) if category_node else "综合",
                "crawl_timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
            })

    logging.info(f"采集成功，共获取 {len(records)} 条原始记录")
    return records

三、模块二：数据清洗与结构化转换

爬虫抓取到的文本字段通常混合了货币符号（如 ¥、$）、中文单位（如 万、件）以及不可见的空白符（\r\n、\t）。如果直接将这些文本写入 Excel，后续公式无法进行数学求和与均值计算。

使用 Pandas 对原始字典列表做三步标准化：

1. 数值规范化：使用正则表达式清除符号，按单位统一换算为浮点数；
2. 缺失与异常处理：剔除名称为空的脏数据，对缺失销量赋予默认值；
3. 计算统计指标：计算单品总销售额、类别汇总均值以及按贡献率排序。

import re
import pandas as pd
import numpy as np

def clean_and_transform(raw_records: list) -> pd.DataFrame:
    """使用 Pandas 将爬虫原始文本转化为结构化商业报表数据"""
    if not raw_records:
        return pd.DataFrame()

    df = pd.DataFrame(raw_records)

    # 1. 清洗价格列：提取纯数字并转换为 float
    def parse_number(val_str):
        if not isinstance(val_str, str):
            return 0.0
        cleaned = re.sub(r"[^\d.]", "", val_str)
        try:
            return float(cleaned) if cleaned else 0.0
        except ValueError:
            return 0.0

    df["unit_price"] = df["price_raw"].apply(parse_number)

    # 2. 清洗销量列：处理包含'万'的情况
    def parse_volume(val_str):
        if not isinstance(val_str, str):
            return 0
        val_str = val_str.strip()
        is_wan = "万" in val_str
        cleaned = re.sub(r"[^\d.]", "", val_str)
        try:
            num = float(cleaned) if cleaned else 0.0
            return int(num * 10000) if is_wan else int(num)
        except ValueError:
            return 0

    df["sales_volume"] = df["volume_raw"].apply(parse_volume)

    # 3. 衍生计算：计算单品总额
    df["total_amount"] = (df["unit_price"] * df["sales_volume"]).round(2)

    # 4. 去重与空值过滤
    df = df.dropna(subset=["item_name"])
    df = df[df["unit_price"] > 0]
    df = df.drop_duplicates(subset=["item_name"])

    # 5. 排序：按销售总额降序排列
    df = df.sort_values(by="total_amount", ascending=False).reset_index(drop=True)

    return df

四、模块三：自动化生成高颜值 Excel 与 Word 报表

纯黑白且列宽挤压的 Excel 不符合商业交付标准。通过 openpyxl 库，我们可以对标题行填充工程蓝背景、对数值单元格赋予右对齐与千分位格式（#,##0.00）、自适应调整每列宽度并插入求和公式。

4.1 openpyxl 样式美化与自动化输出

下面的函数接收清洗好的 DataFrame，将其写为一个结构规整、带有格式的 Excel 工作簿。

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter

def export_styled_excel(df: pd.DataFrame, output_path: str):
    """将 DataFrame 格式化输出为带有专业商务配色的 Excel 报表"""
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "销售监测周报"

    # 1. 样式定义：符合工程蓝与铜棕的主色调
    header_fill = PatternFill(start_color="345E82", end_color="345E82", fill_type="solid")
    header_font = Font(name="Microsoft YaHei", size=11, bold=True, color="FFFFFF")
    data_font = Font(name="Microsoft YaHei", size=10)
    summary_font = Font(name="Microsoft YaHei", size=10, bold=True, color="AD8060")

    thin_border = Border(
        left=Side(style="thin", color="CBD5E1"),
        right=Side(style="thin", color="CBD5E1"),
        top=Side(style="thin", color="CBD5E1"),
        bottom=Side(style="thin", color="CBD5E1")
    )
    double_bottom_border = Border(
        left=Side(style="thin", color="CBD5E1"),
        right=Side(style="thin", color="CBD5E1"),
        top=Side(style="thin", color="CBD5E1"),
        bottom=Side(style="double", color="345E82")
    )

    # 2. 写入表头
    columns = ["商品/项目名称", "业务类目", "单价 (元)", "销售数量 (件)", "销售总额 (元)", "抓取时间"]
    ws.append(columns)
    for col_idx in range(1, len(columns) + 1):
        cell = ws.cell(row=1, column=col_idx)
        cell.fill = header_fill
        cell.font = header_font
        cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
        cell.border = thin_border
    ws.row_dimensions[1].height = 26

    # 3. 写入数据行
    for row_idx, row in df.iterrows():
        current_row = row_idx + 2
        ws.append([
            row["item_name"],
            row["category"],
            row["unit_price"],
            row["sales_volume"],
            row["total_amount"],
            row.get("crawl_timestamp", "")
        ])
        
        # 逐单元格设置字体、对齐和格式
        for col_idx in range(1, len(columns) + 1):
            cell = ws.cell(row=current_row, column=col_idx)
            cell.font = data_font
            cell.border = thin_border
            
            # 单价、销售额靠右对齐并设千分位
            if col_idx in (3, 5):
                cell.alignment = Alignment(horizontal="right", vertical="center")
                cell.number_format = "¥#,##0.00"
            elif col_idx == 4:
                cell.alignment = Alignment(horizontal="right", vertical="center")
                cell.number_format = "#,##0"
            elif col_idx == 2:
                cell.alignment = Alignment(horizontal="center", vertical="center")
            else:
                cell.alignment = Alignment(horizontal="left", vertical="center")

    # 4. 插入汇总统计行
    total_row = len(df) + 2
    ws.cell(row=total_row, column=1, value="合计 / 汇总")
    ws.cell(row=total_row, column=4, value=f"=SUM(D2:D{total_row-1})")
    ws.cell(row=total_row, column=5, value=f"=SUM(E2:E{total_row-1})")

    for col_idx in range(1, len(columns) + 1):
        cell = ws.cell(row=total_row, column=col_idx)
        cell.font = summary_font
        cell.border = double_bottom_border
        if col_idx in (4, 5):
            cell.alignment = Alignment(horizontal="right", vertical="center")
            cell.number_format = "¥#,##0.00" if col_idx == 5 else "#,##0"

    # 5. 列宽自适应：考虑中文双字节字符宽度
    for col in ws.columns:
        max_len = 0
        col_letter = get_column_letter(col[0].column)
        for cell in col:
            val_str = str(cell.value or "")
            # 计算包含中文字符的显示宽度
            char_len = sum(2 if ord(c) > 127 else 1 for c in val_str)
            if char_len > max_len:
                max_len = char_len
        ws.column_dimensions[col_letter].width = max(max_len + 3, 12)

    # 冻结首行
    ws.freeze_panes = "A2"
    wb.save(output_path)
    print(f"Excel 自动化报表保存成功: {output_path}")

4.2 python-docx 自动编排文字与周报

在管理层汇报中，纯表格往往缺乏直观结论。利用 python-docx 可以把 Pandas 提炼的关键数据指标转换为一段自然流畅的分析摘要和 TOP 榜单表格。

from docx import Document
from docx.shared import Pt, Inches, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.enum.table import WD_TABLE_ALIGNMENT

def generate_word_summary(df: pd.DataFrame, output_path: str):
    """自动生成周度经营分析 Word 简报文档"""
    doc = Document()

    # 1. 设置正文标题
    title = doc.add_heading(level=0)
    title_run = title.add_run("本周市场数据采集与业务分析周报")
    title_run.font.name = "Microsoft YaHei"
    title_run.font.size = Pt(20)
    title_run.font.color.rgb = RGBColor(0x34, 0x5E, 0x82)
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER

    # 2. 宏观数据概述段落
    doc.add_heading("一、总体业务运行概要", level=1)
    total_sales = df["total_amount"].sum()
    total_volume = df["sales_volume"].sum()
    avg_price = df["unit_price"].mean()
    top_item = df.iloc[0]["item_name"] if not df.empty else "无"

    p = doc.add_paragraph()
    p.paragraph_format.line_spacing = 1.3
    p.add_run(f"本周期内脚本共采集到有效商品/业务记录 {len(df)} 条。")
    p.add_run(f"全网总销售额达到 ¥{total_sales:,.2f} 元，总销量为 {total_volume:,} 件，")
    p.add_run(f"商品平均客单价为 ¥{avg_price:,.2f} 元。")
    p.add_run(f"其中，表现最为突出的单品为【{top_item}】，销售额位列全盘首位。")

    # 3. 插入 TOP5 榜单表格
    doc.add_heading("二、畅销品 TOP 5 结构清单", level=1)
    table = doc.add_table(rows=1, cols=4)
    table.alignment = WD_TABLE_ALIGNMENT.CENTER
    hdr_cells = table.rows[0].cells
    hdr_titles = ["商品名称", "所属类目", "单价 (元)", "销售额 (元)"]
    for i, t in enumerate(hdr_titles):
        hdr_cells[i].text = t
        hdr_cells[i].paragraphs[0].runs[0].font.bold = True
        hdr_cells[i].paragraphs[0].runs[0].font.color.rgb = RGBColor(0x34, 0x5E, 0x82)

    top5 = df.head(5)
    for _, row in top5.iterrows():
        row_cells = table.add_row().cells
        row_cells[0].text = str(row["item_name"])
        row_cells[1].text = str(row["category"])
        row_cells[2].text = f"¥{row['unit_price']:,.2f}"
        row_cells[3].text = f"¥{row['total_amount']:,.2f}"

    doc.save(output_path)
    print(f"Word 周报文档生成成功: {output_path}")

五、模块四：定时自动化调度与闭环机制

完成了数据抓取与报表渲染之后，最后一步是让整套脚本具备“按时自动触发、失败自动告警、日志有据可查”的闭环能力。

5.1 调度与主控程序集成

使用 schedule 库编写主任务调度循环，并在 try...except 块中拦截全局异常，结合钉钉或企业微信群机器人发送即时状态提示。

import os
import time
import schedule
import traceback

def run_pipeline():
    """自动化流水线单次执行闭环"""
    print("\n" + "="*50)
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 触发自动化报表流水线...")
    
    target_url = "https://example.com/market/products"
    excel_path = os.path.abspath("周度业务监测报表.xlsx")
    word_path = os.path.abspath("业务经营分析简报.docx")

    try:
        # 1. 爬取
        records = fetch_product_data(target_url)
        if not records:
            raise ValueError("爬取数据为空，可能目标网站结构发生变动或触发反爬。")

        # 2. 清洗
        df = clean_and_transform(records)
        if df.empty:
            raise ValueError("数据清洗后无有效条目。")

        # 3. 生成 Excel
        export_styled_excel(df, excel_path)

        # 4. 生成 Word
        generate_word_summary(df, word_path)

        print("【成功】所有报表生成完毕，准备分发。")
        # 此处可调用 webhook 发送群消息与附件

    except Exception as e:
        error_msg = f"流水线执行异常: {str(e)}\n{traceback.format_exc()}"
        print(f"【报警】{error_msg}")
        # 此处可调用机器人 Webhook 发送报警消息至运维群

def start_scheduler():
    """启动常驻守护任务"""
    # 每周一上午 09:30 自动执行一次
    schedule.every().monday.at("09:30").do(run_pipeline)
    
    # 也支持开发阶段每 30 分钟轮询一次：
    # schedule.every(30).minutes.do(run_pipeline)

    print("自动化任务守护进程已启动，等待调度触发...")
    while True:
        schedule.run_pending()
        time.sleep(1)

if __name__ == "__main__":
    # 首次启动可立即触发一次验证
    run_pipeline()
    # 随后进入定时调度循环
    # start_scheduler()

六、环境依赖与工程实践要点

在生产环境中部署本套脚本时，建议遵循以下工程规范：