1

历年行政区划码成品下载，欢迎白瞟，拿走不谢 - 顾志兵

 6 months ago
source link: https://www.cnblogs.com/guzb/p/18064214/free-download-administrative-area-code-for-each-year
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

几乎所有业务系统，都会涉及行政区域。国家统计局 官网上公开了所有的区域编码，一年一更新。但只能在线查看，没有提供完整数据库下载的连接。为此，我编写了一个简陋的 python 脚本，抓取了近几年的数据，供大家下载。如果这里的下载成品中没有你需要的数据，可以根据自己的要求，修改脚本，再运行起来去官网抓取即可。

📣 特别说明

  • 本脚本使用的 python 版本为 3.12.2
  • 本脚本仅在 windows11 下验证通过，未在 Linux 上验证过

行政区划码的特点

  • 编码是长度固定为12位的纯数字

    比如北京的编码 110000000000, 长度为12位，并且全部为数字，同时第1位数字不为0，也就是说，在数据库存储区位码时，可以直接使用 number 类型，而不必是 varchar。

  • 区划码共5个等级，见下表：

    等级 行政级别 示例
    1 省/直辖市 · 四川省
    · 北京市
    2 市 · 四川省/成都市
    · 北京市/市辖区
    3 区/县 · 四川省/成都市/武候区
    · 陕西省/咸阳市/泾阳县
    4 街道/乡镇 · 四川省/成都市/武候区/石羊街道
    · 陕西省/咸阳市/泾阳县/永乐镇
    5 社区/村委会 · 四川省/成都市/武候区/石羊街道/府城社区居委会
    · 陕西省/咸阳市/泾阳县/永乐镇/磨子桥村委会

  • 各等级所占数字位数及开始位置如下

    四川省/成都市/武候区/石羊街道/府城社区居委会+----+----+----+-----+-----+| 51 | 01 | 07 | 063 | 009 |+----+----+----+-----+-----+
年份 3级数据 4级数据 5级数据
2023 共3629条 点击下载 共4,4903条 点击下载 敬请期待
2022 共3634条 点击下载 共4,4907条 点击下载 敬请期待
2021 共3640条 点击下载 共4,4918条 点击下载 敬请期待
2020 共3644条 点击下载 共4,5180条 点击下载 敬请期待
2019 共3645条 点击下载 共4,6672条 点击下载 敬请期待

通常下载后，需要将数据保存到 MySql 数据库。假定你的 MySql 数据库信息如下：

  • 用户名：root
  • 密　码：root
  • 端　口：3306
  • 数据库名：my_db
  • 下载后的Sql文件位置为：d:\admin_area_2023_level-4.sql

则执行以下脚本将数据写入到 MySql

mysql -uroot -proot -P3306 my_db < D:\admin_area_2023_level-4.sql

根据实测情况，抓取不同等级范围的数据，耗时差别巨大，详情如下：

  • 3级数据：3分钟左右
  • 4级数据：30分钟左右
  • 5级数据：8个小时以上

Python 脚本

