从爬虫到透视表：构建Python+MySQL+Excel电商数据分析闭环

在实际商业数据分析项目中，数据透视表、数据库操作、Python脚本和网络爬虫是四个紧密关联的核心技能。很多教程将它们分开讲解，导致学习者在面对真实业务需求时，难以将这些工具串联成一个完整的工作流。例如，如何从网页获取原始数据，清洗后存入数据库，再用SQL或Python进行分析，最后用数据透视表进行可视化呈现，这个链条的每个环节都有其技术细节和常见陷阱。

本文旨在构建一个从数据采集到分析呈现的完整闭环。我们将模拟一个常见的商业分析场景：分析某电商平台的商品信息。整个过程将覆盖使用Python爬虫抓取数据、利用Pandas进行清洗、将数据存储到MySQL数据库、通过SQL进行聚合查询，以及最终在Excel中使用数据透视表生成可视化报告。通过这个案例，你将理解每个工具在流程中的角色，掌握它们之间的数据衔接方法，并能够独立处理类似的数据分析任务。

1. 环境准备与工具链搭建

开始任何数据分析项目前，确保开发环境配置正确是避免后续一系列问题的关键。一个混乱的环境会导致依赖冲突、库版本不匹配、数据库连接失败等难以排查的错误。

1.1 Python环境与核心库安装

Python是此工作流的核心引擎，负责爬虫、数据清洗和初步分析。推荐使用Anaconda来管理Python环境，它可以有效隔离不同项目的依赖。

首先，从Anaconda官网下载并安装适合你操作系统的版本。安装完成后，打开终端（Windows系统为Anaconda Prompt或CMD，Mac/Linux为Terminal），创建一个专用于本项目的虚拟环境。

# 创建一个名为data_analysis_env的虚拟环境，并指定Python版本为3.9 conda create -n data_analysis_env python=3.9 # 激活该环境 conda activate data_analysis_env

环境激活后，你需要安装一系列核心库。请严格按照以下顺序和指定版本安装，以最大程度避免兼容性问题。

# 1. 首先升级pip工具本身 python -m pip install --upgrade pip # 2. 安装数据处理与分析库 pip install pandas==1.5.3 numpy==1.24.3 # 3. 安装网络请求与解析库 # requests用于发送HTTP请求，lxml和html5lib是HTML解析器，beautifulsoup4是解析工具 pip install requests==2.28.2 lxml==4.9.2 html5lib==1.1 beautifulsoup4==4.11.2 # 4. 安装数据库连接驱动 # pymysql用于连接MySQL数据库 pip install pymysql==1.0.3 # 5. 安装Jupyter Notebook（可选，用于交互式开发和调试） pip install jupyter==1.0.0

安装完成后，可以通过以下命令验证关键库是否安装成功：

python -c “import pandas; print(f’Pandas version: {pandas.__version__}’)” python -c “import requests; print(f’Requests version: {requests.__version__}’)” python -c “import pymysql; print(f’PyMySQL version: {pymysql.__version__}’)”

1.2 数据库环境配置

我们将使用MySQL作为数据存储和查询的中枢。如果你没有安装MySQL，可以选择以下两种方式之一：

方式一：本地安装MySQL Server访问MySQL官方网站下载社区版安装包。安装过程中，请务必记住你设置的root用户密码。同时，建议创建一个专门用于数据分析的数据库用户，并授予其相应权限，这比直接使用root用户更安全。

方式二：使用Docker快速部署（推荐用于学习和测试）如果你已经安装了Docker，可以通过一条命令快速启动一个MySQL容器。

docker run -d \ --name mysql_data_analysis \ -p 3306:3306 \ -e MYSQL_ROOT_PASSWORD=your_strong_password \ -e MYSQL_DATABASE=ecommerce_analysis \ mysql:8.0

这条命令会下载MySQL 8.0镜像，创建一个名为mysql_data_analysis的容器，将容器的3306端口映射到主机的3306端口，设置root密码，并同时创建一个名为ecommerce_analysis的数据库。

数据库启动后，你需要一个图形化管理工具来执行SQL语句、查看表结构。这里推荐使用DBeaver或MySQL Workbench。以DBeaver连接上述Docker中的MySQL为例，连接参数如下：

• 主机：localhost或127.0.0.1
• 端口：3306
• 数据库：ecommerce_analysis
• 用户名：root
• 密码：your_strong_password

1.3 项目目录结构规划

一个清晰的项目结构有助于管理代码、数据和配置文件。在开始编码前，建议创建如下目录：

ecommerce_data_analysis_project/ │ ├── config/ # 配置文件目录 │ └── database_config.py # 数据库连接配置（敏感信息不提交到Git） │ ├── src/ # 源代码目录 │ ├── spider/ # 爬虫模块 │ │ ├── __init__.py │ │ └── taobao_spider.py │ ├── data_processor/ # 数据处理模块 │ │ ├── __init__.py │ │ └── cleaner.py │ └── database/ # 数据库操作模块 │ ├── __init__.py │ └── db_handler.py │ ├── data/ # 数据目录 │ ├── raw/ # 原始数据（如爬取的JSON/CSV） │ ├── processed/ # 清洗后的数据 │ └── output/ # 最终输出（如Excel报告） │ ├── sql/ # SQL脚本目录 │ └── create_tables.sql │ ├── notebooks/ # Jupyter Notebook文件（用于探索性分析） │ └── exploratory_analysis.ipynb │ ├── requirements.txt # 项目依赖列表 └── README.md # 项目说明

在config/database_config.py中，使用字典存储数据库连接信息，注意不要将此文件提交到版本控制系统（应在.gitignore中忽略）。

# config/database_config.py DB_CONFIG = { ‘host’: ‘localhost’, ‘port’: 3306, ‘user’: ‘root’, # 生产环境应使用专用账户 ‘password’: ‘your_strong_password’, # 从环境变量读取更安全 ‘database’: ‘ecommerce_analysis’, ‘charset’: ‘utf8mb4’ # 支持存储Emoji等特殊字符 }

2. 构建稳健的电商数据爬虫

网络爬虫是数据获取的起点，但也是最容易出问题的环节。一个健壮的爬虫不仅要能获取数据，还要处理反爬机制、网络异常和数据解析失败等情况。

2.1 设计数据抓取策略与反爬应对

我们的目标是模拟抓取电商平台的商品列表页信息。在实际操作中，必须严格遵守网站的robots.txt协议，并仅将技术用于学习目的。对于公开的、允许爬取的数据，也应遵循以下伦理和技术准则：

1. 设置请求头（User-Agent）：模拟真实浏览器访问，这是最基本的反爬绕过措施。
2. 控制请求频率：在请求间添加随机延时，避免对目标服务器造成压力。通常建议间隔在2-5秒以上。
3. 处理异常：网络请求可能超时、返回错误状态码（如404、500），代码必须能捕获这些异常并做出相应处理（如重试、跳过或记录日志）。
4. 解析备用方案：主解析方式（如CSS选择器）失败时，应有备用方案（如正则表达式）或至少能记录错误，避免整个程序崩溃。

下面是一个具备基本健壮性的爬虫函数示例，它抓取一个模拟的商品列表页（使用一个公开的测试网站代替真实电商平台）。

