Python爬虫实战：5分钟搞定肯德基全国门店地址抓取（附完整代码）

你是否曾经好奇，一家像肯德基这样的全球连锁巨头，在中国究竟有多少家门店？这些门店又分布在哪些城市和商圈？对于数据分析爱好者、市场研究者，甚至是想要寻找最佳开店位置的创业者来说，这些数据都蕴含着巨大的价值。今天，我们就来聊聊如何用Python，这个被誉为“胶水语言”的神奇工具，在短短几分钟内，自动化地抓取肯德基全国的餐厅地址信息。这不仅仅是一次技术演练，更是一次将网络数据转化为结构化信息的实战之旅，整个过程清晰、直接，即便是Python新手也能轻松跟上。

我们将绕过复杂的网页渲染，直击数据源头——那个默默为前端页面提供数据的API接口。你会发现，获取这些公开的商业位置信息，远比想象中简单。本文面向所有对Python爬虫和数据分析感兴趣的初学者及爱好者，我们将从零开始，手把手带你完成从分析请求、构造参数到存储数据的完整流程，并提供可直接复制、运行并得到结果的代码。准备好了吗？让我们开始这次高效的数据采集之旅。

1. 逆向工程：找到数据的“后门”

在开始写代码之前，最关键的一步是弄清楚数据从哪里来。现代网站为了提升用户体验，大量使用Ajax技术动态加载内容。这意味着，你在页面上看到的餐厅列表，并非直接写在初始的HTML里，而是由浏览器在后台悄悄请求回来的。

1.1 识别Ajax请求

打开你常用的浏览器（Chrome或Edge均可），访问肯德基餐厅查询页面。在页面上随便输入一个城市名，比如“上海”，然后点击查询。此时，页面内容刷新了，但浏览器的地址栏URL很可能没有变化。这就是典型的Ajax局部刷新。

接下来，按下 F12 键打开开发者工具，切换到 Network（网络） 面板。为了更精确地找到我们需要的请求，在筛选类型中选择 XHR 或 Fetch。重新点击一次查询按钮，你会看到网络面板中出现了新的请求记录。

提示：如果网络面板内容太多，可以先点击面板左上角的“清除”按钮（一个带斜杠的圆圈），然后再进行操作，这样能更清晰地看到新产生的请求。

在这些请求中，我们需要找到一个名字看起来像 GetStoreList.ashx 或包含类似关键词的请求。这个请求的 Method（方法） 通常是 POST。点击这个请求，查看其详细信息。

1.2 分析请求参数与响应

找到目标请求后，我们需要关注三个核心部分：

1. Request URL（请求地址）：这是数据接口的真实地址，是我们代码中要访问的URL。
2. Request Headers（请求头）：特别是 User-Agent，它用于告诉服务器我们是一个“浏览器”在访问，而不是一个脚本。有时可能还需要 Referer 等信息来通过简单的反爬检查。
3. Form Data / Payload（请求参数）：这是发送给服务器的查询条件，通常以 x-www-form-urlencoded 或 json 格式存在。

以肯德基的查询接口为例，我们可能会看到类似以下的参数结构：

参数名	示例值	说明
cname	(空)	可能为城市名，但实践中常留空
pid	(空)	省份ID，常留空
keyword	上海	核心参数，查询的城市或区域关键词
pageIndex	1	当前页码，用于分页查询
pageSize	10	每页显示的餐厅数量

在 Response（响应） 标签页，我们可以看到服务器返回的数据。它通常是 JSON 格式，结构清晰，包含了餐厅列表、总数等信息。确认这个响应里确实有我们需要的餐厅名称、地址、联系电话等字段。

至此，逆向工程完成。我们已经掌握了“敲门”所需的所有信息：地址、敲门的方式（POST）、以及要说的话（参数）。

2. 构建你的第一个爬虫脚本

有了前面的分析，编写爬虫代码就变成了一个“填空”练习。我们将使用Python中最受欢迎的HTTP库——requests。如果你还没有安装，只需在命令行中执行一条命令：

pip install requests

2.1 基础请求框架

让我们从最核心的请求部分开始。创建一个新的Python文件，比如 kfc_crawler.py。

import requests
import json