import requestsfrom lxml import etreeimport pymysqlimport tracebackimport time # 关闭 https 相关的警告requests.packages.urllib3.disable_warnings() # 国家统计局 (National Bureau Of Statistics) 行政区划数据抓取的主URLHOME_URL = "https://www.stats.gov.cn/sj/tjbz/tjyqhdmhcxhfdm/2023" # 是否开启打印输出ENABLE_PRINT = False # 最大抓取深度，最抓取到哪一个层级的区域数据，总共5级MAX_GRAB_LEVEL = 4 # 是否开启将数据写入到MySqlENABLE_MYSQL_INSERTION = True # 遇到列值为 null 时，是否跳过这条记录，继续向下执行SKIP_NULL_COLUMN_VALUE = True # 抓取的最大数据条数，主要用于调代码，避免输出内容太多，负数代表抓取所有MAX_GRAB_COUNT = -1 # 当前正在处理的省份，用于判断是否是直辖市current_province_name = None # 当前正在处理的城市名，用于判断提交MySql时，日志输出current_city_name = None # 连接MySql，请根据实际情况修改try: db = pymysql.connect(host='localhost', user='root', passwd='root', port=3306, db="my_db") cursor = db.cursor() print('连接Mysql成功！')except: print('连接MySql失败') exit def print_info(message:str): ''' 自定义一个内容输出方法，主要目的是可以统一控制是否输出，用于调试 ''' if ENABLE_PRINT: print(message) def insert_area_to_mysql(code:str, name:str, level:int, parent_code:str): ''' 插入一条记录到MySql，但不提交 参数： code(str): 区域编码 name(str): 区域名称 level(int): 区域等级， 1: 省/直辖市 2: 市 3: 区/县 4: 乡镇/街道 5: 社区/村委会 parent_code(str): 父级编码 ''' if not ENABLE_MYSQL_INSERTION: return if code is None or name is None: print("发现null值：code={}, name={}, level={}, parent_code={}".format(code, name, level, parent_code)) if SKIP_NULL_COLUMN_VALUE: return else: db.close() print("插入到MySql时遇到 Null 列值，程序将退出") exit() sql = "insert into admin_area_2023(`code`, `name`, `level`, `parent_code`) values ('{}', '{}', {}, '{}')".format(code, name, level, parent_code) sql = sql.replace("'None'", 'NULL') print_info(sql) cursor.execute(sql) def commit_for_mysql(): global db, current_province_name try: db.commit() print("保存<{}·{}>行政区划数据到MySql成功".format(current_province_name, current_city_name)) except Exception as e: db.rollback() print("保存" + current_province_name + "的行政区划数据到MySql失败") print(traceback.format_exc()) def get_admin_area_html(url:str): try_count = 0 while try_count < 3: try_count += 1 try: if try_count == 1: time.sleep(0.1) # 第一次抓取失败 elif try_count == 2: time.sleep(1) else: time.sleep(2) response = requests.get(url) response.encoding = response.apparent_encoding return etree.HTML(response.text) except Exception: if try_count > 3: print(traceback.format_exc()) print("连续 {} 次抓取 {} 页面时发生错误, 将放弃本页面的数据抓取。可能被服务怀疑是爬虫，拒绝了网络连接，因此休息10秒".format(try_count, url)) time.sleep(10) return None else: print("第 {} 次抓取 {} 网页文本失败".format(try_count, url)) def grap_all_provinces(): ''' 抓取所有省份 ''' html = get_admin_area_html(HOME_URL + "/index.html") province_nodes = html.xpath('//*/tr[@class="provincetr"]/td/a') grabed_count = 0 for province_node in province_nodes: grabed_count += 1 province_city_link = HOME_URL + "/" + province_node.attrib["href"] province_code = province_node.attrib["href"][0:2] + '0000000000' province_name = province_node.text.strip() global current_province_name current_province_name = province_name print_info("province_code={}, province_name={}".format(province_code, province_name)) insert_area_to_mysql(province_code, province_name, 1, None) if MAX_GRAB_LEVEL >= 2: grab_province_cities(province_city_link, province_code, province_name) if MAX_GRAB_COUNT > 0 and grabed_count >= MAX_GRAB_COUNT: break def grab_province_cities(province_city_link:str, province_code:str, province_name:str): ''' 抓取单个省/直辖市下的城市/区县 参数: province_city_link(str): 省/直辖市区域页面的完整 url province_code(str): 城市所属的省份编码 province_name(str): 城市所属的省份名称 ''' print("开始抓取省份（{}）的城市列表, URL={}".format(province_name, province_city_link)) html = get_admin_area_html(province_city_link) if html is None: print("抓取省份（{}）的城市列表失败".format(province_name)) return cityNodes = html.xpath('//*/tr[@class="citytr"]') grabed_count = 0 global current_city_name for cityNode in cityNodes: link_nodes = cityNode.xpath('./*/a') city_code = link_nodes[0].text city_name = link_nodes[1].text.strip() current_city_name = city_name insert_area_to_mysql(city_code, city_name, 2, province_code) print_info("city_code={}, city_name={}".format(city_code, city_name)) if MAX_GRAB_LEVEL >= 3 and link_nodes[1].attrib.has_key("href"): county_link = province_city_link[0:province_city_link.rfind('/')] + "/" + link_nodes[1].attrib["href"] grap_city_couties(county_link, city_code, city_name) # 以城市为最小提交单位 commit_for_mysql() if MAX_GRAB_COUNT > 0 and grabed_count >= MAX_GRAB_COUNT: break def grap_city_couties(city_county_link:str, city_code:str, city_name:str): ''' 抓取单个城市下的区/县 参数: city_county_link(str): 城市区/县页面的完整 url city_code(str): 城市的编码 city_name(str): 城市的名称 ''' print("开始抓取城市（{}）的区/县列表, URL={}".format(city_name, city_county_link)) html = get_admin_area_html(city_county_link) if html is None: print("抓取城市（{}）的区/县列表失败".format(city_name)) return county_nodes = html.xpath('//*/tr[@class="countytr"]') grabed_count = 0 global current_province_name for county_node in county_nodes: grabed_count += 1 county_link_nodes = county_node.xpath("./*/a") if len(county_link_nodes) == 0: # 没有<a>标签，通常是直辖市的市辖区，内容抓取方式不同 county_code = county_node.xpath("./td")[0].text county_name = county_node.xpath("./td")[1].text insert_area_to_mysql(county_code, county_name, 3, city_code) print_info("county_code={}, county_name={}, parent_code={}".format(county_code, county_name, city_code)) else: county_code = county_link_nodes[0].text county_name = county_link_nodes[1].text insert_area_to_mysql(county_code, county_name, 3, city_code) print_info("county_code={}, county_name={}, level=2, parent_code = {}".format(county_code, county_name, city_code)) if MAX_GRAB_LEVEL >= 4 and county_link_nodes[1].attrib.has_key("href"): town_link = city_county_link[0:city_county_link.rfind("/")] + "/" + county_link_nodes[1].attrib["href"] grap_county_towns(town_link, county_code, county_name) if MAX_GRAB_COUNT > 0 and grabed_count >= MAX_GRAB_COUNT: break def grap_county_towns(county_town_link:str, county_code:str, county_name:str): ''' 抓取单个区/县下的乡镇/街道 参数: county_town_link(str): 乡镇/街道数据页面完整的 url county_code(str): 区/县的编码 county_name(str): 区/县的名称 ''' print("开始抓取区县（{}）的街道/乡镇列表, URL={}".format(county_name, county_town_link)) html = get_admin_area_html(county_town_link) if html is None: print("抓取区县（{}）的街道/乡镇列表失败".format(county_name)) return town_nodes = html.xpath('//*/tr[@class="towntr"]') grabed_count = 0 for town_node in town_nodes: grabed_count += 1 village_link_nodes = town_node.xpath('./*/a') town_code = village_link_nodes[0].text town_name = village_link_nodes[1].text print_info("town_code={}, town_name={}".format(town_code, town_name)) insert_area_to_mysql(town_code, town_name, 4, county_code) if MAX_GRAB_LEVEL >= 5 and village_link_nodes[1].attrib.has_key("href"): village_link = county_town_link[0:county_town_link.rfind("/")] + "/" + village_link_nodes[1].attrib["href"] grap_town_villages(village_link, town_code, town_name) if MAX_GRAB_COUNT > 0 and grabed_count >= MAX_GRAB_COUNT: break def grap_town_villages(town_village_url:str, town_code:str, town_name:str): ''' 抓取单个街道/乡镇下的社区/村委会 参数: town_village_url(str): 社区/村委会数据页面完整的 url town_code(str): 街道/乡镇的编码 town_name(str): 街道/乡镇的名称 ''' print_info("开始抓取街道/乡镇下（{}）的社区/村委会列表, URL={}".format(town_name, town_village_url)) html = get_admin_area_html(town_village_url) if html is None: print("抓取街道/乡镇下（{}）的社区/村委会列表失败".format(town_name)) return village_nodes = html.xpath('//*/tr[@class="villagetr"]') grabed_count = 0 for village_node in village_nodes: grabed_count += 1 village_info_columns = village_node.xpath('./td') village_code = village_info_columns[0].text village_name = village_info_columns[2].text insert_area_to_mysql(village_code, village_name, 5, town_code) print_info("village_code={}, village_code={}".format(village_code, village_name)) if MAX_GRAB_COUNT > 0 and grabed_count >= MAX_GRAB_COUNT: break # 正式执行数据抓取任务grap_all_provinces() db.close()

  1. 下载 Window版Python 的安装程序，并在本机安装

    如果你打算在 Linux 下运行这个程序，你也可以直接下载 Linux 版本的 python，但我还没在 Linux 环境下验证过这个脚本

  2. 打开命令行窗口，依次执行以下脚本，以安装本脚本的依赖库

    pip install requestspip install lxmlpip install pymysql

  3. 安装 MySql 服务器并创建好相应的表

    你需要在本机上安装 MySql 数据库，并创建用于存储区划码数据的表。建表语句如下：

    CREATE TABLE `admin_area_2023` ( `code` char(12) NOT NULL COMMENT '区域编码', `name` varchar(60) NOT NULL COMMENT '区域名称', `level` tinyint(4) NOT NULL COMMENT '区域等级：\r\n1 : 省/直辖市\r\n2 : 市\r\n3 : 区/县\r\n4 : 乡镇/街道\r\n5 : 社区/村委会', `parent_code` char(12) DEFAULT NULL COMMENT '父级区域编码', PRIMARY KEY (`code`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

  4. 根据实际情况修改脚本

    一般来说，你需要修改以下几项参数

    • 要抓取哪一年的数据。搜索 HOME_URL = 即可找到代码所在行

    • 连接 MySql 的用户名、密码、端口、数据库名称。搜索 pymysql.connect 即可找到代码所在行

    • 修改SQL语句，设置要插入的表名。搜索 insert into 即可找到代码所在行

    • 设置要抓取的数据等级，默认为4级。搜索 MAX_GRAB_LEVEL 即可找到代码所在行

    还支持一些其它的冷门设置，就请自行阅读源码吧。

  5. 假设本机的 python 脚本命名为 admin-area-data-spider.py, 且位于 D 盘根目录，则执行以下命令运行程序：

    python d:\admin-area-data-spider.py

这是我花了一上午时间，利用网友分享的 python 知识，临时编写的脚本。但由于之前从没有接触过 python，因此代码质量无法保障，请各位老鸟见量。可以确保的是，它当前在 windows 下是可以工作的。

不过通过这次临时的 python 体验后，非常喜欢这门语言，用它来快速开发各种工具和快速构建原型项目，以验证业务可行性是两个很不错的应用领域。当然，它当前在科学计算和人工智能领域的应用更广泛。

上面已经贴出了数据抓取的 python 代码，但后期我可能还会修改，比如:

  • 拆分成多个 module
  • 添加抓取统计功能
  • 参数化脚本的执行

故这里再附上 行政区划码抓取的Gitee工程 地址，感兴趣的朋友可 watch、star、fork。


report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK