手把手教你构建统计局地区经济数据爬虫：从环境搭建到数据持久化全指南

一、为什么要爬取统计局数据

在数据分析、宏观经济研究、区域经济对比等场景中，国家统计局（NBS）发布的地区经济数据是最权威、最系统的公开数据源之一。其官网提供分省、市、县的GDP、人口、固定资产投资、居民收入等数百项指标，时间跨度从建国初期至今。然而，官方页面通常以动态表格和分页查询形式呈现，手动复制效率极低，且无法实现自动化监控与更新。因此，编写一个稳定、高效、合规的Python爬虫，成为数据从业者的必备技能。

本文将带您从零开始，使用2025-2026年最新Python生态工具，构建一个完整的国家统计局地区经济数据爬取系统。全文不仅提供可运行的代码，更会深入讲解反爬策略、动态渲染处理、数据清洗、异常重试、代理轮换、增量存储等生产级细节。

目录

一、为什么要爬取统计局数据

二、技术选型：为什么选择这些库

三、目标网站分析与接口破解

3.1 国家统计局数据查询入口

3.2 指标代码与地区代码的获取

四、环境搭建与项目结构

4.1 创建虚拟环境

4.2 安装依赖

4.3 项目目录结构

五、核心代码逐段实现

5.1 配置文件 config/settings.py

5.2 日志与异常处理 utils/helpers.py

5.3 请求抓取器 core/fetcher.py

5.4 数据解析器 core/parser.py

5.5 数据管道 core/pipeline.py

5.6 ORM模型 models/tables.py

5.7 爬虫主程序 main.py

六、反爬策略深度剖析与应对

七、性能优化与异步改造

二、技术选型：为什么选择这些库

此外，我们还会使用Redis作为去重过滤器（可选），Scrapy框架思想但轻量级实现，避免过度工程化。

三、目标网站分析与接口破解

3.1 国家统计局数据查询入口

主要入口为：https://data.stats.gov.cn/easyquery.htm

该页面提供“宏观数据”、“地区数据”、“年度数据”等标签。我们需要重点关注的是“地区数据”下的分省月度/季度指标。实际数据并非直接嵌入HTML，而是通过Ajax请求获取JSON。

打开浏览器开发者工具（F12） -> Network -> XHR，选择“地区数据” -> “GDP” -> “北京市”，可以看到类似请求：

text

GET /easyquery.htm?m=QueryData&dbcode=fsnd&rowcode=reg&colcode=sj&wds=[]&dfwds=[{"wdcode":"zb","valuecode":"A0201"}]

关键参数解析：

• m=QueryData：固定操作类型

• dbcode：数据库代码，fsnd表示分省年度，fsmonth表示分省月度，hgnd表示宏观年度

• rowcode：行维度，reg表示地区

• colcode：列维度，sj表示时间

• wds：普通筛选条件，如地区编码

• dfwds：指标筛选，zb表示指标代码，valuecode为具体指标ID

3.2 指标代码与地区代码的获取

国家统计局使用一套内部编码体系。例如：

• A0201：地区生产总值（GDP）

• A0202：第一产业增加值

• A0203：第二产业增加值

• A0204：第三产业增加值

• A0301：人口数（年末）

• A0401：固定资产投资（不含农户）

地区代码采用12位统计用行政区划代码，如110000代表北京市，310000代表上海市，440000代表广东省。

我们可以通过另一个接口获取全量地区列表：

text

GET /easyquery.htm?m=GetTree&dbcode=fsnd&id=zb

但更稳健的方式是直接使用官方提供的“地区列表”下拉框，其中包含所有省、自治区、直辖市及部分计划单列市。

四、环境搭建与项目结构

4.1 创建虚拟环境

bash

python -m venv nbs_spider
source nbs_spider/bin/activate # Linux/Mac
# 或 nbs_spider\\Scripts\\activate # Windows

4.2 安装依赖

bash

pip install requests httpx selenium webdriver-manager pandas parsel retry fake-useragent loguru sqlalchemy pymysql redis-py lxml

4.3 项目目录结构

text

nbs_spider/
├── config/
│ ├── __init__.py
│ └── settings.py # 全局配置（超时、重试、数据库连接）
├── core/
│ ├── __init__.py
│ ├── fetcher.py # 请求发送与重试
│ ├── parser.py # JSON/HTML解析
│ ├── pipeline.py # 数据清洗与存储
│ └── middleware.py # 代理、UA、日志中间件
├── models/
│ ├── __init__.py
│ └── tables.py # SQLAlchemy ORM模型
├── utils/
│ ├── __init__.py
│ ├── helpers.py # 工具函数（日期转换、编码映射）
│ └── exceptions.py # 自定义异常类
├── logs/ # 日志目录
├── data/ # CSV/Excel备份
├── main.py # 程序入口
└── README.md

五、核心代码逐段实现

5.1 配置文件 config/settings.py

python

# config/settings.py
import os
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent

# 请求配置
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36"
TIMEOUT = 30
RETRY_TIMES = 3
RETRY_BACKOFF = 2 # 指数退避基数

# 代理配置（可选，使用免费代理池或付费代理）
PROXY_ENABLED = False
PROXY_LIST = [
"http://127.0.0.1:7890",
# 更多代理…
]

# 目标接口
BASE_URL = "https://data.stats.gov.cn/easyquery.htm"
REFERER = "https://data.stats.gov.cn/easyquery.htm?cn=E0103"

# 数据库配置（MySQL示例）
DB_URI = "mysql+pymysql://username:password@localhost:3306/nbs_data?charset=utf8mb4"

# 日志配置
LOG_LEVEL = "INFO"
LOG_FILE = BASE_DIR / "logs" / "spider.log"

# 下载延迟（秒）
DOWNLOAD_DELAY = 0.5

# 最大重试次数
MAX_RETRIES = 5

# 地区代码列表（可动态获取，此处预置部分）
REGION_CODES = [
"110000", "120000", "130000", "140000", "150000",
"210000", "220000", "230000", "310000", "320000",
"330000", "340000", "350000", "360000", "370000",
"410000", "420000", "430000", "440000", "450000",
"460000", "500000", "510000", "520000", "530000",
"540000", "610000", "620000", "630000", "640000",
"650000", "710000", "810000", "820000"
]

# 指标代码映射
INDICATOR_MAP = {
"gdp": "A0201",
"primary_industry": "A0202",
"secondary_industry": "A0203",
"tertiary_industry": "A0204",
"population": "A0301",
"fixed_asset_invest": "A0401",
"retail_sales": "A0501",
}

5.2 日志与异常处理 utils/helpers.py

python

# utils/helpers.py
import json
import time
from datetime import datetime
from functools import wraps
from loguru import logger
from config.settings import LOG_FILE

# 配置loguru
logger.add(LOG_FILE, rotation="500 MB", retention="10 days", level="INFO")
logger.add(lambda msg: print(msg, end=""), level="DEBUG")

