Day 14 · 实战：电商竞品价格监控全流程

梅雅达编程笔记

336人浏览 · 2026-08-17 17:09:10

梅雅达编程笔记 · 2026-08-17 17:09:10 发布

「AI Python 系列」第 03 栏 · Python 爬虫实战
全栏 15 篇 · 零成本跟完 🍃
作者：梅雅达编程笔记
首发：CSDN

摘要： 前面13篇学了一堆零散技能，今天全部串起来。从零搭建一个电商竞品价格监控系统——以当当网图书为例，完整走完"分析页面→编写爬虫→数据存储→AI清洗→生成报告"的全流程。代码模块化拆分成spider/parser/storage/cleaner/report五个模块，加上定时自动运行。这是一套可以直接拿去接私活的完整方案。

前面 13 篇学的东西，今天全部串起来。

做一个真实的场景：电商竞品价格监控。

你开了一家网上书店，想知道竞争对手的同类书籍定价。手动去看？几十本书每天看一遍，人废了。写个爬虫自动监控，每天跑一次，价格变动自动报告。

一、目标拆解

整个系统要完成的事：

1. 分析当当网图书列表页结构
2. 编写爬虫抓取图书信息（书名、作者、价格、评分、出版社）
3. 数据存入 SQLite
4. AI 清洗：标准化出版社名称、作者名
5. 生成报告：价格变动、新书上架、评分排名
6. 定时运行

代码结构：

price_monitor/
├── spider.py       # 爬虫：抓取页面
├── parser.py       # 解析：提取数据
├── storage.py      # 存储：SQLite 入库
├── cleaner.py      # 清洗：AI 数据清洗
├── reporter.py     # 报告：生成分析报告
├── main.py         # 主入口：串联全流程
├── requirements.txt
└── data/
    └── books.db

二、模块 1：`spider.py` - 爬虫模块

负责抓取页面，处理反爬。

"""
spider.py - 爬虫模块
负责抓取页面，处理反爬
"""
import requests
import random
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0",
]


def create_session():
    """创建带重试的 Session"""
    session = requests.Session()
    retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503])
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


def fetch_page(url, session=None):
    """抓取单个页面"""
    if session is None:
        session = create_session()
    headers = {
        "User-Agent": random.choice(USER_AGENTS),
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9",
    }
    try:
        time.sleep(random.uniform(0.5, 1.5))
        response = session.get(url, headers=headers, timeout=10)
        if response.status_code == 200:
            response.encoding = response.apparent_encoding
            return response.text
        return None
    except Exception:
        return None


def fetch_book_list(keyword, max_pages=3):
    """抓取关键词分类下的图书列表"""
    session = create_session()
    all_pages = []
    for page in range(1, max_pages + 1):
        url = f"http://search.dangdang.com/?key={keyword}&act=input&page_index={page}"
        html = fetch_page(url, session)
        if html:
            all_pages.append(html)
        else:
            break
    return all_pages

三、模块 2：`parser.py` - 解析模块

负责从 HTML 中提取数据。

"""
parser.py - 解析模块
负责从 HTML 中提取图书数据
"""
from bs4 import BeautifulSoup
import re


def parse_book_list(html):
    """从当当网图书列表页提取图书信息"""
    soup = BeautifulSoup(html, "lxml")
    books = []
    seen_urls = set()

    # 用 a.pic 定位图书链接
    pic_links = soup.select("a.pic")
    for link in pic_links:
        try:
            href = link.get("href", "")
            if not href or href in seen_urls:
                continue
            seen_urls.add(href)

            # 向上找包含完整信息的父级容器
            parent = link
            for _ in range(10):
                parent = parent.parent
                if parent is None:
                    break
                if parent.select_one("span.price_n") or parent.select_one("p.price span"):
                    break
                if parent.name == "li":
                    break

            book = {}
            book["url"] = href

            # 书名
            book["title"] = link.get("title", "") or link.get_text(strip=True)
            if not book["title"]:
                title_tag = parent.select_one("h3") or parent.select_one("a[title]")
                if title_tag:
                    book["title"] = title_tag.get_text(strip=True)

            # 作者
            author_tag = parent.select_one("span.search_book_author") or parent.select_one("p.authors")
            book["author"] = author_tag.get_text(strip=True) if author_tag else ""

            # 出版社
            publisher_tag = parent.select_one("span.publisher") or parent.select_one("p.publisher")
            book["publisher"] = publisher_tag.get_text(strip=True) if publisher_tag else ""

            # 价格
            price_tag = parent.select_one("span.price_n") or parent.select_one("p.price span")
            if price_tag:
                price_text = price_tag.get_text(strip=True)
                price_match = re.search(r'[\d.]+', price_text)
                book["price"] = float(price_match.group()) if price_match else None
            else:
                book["price"] = None

            # 评分
            rating_tag = parent.select_one("span.search_comment_num") or parent.select_one("div.star")
            book["rating"] = rating_tag.get_text(strip=True) if rating_tag else ""

            # 简介
            desc_tag = parent.select_one("p.detail") or parent.select_one("div.describe")
            book["description"] = desc_tag.get_text(strip=True) if desc_tag else ""

            if book.get("title"):
                books.append(book)
        except Exception:
            continue

    return books


def parse_all_pages(html_pages):
    """解析多个页面并去重"""
    all_books = []
    for html in html_pages:
        books = parse_book_list(html)
        all_books.extend(books)

    # 去重（按 URL）
    seen_urls = set()
    unique_books = []
    for book in all_books:
        url = book.get("url", "")
        if url and url not in seen_urls:
            seen_urls.add(url)
            unique_books.append(book)

    return unique_books

四、模块 3：`storage.py` - 存储模块

负责数据入库，支持增量更新。

"""
storage.py - 存储模块
负责数据入库，支持增量更新
"""
import sqlite3
import pandas as pd
import os

DB_PATH = "data/books.db"


def init_db():
    """初始化数据库"""
    os.makedirs("data", exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS books (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            author TEXT,
            publisher TEXT,
            price REAL,
            rating TEXT,
            description TEXT,
            url TEXT UNIQUE,
            first_seen DATE DEFAULT CURRENT_DATE,
            last_updated DATE DEFAULT CURRENT_DATE
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            book_url TEXT NOT NULL,
            price REAL,
            recorded_at DATE DEFAULT CURRENT_DATE,
            FOREIGN KEY (book_url) REFERENCES books(url)
        )
    """)
    conn.commit()
    conn.close()


def save_books(books):
    """保存图书数据（增量更新）"""
    conn = sqlite3.connect(DB_PATH)
    new_count = 0
    update_count = 0

    for book in books:
        cursor = conn.execute("SELECT id, price FROM books WHERE url = ?", (book.get("url", ""),))
        existing = cursor.fetchone()

        if existing:
            book_id, old_price = existing
            conn.execute("""
                UPDATE books SET price = ?, rating = ?, last_updated = CURRENT_DATE
                WHERE id = ?
            """, (book.get("price"), book.get("rating"), book_id))

            if old_price != book.get("price"):
                conn.execute("""
                    INSERT INTO price_history (book_url, price) VALUES (?, ?)
                """, (book.get("url"), book.get("price")))
            update_count += 1
        else:
            conn.execute("""
                INSERT INTO books (title, author, publisher, price, rating, description, url)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (
                book.get("title"), book.get("author"), book.get("publisher"),
                book.get("price"), book.get("rating"), book.get("description"),
                book.get("url")
            ))
            new_count += 1

    conn.commit()
    conn.close()
    return new_count, update_count


def get_all_books():
    """读取所有图书数据"""
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("SELECT * FROM books ORDER BY price", conn)
    conn.close()
    return df


def get_price_changes():
    """获取价格变动"""
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("""
        SELECT b.title, b.author, b.url,
               h.price as old_price,
               b.price as new_price,
               (b.price - h.price) as change,
               h.recorded_at
        FROM price_history h
        JOIN books b ON b.url = h.book_url
        ORDER BY h.recorded_at DESC
        LIMIT 20
    """, conn)
    conn.close()
    return df

五、模块 4：`cleaner.py` - AI 清洗模块

"""
cleaner.py - AI 数据清洗模块
"""
import json
from openai import OpenAI

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="https://open.bigmodel.cn/api/paas/v4/"
)

MODEL = "glm-4-flash"

def clean_book_data(books):
    """用 AI 清洗图书数据"""
    
    if not books:
        return books
    
    # 分批处理
    batch_size = 20
    all_cleaned = []
    
    for i in range(0, len(books), batch_size):
        batch = books[i:i + batch_size]
        print(f"正在清洗第 {i+1}-{min(i+batch_size, len(books))} 条...")
        
        prompt = f"""请清洗以下图书数据，执行以下操作：
1. 标准化作者名（去掉"著"、"编著"等后缀，只保留人名）
2. 标准化出版社名（如"人民邮电出版社有限公司"→"人民邮电出版社"）
3. 如果作者有多个，用"、"分隔
4. 检查价格是否合理（0-1000范围内）
5. 去掉书名中的多余空格和特殊字符

返回 JSON 数组，保持原有字段，加上 "clean_note" 字段说明改了什么。
只返回 JSON。

数据：
{json.dumps(batch, ensure_ascii=False, indent=2)}"""

        try:
            response = client.chat.completions.create(
                model=MODEL,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.1
            )
            
            content = response.choices[0].message.content.strip()
            
            # 清理 markdown
            if "```json" in content:
                content = content.split("```json")[1].split("```")[0]
            elif "```" in content:
                content = content.split("```")[1].split("```")[0]
            
            cleaned_batch = json.loads(content)
            all_cleaned.extend(cleaned_batch)
            
        except Exception as e:
            print(f"  AI 清洗失败，保留原始数据：{e}")
            all_cleaned.extend(batch)
        
        import time
        time.sleep(1)
    
    return all_cleaned

六、模块 5：`reporter.py` - 报告模块

"""
reporter.py - 报告生成模块
使用 GLM-4.5-AIR 生成智能分析报告
"""
import pandas as pd
from datetime import datetime
import os
import json
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env"))

client = OpenAI(
    api_key=os.getenv("GLM_API_KEY"),
    base_url=os.getenv("BIGMODEL_BASE_URL", "https://open.bigmodel.cn/api/paas/v4/"),
    timeout=60
)

MODEL = os.getenv("GLM_REPORTER_MODEL", "glm-4.5-air")


def generate_ai_report(df, changes_df):
    """用 GLM-4.5-AIR 生成智能分析报告"""
    if len(df) == 0:
        return "（无数据）"

    summary = {
        "total_books": len(df),
        "avg_price": round(df["price"].mean(), 2) if "price" in df else 0,
        "min_price": round(df["price"].min(), 2) if "price" in df else 0,
        "max_price": round(df["price"].max(), 2) if "price" in df else 0,
        "median_price": round(df["price"].median(), 2) if "price" in df else 0,
        "top5_expensive": df.nlargest(5, "price")[["title", "price", "author"]].to_dict("records") if "price" in df else [],
        "top5_cheap": df.nsmallest(5, "price")[["title", "price", "author"]].to_dict("records") if "price" in df else [],
        "price_changes_count": len(changes_df),
        "recent_changes": changes_df.head(5).to_dict("records") if len(changes_df) > 0 else [],
        "publishers_top5": df["publisher"].value_counts().head(5).to_dict() if "publisher" in df else {},
    }

    prompt = f"""你是资深电商竞品分析师。基于以下图书价格监控数据，生成一份专业的中文分析报告。

【数据摘要】
{json.dumps(summary, ensure_ascii=False, indent=2, default=str)}

【报告要求】
1. 用中文，分 5 个段落，每段用 --- 分隔
2. 段落：一、市场概览 | 二、价格分布分析 | 三、TOP商家与品类 | 四、价格变动追踪 | 五、竞品策略建议
3. 数据要具体（引用实际数字），不要空话
4. 第五段给出 3-5 条可操作建议（针对书店经营者）
5. 报告 500-800 字
"""

    try:
        response = client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": "你是资深电商数据分析师，输出专业中文报告，段落用---分隔。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.7,
            max_tokens=2000
        )
        return response.choices[0].message.content.strip()
    except Exception:
        # AI 失败则用基础模板
        return generate_basic_report(df, changes_df)