def fetch_kfc_stores(city, page=1):
    """
    抓取指定城市和页码的肯德基门店信息
    :param city: 城市名，如‘北京’
    :param page: 页码，从1开始
    :return: 解析后的JSON数据，如果请求失败返回None
    """
    # 1. 目标URL (直接从开发者工具中复制)
    url = 'http://www.kfc.com.cn/kfccda/ashx/GetStoreList.ashx?op=keyword'

    # 2. 构造请求参数 (与我们分析的表单数据一致)
    data = {
        'cname': '',
        'pid': '',
        'keyword': city,
        'pageIndex': str(page),  # 注意：接口可能需要字符串类型的页码
        'pageSize': '10',
    }

    # 3. 设置请求头，进行简单的UA伪装
    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',
        'Referer': 'http://www.kfc.com.cn/kfccda/storelist/index.aspx', # 有时需要添加来源页
    }

    try:
        # 4. 发送POST请求
        response = requests.post(url=url, data=data, headers=headers, timeout=10)
        # 5. 检查请求是否成功
        response.raise_for_status()
        # 6. 解析JSON响应
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败，城市：{city}, 页码：{page}。错误信息：{e}")
        return None
    except json.JSONDecodeError:
        print(f"响应内容不是有效的JSON格式，城市：{city}, 页码：{page}。")
        return None

# 简单测试一下
if __name__ == '__main__':
    result = fetch_kfc_stores('北京', 1)
    if result:
        print(f"请求成功！共找到 {result.get('Table', [{}])[0].get('rowcount', 0)} 条记录。")
        # 打印第一条餐厅信息看看结构
        stores = result.get('Table1', [])
        if stores:
            first_store = stores[0]
            print(f"示例 - 店名：{first_store.get('storeName')}, 地址：{first_store.get('addressDetail')}")

这段代码定义了一个函数，它完成了单次请求的核心逻辑。我们添加了异常处理，让程序更健壮；设置了超时，避免无限等待；并解析了JSON响应。运行它，你应该能看到控制台打印出北京的餐厅数量和第一条餐厅信息。

2.2 处理分页与多城市

一家之言，一个城市一页的数据显然不够。我们需要处理分页，并可能遍历多个城市。

首先，从第一页的响应数据中，我们通常能获取到总记录数 (rowcount) 和每页大小 (pageSize)。通过这两个值，就能计算出总页数。

def crawl_all_stores_in_city(city):
    """
    抓取指定城市的所有门店
    :param city: 城市名
    :return: 该城市所有门店的列表
    """
    all_stores = []
    page = 1
    total_pages = None

    print(f"开始抓取城市：【{city}】的门店信息...")

    while True:
        # 获取当前页数据
        result = fetch_kfc_stores(city, page)
        if not result:
            break  # 如果请求失败，终止循环

        # 首次请求时，获取总页数
        if total_pages is None:
            table0 = result.get('Table', [])
            if table0:
                total_count = table0[0].get('rowcount', 0)
                page_size = 10  # 从参数可知，或从响应中解析
                total_pages = (total_count + page_size - 1) // page_size  # 向上取整
                print(f"城市【{city}】共有 {total_count} 家门店，预计 {total_pages} 页。")

        # 提取当前页的门店列表
        current_stores = result.get('Table1', [])
        if not current_stores:
            break  # 如果当前页没有数据，终止循环

        all_stores.extend(current_stores)
        print(f"  已抓取第 {page}/{total_pages if total_pages else '?'} 页，累计 {len(all_stores)} 条。")

        # 判断是否已抓取完所有页
        if total_pages and page >= total_pages:
            break
        page += 1

    print(f"城市【{city}】抓取完成，共获得 {len(all_stores)} 条门店信息。")
    return all_stores

对于多城市抓取，我们可以准备一个城市列表。但这里有一个非常重要的注意事项：必须尊重网站的负载，在请求间添加延时，避免请求过快给对方服务器造成压力，这既是道德要求，也能防止你的IP被暂时封锁。

import time

def crawl_multiple_cities(city_list, delay=2):
    """
    抓取多个城市的数据
    :param city_list: 城市名称列表
    :param delay: 每次请求后的延迟秒数，建议至少2秒
    :return: 所有城市门店的字典 {城市名: 门店列表}
    """
    all_data = {}
    for city in city_list:
        stores = crawl_all_stores_in_city(city)
        all_data[city] = stores
        if delay > 0:
            print(f"等待 {delay} 秒，避免请求过快...")
            time.sleep(delay)  # 关键：请求间隔
    return all_data

# 使用示例
if __name__ == '__main__':
    cities_to_crawl = ['北京', '上海', '广州', '深圳'] # 可以扩展更多城市
    national_data = crawl_multiple_cities(cities_to_crawl, delay=3)
    print(f"全国部分城市抓取完成，总计抓取 {sum(len(stores) for stores in national_data.values())} 条数据。")

注意：time.sleep(delay) 是爬虫礼仪的核心。将 delay 设置为2秒或以上，能显著降低被封风险，让爬虫行为更“像人”。

3. 数据存储：从内存到硬盘

数据抓取到内存中只是第一步，我们需要将其持久化保存，以便后续分析。根据数据量和使用场景，有几种常见的选择。

3.1 保存为JSON文件

JSON格式通用、易读，非常适合存储这种结构化的列表数据。Python的 json 模块让这一切变得非常简单。

import json
from datetime import datetime

def save_to_json(data, filename=None):
    """
    将数据保存为JSON文件
    :param data: 要保存的Python数据结构（如字典、列表）
    :param filename: 文件名，如果为None则自动生成
    """
    if filename is None:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f'kfc_stores_{timestamp}.json'

    try:
        with open(filename, 'w', encoding='utf-8') as f:
            # ensure_ascii=False 确保中文正常显示，indent=2 让文件有缩进，更美观
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"数据已成功保存至文件：{filename}")
    except IOError as e:
        print(f"保存文件时出错：{e}")

# 集成到主流程中
if __name__ == '__main__':
    city_data = crawl_all_stores_in_city('杭州')
    if city_data:
        save_to_json(city_data, 'kfc_hangzhou.json')

    # 或者保存多城市数据
    # national_data = crawl_multiple_cities(['南京', '苏州'])
    # save_to_json(national_data)

3.2 导入到CSV/Excel

对于数据分析，CSV（逗号分隔值）文件是更常见的选择，它可以被Excel、Pandas、数据库等轻松导入。我们可以将门店列表扁平化，存储到一个二维表格中。

import csv

def save_to_csv(store_list, filename=None):
    """
    将门店列表保存为CSV文件
    :param store_list: 门店字典的列表
    :param filename: 文件名
    """
    if not store_list:
        print("门店列表为空，无法保存CSV。")
        return

    if filename is None:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f'kfc_stores_{timestamp}.csv'

    # 确定CSV文件的表头（列名）。根据实际API返回的字段调整。
    # 这里是一个示例字段集，你需要根据 response.json() 的实际结构来修改。
    fieldnames = [
        'storeName',        # 门店名称
        'addressDetail',    # 详细地址
        'cityName',         # 城市
        'provinceName',     # 省份
        'postalCode',       # 邮编
        'longitude',        # 经度
        'latitude',         # 纬度
        'pro',              # 可能为省份缩写
    ]

    try:
        with open(filename, 'w', newline='', encoding='utf-8-sig') as csvfile: # utf-8-sig 让Excel正确识别中文
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            for store in store_list:
                # 确保每一行字典都包含所有字段，缺失的用空字符串填充
                row = {field: store.get(field, '') for field in fieldnames}
                writer.writerow(row)
        print(f"数据已成功保存至CSV文件：{filename}")
    except IOError as e:
        print(f"保存CSV文件时出错：{e}")

# 使用示例：先抓取，再保存为CSV
if __name__ == '__main__':
    stores = crawl_all_stores_in_city('武汉')
    save_to_csv(stores)

保存为CSV后，你可以用Excel打开，进行排序、筛选，或者用Pandas进行更复杂的分析，比如计算各城市的门店密度。

3.3 进阶选择：存入数据库

如果数据量非常大，或者你需要频繁地更新和查询，那么使用数据库（如SQLite、MySQL、MongoDB）是更专业的选择。这里以轻量级的SQLite为例：

import sqlite3

def save_to_sqlite(store_list, db_path='kfc_stores.db'):
    """
    将门店数据存入SQLite数据库
    :param store_list: 门店字典列表
    :param db_path: 数据库文件路径
    """
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # 创建表（如果不存在）
    create_table_sql = '''
    CREATE TABLE IF NOT EXISTS stores (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        store_name TEXT,
        address_detail TEXT,
        city_name TEXT,
        province_name TEXT,
        longitude REAL,
        latitude REAL,
        crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    '''
    cursor.execute(create_table_sql)

    # 插入数据
    insert_sql = '''
    INSERT INTO stores (store_name, address_detail, city_name, province_name, longitude, latitude)
    VALUES (?, ?, ?, ?, ?, ?)
    '''
    for store in store_list:
        cursor.execute(insert_sql, (
            store.get('storeName'),
            store.get('addressDetail'),
            store.get('cityName'),
            store.get('provinceName'),
            store.get('longitude'),
            store.get('latitude')
        ))

    conn.commit()
    print(f"成功插入 {len(store_list)} 条记录到数据库 {db_path}")
    conn.close()

# 使用示例
if __name__ == '__main__':
    stores = crawl_all_stores_in_city('成都')
    save_to_sqlite(stores)

4. 优化、伦理与常见问题

一个能跑起来的爬虫只是开始，一个健壮、可持续、负责任的爬虫才是我们的目标。

4.1 提升爬虫的健壮性

• 更完善的错误处理：我们之前的代码已经有了基础异常处理，但可以更细致。例如，针对HTTP状态码429（请求过多）、503（服务不可用）进行重试或长时间等待。
• 设置请求超时与重试：网络不稳定是常事。使用 requests 的 timeout 参数，并结合 tenacity 或 retrying 库实现自动重试机制。
• 使用Session对象：requests.Session() 可以复用TCP连接，并在多次请求间保持一些参数（如cookies），效率更高。
• 解析数据前先验证：在 response.json() 之前，可以先打印或检查 response.text 的前几百个字符，确保返回的是预期的JSON，而不是一个错误页面（如验证码挑战）。

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

session = requests.Session()
session.headers.update({
    'User-Agent': '你的UA',
    'Accept-Language': 'zh-CN,zh;q=0.9',
})

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def robust_fetch(url, data):
    try:
        resp = session.post(url, data=data, timeout=15)
        resp.raise_for_status()
        # 简单检查响应内容是否包含预期关键词
        if 'storeName' not in resp.text and 'rowcount' not in resp.text:
            raise ValueError("响应内容可能不是有效的门店数据")
        return resp.json()
    except (requests.exceptions.RequestException, ValueError) as e:
        print(f"请求异常：{e}，进行重试...")
        raise  # 触发重试装饰器

4.2 遵守Robots协议与法律伦理

这是爬虫开发者的必修课。

• 查看Robots.txt：访问 http://www.kfc.com.cn/robots.txt，查看网站是否允许爬虫抓取相关路径。虽然本例中的API接口可能未被明确禁止，但遵守此协议是基本规范。
• 控制请求频率：如前所述，使用 time.sleep() 是必须的。更友好的做法是随机化延迟时间，例如 time.sleep(random.uniform(1, 3))。
• 识别并尊重反爬机制：如果网站返回了验证码或要求登录，通常意味着你的请求被识别为爬虫。此时应停止爬取，考虑是否真的需要这些数据，或者寻找官方数据渠道。
• 数据用途：将抓取的数据用于个人学习、研究或非商业性的数据分析是通常可接受的。但严禁用于商业竞争、垃圾营销、侵犯隐私或任何违法活动。

4.3 你可能遇到的问题与解决思路

1. 返回空数据或错误数据：

   • 检查参数：确认 keyword 参数格式是否正确。有些接口可能需要城市拼音或特定编码。
   • 检查请求头：尝试添加更多的请求头，如 Accept, Accept-Encoding, Content-Type（对于POST请求，可能是 application/x-www-form-urlencoded），以及之前提到的 Referer。
   • 模拟完整流程：有些网站需要先访问一个初始页面获取Cookie。你可以先用 session.get() 访问一下查询主页，再用同一个 session 去请求API。

2. 被封IP：

   • 增加延迟：这是首要措施。
   • 使用代理IP池：对于大规模抓取，这是必备方案。可以从一些服务商购买高质量的HTTP代理，并在请求中轮换使用。代码示例：

     proxies = {
         'http': 'http://your-proxy-ip:port',
         'https': 'http://your-proxy-ip:port',
     }
     response = requests.post(url, data=data, headers=headers, proxies=proxies)
     

3. 数据结构变化：

   • 网站改版是常态。定期运行你的脚本，如果发现解析失败，重新进行第一步的“逆向工程”分析，更新URL、参数或解析逻辑。

抓取肯德基门店地址这个项目，麻雀虽小，五脏俱全。它涵盖了现代爬虫从分析、请求、解析到存储的核心流程，也触及了伦理和优化的边界。我自己的经验是，把每次爬虫任务都当成一次与服务器的“对话”，礼貌（遵守规则）、清晰（构造正确的请求）、有耐心（控制速度），通常都能顺利拿到想要的数据。最关键的是动手尝试，把上面的代码复制下来，从你所在的城市开始运行，看看能发现什么。数据处理的过程中，你可能会对商业布局有更直观的感受，这才是技术学习之外更有趣的收获。