本系统基于Spark，通过Web进行访问+源代码+文档说明

# -*- coding: utf-8 -*- import csv import datetime import json import math import time import redis from sshtunnel import SSHTunnelForwarder from lib.crawler import StatsGovCn from lib.util import DBUtilStatsGovCn def fetch_stats_gov_cn(url, db_path, show_log=True, sleep_time=0): """ 采集统计局信息 :param url: 统计局信息根网址 :type url: str :param db_path: SQLite数据库路径 :type db_path: str :param show_log: 是否显示日志 :type show_log: bool :param sleep_time: 爬虫每次爬取后的休眠时间，单位为秒。 :type sleep_time: int :return: """ # 程序开始时间 begin_time = time.time() stats_gov_cn_crawler = StatsGovCn() stats_gov_cn_crawler.sleep_time = sleep_time if stats_gov_cn_crawler.check(url.replace('$ROUTE$', 'index.html'))[0] != 'province': raise Exception('不是省级信息页面') # 数据库操作对象 db_util = DBUtilStatsGovCn(db_path + 'db_stats.gov.cn.sqlite') # 上级 URL 地址 url_base = {} url_base_temp = '' # 抓取并保存省级信息 if show_log: print(f'[Log][{datetime.datetime.now()}] 开始抓取并保存省级信息') provinces = stats_gov_cn_crawler.province(url.replace('$ROUTE$', 'index.html')) db_util.truncate_province() for province in provinces: db_util.insert_province(province['statistical_code'], province['code'], province['name']) url_base[province['statistical_code']] = url.replace('$ROUTE$', '') if show_log: print(f'[Log][{datetime.datetime.now()}] 完成抓取并保存省级信息') print(f'[REPORT] 省级信息 {len(provinces)} 个') # 抓取并保存地级信息 if show_log: print(f'[Log][{datetime.datetime.now()}] 开始抓取并保存地级信息') cities = [] db_util.truncate_city() for province in provinces: if show_log: province_name_temp = province['name'] print(f'[Log][{datetime.datetime.now()}] [{provinces.index(province) + 1}/{len(provinces)}] ' f'开始抓取并保存【{province_name_temp}】') if province['href'] != '': url_base_temp = url_base[province['statistical_code']] + province['href'] cities_temp = stats_gov_cn_crawler.city(url_base_temp) for city in cities_temp: cities.append(city) db_util.insert_city(city['statistical_code'], city['code'], city['name'], province['statistical_code']) url_base[city['statistical_code']] = url_base_temp[0:url_base_temp.rfind('/')+1] if show_log: province_name_temp = province['name'] print(f'[Log][{datetime.datetime.now()}] [{provinces.index(province) + 1}/{len(provinces)}] ' f'完成抓取并保存【{province_name_temp}】') if show_log: print(f'[Log][{datetime.datetime.now()}] 完成抓取并保存地级信息') print(f'[REPORT] 地级信息 {len(cities)} 个') # 抓取并保存县级信息 if show_log: print(f'[Log][{datetime.datetime.now()}] 开始抓取并保存县级信息') counties = [] db_util.truncate_county() for city in cities: city_db_temp = db_util.select_city(city['statistical_code']) if show_log: province_name_temp = db_util.select_province(city_db_temp['province_statistical_code'])['name'] city_name_temp = city['name'] print(f'[Log][{datetime.datetime.now()}] [{cities.index(city) + 1}/{len(cities)}] ' f'开始抓取并保存【{province_name_temp}】【{city_name_temp}】') if city['href'] != '': try: url_base_temp = url_base[city['statistical_code']] + city['href'] counties_temp = stats_gov_cn_crawler.county(url_base_temp) except Exception as e: if e.args[0] == '不是县级信息页面': counties_temp = [{ 'href': city['href'][city['href'].find('/')+1:], 'statistical_code': city['statistical_code'], 'code': city['statistical_code'][0:6], 'name': city['name'] }] else: raise e for county in counties_temp: counties.append(county) db_util.insert_county( county['statistical_code'], county['code'], county['name'], city_db_temp['province_statistical_code'], city['statistical_code'] ) url_base[county['statistical_code']] = url_base_temp[0:url_base_temp.rfind('/')+1] if show_log: province_name_temp = db_util.select_province(city_db_temp['province_statistical_code'])['name'] city_name_temp = city['name'] print(f'[Log][{datetime.datetime.now()}] [{cities.index(city) + 1}/{len(cities)}] ' f'完成抓取并保存【{province_name_temp}】【{city_name_temp}】') if show_log: print(f'[Log][{datetime.datetime.now()}] 完成抓取并保存县级信息') print(f'[REPORT] 县级信息 {len(counties)} 个') # 抓取并保存乡级信息 if show_log: print(f'[Log][{datetime.datetime.now()}] 开始抓取并保存乡级信息') towns = [] db_util.truncate_town() for county in counties: county_db_temp = db_util.select_county(county['statistical_code']) if show_log: province_name_temp = db_util.select_province(county_db_temp['province_statistical_code'])['name'] city_name_temp = db_util.select_city(county_db_temp['city_statistical_code'])['name'] county_name_temp = county['name'] print(f'[Log][{datetime.datetime.now()}] [{counties.index(county) + 1}/{len(counties)}] ' f'开始抓取并保存【{province_name_temp}】【{city_name_temp}】【{county_name_temp}】') if county['href'] != '': try: url_base_temp = url_base[county['statistical_code']] + county['href'] towns_temp = stats_gov_cn_crawler.town(url_base_temp) except Exception as e: if e.args[0] == '不是乡级信息页面': towns_temp = [{ 'href': county['href'][county['href'].find('/')+1:], 'statistical_code': county['statistical_code'], 'code': county['statistical_code'][0:9], 'name': county['name'] }] else: raise e for town in towns_temp: towns.append(town) db_util.insert_town( town['statistical_code'], town['code'], town['name'], county_db_temp['province_statistical_code'], county_db_temp['city_statistical_code'], county['statistical_code'] ) url_base[town['statistical_code']] = url_base_temp[0:url_base_temp.rfind('/')+1] if show_log: province_name_temp = db_util.select_province(county_db_temp['province_statistical_code'])['name'] city_name_temp = db_util.select_city(county_db_temp['city_statistical_code'])['name'] county_name_temp = county['name'] print(f'[Log][{datetime.datetime.now()}] [{counties.index(county) + 1}/{len(counties)}] ' f'完成抓取并保存【{province_name_temp}】【{city_name_temp}】【{county_name_temp}】') if show_log: print(f'[Log][{datetime.datetime.now()}] 完成抓取并保存乡级信息') print(f'[REPORT] 乡级信息 {len(towns)} 个') # 抓取并保存村级信息 if show_log: print(f'[Log][{