def generate_basic_report(df, changes_df):
    """基础报告（备用）"""
    lines = [f"# 图书竞品价格监控报告",
             f"\n**生成时间：** {datetime.now().strftime('%Y-%m-%d %H:%M')}",
             f"**监控图书数：** {len(df)} 本",
             "\n## 一、价格概览\n"]

    if len(df) > 0:
        lines.append(f"- **平均价格：** ¥{df['price'].mean():.2f}")
        lines.append(f"- **最低价：** ¥{df['price'].min():.2f}")
        lines.append(f"- **最高价：** ¥{df['price'].max():.2f}")

    lines.append("\n## 二、近期价格变动\n")
    if len(changes_df) > 0:
        for _, row in changes_df.head(10).iterrows():
            direction = "📈" if row["change"] > 0 else "📉"
            sign = "+" if row["change"] > 0 else ""
            lines.append(f"- {direction} **{row['title']}**：¥{row['old_price']:.2f} → ¥{row['new_price']:.2f}（{sign}{row['change']:.2f}）")
    else:
        lines.append("- 暂无价格变动")

    return "\n".join(lines)


def generate_report(storage_module):
    """生成价格监控报告（主入口）"""
    df = storage_module.get_all_books()
    changes_df = storage_module.get_price_changes()

    report_text = generate_ai_report(df, changes_df)

    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_file = f"data/report_{ts}.md"

    with open(report_file, "w", encoding="utf-8") as f:
        f.write(f"# 图书竞品价格监控报告\n\n")
        f.write(f"**生成时间：** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
        f.write(f"**监控图书数：** {len(df)} 本\n\n")
        f.write("---\n\n")
        f.write(report_text)
        f.write("\n\n---\n\n")
        f.write("## 附：价格 Top 10\n\n")

        if len(df) > 0:
            top10 = df.nlargest(10, "price")[["title", "author", "price", "publisher"]]
            f.write(top10.to_markdown(index=False))

        if len(changes_df) > 0:
            f.write("\n\n## 附：近期价格变动\n\n")
            f.write(changes_df.head(10).to_markdown(index=False))

    return report_file

七、`main.py` - 主入口

"""
main.py - 主入口：串联全流程
"""
import sys
import time
from datetime import datetime

from spider import fetch_book_list
from parser import parse_all_pages
from storage import init_db, save_books
from cleaner import clean_book_data
from reporter import generate_report
import storage as storage_module


def run_pipeline(keyword="Python编程", max_pages=3):
    """执行完整的采集流程"""
    start_time = time.time()

    # 1. 初始化数据库
    init_db()

    # 2. 抓取页面
    html_pages = fetch_book_list(keyword, max_pages=max_pages)
    if not html_pages:
        return None

    # 3. 解析数据
    books = parse_all_pages(html_pages)
    if not books:
        return None

    # 4. AI 清洗
    cleaned_books = clean_book_data(books)

    # 5. 存储
    new_count, update_count = save_books(cleaned_books)

    # 6. 生成报告
    report_file = generate_report(storage_module)

    elapsed = time.time() - start_time
    return {
        "books_count": len(books),
        "new_count": new_count,
        "update_count": update_count,
        "report_file": report_file,
        "elapsed": elapsed
    }


def run_scheduled(interval_hours=24):
    """定时运行"""
    import schedule
    schedule.every(interval_hours).hours.do(run_pipeline)
    run_pipeline()
    while True:
        schedule.run_pending()
        time.sleep(60)


if __name__ == "__main__":
    if "--scheduled" in sys.argv:
        run_scheduled(interval_hours=24)
    else:
        run_pipeline()

八、`requirements.txt`

requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=4.9.0
pandas>=2.0.0
openai>=1.12.0
schedule>=1.2.0

安装：

pip install -r requirements.txt

九、运行

单次运行

python main.py

定时运行（每 24 小时）

python main.py --scheduled

用 `cron` 定时（Linux/Mac）

# 每天早上 8 点运行
crontab -e
0 8 * * * cd /path/to/price_monitor && python main.py >> data/cron.log 2>&1

十、扩展思路

这个框架可以很容易扩展：

1. 加通知

价格大幅变动时发微信/邮件通知：

# 在 reporter.py 中加入
def check_price_alerts(changes_df, threshold=0.2):
    """价格变动超过 threshold 比例时发出警告"""
    alerts = changes_df[abs(changes_df["change"] / changes_df["old_price"]) > threshold]
    
    if len(alerts) > 0:
        message = "⚠️ 价格大幅变动：\n"
        for _, row in alerts.iterrows():
            message += f"  {row['title']}：¥{row['old_price']} → ¥{row['new_price']}\n"
        # 发送通知（邮件/微信/钉钉）
        send_notification(message)

2. 多品类监控

CATEGORIES = {
    "Python编程": "http://search.dangdang.com/?key=Python...",
    "Java编程": "http://search.dangdang.com/?key=Java...",
    "人工智能": "http://search.dangdang.com/?key=人工智能...",
}

for name, url in CATEGORIES.items():
    print(f"\n{'='*40}")
    print(f"监控分类：{name}")
    html_pages = fetch_book_list(url, max_pages=3)
    # ...

3. 可视化

# 用 matplotlib 画价格趋势图
import matplotlib.pyplot as plt

def plot_price_trend(book_url):
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("""
        SELECT recorded_at, price FROM price_history 
        WHERE book_url = ? ORDER BY recorded_at
    """, conn, params=(book_url,))
    conn.close()
    
    plt.figure(figsize=(10, 5))
    plt.plot(df["recorded_at"], df["price"], marker="o")
    plt.title("价格趋势")
    plt.xlabel("日期")
    plt.ylabel("价格（元）")
    plt.savefig("data/price_trend.png")

📊 本篇成本透明栏

项目	数值
API 调用次数	~5-10 次（AI 清洗）
消耗 Token	约 1-3 万 tokens
成本	¥0（GLM-4.7-Flash 免费额度内）

练手改造题

改造 1（基础）： 把这个框架跑起来，换成你关注的品类（比如"机器学习"、“产品经理”），生成你的第一份监控报告。

改造 2（进阶）： 加上邮件通知功能——当某本书价格下降超过 20% 时，自动发邮件提醒你。

下期预告

Day 15 · 实战：AI 驱动的智能数据助手

最后一篇，做一个"对话式数据查询工具"——用自然语言告诉它你要什么数据，它自动决定怎么爬、怎么解析、怎么存储。把前面所有知识串成一个完整的产品。

📚 资源与工具

当当网： https://www.dangdang.com/
schedule 库文档： https://schedule.readthedocs.io/
本专栏配套代码： CSDN 下载区（每篇更新）
梅雅达编程笔记： 专注 Python + AI 实战教程，提供数据采集与自动化定制服务

往期回顾

Day 01 · 爬虫能干啥不能干啥

Day 02 · 环境搭建与第一个爬虫

Day 03 · 列表页抓取：批量拿数据

Day 04 · 详情页 + 翻页：从一条到一百条

Day 05 · Ajax 请求拦截：抓看不到的数据

Day 06 · 浏览器自动化：DrissionPage 实战

Day 07 · Cookie 与 Session：处理登录状态

Day 08 · 反爬对策：Headers + 限速 + 代理

Day 09 · 存成文件：CSV / Excel / JSON

Day 10 · 存进数据库：SQLite 从零上手

Day 11 · 用 AI 替代正则：智能提取结构化数据

Day 12 · AI 数据清洗：脏数据一键变干净

Day 13 · AI 辅助解析：复杂表格和嵌套结构

专栏推荐

「AI Python 系列」第 03 栏 · AI+Python 爬虫实战（连载中）

「AI Python 系列」第 01 栏 ·AI+办公自动化（连载中）

「AI Python 系列」第 02 栏 ·[AI+数据可视化（连载中）

「AI Python 系列」第 05 栏 · AI Agent实战（连载中）