简介本资源是一套面向数据分析初学者与政务信息化研究者的Python政府数据爬取实践方案聚焦赣州、吐鲁番、大理、太原、大庆五地政府官网解决多源地方政府公开信息自动化采集难题适用于政策分析、区域比较研究及基层数据治理等场景。压缩包共35个文件含22个核心Python爬虫脚本如crawGanzhouGov.py、crawTaiyuanWjw.py等覆盖卫健、发改、教育、环保等十余类部门站点4份docx文档提供各地需求说明与使用指引4张jpeg/png图像用于界面示意或流程图解2个log文件记录运行轨迹便于调试追踪另有readme.txt和.gitignore等辅助文件结构清晰、模块可复用。资源大小为7.08MBZIP格式即下即用。目前已有295人学习下载读者可直接部署运行、按需修改目标URL与解析逻辑并基于现成日志与分城市脚本快速定位问题、迁移适配其他地市站点。1. 项目缘起从“数据孤岛”到“价值挖掘”的实践最近在做一个区域经济分析的课题需要对比几个典型城市在特定领域的公开数据。我首先想到的就是各地的政府数据开放平台这些平台汇聚了交通、气象、民生、经济等大量一手数据是绝佳的分析素材。然而当我真正开始动手时发现事情没那么简单赣州、吐鲁番、大理、太原、大庆这五个城市分属不同省份发展水平各异其政府数据开放的程度、平台的技术架构、数据的呈现方式可以说是“五花八门”。有的提供了清晰的API接口有的则隐藏在复杂的网页动态加载背后还有的甚至只是以PDF附件的形式存在。手动收集不仅效率低下而且难以保证数据的持续更新和格式统一。这正是“基于Python的政府数据爬取”项目诞生的背景。它不是一个简单的、针对单一网站的脚本而是一套需要应对多种技术栈、反爬策略和数据结构的综合性解决方案。我的目标很明确设计一套健壮、可扩展的代码框架能够自动化地从这五个差异巨大的政府数据平台上稳定、合规地获取所需的结构化数据。这个过程远不止是写几个requests.get那么简单它涉及到对目标网站架构的逆向工程、对反爬机制的应对、对异构数据的清洗规整以及最终构建一个可维护的数据管道。接下来我就把这套设计思路和核心源码的实现细节结合踩过的坑和总结的经验完整地分享出来。2. 目标网站分析与技术选型知己知彼百战不殆在动手写一行代码之前充分的侦察是成功的一半。对于政府网站尤其需要谨慎分析确保爬取行为的合法合规与技术可行性。2.1 五城政府数据平台特征分析我首先对五个目标平台进行了手动浏览和初步的技术探查赣州市其数据开放平台相对现代部分数据集提供了直接的API查询接口返回格式为JSON这是最理想的情况。但更多数据是以“数据目录”列表形式存在点击后进入详情页数据可能以表格形式嵌入在HTML中或提供CSV/Excel文件下载链接。吐鲁番市网站风格较为传统数据多以静态HTML表格呈现。难点在于分页逻辑可能不是标准的page参数而是通过POST表单提交或JavaScript生成的特殊令牌。此外页面编码需要特别注意可能存在非UTF-8的情况。大理白族自治州部分数据通过“数据超市”的概念展示界面交互较多数据很可能通过Ajax动态加载。需要分析XHRXMLHttpRequest网络请求找到真正的数据接口。这类接口往往带有用于身份验证或防篡改的签名参数。太原市作为省会城市其平台建设较为规范提供了数据集目录和预览。但下载数据时常需要“申请”或“验证”流程中可能涉及模拟登录、填写申请理由等步骤。数据文件格式也可能多样包括ZIP压缩包内的多个文件。大庆市平台数据量可能相对较少但同样不可轻视。其网页可能使用了前端框架如Vue.js、React导致初始HTML中不包含数据必须执行JavaScript才能渲染。这需要用到Selenium或Playwright这类浏览器自动化工具。2.2 核心工具链选型与理由基于以上分析我选择了以下技术栈它们共同构成了一个从简单到复杂、层层递进的武器库核心请求库Requests httpxRequests是Python生态的基石简单易用处理静态页面和简单API足矣。我将其作为首选。httpx支持HTTP/2和异步请求。当需要高效并发抓取多个列表页或API时httpx的异步特性将带来巨大的性能提升。我计划在架构上预留异步支持。HTML解析BeautifulSoup4 lxmlBeautifulSoup4提供了非常友好的Pythonic API来解析和遍历HTML/XML文档适合快速开发和数据提取逻辑的原型构建。lxml是一个解析速度极快的C语言库。BeautifulSoup可以指定lxml作为后端解析器兼顾了易用性与性能。对于大规模页面解析这一点至关重要。动态页面渲染Selenium当目标数据由JavaScript动态渲染时无头浏览器是唯一可靠的选择。我选择Selenium因为它生态成熟社区资源丰富。配合ChromeDriver或GeckoDriver可以模拟真实用户操作。重要心得Selenium很强大但也很重、很慢。我的原则是“非必要不使用”。只有在前端分析确认数据确实由JS渲染且找不到隐藏的Ajax接口时才会启用Selenium。数据存储与中间状态SQLite JSONSQLite用于存储最终清洗后的结构化数据。它无需单独服务器一个文件就是一个数据库非常适合本项目。我可以为每个城市或每类数据建立单独的表方便后续分析。JSON用于存储爬取任务的配置如URL列表、解析规则、中间状态如已爬取的页码以及从API直接获取的原始数据。JSON文件易于阅读和调试。调度与健壮性Schedule Retryingschedule库用于实现定时爬取任务。例如可以设定每天凌晨2点自动运行爬虫获取最新的数据。retrying库网络请求充满不确定性偶尔的超时、连接错误是正常的。用retrying装饰器可以轻松实现“遇到异常时自动重试N次”极大增强爬虫的健壮性。这个选型策略的核心思想是**“轻重结合按需启用”**。一个典型的爬虫流程会优先尝试用最轻量的RequestsBeautifulSoup方案失败或确认无效后再逐步升级到更复杂的方案。3. 爬虫架构设计与核心模块实现面对五个不同的网站写五个独立的、互不关联的脚本是下策。一旦某个网站的改版或者需要增加第六个城市维护成本会急剧上升。因此我设计了一个模块化的架构旨在提高代码的复用性和可维护性。3.1 项目目录结构规划一个清晰的项目结构是良好架构的开始。gov_data_crawler/ ├── config/ # 配置文件 │ ├── ganzhou.json # 赣州站点的配置起始URL 解析规则等 │ ├── tulufan.json # 吐鲁番站点配置 │ └── ... ├── core/ # 核心引擎 │ ├── __init__.py │ ├── downloader.py # 下载器封装请求逻辑处理重试、代理、头信息 │ ├── parser.py # 解析器基类与各站点解析器 │ ├── scheduler.py # 任务调度器管理待抓取URL队列 │ └── storage.py # 数据存储器负责存入SQLite或文件 ├── spiders/ # 各站点爬虫主逻辑 │ ├── base_spider.py # 爬虫基类定义爬取流程模板 │ ├── ganzhou_spider.py # 赣州站点爬虫 │ ├── tulufan_spider.py # 吐鲁番站点爬虫 │ └── ... ├── utils/ # 工具函数 │ ├── logger.py # 日志配置 │ ├── tools.py # 通用工具函数编码处理、字符串清洗等 │ └── ... ├── data/ # 数据存放目录 │ ├── raw/ # 原始HTML、JSON数据 │ ├── processed/ # 清洗后的结构化数据CSV │ └── database.db # SQLite数据库文件 ├── main.py # 主程序入口 └── requirements.txt # 项目依赖3.2 核心模块源码剖析接下来我们深入几个最关键模块的代码实现。1. 下载器 (core/downloader.py)网络请求的稳健基石下载器的职责是处理所有HTTP请求内置重试、超时、随机User-Agent、代理支持等功能确保网络层面的稳定性。import requests from retrying import retry import random import time from fake_useragent import UserAgent from urllib.parse import urljoin class Downloader: def __init__(self, delay_range(1, 3), use_proxyFalse, proxy_poolNone): 初始化下载器 :param delay_range: 请求间隔延迟范围秒用于礼貌爬取 :param use_proxy: 是否使用代理IP :param proxy_pool: 代理IP池列表 self.delay_range delay_range self.use_proxy use_proxy self.proxy_pool proxy_pool or [] self.ua UserAgent() self.session requests.Session() # 设置一个默认的、看起来像浏览器的请求头 self.session.headers.update({ Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Connection: keep-alive, }) def _get_random_delay(self): 生成随机延迟时间 return random.uniform(*self.delay_range) def _get_random_headers(self): 生成随机User-Agent头信息 return {User-Agent: self.ua.random} def _get_random_proxy(self): 随机选择一个代理如果池为空则返回None if self.use_proxy and self.proxy_pool: return {http: random.choice(self.proxy_pool), https: random.choice(self.proxy_pool)} return None retry(stop_max_attempt_number3, wait_fixed2000) def fetch(self, url, methodGET, paramsNone, dataNone, jsonNone, **kwargs): 执行HTTP请求内置重试机制 :param url: 目标URL :return: requests.Response对象 # 每次请求前延迟模拟人类操作避免被封IP time.sleep(self._get_random_delay()) headers self._get_random_headers() proxies self._get_random_proxy() try: response self.session.request( methodmethod, urlurl, paramsparams, datadata, jsonjson, headers{**self.session.headers, **headers}, proxiesproxies, timeout10, # 连接和读取超时设为10秒 **kwargs ) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常触发重试 # 检查编码避免乱码 if response.encoding ISO-8859-1: response.encoding response.apparent_encoding or utf-8 return response except requests.exceptions.RequestException as e: print(f请求失败: {url}, 错误: {e}) # 重试装饰器会处理重试第三次失败后抛出异常 raise # 使用示例 if __name__ __main__: dl Downloader(delay_range(2, 5)) try: resp dl.fetch(https://data.ganzhou.gov.cn/api/dataset/list) print(resp.json()) # 假设返回JSON except Exception as e: print(f最终抓取失败: {e})注意fake_useragent库需要在线更新UA列表在无网络环境可能报错。生产环境中可以考虑维护一个本地的UA列表文件。另外代理IP池需要自己维护或购买服务对于政府公开数据在控制频率的前提下通常不需要代理。2. 解析器基类与站点解析器 (core/parser.pyspiders/xxx_spider.py)解析器负责从下载到的HTML或JSON中提取目标数据。我采用基类定义接口各站点继承并实现具体逻辑的模式。# core/parser.py from abc import ABC, abstractmethod from bs4 import BeautifulSoup import json class BaseParser(ABC): 解析器抽象基类 abstractmethod def parse_list(self, html_content, base_url): 解析列表页提取详情页链接和可能的分页信息 :param html_content: 列表页HTML内容 :param base_url: 基础URL用于拼接相对链接 :return: tuple (detail_urls, next_page_url) pass abstractmethod def parse_detail(self, html_content): 解析详情页提取结构化数据 :param html_content: 详情页HTML内容 :return: dict 结构化数据 pass staticmethod def make_soup(html_content, parserlxml): 创建BeautifulSoup对象 return BeautifulSoup(html_content, parser) # spiders/ganzhou_spider.py from core.parser import BaseParser from urllib.parse import urljoin import re class GanzhouParser(BaseParser): 赣州市数据平台解析器示例针对API接口和HTML混合情况 def parse_list(self, html_content, base_url): 赣州平台列表页可能是API返回的JSON也可能是HTML。 这里假设我们分析的是HTML目录页。 soup self.make_soup(html_content) detail_urls [] next_page_url None # 假设数据条目在 classdataset-list 的div中每个条目链接在a标签里 list_container soup.find(div, class_dataset-list) if list_container: for item in list_container.find_all(a, hrefTrue): link item[href] # 拼接完整URL full_url urljoin(base_url, link) # 通过链接特征过滤只保留数据详情页链接 if /dataset/ in full_url and resource not in full_url: detail_urls.append(full_url) # 查找下一页链接假设它在 classnext 的a标签里 next_tag soup.find(a, class_next, hrefTrue) if next_tag: next_page_url urljoin(base_url, next_tag[href]) return detail_urls, next_page_url def parse_detail(self, html_content): 解析数据详情页提取元数据和资源链接。 实际情况可能更复杂需要处理表格预览、文件下载链接等。 soup self.make_soup(html_content) data {} # 提取数据集标题 title_tag soup.find(h1, class_page-heading) data[title] title_tag.get_text(stripTrue) if title_tag else # 提取描述信息 desc_tag soup.find(div, class_notes) data[description] desc_tag.get_text(stripTrue) if desc_tag else # 提取资源部分如CSV、Excel文件链接 resources [] resource_section soup.find(div, class_resource-list) if resource_section: for li in resource_section.find_all(li): link_tag li.find(a, class_resource-url-analytics, hrefTrue) if link_tag: resource { name: link_tag.get_text(stripTrue), url: link_tag[href], format: li.find(span, class_format-label).get_text(stripTrue) if li.find(span, class_format-label) else } resources.append(resource) data[resources] resources # 提取其他元数据如发布部门、更新时间等 # ... 根据实际页面结构编写更多提取逻辑 return data # 针对API接口的解析假设列表API返回JSON def parse_api_list(self, json_content): 解析API返回的列表JSON data_list json.loads(json_content) detail_urls [] for item in data_list.get(result, []): # 假设每个item里有id字段可以拼接出详情页API地址 detail_api_url fhttps://data.ganzhou.gov.cn/api/dataset/{item[id]} detail_urls.append(detail_api_url) # API分页信息通常包含在JSON里 next_page data_list.get(next_page) next_page_url fhttps://data.ganzhou.gov.cn/api/dataset/list?page{next_page} if next_page else None return detail_urls, next_page_url3. 爬虫基类与调度 (spiders/base_spider.py)基类定义了爬取的标准工作流初始化配置 - 处理列表页 - 循环抓取详情页 - 存储数据。# spiders/base_spider.py import json import os from core.downloader import Downloader from core.storage import Storage import logging class BaseSpider: def __init__(self, name, start_urls, parser, config_path): 初始化爬虫 :param name: 爬虫名称如 ganzhou :param start_urls: 起始URL列表 :param parser: 该站点对应的解析器实例 :param config_path: 配置文件路径 self.name name self.start_urls start_urls self.parser parser self.downloader Downloader(delay_range(2, 5)) # 针对政府网站延迟稍长以示友好 self.storage Storage(db_pathfdata/database.db) self.logger logging.getLogger(name) # 加载站点特定配置 with open(config_path, r, encodingutf-8) as f: self.config json.load(f) # 创建数据目录 os.makedirs(fdata/raw/{self.name}, exist_okTrue) os.makedirs(fdata/processed/{self.name}, exist_okTrue) def crawl_list_page(self, list_url): 抓取并解析一个列表页 self.logger.info(f开始抓取列表页: {list_url}) try: resp self.downloader.fetch(list_url) # 根据内容类型决定解析方式 content_type resp.headers.get(Content-Type, ) if application/json in content_type: detail_urls, next_page_url self.parser.parse_api_list(resp.text) else: detail_urls, next_page_url self.parser.parse_list(resp.text, list_url) # 保存原始列表页HTML/JSON raw_filename fdata/raw/{self.name}/list_{hash(list_url)}.html with open(raw_filename, w, encodingutf-8) as f: f.write(resp.text) return detail_urls, next_page_url except Exception as e: self.logger.error(f抓取列表页失败 {list_url}: {e}) return [], None def crawl_detail_page(self, detail_url): 抓取并解析一个详情页 self.logger.info(f开始抓取详情页: {detail_url}) try: resp self.downloader.fetch(detail_url) data self.parser.parse_detail(resp.text) # 为数据添加来源信息 data[source_url] detail_url data[crawled_time] datetime.now().isoformat() # 保存原始详情页 raw_filename fdata/raw/{self.name}/detail_{hash(detail_url)}.html with open(raw_filename, w, encodingutf-8) as f: f.write(resp.text) # 存储结构化数据到数据库 self.storage.save_data(self.name, data) # 同时保存一份JSON到本地便于调试 processed_filename fdata/processed/{self.name}/{hash(detail_url)}.json with open(processed_filename, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) self.logger.info(f详情页抓取成功: {detail_url}) return data except Exception as e: self.logger.error(f抓取详情页失败 {detail_url}: {e}) return None def run(self): 爬虫主运行逻辑 self.logger.info(f爬虫 {self.name} 开始运行...) # 待抓取列表页队列 list_url_queue list(self.start_urls) visited_list_urls set() while list_url_queue: current_list_url list_url_queue.pop(0) if current_list_url in visited_list_urls: continue visited_list_urls.add(current_list_url) # 1. 抓取当前列表页 detail_urls, next_page_url self.crawl_list_page(current_list_url) # 2. 抓取当前列表页中的所有详情页 for detail_url in detail_urls: self.crawl_detail_page(detail_url) # 可以在这里加入更细粒度的延迟 # 3. 将下一页列表URL加入队列 if next_page_url and next_page_url not in visited_list_urls: list_url_queue.append(next_page_url) self.logger.info(f爬虫 {self.name} 运行结束。) self.storage.close()4. 实战中的挑战与解决方案五个城市的“攻坚战”有了框架真正的挑战在于适配每个独特的网站。下面分享针对每个城市平台遇到的具体问题及解决代码。4.1 赣州处理API与HTML的混合架构赣州平台部分数据有API部分只有HTML。需要能自动判断并切换解析模式。解决方案在Downloader.fetch方法中根据URL特征或响应头Content-Type自动选择解析路径。在GanzhouParser中实现两个解析方法如上述代码所示并在爬虫主逻辑中做分支判断。关键代码补充在GanzhouSpider中class GanzhouSpider(BaseSpider): def crawl_list_page(self, list_url): resp self.downloader.fetch(list_url) # 判断是否为API if api in list_url or resp.headers.get(Content-Type, ).startswith(application/json): detail_urls, next_page self.parser.parse_api_list(resp.text) else: detail_urls, next_page self.parser.parse_list(resp.text, list_url) # ... 后续处理4.2 吐鲁番应对非标准分页与表单提交吐鲁番网站的分页可能使用POST方法参数隐藏在表单中。解决方案使用浏览器开发者工具的“网络”选项卡记录点击“下一页”时发出的真实请求。发现是POST /listData.do携带pageNo和pageSize等参数。关键代码# 在吐鲁番的解析器或爬虫中 def get_list_page(self, page_no): list_url http://www.tlf.gov.cn/listData.do # 模拟表单提交 form_data { pageNo: page_no, pageSize: 20, param1: value1, # 其他可能需要的固定参数 } # 注意可能需要先获取一个初始页面的Cookie或Token resp self.downloader.fetch(list_url, methodPOST, dataform_data) # 解析resp.text (可能是HTML或JSON)4.3 大理破解Ajax动态加载与参数签名大理平台的数据通过Ajax加载请求URL类似/api/data/search并且请求参数里有一个不断变化的sign或token字段。解决方案逆向JavaScript在“网络”选项卡中找到这个Ajax请求查看它的“发起程序” (Initiator)追踪到前端生成sign的JavaScript代码。模拟计算如果算法简单如对固定字符串进行MD5可以用Python的hashlib复现。如果复杂且混淆严重可以考虑使用PyExecJS或js2py来执行关键的JS代码片段。备用方案如果逆向工程太困难最后的办法是使用Selenium完整渲染页面然后从渲染后的HTML中提取数据。虽然慢但能保证拿到数据。简化示例假设sign是时间戳的MD5import hashlib import time def generate_sign(params): # 假设签名算法是将参数按key排序后拼接加上一个盐值再做MD5 salt some_secret_salt param_str .join([f{k}{v} for k, v in sorted(params.items())]) sign_str param_str salt return hashlib.md5(sign_str.encode(utf-8)).hexdigest() params { keyword: 经济, page: 1, timestamp: int(time.time()*1000) # 毫秒时间戳 } params[sign] generate_sign(params) # 然后将params作为查询字符串或POST data发送4.4 太原模拟登录与文件下载处理太原平台下载数据需要登录或申请。模拟登录是关键。解决方案会话保持使用requests.Session()对象它会自动处理Cookies。分析登录请求手动登录一次抓包分析登录POST请求的URL、表单字段除了用户名密码可能还有csrf_token。获取并传递Token通常登录前需要先访问登录页从HTML中提取一个隐藏的csrf_token随登录请求一起提交。下载文件登录后访问文件下载链接使用session.get(url, streamTrue)来流式下载大文件避免内存溢出。关键代码class TaiyuanSpider: def __init__(self): self.session requests.Session() self.login_url https://data.taiyuan.gov.cn/login self.download_url_template https://data.taiyuan.gov.cn/dataset/download/{id} def login(self, username, password): # 1. 获取登录页提取csrf_token login_page self.session.get(self.login_url) soup BeautifulSoup(login_page.text, lxml) csrf_token soup.find(input, {name: csrf_token})[value] # 2. 构造登录数据 login_data { username: username, password: password, csrf_token: csrf_token } # 3. 提交登录 resp self.session.post(self.login_url, datalogin_data) if 登录成功 in resp.text or resp.url ! self.login_url: print(登录成功) return True else: print(登录失败) return False def download_file(self, file_id, save_path): url self.download_url_template.format(idfile_id) resp self.session.get(url, streamTrue) if resp.status_code 200: with open(save_path, wb) as f: for chunk in resp.iter_content(chunk_size8192): f.write(chunk) print(f文件已保存至: {save_path}) else: print(f下载失败状态码: {resp.status_code})4.5 大庆使用Selenium应对JavaScript渲染当所有分析都指向数据由前端JS框架动态渲染时Selenium是最终手段。解决方案配置无头浏览器使用headless模式不显示GUI节省资源。智能等待使用WebDriverWait和expected_conditions等待特定元素出现后再操作避免因网络延迟导致的错误。提取数据等待页面加载完成后使用driver.page_source获取完整HTML再交给BeautifulSoup解析。或者直接使用Selenium的find_element方法提取数据。关键代码from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import time class DaqingSeleniumCrawler: def __init__(self): chrome_options Options() chrome_options.add_argument(--headless) # 无头模式 chrome_options.add_argument(--disable-gpu) chrome_options.add_argument(--no-sandbox) # 可添加其他选项如屏蔽图片加载以加速 # chrome_options.add_experimental_option(prefs, {profile.managed_default_content_settings.images: 2}) self.driver webdriver.Chrome(optionschrome_options) self.wait WebDriverWait(self.driver, 10) def crawl_data_list(self, url): self.driver.get(url) data_items [] try: # 等待数据表格加载出来 table self.wait.until( EC.presence_of_element_located((By.CSS_SELECTOR, .data-table)) ) # 假设每行数据在tr标签内 rows table.find_elements(By.TAG_NAME, tr) for row in rows[1:]: # 跳过表头 cols row.find_elements(By.TAG_NAME, td) if len(cols) 2: item { name: cols[0].text, value: cols[1].text, # ... 其他字段 } data_items.append(item) return data_items except Exception as e: print(f使用Selenium抓取失败: {e}) return [] finally: self.driver.quit() # 使用 crawler DaqingSeleniumCrawler() data crawler.crawl_data_list(http://data.daqing.gov.cn/) print(data)5. 数据清洗、存储与任务调度爬取到的原始数据往往是脏的、不规整的。存储和后续的自动化调度是让项目从脚本升级为系统的关键。5.1 数据清洗与规整在storage.save_data方法调用前或后应有一个数据清洗的环节。我通常在解析器parse_detail返回数据后立即进行清洗。清洗常见任务去除空白字符str.strip()统一日期格式使用datetime或dateutil.parser将“2023年1月1日”、“2023-01-01”等统一为ISO格式“2023-01-01”。处理缺失值将“暂无”、“N/A”、“-”等替换为None或空字符串。拆分合并字段例如将“地址江西省赣州市章贡区”拆分为province、city、district字段。数值提取从字符串中提取数字如从“约123.5万元”中提取123.5。示例清洗函数# utils/data_cleaner.py import re from datetime import datetime def clean_text(text): if not isinstance(text, str): return text # 去除首尾空白、换行符、多个连续空格 text re.sub(r\s, , text.strip()) return text if text else None def parse_chinese_date(date_str): 尝试解析中文日期字符串 if not date_str: return None patterns [ r(\d{4})年(\d{1,2})月(\d{1,2})日, r(\d{4})-(\d{1,2})-(\d{1,2}), r(\d{4})/(\d{1,2})/(\d{1,2}), ] for pattern in patterns: match re.search(pattern, date_str) if match: year, month, day match.groups() try: return datetime(int(year), int(month), int(day)).date().isoformat() except ValueError: continue # 如果都无法解析返回原始字符串或None return date_str def extract_number_from_string(text): 从字符串中提取第一个浮点数或整数 if not text: return None match re.search(r[-]?\d*\.\d|\d, text.replace(,, )) # 移除千分位逗号 return float(match.group()) if match else None5.2 结构化存储到SQLitecore/storage.py负责将清洗后的数据持久化。# core/storage.py import sqlite3 import json from datetime import datetime class Storage: def __init__(self, db_pathgov_data.db): self.conn sqlite3.connect(db_path) self.cursor self.conn.cursor() self._init_tables() def _init_tables(self): # 为每个城市创建一个表示例。更优设计可能是统一结构用city字段区分。 create_table_sql CREATE TABLE IF NOT EXISTS ganzhou_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, description TEXT, department TEXT, publish_date DATE, update_date DATE, resource_links TEXT, -- 存储JSON字符串如 [{name:xx.csv, url:..., format:CSV}] source_url TEXT UNIQUE, crawled_time TIMESTAMP, raw_data TEXT -- 可选存储原始JSON或关键字段便于回溯 ); self.cursor.execute(create_table_sql) # 类似地创建其他城市的表... self.conn.commit() def save_data(self, city, data_dict): table_name f{city}_data # 确保资源链接以JSON字符串存储 if resources in data_dict and isinstance(data_dict[resources], list): data_dict[resources_json] json.dumps(data_dict[resources], ensure_asciiFalse) del data_dict[resources] # 移除原字段避免插入错误 # 动态构建SQL插入语句处理可能缺失的字段 columns , .join(data_dict.keys()) placeholders , .join([?] * len(data_dict)) sql fINSERT OR REPLACE INTO {table_name} ({columns}) VALUES ({placeholders}) try: self.cursor.execute(sql, list(data_dict.values())) self.conn.commit() print(f数据成功存入{table_name}) except sqlite3.IntegrityError as e: print(f数据可能重复唯一约束冲突: {e}) except Exception as e: print(f数据存储失败: {e}) def close(self): self.conn.close()5.3 使用Schedule实现定时任务最后我们可以用schedule库让爬虫定时自动运行。# main_scheduler.py import schedule import time from spiders.ganzhou_spider import GanzhouSpider from spiders.tulufan_spider import TulufanSpider # ... 导入其他爬虫 import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) def job_ganzhou(): logging.info(开始执行赣州数据爬取任务) spider GanzhouSpider( nameganzhou, start_urls[https://data.ganzhou.gov.cn/catalog], parserGanzhouParser(), config_pathconfig/ganzhou.json ) spider.run() def job_tulufan(): logging.info(开始执行吐鲁番数据爬取任务) # ... 初始化并运行吐鲁番爬虫 # 定义定时规则例如每天凌晨2点运行 schedule.every().day.at(02:00).do(job_ganzhou) schedule.every().day.at(02:30).do(job_tulufan) # 错开时间避免并发和IP压力 if __name__ __main__: logging.info(爬虫调度器已启动...) # 立即运行一次可选 # job_ganzhou() while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次是否有任务需要执行6. 伦理、合规与最佳实践在爬取政府公开数据时必须将合法合规与对目标网站的尊重放在首位。遵守robots.txt在爬取任何网站前首先访问其/robots.txt文件如https://data.ganzhou.gov.cn/robots.txt查看是否允许爬虫访问目标路径。即使没有明确禁止也应遵循善意爬取的准则。控制请求频率这是最重要的原则。在Downloader中设置delay_range如2-5秒模拟人类浏览速度避免对服务器造成瞬时高负载。对于政府网站建议将延迟设置得更高一些。设置合理的超时和重试使用retrying库和timeout参数避免因单次请求卡住而阻塞整个进程也避免因网络波动导致的数据丢失。使用识别性User-Agent在请求头中明确标识你的爬虫例如YourProjectName-Bot/1.0 (contactyour-email.com)。这既是礼貌也方便网站管理员在需要时联系你。只爬取公开数据绝对不要尝试绕过登录去获取非公开数据除非已获得明确授权。对于需要登录才能访问的数据确保你的爬取行为符合该平台的服务条款。缓存与增量更新对已爬取的数据进行记录如存储source_url和crawled_time下次运行时可以先检查数据是否已存在或已更新实现增量爬取避免重复请求。错误处理与日志完善的日志记录如使用Python的logging模块至关重要。记录下每次请求的URL、状态码、耗时以及解析失败的具体位置和原因。这能帮助你在网站改版后快速定位问题。数据使用目的将爬取的数据用于个人学习、研究或公益性的数据分析是通常可以接受的。但如果用于商业用途务必仔细研究相关网站的版权声明和数据使用政策。这个项目从设计到实现贯穿了分析、设计、编码、测试、部署的完整流程。它教会我的不仅是Python爬虫技术更是如何系统性地解决一个多变的、真实世界的问题。每个城市的数据平台就像一座堡垒你需要用不同的“钥匙”和“策略”去打开它。而构建一个可扩展的框架则让你在面对新的“堡垒”时能够快速装备高效攻坚。希望这份详细的设计与源码剖析能为你处理类似的异构数据源采集任务提供一个坚实的起点。本文还有配套的精品资源点击获取