# src/spider/taobao_spider.py import requests import pandas as pd from bs4 import BeautifulSoup import time import random import logging from typing import List, Dict, Optional # 配置日志，便于排查问题 logging.basicConfig(level=logging.INFO, format=‘%(asctime)s - %(levelname)s - %(message)s’) logger = logging.getLogger(__name__) class EcommerceSpider: def __init__(self, base_url: str): self.base_url = base_url # 定义请求头，模拟Chrome浏览器 self.headers = { ‘User-Agent’: ‘Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36’, ‘Accept-Language’: ‘zh-CN,zh;q=0.9’, } self.session = requests.Session() # 使用Session可以保持一些连接状态，提升效率 def _make_request(self, url: str, params: Optional[Dict] = None, max_retries: int = 3) -> Optional[requests.Response]: “”“发送HTTP请求，包含重试机制”“” for attempt in range(max_retries): try: # 添加随机延时，模拟人工操作 time.sleep(random.uniform(1, 3)) resp = self.session.get(url, headers=self.headers, params=params, timeout=10) resp.raise_for_status() # 如果状态码不是200，抛出HTTPError异常 # 检查返回内容是否有效，例如是否包含特定关键字或不是空页面 if len(resp.content) < 100: # 简单示例：检查内容是否过短 logger.warning(f”页面内容可能为空: {url}“) return None return resp except requests.exceptions.RequestException as e: logger.error(f”请求失败 (尝试 {attempt + 1}/{max_retries}): {url}, 错误: {e}“) if attempt == max_retries - 1: return None time.sleep(2 ** attempt) # 指数退避策略等待 return None def parse_product_list(self, page_num: int = 1) -> List[Dict]: “”“解析商品列表页，提取每个商品的基本信息”“” # 注意：此处URL替换为一个公开的、用于测试的电商数据模拟网站 # 实际项目中，此处应为目标电商平台的列表页URL构造逻辑 list_url = f”{self.base_url}/products?page={page_num}“ logger.info(f”开始抓取列表页: {list_url}“) resp = self._make_request(list_url) if not resp: return [] soup = BeautifulSoup(resp.text, ‘lxml’) products = [] # 假设商品项包裹在 class=‘product-item’ 的div中 # 实际CSS选择器需要根据目标网站的实际HTML结构进行调整 product_items = soup.select(‘div.product-item’) if not product_items: logger.warning(f”未在页面 {list_url} 中找到商品项，可能页面结构已变化或选择器错误。”) # 可以尝试备用选择器或记录HTML片段用于调试 # with open(f’debug_page_{page_num}.html’, ‘w’, encoding=‘utf-8’) as f: # f.write(resp.text[:2000]) for item in product_items: try: product_info = {} # 提取商品名称 name_elem = item.select_one(‘h3.product-title a’) product_info[‘product_name’] = name_elem.get_text(strip=True) if name_elem else ‘N/A’ # 提取价格 price_elem = item.select_one(‘span.price’) price_text = price_elem.get_text(strip=True) if price_elem else ” # 清理价格字符串，移除货币符号和逗号 try: product_info[‘price’] = float(price_text.replace(‘¥’, ”).replace(‘,’, ”)) except ValueError: product_info[‘price’] = 0.0 # 提取销量（或评价数） sales_elem = item.select_one(‘span.sales’) sales_text = sales_elem.get_text(strip=True) if sales_elem else ‘0’ # 处理“1.2万”这样的中文单位 if ‘万’ in sales_text: product_info[‘sales_volume’] = int(float(sales_text.replace(‘万’, ”)) * 10000) else: product_info[‘sales_volume’] = int(sales_text) if sales_text.isdigit() else 0 # 提取店铺名称 shop_elem = item.select_one(‘div.shop-name’) product_info[‘shop_name’] = shop_elem.get_text(strip=True) if shop_elem else ‘N/A’ # 提取商品链接 link_elem = item.select_one(‘h3.product-title a’) product_info[‘product_url’] = f”{self.base_url}{link_elem[‘href’]}“ if link_elem and link_elem.get(‘href’) else ” # 添加抓取时间戳 product_info[‘crawl_time’] = pd.Timestamp.now().strftime(‘%Y-%m-%d %H:%M:%S’) products.append(product_info) except Exception as e: logger.error(f”解析单个商品项时出错: {e}“) continue # 跳过当前出错项，继续解析下一个 logger.info(f”页面 {page_num} 解析完成，共获取 {len(products)} 个商品信息。”) return products def crawl_multiple_pages(self, start_page: int = 1, end_page: int = 5) -> pd.DataFrame: “”“抓取多页数据并合并为DataFrame”“” all_products = [] for page in range(start_page, end_page + 1): products = self.parse_product_list(page) if products: all_products.extend(products) else: logger.warning(f”第 {page} 页未获取到数据，可能已无更多页面。”) break # 如果某一页没数据，假设已到末页，停止抓取 df = pd.DataFrame(all_products) logger.info(f”总计抓取 {len(df)} 条商品记录。”) return df if __name__ == ‘__main__’: # 使用一个公开的测试网站URL，实际项目中替换为目标网站 spider = EcommerceSpider(base_url=‘https://httpbin.org’) # httpbin.org仅用于测试请求 # 实际运行时应注释掉下一行，并使用真实的基础URL和解析逻辑 # df_products = spider.crawl_multiple_pages(1, 3) # df_products.to_csv(‘../data/raw/products_raw.csv’, index=False, encoding=‘utf-8-sig’) print(“爬虫类定义完成，请根据目标网站结构调整解析逻辑后运行。”)

2.2 数据清洗与格式化