def retry_on_exception(max_retries=3, delay=1, backoff=2, exceptions=(Exception,)):
"""重试装饰器，支持指数退避"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
_delay = delay
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_retries:
logger.error(f"Function {func.__name__} failed after {max_retries} retries: {e}")
raise
logger.warning(f"Retry {attempt+1}/{max_retries} for {func.__name__} due to {e}")
time.sleep(_delay)
_delay *= backoff
return None
return wrapper
return decorator

def safe_parse_json(text: str) -> dict:
"""安全解析JSON，处理BOM和多余字符"""
try:
# 去除BOM头
if text.startswith('\\ufeff'):
text = text[1:]
return json.loads(text)
except json.JSONDecodeError as e:
logger.error(f"JSON parse error: {e}, raw: {text[:200]}")
return {}

def timestamp_to_str(ts: int, fmt="%Y-%m-%d %H:%M:%S") -> str:
"""时间戳转字符串"""
return datetime.fromtimestamp(ts / 1000).strftime(fmt) if ts else None

5.3 请求抓取器 core/fetcher.py

我们采用requests.Session保持Cookie，并集成fake_useragent自动轮换UA。对于需要JavaScript渲染的页面，我们使用Selenium作为降级方案，但主数据接口为纯JSON，因此优先用requests。

python

# core/fetcher.py
import requests
import time
from fake_useragent import UserAgent
from retry import retry
from loguru import logger
from config.settings import TIMEOUT, RETRY_TIMES, PROXY_ENABLED, PROXY_LIST, BASE_URL, REFERER
from utils.helpers import retry_on_exception

class Fetcher:
def __init__(self, use_proxy=False):
self.session = requests.Session()
self.ua = UserAgent()
self.use_proxy = use_proxy and PROXY_ENABLED
self.proxy_list = PROXY_LIST if use_proxy else []
self.proxy_index = 0
self.session.headers.update({
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Connection": "keep-alive",
"Referer": REFERER,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"User-Agent": self.ua.random,
"X-Requested-With": "XMLHttpRequest",
})
# 适配国家统计局特定Cookie（从浏览器复制可能有效）
self.session.cookies.set("JSESSIONID", "随机或空", domain="data.stats.gov.cn")

@retry_on_exception(max_retries=RETRY_TIMES, delay=1, backoff=2)
def get_json(self, params: dict) -> dict:
"""发送GET请求并返回JSON"""
url = BASE_URL
# 轮换UA
self.session.headers.update({"User-Agent": self.ua.random})

proxies = None
if self.use_proxy and self.proxy_list:
proxy = self.proxy_list[self.proxy_index % len(self.proxy_list)]
proxies = {"http": proxy, "https": proxy}
self.proxy_index += 1
logger.debug(f"Using proxy: {proxy}")

try:
response = self.session.get(
url,
params=params,
timeout=TIMEOUT,
proxies=proxies,
allow_redirects=False
)
response.raise_for_status()
# 检查是否返回了反爬页面（通常包含"验证"字样）
if "验证" in response.text or "blocked" in response.text:
logger.warning("可能触发反爬机制，尝试更换UA或代理")
# 此处可抛出异常触发重试
raise Exception("Anti-crawler detected")
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"Request failed: {e}")
raise
except ValueError as e:
logger.error(f"JSON decode failed: {e}, response: {response.text[:200]}")
raise

def close(self):
self.session.close()

# 若需要使用Selenium获取页面，可增加以下方法（但本爬虫主要用JSON接口）
class SeleniumFetcher:
"""用于偶尔需要JS渲染的页面，比如获取地区树"""
def __init__(self, headless=True):
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service

chrome_options = Options()
if headless:
chrome_options.add_argument("–headless=new")
chrome_options.add_argument("–disable-gpu")
chrome_options.add_argument("–no-sandbox")
chrome_options.add_argument("–disable-dev-shm-usage")
# 规避检测
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)

service = Service(ChromeDriverManager().install())
self.driver = webdriver.Chrome(service=service, options=chrome_options)
self.driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
'source': 'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
})

def get_html(self, url):
self.driver.get(url)
time.sleep(2) # 等待JS加载
return self.driver.page_source

def close(self):
self.driver.quit()

5.4 数据解析器 core/parser.py

统计局返回的JSON结构通常为：

json

{
"retcode": 0,
"retmsg": "success",
"data": {
"cached": false,
"cachename": "",
"datanodes": [
{
"code": "zb.A0201.reg.110000.sj.2020",
"data": {
"strdata": "36102.6",
"unit": "亿元"
},
"wds": [
{"wdcode": "zb", "valuecode": "A0201"},
{"wdcode": "reg", "valuecode": "110000"},
{"wdcode": "sj", "valuecode": "2020"}
]
}
],
"cols": […],
"rows": […]
}
}

我们需要将datanodes转换为扁平化的表格。

python

# core/parser.py
from typing import List, Dict, Any
from loguru import logger
from utils.helpers import safe_parse_json

class Parser:
@staticmethod
def parse_region_data(json_data: dict, indicator_name: str) -> List[Dict[str, Any]]:
"""
解析地区经济数据，返回列表字典
每个字典包含：region_code, year, indicator, value, unit
"""
if json_data.get("retcode") != 0:
logger.error(f"API returned error: {json_data.get('retmsg')}")
return []

datanodes = json_data.get("data", {}).get("datanodes", [])
if not datanodes:
logger.warning("No datanodes found in response")
return []

results = []
for node in datanodes:
wds = node.get("wds", [])
# 从wds中提取地区、时间、指标
region_code = None
year = None
zb_code = None
for item in wds:
if item.get("wdcode") == "reg":
region_code = item.get("valuecode")
elif item.get("wdcode") == "sj":
year = item.get("valuecode") # 可能是"2020"或"2020-01"
elif item.get("wdcode") == "zb":
zb_code = item.get("valuecode")

if not region_code or not year:
# 尝试从code字段解析
code_parts = node.get("code", "").split(".")
if len(code_parts) >= 4:
# 格式: zb.A0201.reg.110000.sj.2020
for i, part in enumerate(code_parts):
if part == "reg" and i+1 < len(code_parts):
region_code = code_parts[i+1]
elif part == "sj" and i+1 < len(code_parts):
year = code_parts[i+1]

data_obj = node.get("data", {})
value_str = data_obj.get("strdata")
unit = data_obj.get("unit", "")

# 转换数值
try:
value = float(value_str) if value_str else None
except (ValueError, TypeError):
value = None
logger.warning(f"Value conversion failed: {value_str}")

results.append({
"region_code": region_code,
"year": year,
"indicator_code": zb_code,
"indicator_name": indicator_name,
"value": value,
"unit": unit,
"raw_data": node # 保留原始数据以便调试
})

return results

@staticmethod
def parse_region_list(html: str) -> List[Dict[str, str]]:
"""从HTML中提取地区列表（如果使用Selenium）"""
from parsel import Selector
sel = Selector(text=html)
options = sel.xpath("//select[@id='region']/option")
regions = []
for opt in options:
code = opt.xpath("./@value").get()
name = opt.xpath("./text()").get()
if code and name:
regions.append({"code": code, "name": name.strip()})
return regions

5.5 数据管道 core/pipeline.py

数据清洗、去重、格式标准化，并支持写入CSV和数据库。

python

# core/pipeline.py
import pandas as pd
from sqlalchemy import create_engine, and_
from sqlalchemy.orm import sessionmaker
from loguru import logger
from config.settings import DB_URI
from models.tables import RegionEconomicData, Base
from typing import List, Dict, Any

class DataPipeline:
def __init__(self, use_db=True, use_csv=True, csv_path="data/region_economic.csv"):
self.use_db = use_db
self.use_csv = use_csv
self.csv_path = csv_path
self.engine = None
self.Session = None
if use_db and DB_URI:
self.engine = create_engine(DB_URI, echo=False, pool_pre_ping=True)
# 创建表（如果不存在）
Base.metadata.create_all(self.engine)
self.Session = sessionmaker(bind=self.engine)
self.cache = set() # 用于去重（内存版，生产环境建议Redis）

def process_items(self, items: List[Dict[str, Any]]):
"""批量处理数据：清洗+去重+存储"""
if not items:
return

# 1. 清洗（去除无效值）
clean_items = self.clean(items)

# 2. 去重（基于region_code+year+indicator_code）
unique_items = self.deduplicate(clean_items)

if not unique_items:
logger.info("No new items after deduplication")
return

# 3. 存储
if self.use_db:
self.save_to_db(unique_items)
if self.use_csv:
self.save_to_csv(unique_items)

logger.info(f"Saved {len(unique_items)} new records")

def clean(self, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""数据清洗：标准化字段、处理缺失值"""
cleaned = []
for item in items:
# 确保年份为字符串
year = str(item.get("year", ""))
if len(year) == 4 and year.isdigit():
pass
elif "-" in year:
year = year.split("-")[0] # 取年份部分
else:
continue # 跳过无效年份

# 地区编码补零到6位
region_code = str(item.get("region_code", "")).zfill(6)
if not region_code or len(region_code) < 6:
continue

# 数值处理
value = item.get("value")
if value is None or (isinstance(value, float) and pd.isna(value)):
# 可保留None，但建议填充或跳过
pass

cleaned.append({
"region_code": region_code,
"year": year,
"indicator_code": item.get("indicator_code", ""),
"indicator_name": item.get("indicator_name", ""),
"value": value,
"unit": item.get("unit", ""),
})
return cleaned

def deduplicate(self, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""基于内存集合去重（生产环境建议布隆过滤器或Redis）"""
unique = []
for item in items:
key = (item["region_code"], item["year"], item["indicator_code"])
if key not in self.cache:
self.cache.add(key)
unique.append(item)
return unique

def save_to_db(self, items: List[Dict[str, Any]]):
"""使用SQLAlchemy批量插入，处理重复冲突"""
session = self.Session()
try:
for item in items:
# 检查是否存在
exists = session.query(RegionEconomicData).filter(
and_(
RegionEconomicData.region_code == item["region_code"],
RegionEconomicData.year == item["year"],
RegionEconomicData.indicator_code == item["indicator_code"]
)
).first()
if exists:
# 更新数值（若变化）
if exists.value != item["value"]:
exists.value = item["value"]
exists.unit = item["unit"]
logger.debug(f"Update record: {item['region_code']}-{item['year']}-{item['indicator_code']}")
else:
# 插入新记录
record = RegionEconomicData(**item)
session.add(record)
session.commit()
logger.info(f"DB commit success, {len(items)} records processed")
except Exception as e:
session.rollback()
logger.error(f"DB error: {e}")
finally:
session.close()

def save_to_csv(self, items: List[Dict[str, Any]]):
"""追加到CSV文件"""
df = pd.DataFrame(items)
# 如果文件已存在则追加，否则写入头
try:
existing = pd.read_csv(self.csv_path)
combined = pd.concat([existing, df], ignore_index=True)
combined.to_csv(self.csv_path, index=False, encoding="utf-8-sig")
except FileNotFoundError:
df.to_csv(self.csv_path, index=False, encoding="utf-8-sig")
logger.info(f"CSV saved to {self.csv_path}")

5.6 ORM模型 models/tables.py

python

# models/tables.py
from sqlalchemy import Column, Integer, String, Float, DateTime, UniqueConstraint, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func

Base = declarative_base()

class RegionEconomicData(Base):
__tablename__ = "region_economic_data"

id = Column(Integer, primary_key=True, autoincrement=True)
region_code = Column(String(12), nullable=False, comment="地区代码")
year = Column(String(10), nullable=False, comment="年份")
indicator_code = Column(String(20), nullable=False, comment="指标代码")
indicator_name = Column(String(100), comment="指标名称")
value = Column(Float, comment="数值")
unit = Column(String(20), comment="单位")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(DateTime, onupdate=func.now(), comment="更新时间")

__table_args__ = (
UniqueConstraint('region_code', 'year', 'indicator_code', name='uix_region_year_indicator'),
Index('idx_region_code', 'region_code'),
Index('idx_year', 'year'),
)

def __repr__(self):
return f"<RegionEconomicData(region={self.region_code}, year={self.year}, indicator={self.indicator_code}, value={self.value})>"

5.7 爬虫主程序 main.py

整合所有模块，构建工作流：获取地区列表 -> 遍历指标 -> 遍历年份 -> 请求数据 -> 解析 -> 存储。

python

# main.py
import time
import random
from loguru import logger
from config.settings import REGION_CODES, INDICATOR_MAP, DOWNLOAD_DELAY
from core.fetcher import Fetcher
from core.parser import Parser
from core.pipeline import DataPipeline
from utils.helpers import retry_on_exception

class NBSSpider:
def __init__(self):
self.fetcher = Fetcher(use_proxy=False)
self.parser = Parser()
self.pipeline = DataPipeline(use_db=True, use_csv=True)
self.year_range = range(2000, 2026) # 2000-2025年

def build_params(self, indicator_code: str, region_code: str, year: int) -> dict:
"""构建请求参数"""
return {
"m": "QueryData",
"dbcode": "fsnd", # 分省年度
"rowcode": "reg",
"colcode": "sj",
"wds": "[]",
"dfwds": f'[{{"wdcode":"zb","valuecode":"{indicator_code}"}},{{"wdcode":"reg","valuecode":"{region_code}"}},{{"wdcode":"sj","valuecode":"{year}"}}]'
}

def crawl_region_indicator(self, region_code: str, indicator_name: str, indicator_code: str):
"""爬取单个地区单个指标的所有年份"""
logger.info(f"Starting crawl: region={region_code}, indicator={indicator_name}({indicator_code})")

for year in self.year_range:
params = self.build_params(indicator_code, region_code, year)
try:
json_data = self.fetcher.get_json(params)
# 检查是否有数据
if json_data.get("data", {}).get("datanodes"):
items = self.parser.parse_region_data(json_data, indicator_name)
if items:
self.pipeline.process_items(items)
logger.info(f"Fetched {len(items)} records for {region_code} {indicator_name} {year}")
else:
logger.debug(f"No valid items for {region_code} {indicator_name} {year}")
else:
logger.debug(f"No datanodes for {region_code} {indicator_name} {year}")
except Exception as e:
logger.error(f"Error crawling {region_code} {indicator_name} {year}: {e}")

# 延迟，避免请求过快
time.sleep(DOWNLOAD_DELAY + random.uniform(0, 0.5))

def run(self):
"""主运行入口"""
logger.info("Spider started")
total_regions = len(REGION_CODES)
total_indicators = len(INDICATOR_MAP)
logger.info(f"Total regions: {total_regions}, indicators: {total_indicators}")

for region_code in REGION_CODES:
for indicator_name, indicator_code in INDICATOR_MAP.items():
self.crawl_region_indicator(region_code, indicator_name, indicator_code)
# 每个指标切换时稍作休息
time.sleep(0.5)

self.fetcher.close()
logger.info("Spider finished successfully")

if __name__ == "__main__":
spider = NBSSpider()
spider.run()

六、反爬策略深度剖析与应对

国家统计局虽然不像金融类网站有极强反爬，但仍有以下防护措施：

若遇到JS生成加密参数（如signature），则需要逆向分析JS代码，或改用Selenium模拟完整浏览器环境。但当前接口为开放API，暂不需如此复杂。

七、性能优化与异步改造

对于大量地区和指标，串行请求效率较低。我们可以利用asyncio + httpx实现异步并发，但需注意控制并发数避免被封。

以下是异步版Fetcher示例（选读）：

python

# core/async_fetcher.py
import httpx
import asyncio
from loguru import logger
from config.settings import TIMEOUT

class AsyncFetcher:
def __init__(self, max_concurrent=10):
self.client = httpx.AsyncClient(timeout=TIMEOUT, follow_redirects=False)
self.semaphore = asyncio.Semaphore(max_concurrent)

async def get_json(self, params: dict):
async with self.semaphore:
try:
response = await self.client.get(BASE_URL, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Async request error: {e}")
return None