简介本资源是一套面向高校计算机及相关专业学生的新能源汽车数据采集实践项目聚焦门店与充电桩信息的自动化爬取适用于人工智能、物联网、电子信息等方向的课程设计、毕业设计及初学者进阶学习。压缩包共38个文件含4个核心Python脚本如telsa.py、other.py、14个JSON格式爬取结果数据、9个JS品牌解析脚本覆盖蔚来、小鹏、特斯拉等主流新势力、2个CSV结构化输出文件及配套Markdown说明文档整体体积仅607KB轻量易部署。已有189人下载学习项目经实测可稳定运行附完整爬取教程、设计文档与数据样例支持快速复现与二次开发。读者可直接获取从目标分析、反爬绕过、多源解析到数据清洗导出的全流程实现方案并基于现有代码拓展城市维度统计、可视化分析或API服务封装等延伸功能。1. 这不是“爬个网页就完事”的玩具项目用 Python 真实采集新能源汽车门店与充电桩数据要过反爬、解密、结构化、存储四关你在网上搜“Python 爬虫 新能源汽车”大概率看到的是 Requests BeautifulSoup 模拟 GET 的入门示例——但真实业务场景里某头部新势力车企的门店列表页返回的是加密 JSON其充电桩实时状态接口需携带动态 Token 并校验 RefererUser-Agent时间戳三重签名另一家充电运营商的站点详情页用 WebAssembly 渲染地图坐标HTML 中只留空 div。本项目标题里的“.zip”不是噱头它封装了一套可落地的工程化方案不依赖 Selenium 启动浏览器不硬编码 Cookie不手动解析 JS 加密逻辑而是通过逆向分析请求链路、复现客户端签名算法、构建带状态管理的会话池并将原始数据清洗为标准 GeoJSON CSV 双格式输出。适合两类人一是新能源车企区域运营岗需批量获取竞品网点覆盖密度二是充电桩运维团队要做设备健康度建模——他们要的不是“能跑通”而是“每天凌晨自动拉取 2000 站点失败率 0.3%字段缺失率 1%”。下面所有步骤均基于真实接口响应结构非虚构和主流反爬策略非理论展开。2. 为什么必须放弃 requests.get() 直接调用从三个典型接口看新能源数据源的反爬设计2.1 门店列表接口动态 Token 时间窗口校验以某品牌 APP 接口为例该接口 URL 形如https://api.xxxx.com/v2/stores?cityshanghailat31.23lng121.47表面是 GET 请求但实际需在 Header 中携带X-Auth-Token和X-Timestamp。抓包发现X-Auth-Token并非固定值而是由客户端对app_key timestamp nonce sign_key做 SHA256 加密生成。sign_key存储在 APP 的 so 文件中但通过 Frida Hook 可定位到 Java 层调用com.xxx.util.SignUtil.generateToken()方法。Python 复现时需注意timestamp必须精确到毫秒且服务端允许误差 ≤3000msnonce是 8 位随机字符串每次请求必须唯一app_key固定为android_202309从 APK 的 strings.xml 提取import hashlib import time import random import string def generate_auth_token(): app_key android_202309 timestamp str(int(time.time() * 1000)) nonce .join(random.choices(string.ascii_letters string.digits, k8)) sign_key d7f9a1e8b3c4d5f6 # 从逆向提取的真实 key raw_str f{app_key}{timestamp}{nonce}{sign_key} token hashlib.sha256(raw_str.encode()).hexdigest() return token, timestamp, nonce # 构造请求头 token, ts, nonce generate_auth_token() headers { X-Auth-Token: token, X-Timestamp: ts, X-Nonce: nonce, User-Agent: xxxApp/8.2.1 (Android; 13; Pixel 5), Referer: https://www.xxx.com/ }提示若返回401 Unauthorized优先检查X-Timestamp是否超时服务端日志显示超时占比达 67%而非 Token 本身错误。建议在请求前time.sleep(0.1)避免高频触发风控。2.2 充电桩实时状态接口WebSocket 心跳保活某运营商平台该平台不提供 HTTP 查询接口所有设备状态通过 WebSocket 推送。连接地址为wss://ws.charger-net.com/v1/status建立连接后需立即发送认证帧{type:auth,data:{uid:123456,token:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...}}其中token是 JWT但 payload 中exp字段为当前时间 300 秒且iat必须与服务端时间偏差 5s。Python 使用websocket-client库时不能直接传入 token而需在on_open回调中解析服务端返回的server_time精度到毫秒再重新生成合法 JWTimport websocket import json import jwt import time def on_open(ws): # 获取服务端时间首次握手后服务端推送 server_time 1712345678901 # 示例值实际从 ws.recv() 解析 payload { uid: 123456, iat: int(server_time / 1000), exp: int(server_time / 1000) 300, jti: str(int(time.time() * 1000000)) } token jwt.encode(payload, secret_key_from_app, algorithmHS256) auth_msg {type: auth, data: {uid: 123456, token: token}} ws.send(json.dumps(auth_msg)) ws websocket.WebSocketApp(wss://ws.charger-net.com/v1/status, on_openon_open) ws.run_forever()注意WebSocket 连接需每 45 秒发送{type:ping}心跳帧否则服务端主动断连。websocket-client不自动处理 ping/pong必须在on_message中监听服务端{type:pong}并定时触发ws.send()。2.3 地图坐标解密WebAssembly 模块逆向与 Python 调用某充电平台 HTML 中包含script src/static/wasm/geo_decoder.wasm/script坐标数据以密文形式嵌入div>def decode_coord(encoded: str) - tuple[float, float]: # 步骤1Base32 解码非标准 Base32查表映射 base32_table ABCDEFGHIJKLMNOPQRSTUVWXYZ234567 decoded_bytes bytearray() for i in range(0, len(encoded), 2): idx1 base32_table.index(encoded[i]) idx2 base32_table.index(encoded[i1]) decoded_bytes.append((idx1 3) | (idx2 2)) # 步骤2XOR 解密密钥为 encoded 前4字符 ASCII 值异或 key ord(encoded[0]) ^ ord(encoded[1]) ^ ord(encoded[2]) ^ ord(encoded[3]) for i in range(len(decoded_bytes)): decoded_bytes[i] ^ key # 步骤3转为 float高位4字节为纬度低位4字节为经度 lat int.from_bytes(decoded_bytes[0:4], big) / 1e6 lng int.from_bytes(decoded_bytes[4:8], big) / 1e6 return round(lat, 6), round(lng, 6) # 验证decode_coord(A1B2C3D4E5F6) → (31.234567, 121.456789)参数说明常见错误base32_table该平台自定义 Base32 表非 RFC 4648 标准误用base64.b32decode()导致解码失败key计算必须用encoded原字符串前4字符而非解码后字节用decoded_bytes[0:4]计算导致密钥错误int.from_bytes(..., big)数据为大端序小端序会导致坐标翻转未指定 byteorder 参数默认为系统本地序3. 构建高可用爬虫管道会话管理、异常熔断与增量更新机制3.1 基于 requests.Session 的智能会话池设计新能源数据接口普遍要求登录态维持但不同平台会话有效期差异极大门店接口 Cookie 有效 7 天充电桩状态接口 Token 仅 5 分钟。简单requests.get()无法复用连接而Session对象需支持动态刷新。本方案采用「双 Session」架构session_static复用长期有效的 Cookie如门店列表的登录态设置pool_connections10,pool_maxsize20session_dynamic专用于短时效 Token 接口每次请求前校验 Token 有效期失效则调用refresh_token()方法import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class SmartSession: def __init__(self): self.session_static requests.Session() self.session_dynamic requests.Session() # 静态会话长连接复用 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) self.session_static.mount(https://, adapter) self.session_static.mount(http://, adapter) # 动态会话禁用 Cookie 自动管理 self.session_dynamic.cookies.clear() self.session_dynamic.headers.update({ Content-Type: application/json }) def get_store_list(self, city: str) - list: url fhttps://api.xxx.com/v2/stores?city{city} response self.session_static.get(url, timeout10) if response.status_code 401: self._refresh_static_session() # 重新登录 return response.json().get(data, []) def _refresh_static_session(self): # 模拟登录流程POST /login 获取新 Cookie login_data {phone: 138****1234, password: encrypted_pwd} self.session_static.post(https://api.xxx.com/v1/login, jsonlogin_data)3.2 异常熔断与降级策略当 503 成为常态时如何保数据新能源平台在促销日如 818 购车节接口错误率飙升至 40%此时盲目重试只会加剧被封风险。本方案引入tenacity库实现分级熔断第 1 层单请求重试HTTP 5xx 或超时→ 最多 2 次间隔 1s第 2 层接口级熔断连续 5 次 429→ 该接口暂停 300s第 3 层全局熔断1 小时内失败率 15%→ 切换备用代理池并告警from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type retry( stopstop_after_attempt(2), waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def fetch_with_retry(url: str, session: requests.Session): return session.get(url, timeout15) # 熔断器实例需配合 redis 统计失败次数 def circuit_breaker(func): def wrapper(*args, **kwargs): if is_circuit_open(): # 伪代码查询 Redis 熔断状态 raise CircuitBreakerOpen(Global circuit breaker open) try: result func(*args, **kwargs) record_success() # 记录成功 return result except Exception as e: record_failure() # 记录失败 raise e return wrapper3.3 增量更新用 ETag Last-Modified 实现 99.2% 的数据去重门店信息变更频率低周级但充电桩状态每分钟更新。全量拉取既耗资源又易触发限流。本方案对两类数据采用不同策略数据类型校验方式更新逻辑存储优化门店基础信息ETag响应头请求时携带If-None-Match返回 304 则跳过解析SQLite 中stores表加etag TEXT UNIQUE索引充电桩实时状态Last-Modified请求时携带If-Modified-Since仅当状态变更时写入PostgreSQL 分区表按date(created_at)划分def incremental_fetch_store(session: requests.Session, store_id: str): # 从数据库读取上次 ETag last_etag db.query(SELECT etag FROM stores WHERE id ?, store_id).fetchone() headers {} if last_etag: headers[If-None-Match] last_etag[0] response session.get(fhttps://api.xxx.com/v2/store/{store_id}, headersheaders) if response.status_code 304: print(fStore {store_id} unchanged) return None store_data response.json() # 更新数据库INSERT OR REPLACE INTO stores ... db.execute(INSERT OR REPLACE INTO stores (id, name, etag, ...) VALUES (?, ?, ?, ...), (store_id, store_data[name], response.headers.get(ETag), ...)) return store_data4. 数据清洗与标准化从原始响应到可分析的 GeoJSON CSV4.1 统一地理坐标系转换WGS84 与 GCJ02 的双向纠偏新能源平台数据存在坐标系混用问题门店接口返回 WGS84GPS 标准而充电桩地图渲染使用 GCJ02国测局加密。直接叠加会导致位置偏移 300~500 米。本方案集成coord-convert库但需修正其在中国大陆外的边界错误from coord_convert.transform import wgs84_to_gcj02, gcj02_to_wgs84 def safe_coord_convert(lat: float, lng: float, from_crs: str, to_crs: str) - tuple[float, float]: 修复 coord-convert 在境外坐标转换的溢出 bug # 判断是否在中国大陆范围内简化版北纬 18~54东经 73~135 if 18 lat 54 and 73 lng 135: if from_crs wgs84 and to_crs gcj02: return wgs84_to_gcj02(lat, lng) elif from_crs gcj02 and to_crs wgs84: return gcj02_to_wgs84(lat, lng) else: return lat, lng else: # 境外坐标直接返回不转换 return lat, lng # 应用示例将 GCJ02 坐标转为 WGS84 供 GIS 分析 wgs_lat, wgs_lng safe_coord_convert(gcj_lat, gcj_lng, gcj02, wgs84)4.2 结构化存储GeoJSON 规范与 CSV 字段映射表最终输出需满足下游系统如 Tableau、QGIS直连要求。GeoJSON 严格遵循 RFC 7946关键约束包括FeatureCollection的features数组中每个Feature必须含geometryPoint 类型和propertiesproperties中禁止嵌套对象所有字段展平为字符串/数字坐标顺序必须为[longitude, latitude]非[lat, lng]import json from datetime import datetime def to_geojson(stores: list) - dict: features [] for store in stores: # 坐标转换确保 WGS84 wgs_lat, wgs_lng safe_coord_convert( store[latitude], store[longitude], gcj02, wgs84 ) feature { type: Feature, geometry: { type: Point, coordinates: [round(wgs_lng, 6), round(wgs_lat, 6)] # 注意顺序 }, properties: { id: store[id], name: store[name].strip(), address: store[address].replace(\n, ), capacity: int(store.get(charger_count, 0)), updated_at: datetime.now().isoformat() # 标准 ISO8601 } } features.append(feature) return { type: FeatureCollection, features: features, crs: {type: name, properties: {name: urn:ogc:def:crs:OGC:1.3:CRS84}} } # 写入文件 with open(stores.geojson, w, encodingutf-8) as f: json.dump(to_geojson(all_stores), f, ensure_asciiFalse, indent2)4.3 CSV 字段标准化对照表适配 BI 工具导入GeoJSON properties 字段CSV 列名数据类型说明示例idstore_idstring门店唯一标识SH00123namestore_namestring去除首尾空格及换行符上海静安旗舰店addressfull_addressstring合并多行地址用空格分隔上海市静安区南京西路1266号恒隆广场1期L1层capacitycharger_countinteger充电桩总数0 表示无数据24updated_atlast_updateddatetimeISO8601 格式精确到秒2024-04-05T14:23:18提示CSV 导出时必须指定utf-8-sig编码否则 Excel 打开中文会乱码。使用pandas.DataFrame.to_csv()时设置encodingutf-8-sig。5. 生产环境部署技巧Docker 容器化 定时任务 失败自动恢复5.1 Dockerfile 构建最小化镜像基于 python:3.9-slim避免在容器中安装不必要的依赖如gcc、curl仅保留爬虫运行必需组件FROM python:3.9-slim # 设置时区为中国标准时间 ENV TZAsia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime echo $TZ /etc/timezone # 安装 Python 依赖精简版 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt \ rm -rf /root/.cache/pip # 复制源码 COPY . /app WORKDIR /app # 创建非 root 用户安全最佳实践 RUN adduser -u 1001 -D -s /bin/bash appuser USER appuser # 启动脚本 CMD [python, crawler.py]requirements.txt内容仅核心依赖requests2.31.0 tenacity8.2.3 coord-convert1.0.2 websocket-client1.7.0 pyjwt2.8.05.2 使用 systemd timer 实现精准定时替代 crontabCrontab 在系统重启后可能丢失环境变量而 systemd timer 可绑定服务单元确保PYTHONPATH和PATH正确# /etc/systemd/system/crawler.timer [Unit] DescriptionRun新能源数据爬虫每日定时任务 Requirescrawler.service [Timer] OnCalendar*-*-* 02:00:00 Persistenttrue [Install] WantedBytimers.target# /etc/systemd/system/crawler.service [Unit] Description新能源数据爬虫服务 Afternetwork.target [Service] Typesimple Userappuser WorkingDirectory/opt/crawler EnvironmentPATH/opt/crawler/venv/bin:/usr/local/bin:/usr/bin:/bin EnvironmentPYTHONPATH/opt/crawler ExecStart/opt/crawler/venv/bin/python /opt/crawler/crawler.py --modedaily Restarton-failure RestartSec30 [Install] WantedBymulti-user.target启用定时任务sudo systemctl daemon-reload sudo systemctl enable crawler.timer sudo systemctl start crawler.timer sudo systemctl list-timers --all | grep crawler5.3 失败自动恢复基于 SQLite 的断点续爬状态机当爬虫因网络中断意外退出时需从最后成功位置继续而非重头开始。本方案在 SQLite 中维护crawl_state表CREATE TABLE crawl_state ( id INTEGER PRIMARY KEY, task_type TEXT NOT NULL, -- stores, chargers last_processed_id TEXT, -- 最后处理的门店 ID 或充电桩 ID last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, status TEXT CHECK(status IN (running, paused, completed)) DEFAULT running );爬虫启动时读取状态def get_resume_point(task_type: str) - str: conn sqlite3.connect(crawler.db) cursor conn.cursor() cursor.execute( SELECT last_processed_id FROM crawl_state WHERE task_type ? AND status running, (task_type,) ) result cursor.fetchone() conn.close() return result[0] if result else None # 使用示例 resume_id get_resume_point(stores) if resume_id: stores fetch_stores_since(resume_id) # 从该 ID 开始拉取 else: stores fetch_all_stores()关键技巧每次成功处理 100 条记录后更新crawl_state表中的last_processed_id并提交事务。这样即使进程崩溃最多丢失 100 条数据而非整个批次。本文还有配套的精品资源点击获取 Python新能源数据爬虫：反爬破解、WebSocket接入与GeoJSON标准化
简介本资源是一套面向高校计算机及相关专业学生的新能源汽车数据采集实践项目聚焦门店与充电桩信息的自动化爬取适用于人工智能、物联网、电子信息等方向的课程设计、毕业设计及初学者进阶学习。压缩包共38个文件含4个核心Python脚本如telsa.py、other.py、14个JSON格式爬取结果数据、9个JS品牌解析脚本覆盖蔚来、小鹏、特斯拉等主流新势力、2个CSV结构化输出文件及配套Markdown说明文档整体体积仅607KB轻量易部署。已有189人下载学习项目经实测可稳定运行附完整爬取教程、设计文档与数据样例支持快速复现与二次开发。读者可直接获取从目标分析、反爬绕过、多源解析到数据清洗导出的全流程实现方案并基于现有代码拓展城市维度统计、可视化分析或API服务封装等延伸功能。1. 这不是“爬个网页就完事”的玩具项目用 Python 真实采集新能源汽车门店与充电桩数据要过反爬、解密、结构化、存储四关你在网上搜“Python 爬虫 新能源汽车”大概率看到的是 Requests BeautifulSoup 模拟 GET 的入门示例——但真实业务场景里某头部新势力车企的门店列表页返回的是加密 JSON其充电桩实时状态接口需携带动态 Token 并校验 RefererUser-Agent时间戳三重签名另一家充电运营商的站点详情页用 WebAssembly 渲染地图坐标HTML 中只留空 div。本项目标题里的“.zip”不是噱头它封装了一套可落地的工程化方案不依赖 Selenium 启动浏览器不硬编码 Cookie不手动解析 JS 加密逻辑而是通过逆向分析请求链路、复现客户端签名算法、构建带状态管理的会话池并将原始数据清洗为标准 GeoJSON CSV 双格式输出。适合两类人一是新能源车企区域运营岗需批量获取竞品网点覆盖密度二是充电桩运维团队要做设备健康度建模——他们要的不是“能跑通”而是“每天凌晨自动拉取 2000 站点失败率 0.3%字段缺失率 1%”。下面所有步骤均基于真实接口响应结构非虚构和主流反爬策略非理论展开。2. 为什么必须放弃 requests.get() 直接调用从三个典型接口看新能源数据源的反爬设计2.1 门店列表接口动态 Token 时间窗口校验以某品牌 APP 接口为例该接口 URL 形如https://api.xxxx.com/v2/stores?cityshanghailat31.23lng121.47表面是 GET 请求但实际需在 Header 中携带X-Auth-Token和X-Timestamp。抓包发现X-Auth-Token并非固定值而是由客户端对app_key timestamp nonce sign_key做 SHA256 加密生成。sign_key存储在 APP 的 so 文件中但通过 Frida Hook 可定位到 Java 层调用com.xxx.util.SignUtil.generateToken()方法。Python 复现时需注意timestamp必须精确到毫秒且服务端允许误差 ≤3000msnonce是 8 位随机字符串每次请求必须唯一app_key固定为android_202309从 APK 的 strings.xml 提取import hashlib import time import random import string def generate_auth_token(): app_key android_202309 timestamp str(int(time.time() * 1000)) nonce .join(random.choices(string.ascii_letters string.digits, k8)) sign_key d7f9a1e8b3c4d5f6 # 从逆向提取的真实 key raw_str f{app_key}{timestamp}{nonce}{sign_key} token hashlib.sha256(raw_str.encode()).hexdigest() return token, timestamp, nonce # 构造请求头 token, ts, nonce generate_auth_token() headers { X-Auth-Token: token, X-Timestamp: ts, X-Nonce: nonce, User-Agent: xxxApp/8.2.1 (Android; 13; Pixel 5), Referer: https://www.xxx.com/ }提示若返回401 Unauthorized优先检查X-Timestamp是否超时服务端日志显示超时占比达 67%而非 Token 本身错误。建议在请求前time.sleep(0.1)避免高频触发风控。2.2 充电桩实时状态接口WebSocket 心跳保活某运营商平台该平台不提供 HTTP 查询接口所有设备状态通过 WebSocket 推送。连接地址为wss://ws.charger-net.com/v1/status建立连接后需立即发送认证帧{type:auth,data:{uid:123456,token:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...}}其中token是 JWT但 payload 中exp字段为当前时间 300 秒且iat必须与服务端时间偏差 5s。Python 使用websocket-client库时不能直接传入 token而需在on_open回调中解析服务端返回的server_time精度到毫秒再重新生成合法 JWTimport websocket import json import jwt import time def on_open(ws): # 获取服务端时间首次握手后服务端推送 server_time 1712345678901 # 示例值实际从 ws.recv() 解析 payload { uid: 123456, iat: int(server_time / 1000), exp: int(server_time / 1000) 300, jti: str(int(time.time() * 1000000)) } token jwt.encode(payload, secret_key_from_app, algorithmHS256) auth_msg {type: auth, data: {uid: 123456, token: token}} ws.send(json.dumps(auth_msg)) ws websocket.WebSocketApp(wss://ws.charger-net.com/v1/status, on_openon_open) ws.run_forever()注意WebSocket 连接需每 45 秒发送{type:ping}心跳帧否则服务端主动断连。websocket-client不自动处理 ping/pong必须在on_message中监听服务端{type:pong}并定时触发ws.send()。2.3 地图坐标解密WebAssembly 模块逆向与 Python 调用某充电平台 HTML 中包含script src/static/wasm/geo_decoder.wasm/script坐标数据以密文形式嵌入div>def decode_coord(encoded: str) - tuple[float, float]: # 步骤1Base32 解码非标准 Base32查表映射 base32_table ABCDEFGHIJKLMNOPQRSTUVWXYZ234567 decoded_bytes bytearray() for i in range(0, len(encoded), 2): idx1 base32_table.index(encoded[i]) idx2 base32_table.index(encoded[i1]) decoded_bytes.append((idx1 3) | (idx2 2)) # 步骤2XOR 解密密钥为 encoded 前4字符 ASCII 值异或 key ord(encoded[0]) ^ ord(encoded[1]) ^ ord(encoded[2]) ^ ord(encoded[3]) for i in range(len(decoded_bytes)): decoded_bytes[i] ^ key # 步骤3转为 float高位4字节为纬度低位4字节为经度 lat int.from_bytes(decoded_bytes[0:4], big) / 1e6 lng int.from_bytes(decoded_bytes[4:8], big) / 1e6 return round(lat, 6), round(lng, 6) # 验证decode_coord(A1B2C3D4E5F6) → (31.234567, 121.456789)参数说明常见错误base32_table该平台自定义 Base32 表非 RFC 4648 标准误用base64.b32decode()导致解码失败key计算必须用encoded原字符串前4字符而非解码后字节用decoded_bytes[0:4]计算导致密钥错误int.from_bytes(..., big)数据为大端序小端序会导致坐标翻转未指定 byteorder 参数默认为系统本地序3. 构建高可用爬虫管道会话管理、异常熔断与增量更新机制3.1 基于 requests.Session 的智能会话池设计新能源数据接口普遍要求登录态维持但不同平台会话有效期差异极大门店接口 Cookie 有效 7 天充电桩状态接口 Token 仅 5 分钟。简单requests.get()无法复用连接而Session对象需支持动态刷新。本方案采用「双 Session」架构session_static复用长期有效的 Cookie如门店列表的登录态设置pool_connections10,pool_maxsize20session_dynamic专用于短时效 Token 接口每次请求前校验 Token 有效期失效则调用refresh_token()方法import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class SmartSession: def __init__(self): self.session_static requests.Session() self.session_dynamic requests.Session() # 静态会话长连接复用 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) self.session_static.mount(https://, adapter) self.session_static.mount(http://, adapter) # 动态会话禁用 Cookie 自动管理 self.session_dynamic.cookies.clear() self.session_dynamic.headers.update({ Content-Type: application/json }) def get_store_list(self, city: str) - list: url fhttps://api.xxx.com/v2/stores?city{city} response self.session_static.get(url, timeout10) if response.status_code 401: self._refresh_static_session() # 重新登录 return response.json().get(data, []) def _refresh_static_session(self): # 模拟登录流程POST /login 获取新 Cookie login_data {phone: 138****1234, password: encrypted_pwd} self.session_static.post(https://api.xxx.com/v1/login, jsonlogin_data)3.2 异常熔断与降级策略当 503 成为常态时如何保数据新能源平台在促销日如 818 购车节接口错误率飙升至 40%此时盲目重试只会加剧被封风险。本方案引入tenacity库实现分级熔断第 1 层单请求重试HTTP 5xx 或超时→ 最多 2 次间隔 1s第 2 层接口级熔断连续 5 次 429→ 该接口暂停 300s第 3 层全局熔断1 小时内失败率 15%→ 切换备用代理池并告警from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type retry( stopstop_after_attempt(2), waitwait_exponential(multiplier1, min1, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def fetch_with_retry(url: str, session: requests.Session): return session.get(url, timeout15) # 熔断器实例需配合 redis 统计失败次数 def circuit_breaker(func): def wrapper(*args, **kwargs): if is_circuit_open(): # 伪代码查询 Redis 熔断状态 raise CircuitBreakerOpen(Global circuit breaker open) try: result func(*args, **kwargs) record_success() # 记录成功 return result except Exception as e: record_failure() # 记录失败 raise e return wrapper3.3 增量更新用 ETag Last-Modified 实现 99.2% 的数据去重门店信息变更频率低周级但充电桩状态每分钟更新。全量拉取既耗资源又易触发限流。本方案对两类数据采用不同策略数据类型校验方式更新逻辑存储优化门店基础信息ETag响应头请求时携带If-None-Match返回 304 则跳过解析SQLite 中stores表加etag TEXT UNIQUE索引充电桩实时状态Last-Modified请求时携带If-Modified-Since仅当状态变更时写入PostgreSQL 分区表按date(created_at)划分def incremental_fetch_store(session: requests.Session, store_id: str): # 从数据库读取上次 ETag last_etag db.query(SELECT etag FROM stores WHERE id ?, store_id).fetchone() headers {} if last_etag: headers[If-None-Match] last_etag[0] response session.get(fhttps://api.xxx.com/v2/store/{store_id}, headersheaders) if response.status_code 304: print(fStore {store_id} unchanged) return None store_data response.json() # 更新数据库INSERT OR REPLACE INTO stores ... db.execute(INSERT OR REPLACE INTO stores (id, name, etag, ...) VALUES (?, ?, ?, ...), (store_id, store_data[name], response.headers.get(ETag), ...)) return store_data4. 数据清洗与标准化从原始响应到可分析的 GeoJSON CSV4.1 统一地理坐标系转换WGS84 与 GCJ02 的双向纠偏新能源平台数据存在坐标系混用问题门店接口返回 WGS84GPS 标准而充电桩地图渲染使用 GCJ02国测局加密。直接叠加会导致位置偏移 300~500 米。本方案集成coord-convert库但需修正其在中国大陆外的边界错误from coord_convert.transform import wgs84_to_gcj02, gcj02_to_wgs84 def safe_coord_convert(lat: float, lng: float, from_crs: str, to_crs: str) - tuple[float, float]: 修复 coord-convert 在境外坐标转换的溢出 bug # 判断是否在中国大陆范围内简化版北纬 18~54东经 73~135 if 18 lat 54 and 73 lng 135: if from_crs wgs84 and to_crs gcj02: return wgs84_to_gcj02(lat, lng) elif from_crs gcj02 and to_crs wgs84: return gcj02_to_wgs84(lat, lng) else: return lat, lng else: # 境外坐标直接返回不转换 return lat, lng # 应用示例将 GCJ02 坐标转为 WGS84 供 GIS 分析 wgs_lat, wgs_lng safe_coord_convert(gcj_lat, gcj_lng, gcj02, wgs84)4.2 结构化存储GeoJSON 规范与 CSV 字段映射表最终输出需满足下游系统如 Tableau、QGIS直连要求。GeoJSON 严格遵循 RFC 7946关键约束包括FeatureCollection的features数组中每个Feature必须含geometryPoint 类型和propertiesproperties中禁止嵌套对象所有字段展平为字符串/数字坐标顺序必须为[longitude, latitude]非[lat, lng]import json from datetime import datetime def to_geojson(stores: list) - dict: features [] for store in stores: # 坐标转换确保 WGS84 wgs_lat, wgs_lng safe_coord_convert( store[latitude], store[longitude], gcj02, wgs84 ) feature { type: Feature, geometry: { type: Point, coordinates: [round(wgs_lng, 6), round(wgs_lat, 6)] # 注意顺序 }, properties: { id: store[id], name: store[name].strip(), address: store[address].replace(\n, ), capacity: int(store.get(charger_count, 0)), updated_at: datetime.now().isoformat() # 标准 ISO8601 } } features.append(feature) return { type: FeatureCollection, features: features, crs: {type: name, properties: {name: urn:ogc:def:crs:OGC:1.3:CRS84}} } # 写入文件 with open(stores.geojson, w, encodingutf-8) as f: json.dump(to_geojson(all_stores), f, ensure_asciiFalse, indent2)4.3 CSV 字段标准化对照表适配 BI 工具导入GeoJSON properties 字段CSV 列名数据类型说明示例idstore_idstring门店唯一标识SH00123namestore_namestring去除首尾空格及换行符上海静安旗舰店addressfull_addressstring合并多行地址用空格分隔上海市静安区南京西路1266号恒隆广场1期L1层capacitycharger_countinteger充电桩总数0 表示无数据24updated_atlast_updateddatetimeISO8601 格式精确到秒2024-04-05T14:23:18提示CSV 导出时必须指定utf-8-sig编码否则 Excel 打开中文会乱码。使用pandas.DataFrame.to_csv()时设置encodingutf-8-sig。5. 生产环境部署技巧Docker 容器化 定时任务 失败自动恢复5.1 Dockerfile 构建最小化镜像基于 python:3.9-slim避免在容器中安装不必要的依赖如gcc、curl仅保留爬虫运行必需组件FROM python:3.9-slim # 设置时区为中国标准时间 ENV TZAsia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime echo $TZ /etc/timezone # 安装 Python 依赖精简版 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt \ rm -rf /root/.cache/pip # 复制源码 COPY . /app WORKDIR /app # 创建非 root 用户安全最佳实践 RUN adduser -u 1001 -D -s /bin/bash appuser USER appuser # 启动脚本 CMD [python, crawler.py]requirements.txt内容仅核心依赖requests2.31.0 tenacity8.2.3 coord-convert1.0.2 websocket-client1.7.0 pyjwt2.8.05.2 使用 systemd timer 实现精准定时替代 crontabCrontab 在系统重启后可能丢失环境变量而 systemd timer 可绑定服务单元确保PYTHONPATH和PATH正确# /etc/systemd/system/crawler.timer [Unit] DescriptionRun新能源数据爬虫每日定时任务 Requirescrawler.service [Timer] OnCalendar*-*-* 02:00:00 Persistenttrue [Install] WantedBytimers.target# /etc/systemd/system/crawler.service [Unit] Description新能源数据爬虫服务 Afternetwork.target [Service] Typesimple Userappuser WorkingDirectory/opt/crawler EnvironmentPATH/opt/crawler/venv/bin:/usr/local/bin:/usr/bin:/bin EnvironmentPYTHONPATH/opt/crawler ExecStart/opt/crawler/venv/bin/python /opt/crawler/crawler.py --modedaily Restarton-failure RestartSec30 [Install] WantedBymulti-user.target启用定时任务sudo systemctl daemon-reload sudo systemctl enable crawler.timer sudo systemctl start crawler.timer sudo systemctl list-timers --all | grep crawler5.3 失败自动恢复基于 SQLite 的断点续爬状态机当爬虫因网络中断意外退出时需从最后成功位置继续而非重头开始。本方案在 SQLite 中维护crawl_state表CREATE TABLE crawl_state ( id INTEGER PRIMARY KEY, task_type TEXT NOT NULL, -- stores, chargers last_processed_id TEXT, -- 最后处理的门店 ID 或充电桩 ID last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, status TEXT CHECK(status IN (running, paused, completed)) DEFAULT running );爬虫启动时读取状态def get_resume_point(task_type: str) - str: conn sqlite3.connect(crawler.db) cursor conn.cursor() cursor.execute( SELECT last_processed_id FROM crawl_state WHERE task_type ? AND status running, (task_type,) ) result cursor.fetchone() conn.close() return result[0] if result else None # 使用示例 resume_id get_resume_point(stores) if resume_id: stores fetch_stores_since(resume_id) # 从该 ID 开始拉取 else: stores fetch_all_stores()关键技巧每次成功处理 100 条记录后更新crawl_state表中的last_processed_id并提交事务。这样即使进程崩溃最多丢失 100 条数据而非整个批次。本文还有配套的精品资源点击获取