爬取到的原始数据通常包含缺失值、格式不一致、重复记录等问题，必须经过清洗才能用于分析。Pandas是完成这项工作的利器。

# src/data_processor/cleaner.py import pandas as pd import numpy as np import re class DataCleaner: @staticmethod def clean_product_data(raw_df: pd.DataFrame) -> pd.DataFrame: “”“清洗商品数据DataFrame”“” df = raw_df.copy() # 1. 处理重复数据：基于商品名称和店铺去重，保留最新抓取的一条 df[‘crawl_time’] = pd.to_datetime(df[‘crawl_time’]) df = df.sort_values(‘crawl_time’, ascending=False).drop_duplicates(subset=[‘product_name’, ‘shop_name’], keep=‘first’) # 2. 处理缺失值 # 价格缺失可能意味着商品已下架或信息不全，这里用中位数填充（需根据业务判断） if df[‘price’].notna().sum() > 0: # 确保有非空值才计算中位数 median_price = df[‘price’].median() else: median_price = 0 df[‘price’] = df[‘price’].fillna(median_price) # 销量缺失用0填充 df[‘sales_volume’] = df[‘sales_volume’].fillna(0) # 文本字段缺失用‘未知’填充 text_columns = [‘product_name’, ‘shop_name’, ‘product_url’] for col in text_columns: df[col] = df[col].fillna(‘未知’) # 3. 格式化与类型转换 df[‘price’] = df[‘price’].astype(float).round(2) # 价格保留两位小数 df[‘sales_volume’] = df[‘sales_volume’].astype(int) # 4. 数据修正：例如，清理商品名称中的多余空格和换行符 df[‘product_name’] = df[‘product_name’].apply(lambda x: re.sub(r’\s+’, ‘ ‘, str(x)).strip()) df[‘shop_name’] = df[‘shop_name’].apply(lambda x: re.sub(r’\s+’, ‘ ‘, str(x)).strip()) # 5. 衍生字段计算：例如，根据价格划分档次 def price_category(price): if price < 50: return ‘低价’ elif price < 200: return ‘中价’ else: return ‘高价’ df[‘price_category’] = df[‘price’].apply(price_category) # 6. 重置索引 df = df.reset_index(drop=True) return df @staticmethod def validate_data(cleaned_df: pd.DataFrame) -> bool: “”“简单的数据质量校验”“” # 检查关键字段是否存在空值（经过填充后不应有） critical_cols = [‘product_name’, ‘price’, ‘sales_volume’] if cleaned_df[critical_cols].isnull().any().any(): print(“警告：关键字段仍存在空值！”) return False # 检查价格和销量是否为非负数 if (cleaned_df[‘price’] < 0).any() or (cleaned_df[‘sales_volume’] < 0).any(): print(“警告：价格或销量存在负数！”) return False # 检查数据量 if len(cleaned_df) == 0: print(“警告：清洗后的数据为空！”) return False print(f”数据校验通过。总计 {len(cleaned_df)} 条记录，{cleaned_df[‘shop_name’].nunique()} 个店铺。”) return True

3. 构建数据分析数据库与SQL查询

清洗后的数据需要持久化存储，以便进行复杂的聚合查询和历史追踪。我们将数据存入MySQL，并设计合理的表结构。

3.1 数据库表结构设计

根据商品数据，我们设计一张products表。一个好的表设计应考虑未来可能的分析维度。

-- sql/create_tables.sql -- 创建数据库（如果尚未创建） CREATE DATABASE IF NOT EXISTS ecommerce_analysis CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE ecommerce_analysis; -- 商品信息表 CREATE TABLE IF NOT EXISTS products ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT ‘主键ID’, product_name VARCHAR(500) NOT NULL COMMENT ‘商品名称’, price DECIMAL(10, 2) NOT NULL COMMENT ‘商品价格’, sales_volume INT NOT NULL DEFAULT 0 COMMENT ‘销量’, shop_name VARCHAR(255) NOT NULL COMMENT ‘店铺名称’, product_url VARCHAR(1000) COMMENT ‘商品链接’, price_category VARCHAR(20) COMMENT ‘价格档次（低价/中价/高价）’, crawl_time DATETIME NOT NULL COMMENT ‘数据抓取时间’, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘记录创建时间’, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT ‘记录更新时间’, INDEX idx_shop_name (shop_name), INDEX idx_price_category (price_category), INDEX idx_crawl_time (crawl_time), INDEX idx_price (price) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT=‘电商商品信息表’;

设计要点解释：

• 字符集：使用utf8mb4以支持存储Emoji等所有Unicode字符。
• 数值类型：价格使用DECIMAL(10,2)确保精度，避免浮点数计算误差。销量使用INT。
• 索引：为常用的查询字段（如shop_name,price_category,crawl_time,price）创建索引，可以大幅提升查询速度。
• 时间戳：crawl_time是业务时间，created_at和updated_at是系统维护时间，用途不同。

3.2 使用Python将数据写入数据库

编写一个数据库处理器，负责连接数据库、执行SQL语句（包括插入、查询等）。

# src/database/db_handler.py import pymysql from pymysql import Error import pandas as pd from typing import List, Dict, Any, Optional import logging # 导入配置文件 import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) from config.database_config import DB_CONFIG logger = logging.getLogger(__name__) class DatabaseHandler: def __init__(self, config: Dict[str, Any] = None): self.config = config or DB_CONFIG self.connection = None def __enter__(self): “”“支持with语句，自动管理连接”“” self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def connect(self): “”“建立数据库连接”“” try: self.connection = pymysql.connect(**self.config) logger.info(“成功连接到数据库。”) except Error as e: logger.error(f”数据库连接失败: {e}“) raise def close(self): “”“关闭数据库连接”“” if self.connection and self.connection.open: self.connection.close() logger.info(“数据库连接已关闭。”) def execute_query(self, sql: str, params: Optional[tuple] = None, fetch: bool = True) -> Optional[List[Dict]]: “”“执行查询语句，返回结果列表”“” result = None try: with self.connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(sql, params) if fetch: result = cursor.fetchall() self.connection.commit() except Error as e: logger.error(f”执行查询失败: {e}, SQL: {sql}“) self.connection.rollback() raise return result def execute_many(self, sql: str, data: List[tuple]): “”“批量执行插入/更新语句”“” try: with self.connection.cursor() as cursor: cursor.executemany(sql, data) self.connection.commit() logger.info(f”批量操作成功，影响行数: {cursor.rowcount}“) except Error as e: logger.error(f”批量操作失败: {e}“) self.connection.rollback() raise def insert_dataframe(self, table_name: str, df: pd.DataFrame, batch_size: int = 100): “”“将Pandas DataFrame批量插入到指定表”“” if df.empty: logger.warning(“DataFrame为空，跳过插入。”) return # 确保DataFrame列名与表字段名匹配（这里假设一致） columns = df.columns.tolist() placeholders = ‘, ‘.join([‘%s’] * len(columns)) columns_str = ‘, ‘.join(columns) sql = f”INSERT INTO {table_name} ({columns_str}) VALUES ({placeholders})“ # 将DataFrame转换为元组列表 data = [tuple(row) for row in df.itertuples(index=False, name=None)] # 分批插入 for i in range(0, len(data), batch_size): batch = data[i:i + batch_size] self.execute_many(sql, batch) logger.info(f”已插入第 {i//batch_size + 1} 批数据，共 {len(batch)} 条。”) def table_exists(self, table_name: str) -> bool: “”“检查表是否存在”“” sql = ““” SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = %s ”“” result = self.execute_query(sql, (table_name,), fetch=True) return result[0][‘COUNT(*)’] > 0 if result else False

3.3 执行核心商业分析SQL查询

数据入库后，便可以通过SQL进行多维度的商业分析。以下是一些典型的分析场景和对应的SQL语句。

