从零到接单 13：综合实战——批量抓取 + 自动报表 + 定时运行
pythontutorial爬虫自动化接单综合实战项目
系列目录：本文是「从零到接单：Python 自动化与爬虫实战」系列的第 13 篇。前面 12 篇我们学习了 Python 基础、爬虫、数据存储和办公自动化。这篇把所有技能串联起来，构建一个可以交付给客户的完整自动化项目。
这篇文章是一个完整可运行的项目。读完你应该能：
- 理解一个完整自动化项目的架构
- 照猫画虎写出自己的接单项目
- 学会定时运行脚本（给客户"交钥匙"）
项目需求
假设客户需求：
"帮我每天从某电商网站抓取指定品类的商品信息（价格、销量、评分），自动生成一份 Excel 报表和 Word 分析报告，每天早上 8 点前发到邮箱。"
我们把这个需求拆解为：
┌─────────────────────────────────────────────────┐
│ 1. 数据采集（requests + BS4） │
│ 抓取商品列表页 → 提取字段 → 翻页 │
├─────────────────────────────────────────────────┤
│ 2. 数据清洗与存储 │
│ 去重 → 类型转换 → 存入 SQLite │
├─────────────────────────────────────────────────┤
│ 3. 数据分析 │
│ 统计 → 排序 → 计算变化 │
├─────────────────────────────────────────────────┤
│ 4. 自动生成报表 │
│ Excel：数据明细 + 图表 │
│ Word：分析报告 │
├─────────────────────────────────────────────────┤
│ 5. 定时运行 │
│ cron / 任务计划程序 / schedule 库 │
└─────────────────────────────────────────────────┘
一、完整项目代码
下面逐模块实现，最终合并为一个可运行的主程序。
1. 配置模块 config.py
# config.py —— 项目配置
# 目标网站配置
TARGET_URL = "https://example-shop.com/category/electronics?page={page}"
TOTAL_PAGES = 3
# 请求配置
REQUEST_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
}
REQUEST_DELAY = 2 # 请求间隔（秒）
# 数据库配置
DB_PATH = "shop_data.db"
# 输出配置
OUTPUT_DIR = "reports"
EXCEL_REPORT = "商品数据报表.xlsx"
WORD_REPORT = "商品分析报告.docx"
# 关键词过滤
SEARCH_KEYWORDS = ["Python", "编程", "自动化"]
2. 爬虫模块 crawler.py
# crawler.py —— 数据采集模块
import requests
import time
import random
from bs4 import BeautifulSoup
from typing import List, Dict, Optional
class ProductCrawler:
"""商品信息爬虫"""
def __init__(self, config):
self.config = config
self.session = requests.Session()
self.session.headers.update(config.REQUEST_HEADERS)
def fetch_page(self, url: str, max_retries: int = 3) -> Optional[str]:
"""抓取单个页面"""
for attempt in range(max_retries):
try:
print(f" 请求：{url}")
resp = self.session.get(url, timeout=15)
resp.raise_for_status()
resp.encoding = resp.apparent_encoding
return resp.text
except requests.RequestException as e:
print(f" 第 {attempt + 1} 次失败：{e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
return None
def parse_product(self, card) -> Optional[Dict]:
"""
解析单个商品卡片
注意：这里的选择器需要根据实际网站调整！
"""
try:
# 以下为示例选择器，实际使用时需要修改
title_el = card.select_one(".product-title")
price_el = card.select_one(".product-price")
sales_el = card.select_one(".product-sales")
rating_el = card.select_one(".product-rating")
link_el = card.select_one("a.product-link")
# 清洗数据
title = title_el.get_text(strip=True) if title_el else ""
price_text = price_el.get_text(strip=True) if price_el else "0"
sales_text = sales_el.get_text(strip=True) if sales_el else "0"
rating_text = rating_el.get_text(strip=True) if rating_el else "0"
# 提取数字
import re
price = float(re.findall(r"\d+\.?\d*", price_text)[0]) if re.findall(r"\d+\.?\d*", price_text) else 0
sales = int(re.findall(r"\d+", sales_text)[0]) if re.findall(r"\d+", sales_text) else 0
rating = float(re.findall(r"\d+\.?\d*", rating_text)[0]) if re.findall(r"\d+\.?\d*", rating_text) else 0
href = link_el.get("href", "") if link_el else ""
return {
"title": title,
"price": price,
"sales": sales,
"rating": rating,
"link": href,
}
except Exception as e:
print(f" 解析商品失败：{e}")
return None
def scrape_category(self) -> List[Dict]:
"""抓取整个品类的商品列表"""
all_products = []
for page in range(1, self.config.TOTAL_PAGES + 1):
url = self.config.TARGET_URL.format(page=page)
print(f"\n📄 正在抓取第 {page}/{self.config.TOTAL_PAGES} 页...")
html = self.fetch_page(url)
if html is None:
print(f" ❌ 第 {page} 页抓取失败，跳过")
continue
soup = BeautifulSoup(html, "lxml")
cards = soup.select(".product-card") # 根据实际结构调整
page_products = []
for card in cards:
product = self.parse_product(card)
if product:
page_products.append(product)
all_products.extend(page_products)
print(f" ✅ 第 {page} 页抓取 {len(page_products)} 条，累计 {len(all_products)} 条")
# 翻页延时
if page < self.config.TOTAL_PAGES:
delay = random.uniform(1, self.config.REQUEST_DELAY)
time.sleep(delay)
return all_products
def close(self):
self.session.close()
3. 数据存储模块 storage.py
# storage.py —— 数据持久化模块
import sqlite3
from datetime import datetime
from typing import List, Dict
class DataStorage:
"""数据存储管理器"""
def __init__(self, db_path: str):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_db()
def _init_db(self):
"""初始化数据库表"""
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
price REAL DEFAULT 0,
sales INTEGER DEFAULT 0,
rating REAL DEFAULT 0,
link TEXT,
crawl_date DATE DEFAULT (date('now')),
crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(title, crawl_date)
);
CREATE TABLE IF NOT EXISTS crawl_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
total_items INTEGER,
new_items INTEGER,
status TEXT,
error TEXT
);
-- 价格历史表（追踪价格变化）
CREATE TABLE IF NOT EXISTS price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_title TEXT,
price REAL,
record_date DATE DEFAULT (date('now')),
UNIQUE(product_title, record_date)
);
""")
self.conn.commit()
def save_products(self, products: List[Dict]) -> int:
"""批量保存商品（去重）"""
cursor = self.conn.cursor()
new_count = 0
for product in products:
try:
cursor.execute("""
INSERT OR IGNORE INTO products (title, price, sales, rating, link)
VALUES (?, ?, ?, ?, ?)
""", (product["title"], product["price"],
product["sales"], product["rating"], product["link"]))
if cursor.rowcount > 0:
new_count += 1
except sqlite3.Error as e:
print(f" 保存失败 [{product.get('title', 'unknown')}]：{e}")
self.conn.commit()
return new_count
def save_price_history(self):
"""保存当日价格快照"""
cursor = self.conn.cursor()
cursor.execute("""
INSERT OR IGNORE INTO price_history (product_title, price, record_date)
SELECT title, price, date('now') FROM products
WHERE crawl_date = date('now')
""")
self.conn.commit()
def log_crawl(self, total: int, new: int, status: str, error: str = None):
"""记录爬取日志"""
self.conn.execute(
"INSERT INTO crawl_logs (total_items, new_items, status, error) VALUES (?, ?, ?, ?)",
(total, new, status, error)
)
self.conn.commit()
def get_today_products(self) -> List[Dict]:
"""获取今日商品数据"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT * FROM products WHERE crawl_date = date('now')
""")
return [dict(row) for row in cursor.fetchall()]
def get_stats(self) -> Dict:
"""获取统计信息"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT
COUNT(*) as total,
AVG(price) as avg_price,
MAX(price) as max_price,
MIN(price) as min_price,
AVG(rating) as avg_rating,
MAX(sales) as max_sales
FROM products WHERE crawl_date = date('now')
""")
row = cursor.fetchone()
return dict(row) if row else {}
def get_price_changes(self) -> List[Dict]:
"""获取价格变化（与昨天对比）"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT
a.product_title,
a.price as today_price,
b.price as yesterday_price,
a.price - b.price as change
FROM price_history a
LEFT JOIN price_history b
ON a.product_title = b.product_title
AND b.record_date = date('now', '-1 day')
WHERE a.record_date = date('now')
AND b.price IS NOT NULL
AND a.price != b.price
ORDER BY change DESC
""")
return [dict(row) for row in cursor.fetchall()]
def close(self):
self.conn.close()
4. 报表生成模块 report.py
# report.py —— 报表生成模块
from datetime import datetime
from pathlib import Path
import os
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.chart import BarChart, PieChart, Reference
from openpyxl.utils import get_column_letter
from docx import Document
from docx.shared import Inches, Pt, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.enum.table import WD_TABLE_ALIGNMENT
class ReportGenerator:
"""报表生成器"""
def __init__(self, output_dir: str):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def _style_header(self, ws, row, num_cols):
"""统一表头样式"""
header_fill = PatternFill(start_color="2F5496", end_color="2F5496", fill_type="solid")
header_font = Font(bold=True, color="FFFFFF", size=11)
for col in range(1, num_cols + 1):
cell = ws.cell(row=row, column=col)
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal="center", vertical="center")
def generate_excel(self, products, stats, price_changes):
"""生成 Excel 报表"""
wb = Workbook()
# ── Sheet 1：数据明细 ──
ws1 = wb.active
ws1.title = "商品明细"
# 标题行
ws1.merge_cells("A1:F1")
title_cell = ws1.cell(row=1, column=1,
value=f"商品数据报表 - {datetime.now().strftime('%Y年%m月%d日')}")
title_cell.font = Font(bold=True, size=14, color="2F5496")
# 表头
headers = ["商品名称", "价格(元)", "销量", "评分", "链接", "采集时间"]
for col, header in enumerate(headers, 1):
ws1.cell(row=3, column=col, value=header)
self._style_header(ws1, 3, len(headers))
# 数据
for i, product in enumerate(products, 4):
ws1.cell(row=i, column=1, value=product["title"][:60])
ws1.cell(row=i, column=2, value=product["price"])
ws1.cell(row=i, column=3, value=product["sales"])
ws1.cell(row=i, column=4, value=product["rating"])
ws1.cell(row=i, column=5, value=product.get("link", ""))
ws1.cell(row=i, column=6, value=str(product.get("crawl_time", ""))[:19])
# 调整列宽
widths = [35, 10, 10, 8, 40, 18]
for col, w in enumerate(widths, 1):
ws1.column_dimensions[get_column_letter(col)].width = w
# ── Sheet 2：统计汇总 ──
ws2 = wb.create_sheet("统计汇总")
ws2.merge_cells("A1:B1")
ws2.cell(row=1, column=1, value="数据统计汇总").font = Font(bold=True, size=14)
stat_items = [
("商品总数", stats.get("total", 0)),
("平均价格", f"¥{stats.get('avg_price', 0):.2f}" if stats.get('avg_price') else "N/A"),
("最高价格", f"¥{stats.get('max_price', 0):.2f}" if stats.get('max_price') else "N/A"),
("最低价格", f"¥{stats.get('min_price', 0):.2f}" if stats.get('min_price') else "N/A"),
("平均评分", f"{stats.get('avg_rating', 0):.1f}" if stats.get('avg_rating') else "N/A"),
("最高销量", stats.get("max_sales", 0) or 0),
]
for i, (label, value) in enumerate(stat_items, 3):
ws2.cell(row=i, column=1, value=label).font = Font(bold=True)
ws2.cell(row=i, column=2, value=value)
# ── Sheet 3：价格变化 ──
if price_changes:
ws3 = wb.create_sheet("价格变化")
ws3.append(["商品名称", "今日价格", "昨日价格", "变化幅度"])
for item in price_changes:
ws3.append([
item["product_title"],
item["today_price"],
item["yesterday_price"],
f"{item['change']:+.2f}"
])
filepath = self.output_dir / f"商品报表_{datetime.now().strftime('%Y%m%d')}.xlsx"
wb.save(filepath)
print(f"✅ Excel 报表已生成：{filepath}")
return str(filepath)
def generate_word(self, products, stats, price_changes, top_products, existing_chart_path=None):
"""生成 Word 分析报告"""
doc = Document()
# 标题
title = doc.add_heading("商品数据分析报告", level=0)
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
meta = doc.add_paragraph()
meta.alignment = WD_ALIGN_PARAGRAPH.CENTER
meta.add_run(f"生成日期：{datetime.now().strftime('%Y年%m月%d日')}").font.size = Pt(10)
doc.add_paragraph()
# 一、采集概况
doc.add_heading("一、采集概况", level=1)
doc.add_paragraph(
f"本次共采集商品 {stats.get('total', 0)} 件，"
f"价格区间 ¥{stats.get('min_price', 0):.0f} ~ ¥{stats.get('max_price', 0):.0f}，"
f"均价 ¥{stats.get('avg_price', 0):.2f}，"
f"平均评分 {stats.get('avg_rating', 0):.1f} 分。"
)
# 二、TOP 商品
doc.add_heading("二、热销商品 TOP 5", level=1)
if top_products:
table = doc.add_table(rows=1, cols=4, style="Light Grid Accent 1")
table.alignment = WD_TABLE_ALIGNMENT.CENTER
for i, text in enumerate(["排名", "商品名称", "销量", "评分"]):
table.rows[0].cells[i].text = text
for rank, product in enumerate(top_products[:5], 1):
row = table.add_row()
row.cells[0].text = str(rank)
row.cells[1].text = product["title"][:40]
row.cells[2].text = str(product["sales"])
row.cells[3].text = f"{product['rating']:.1f}"
# 三、价格变动
doc.add_heading("三、价格变动分析", level=1)
if price_changes:
up_count = sum(1 for c in price_changes if c["change"] > 0)
down_count = sum(1 for c in price_changes if c["change"] < 0)
doc.add_paragraph(
f"今日共有 {len(price_changes)} 件商品价格发生变化。"
f"其中 {up_count} 件上涨，{down_count} 件下跌。"
)
else:
doc.add_paragraph("今日无显著价格变动。")
# 四、结论
doc.add_heading("四、结论与建议", level=1)
doc.add_paragraph(
"根据今日数据分析：\n"
"1. 商品价格整体稳定，波动在合理范围内\n"
"2. 高评分商品集中在 ¥{:.0f} 以上价格段\n"
"3. 建议关注评分低于 3.0 的商品，可能存在质量问题".format(
stats.get("avg_price", 0) or 0
)
)
doc.add_paragraph()
footer = doc.add_paragraph()
footer.alignment = WD_ALIGN_PARAGRAPH.CENTER
footer.add_run("— 本报告由 AutoReporter 自动生成 —").font.size = Pt(9)
filepath = self.output_dir / f"分析报告_{datetime.now().strftime('%Y%m%d')}.docx"
doc.save(filepath)
print(f"✅ Word 报告已生成：{filepath}")
return str(filepath)
5. 主程序 main.py
# main.py —— 主控程序
import sys
import traceback
from datetime import datetime
from pathlib import Path
from config import *
from crawler import ProductCrawler
from storage import DataStorage
from report import ReportGenerator
class AutoReporter:
"""自动化报表系统主控"""
def __init__(self):
self.config = type("Config", (), {k: v for k, v in globals().items() if k.isupper()})()
self.crawler = ProductCrawler(self.config)
self.storage = DataStorage(DB_PATH)
self.reporter = ReportGenerator(OUTPUT_DIR)
def run(self):
"""执行完整的自动化流程"""
start_time = datetime.now()
print("=" * 60)
print(f"🚀 AutoReporter 启动 - {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
try:
# 步骤 1：数据采集
print("\n📊 步骤 1/5：数据采集")
products = self.crawler.scrape_category()
print(f"采集完成：共获取 {len(products)} 条商品数据")
if not products:
print("⚠️ 未采集到数据，流程终止")
self.storage.log_crawl(0, 0, "NO_DATA")
return
# 步骤 2：数据存储
print("\n💾 步骤 2/5：数据存储")
new_count = self.storage.save_products(products)
self.storage.save_price_history()
print(f"存储完成：新增 {new_count} 条，共 {len(products)} 条")
# 步骤 3：数据分析
print("\n📈 步骤 3/5：数据分析")
today_products = self.storage.get_today_products()
stats = self.storage.get_stats()
price_changes = self.storage.get_price_changes()
# TOP 5 热销
top_products = sorted(today_products, key=lambda x: x.get("sales", 0), reverse=True)
print(f" 商品总数：{stats.get('total', 0)}")
print(f" 均价：¥{stats.get('avg_price', 0):.2f}" if stats.get('avg_price') else " 均价：N/A")
print(f" 价格变动：{len(price_changes)} 件")
# 步骤 4：生成报表
print("\n📝 步骤 4/5：生成报表")
excel_path = self.reporter.generate_excel(today_products, stats, price_changes)
word_path = self.reporter.generate_word(today_products, stats, price_changes, top_products)
# 步骤 5：记录日志
print("\n📋 步骤 5/5：记录日志")
self.storage.log_crawl(len(products), new_count, "SUCCESS")
# 完成
elapsed = (datetime.now() - start_time).total_seconds()
print("\n" + "=" * 60)
print(f"✅ 所有任务完成！耗时 {elapsed:.1f} 秒")
print(f"📁 Excel：{excel_path}")
print(f"📁 Word：{word_path}")
print("=" * 60)
except Exception as e:
print(f"\n❌ 运行失败：{e}")
traceback.print_exc()
self.storage.log_crawl(0, 0, "FAILED", str(e))
finally:
self.crawler.close()
self.storage.close()
if __name__ == "__main__":
reporter = AutoReporter()
reporter.run()
二、定时运行
macOS / Linux：crontab
# 编辑定时任务
crontab -e
# 每天早上 7:00 运行
0 7 * * * cd /path/to/project && python3 main.py >> logs/cron.log 2>&1
Windows：任务计划程序
# PowerShell（管理员模式）
$action = New-ScheduledTaskAction -Execute "python" -Argument "C:\project\main.py" -WorkingDirectory "C:\project"
$trigger = New-ScheduledTaskTrigger -Daily -At "07:00"
Register-ScheduledTask -TaskName "AutoReporter" -Action $action -Trigger $trigger
Python schedule 库（轻量方案）
pip install schedule
# scheduler_runner.py —— Python 内置定时器
import schedule
import time
def job():
from main import AutoReporter
reporter = AutoReporter()
reporter.run()
# 每天早上 7:00 运行
schedule.every().day.at("07:00").do(job)
print("⏰ 定时任务已启动，等待执行...")
while True:
schedule.run_pending()
time.sleep(60) # 每分钟检查一次
三、项目交付清单
当你把这个项目交给客户时，应包含：
AutoReporter/
├── main.py # 主程序
├── config.py # 配置文件
├── crawler.py # 爬虫模块
├── storage.py # 存储模块
├── report.py # 报表模块
├── scheduler_runner.py # 定时运行
├── requirements.txt # 依赖列表
├── README.md # 使用说明
├── logs/ # 日志目录
└── reports/ # 报表输出目录
requirements.txt：
requests==2.31.0
beautifulsoup4==4.12.2
lxml==5.1.0
openpyxl==3.1.2
python-docx==1.1.0
schedule==1.2.0
四、扩展方向
这个项目是一个"骨架"，可以根据客户需求扩展：
- 多平台监控：不同网站用不同的解析器
- 邮件通知：用
smtplib发送报表到客户邮箱 - Web 仪表盘：用 Flask 做一个简单的查看页面
- 钉钉/微信通知：接入 webhook 实时推送异常
- 数据可视化：生成趋势图、热力图等
总结
这篇文章串联了所有技能，完成了一个"交钥匙"级的自动化项目。现在你已经具备：
- ✅ 用 requests + BS4 抓取网页数据
- ✅ 用 SQLite 持久化存储
- ✅ 用 openpyxl 生成 Excel 报表
- ✅ 用 python-docx 生成 Word 报告
- ✅ 用 cron/schedule 定时运行
这就是接单的核心能力。下一篇，我们进入接单实战指南——教你如何找到客户、写提案、定价和避坑。
练习：修改
crawler.py中的parse_product()方法，让它适配一个你感兴趣的网站。跑通整个流程，生成你的第一份自动化报表。
Comments
Sign in to leave a comment.