导航：首页 > 开发技术 >

国家统计局数据采集过程有哪些

发表于：2026-11-01 作者：千家信息网编辑

千家信息网最后更新 2026年11月01日，这篇文章主要介绍"国家统计局数据采集过程有哪些"，在日常操作中，相信很多人在国家统计局数据采集过程有哪些问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"国家统计局数据采集

千家信息网最后更新 2026年11月01日国家统计局数据采集过程有哪些

这篇文章主要介绍"国家统计局数据采集过程有哪些"，在日常操作中，相信很多人在国家统计局数据采集过程有哪些问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答"国家统计局数据采集过程有哪些"的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

采集过程

采集各种公开数据的第一步就是分析网页。

上面的图是国家统计局年度数据的界面。左边是数据分类的树形菜单，右边是每个菜单点击之后显示的数据，可以设置年份来过滤数据。

采集数据分类树

根据页面的情况，首先，我们需要采集树形菜单中的数据，然后再根据菜单的分类来依次采集右边的数据。这样可以避免采集的遗漏。

爬虫采集数据一般有 2 种情况：

采集 html 页面，然后分析其中的结构，提取出数据
查看是否存在获取数据的 API，直接从 API 中提取数据

通过分析网页的加载过程，发现国际统计局的数据是有 API 的，这就节省了很多时间。

API 信息如下：

host: "https://data.stats.gov.cn/easyquery.htm"method: POSTparams:  id=zb&dbcode=hgnd&wdcode=zb&m=getTree

通过 python 的 requests 库模拟 POST 请求就可以获取到树形菜单中的数据了。

def init_tree(tree_data_path):    data = get_tree_data()    with open(tree_data_path, "wb") as f:        pickle.dump(data, f)def get_tree_data(id="zb"):    r = requests.post(f"{host}?id={id}&dbcode=hgnd&wdcode=zb&m=getTree", verify=False)    logging.debug("access url: %s", r.url)    data = r.json()    for node in data:        if node["isParent"]:            node["children"] = get_tree_data(node["id"])        else:            node["children"] = []    return data

直接调用上面的 init_tree 函数即可，树形菜单会以 json 格式序列化到 tree_data_path 中。

序列化的目的是为了后面采集数据时可以反复使用，不用每次都去采集这个树形菜单。（毕竟菜单是基本不变的）

根据分类采集数据

有了分类的菜单，下一步就是采集具体的数据。同样，通过分析网页，数据也是有 API 的，不用采集 html 页面再提取数据。

host: "https://data.stats.gov.cn/easyquery.htm"method: GETparams: 参数有变量，具体参见代码

采集数据稍微复杂一些，不像采集树形菜单那样访问一次 API 即可，而是遍历树形菜单，根据菜单的信息访问 API。

# -*- coding: utf-8 -*-import loggingimport osimport pickleimport timeimport pandas as pdimport requestshost = "https://data.stats.gov.cn/easyquery.htm"tree_data_path = "./tree.data"data_dir = "./data"def data(sj="1978-"):    tree_data = []    with open(tree_data_path, "rb") as f:        tree_data = pickle.load(f)    traverse_tree_data(tree_data, sj)def traverse_tree_data(nodes, sj):    for node in nodes:        # 叶子节点上获取数据        if node["isParent"]:            traverse_tree_data(node["children"], sj)        else:            write_csv(node["id"], sj)def write_csv(nodeId, sj):    fp = os.path.join(data_dir, nodeId + ".csv")    # 文件是否存在, 如果存在, 不爬取    if os.path.exists(fp):        logging.info("文件已存在: %s", fp)        return    statData = get_stat_data(sj, nodeId)    if statData is None:        logging.error("NOT FOUND data for %s", nodeId)        return    # csv 数据    csvData = {"zb": [], "value": [], "sj": [], "zbCN": [], "sjCN": []}    for node in statData["datanodes"]:        csvData["value"].append(node["data"]["data"])        for wd in node["wds"]:            csvData[wd["wdcode"]].append(wd["valuecode"])    # 指标编码含义    zbDict = {}    sjDict = {}    for node in statData["wdnodes"]:        if node["wdcode"] == "zb":            for zbNode in node["nodes"]:                zbDict[zbNode["code"]] = {                    "name": zbNode["name"],                    "cname": zbNode["cname"],                    "unit": zbNode["unit"],                }        if node["wdcode"] == "sj":            for sjNode in node["nodes"]:                sjDict[sjNode["code"]] = {                    "name": sjNode["name"],                    "cname": sjNode["cname"],                    "unit": sjNode["unit"],                }    # csv 数据中加入 zbCN 和 sjCN    for zb in csvData["zb"]:        zbCN = (            zbDict[zb]["cname"]            if zbDict[zb]["unit"] == ""            else zbDict[zb]["cname"] + "(" + zbDict[zb]["unit"] + ")"        )        csvData["zbCN"].append(zbCN)    for sj in csvData["sj"]:        csvData["sjCN"].append(sjDict[sj]["cname"])    # write csv file    df = pd.DataFrame(        csvData,        columns=["sj", "sjCN", "zb", "zbCN", "value"],    )    df.to_csv(fp, index=False)def get_stat_data(sj, zb):    payload = {        "dbcode": "hgnd",        "rowcode": "zb",        "m": "QueryData",        "colcode": "sj",        "wds": "[]",        "dfwds": '[{"wdcode":"zb","valuecode":"'        + zb        + '"},{"wdcode":"sj","valuecode":"'        + sj        + '"}]',    }    r = requests.get(host, params=payload, verify=False)    logging.debug("access url: %s", r.url)    time.sleep(2)    logging.debug(r.text)    resp = r.json()    if resp["returncode"] == 200:        return resp["returndata"]    else:        logging.error("error: %s", resp)        return None

代码说明：