-- 1. 基础概览：商品总数、店铺数、平均价格、总销售额（估算） SELECT COUNT(*) AS total_products, COUNT(DISTINCT shop_name) AS total_shops, ROUND(AVG(price), 2) AS avg_price, SUM(price * sales_volume) AS estimated_total_sales FROM products; -- 2. 销量Top 10的商品 SELECT product_name, shop_name, price, sales_volume, (price * sales_volume) AS sales_revenue FROM products ORDER BY sales_volume DESC LIMIT 10; -- 3. 各价格档次的商品分布与平均销量 SELECT price_category, COUNT(*) AS product_count, ROUND(AVG(sales_volume), 0) AS avg_sales_volume, ROUND(SUM(price * sales_volume), 2) AS category_revenue FROM products GROUP BY price_category ORDER BY category_revenue DESC; -- 4. 店铺竞争力分析：按店铺统计商品数、总销量、平均价格 SELECT shop_name, COUNT(*) AS product_count, SUM(sales_volume) AS total_sales, ROUND(AVG(price), 2) AS avg_price, ROUND(SUM(price * sales_volume), 2) AS shop_revenue FROM products GROUP BY shop_name HAVING product_count >= 5 -- 只分析商品数大于5的店铺 ORDER BY shop_revenue DESC LIMIT 15; -- 5. 每日抓取数据趋势（假设crawl_time包含日期信息） SELECT DATE(crawl_time) AS crawl_date, COUNT(*) AS new_products, ROUND(AVG(price), 2) AS daily_avg_price FROM products GROUP BY DATE(crawl_time) ORDER BY crawl_date DESC;

在Python中，你可以通过DatabaseHandler执行这些查询，并将结果直接转换为Pandas DataFrame进行进一步处理或可视化。

# 示例：在Python中执行SQL并获取DataFrame with DatabaseHandler() as db: sql = “”“ SELECT shop_name, SUM(sales_volume) as total_sales FROM products GROUP BY shop_name ORDER BY total_sales DESC LIMIT 10 ”“” result = db.execute_query(sql, fetch=True) df_top_shops = pd.DataFrame(result) print(df_top_shops)

4. 使用数据透视表进行多维分析与可视化

SQL提供了强大的数据聚合能力，而Excel的数据透视表则是将聚合结果进行交互式探索和可视化的绝佳工具。我们可以将SQL查询结果导出为CSV或直接通过Python库（如openpyxl或pandas的ExcelWriter）写入Excel，并创建数据透视表。

4.1 将分析结果导出至Excel

首先，将我们关心的几个分析结果保存到同一个Excel文件的不同工作表（Sheet）中。

# src/analysis/report_generator.py import pandas as pd from database.db_handler import DatabaseHandler def generate_excel_report(output_path: str = ‘../data/output/analysis_report.xlsx’): “”“连接数据库，执行多个分析查询，并将结果写入Excel”“” analysis_results = {} with DatabaseHandler() as db: # 查询1：各价格档次分析 sql1 = “”“ SELECT price_category, COUNT(*) as product_count, AVG(price) as avg_price, SUM(sales_volume) as total_sales FROM products GROUP BY price_category ”“” df_category = pd.DataFrame(db.execute_query(sql1, fetch=True)) analysis_results[‘价格档次分析’] = df_category # 查询2：店铺销售额排名 sql2 = “”“ SELECT shop_name, COUNT(*) as product_count, SUM(price * sales_volume) as total_revenue FROM products GROUP BY shop_name ORDER BY total_revenue DESC LIMIT 20 ”“” df_shop_revenue = pd.DataFrame(db.execute_query(sql2, fetch=True)) analysis_results[‘店铺销售额排名’] = df_shop_revenue # 查询3：每日上新与均价趋势 sql3 = “”“ SELECT DATE(crawl_time) as date, COUNT(*) as new_products, AVG(price) as avg_price FROM products GROUP BY DATE(crawl_time) ORDER BY date ”“” df_daily_trend = pd.DataFrame(db.execute_query(sql3, fetch=True)) analysis_results[‘每日趋势’] = df_daily_trend # 使用Pandas的ExcelWriter写入多个Sheet with pd.ExcelWriter(output_path, engine=‘openpyxl’) as writer: for sheet_name, df in analysis_results.items(): df.to_excel(writer, sheet_name=sheet_name, index=False) # 自动调整列宽（近似） worksheet = writer.sheets[sheet_name] for column in worksheet.columns: max_length = 0 column_letter = column[0].column_letter for cell in column: try: if len(str(cell.value)) > max_length: max_length = len(str(cell.value)) except: pass adjusted_width = min(max_length + 2, 50) # 设置最大宽度 worksheet.column_dimensions[column_letter].width = adjusted_width print(f”分析报告已生成: {output_path}“) if __name__ == ‘__main__’: generate_excel_report()

4.2 在Excel中创建数据透视表

生成Excel文件后，手动或通过代码（使用openpyxl库可以创建数据透视表定义，但较复杂）创建数据透视表更为常见。以下是基于“店铺销售额排名”工作表创建数据透视表的手动步骤和思路：

1. 打开Excel文件，定位到“店铺销售额排名”工作表。
2. 选中数据区域（包括表头）。
3. 插入数据透视表：在菜单栏选择“插入” -> “数据透视表”。选择放置在新工作表。
4. 配置字段：
   • 行：将shop_name字段拖入。
   • 值：将total_revenue和product_count字段拖入。默认对total_revenue进行求和，对product_count进行求和。
5. 设置值字段格式：右键点击值字段（如“求和项:total_revenue”）->“值字段设置”，可以将汇总方式改为“求和”、“平均值”等，并设置数字格式（如货币格式）。
6. 添加筛选和切片器（可选）：可以添加price_category（如果数据中有）作为筛选器，或者插入切片器进行交互式筛选。
7. 创建数据透视图：选中数据透视表，在“分析”选项卡中点击“数据透视图”，可以快速生成柱状图、折线图等，直观展示店铺收入分布。

数据透视表的核心价值在于其交互性。你可以轻松地：

• 将行字段和列字段互换，从不同视角观察数据。
• 对销售额进行排序，快速找出头部和尾部店铺。
• 通过分组功能，将销售额按区间分组（如0-1000，1000-5000等）。
• 计算字段，例如添加一个“平均商品收入”（total_revenue/product_count）的新字段。

4.3 常见问题与排查

在从爬虫到数据库再到Excel的整个流程中，你可能会遇到以下典型问题：

5. 生产环境最佳实践与扩展方向

将上述流程应用于实际生产环境时，需要考虑更多关于稳定性、效率、安全和可维护性的问题。

5.1 爬虫工程化建议

• 任务调度：使用APScheduler或Celery等库定时执行爬虫任务，实现数据自动更新。
• 分布式与代理：对于大规模抓取，考虑使用Scrapy框架，并集成代理IP池（如scrapy-proxies）以分散请求和规避封禁。
• 数据去重与增量：在数据库层面，使用ON DUPLICATE KEY UPDATE语句实现增量更新。或在爬虫中记录已抓取URL的指纹（如MD5），避免重复抓取。
• 日志与监控：将日志记录到文件，并设置日志轮转。监控爬虫运行状态、成功率、数据量等指标。
• 遵守robots.txt：始终使用robotparser模块检查目标网站是否允许爬取目标路径，并控制爬取速率。

5.2 数据库优化与维护

• 连接池：在生产Web应用或高频任务中，使用DBUtils或SQLAlchemy的连接池管理数据库连接，避免频繁创建连接的开销。
• 定期备份：设置mysqldump定时任务，对数据库进行定期备份。
• 查询优化：对慢查询日志进行分析，为频繁查询的WHERE和JOIN条件字段建立合适的索引，但避免过度索引。
• 数据归档：对于历史抓取数据，可以定期迁移到归档表或数据仓库（如ClickHouse），保证主表的查询性能。

5.3 分析流程自动化

• Airflow或Prefect：使用工作流编排工具将爬虫、清洗、入库、分析、报告生成等任务串联成一个有向无环图（DAG），实现端到端的自动化流水线。
• Jupyter Notebook自动化：使用papermill或nbconvert工具，参数化运行Notebook，并输出分析报告。
• 替代Excel：对于需要更高自动化程度和更复杂图表的场景，可以考虑使用Plotly Dash、Streamlit或Grafana构建交互式Web仪表板。

5.4 安全与合规

• 配置信息管理：数据库密码等敏感信息绝不应硬编码在代码中。应使用环境变量或专门的密钥管理服务（如AWS Secrets Manager）。
• 数据脱敏：如果分析涉及用户隐私数据，在存储和展示前必须进行脱敏处理。
• 法律合规：确保数据抓取和使用行为符合《网络安全法》、《数据安全法》等相关法律法规，以及目标网站的服务条款。

通过将数据透视表、数据库、Python和爬虫技术串联起来，你构建的不仅仅是一个个孤立的技术点，而是一个能够持续运转、产出商业洞察的数据流水线。这个流程的核心思想——采集、清洗、存储、分析、可视化——是绝大多数数据分析项目的通用范式。掌握它，你就具备了解决真实世界商业数据问题的基本框架。接下来，你可以尝试用这个框架去分析不同的数据源，例如社交媒体舆情、行业报告、公开财报等，不断丰富你的分析维度与模型。