最近帮某大宗商品研究团队重构了煤炭数据采集流程，爬取金投网上的煤炭价格、库存、运价指数等内容。这个项目的难点不在“能不能爬到”，而在“怎么长期稳定地爬到”，以及“爬下来的数据能不能直接用”。如果你只是临时抓一次数据做分析，那随便写个脚本就够了；但如果你的下游是日报、周报、月度复盘，甚至是自动化的数据推送服务，那这套逻辑就要重新设计。
这个“精进版”项目就是想解决三件事：一是把单次爬虫变成可持续运行的采集任务；二是让采集到的脏数据在入库前就被清洗干净；三是在数据源反爬策略调整、字段结构变化时，系统能快速适应而不是原地崩溃。下面我把整个项目的设计思路、关键代码和踩坑经历都拆开来讲，适合已经写过简单爬虫、想往工程化方向走的开发者参考。
1. 精进版要解决的核心问题：从单页抓取到全链路采集
1.1 为什么是煤炭行情数据
煤炭行业的数据有一个特点：公开渠道多、口径不统一、更新节奏差异大。金投网这类行情站点把动力煤、焦煤、焦炭等品种的现货报价、期货价格、港口库存、海运运价指数聚合到一起，页面结构清晰，很适合做数据源。但这些数据的价值在于“连续观测”，断一天数据，趋势分析就出现缺口。
团队需要的不是某一天的静态快照，而是每天定时、自动、完整地采集，并和前一天的数据做增量对比。这决定了项目不能是一个一次性的requests脚本，而是一个带有调度、存储、校验、告警的小型系统。
1.2 精进版与基础版的差异
很多人写爬虫时习惯把“下载页面、解析字段、打印结果”三步写在同一个文件里，跑完就算结束。基础版能应付单次任务，但一旦涉及长期运行，会暴露出一堆问题：
- 没有重试机制，网络抖动一次，当天数据就缺失
- 没有去重逻辑，重复抓取导致库里出现大量重复记录
- 没有频率控制，短时间高频请求容易被数据源限制访问
- 没有字段校验，解析到空值、错位数据时，代码不报错但数据已经污染了
- 没有运行日志，出了问题都不知道是哪个环节出的问题
精进版的核心工作，就是把这些“基础版没考虑的事”全部补上。整体架构可以拆成四层：调度层、下载层、解析层、存储层。每层只干自己那一件事，层与层之间通过任务队列解耦，这样任何一层出问题时都可以单独替换，不至于牵一发动全身。
2. 数据源分析与请求策略：先搞清对手再动手
2.1 页面结构与数据加载方式
动手写代码之前，先花一天时间把数据源的页面结构摸清楚。煤炭行情页面一般分两类：一类是服务端直接渲染的静态表格，HTML里就带全部数据；另一类是前端通过接口异步加载，页面上的数据是JavaScript跑出来的。两种方式的采集策略完全不同。
金投网的煤炭板块大部分是服务端渲染，一个列表页对应一个URL，参数里带品种、日期、页码。这种结构的优点是不需要模拟浏览器，一个带正确请求头的requests就能拿到完整HTML；缺点是一旦网站改版，表格列的顺序变了，解析逻辑就要跟着改。
我的建议是：第一次接触一个新数据源时，先手动用浏览器打开页面，查看源代码，确认数据是直接写在HTML里还是通过接口返回的。如果是接口返回的，去Network面板里找真正的数据接口，解析JSON比解析HTML稳得多。这一步决定了后续所有代码的写法，值得多花时间。
2.2 请求头与访问频率设计
爬虫被限制访问，多数时候不是因为你“爬了”，而是因为你“爬得太像程序了”。默认的Python-requests UA一眼就会被识别，所以请求头至少需要伪装成常用浏览器的样子。我维护了一个UA池，每次请求随机取一个，比固定UA更自然：
import random USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/121.0 Firefox/121.0", ] def random_headers(): return { "User-Agent": random.choice(USER_AGENTS), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": "https://www.gold678.com/", "Connection": "keep-alive", }再说频率控制。很多新手踩过的坑是：用线程池一口气发50个请求，数据源还没反应过来就触发了风控。精进版的下载器引入了一个信号量，把并发数控制在合理范围内，同时每次请求之间加一个随机小延迟：
import threading import time SEMAPHORE = threading.Semaphore(5) # 最多5个并发 def throttled_request(session, url, delay_range=(1, 3)): with SEMAPHORE: time.sleep(random.uniform(*delay_range)) resp = session.get(url, headers=random_headers(), timeout=15) resp.raise_for_status() return resp这里有个细节：信号量放在请求之前而不是请求之后，是为了防止并发请求同时卡在timeout等待上时，新请求仍然涌入。实测下来，5并发、1到3秒随机延迟的组合，对于数据量不大的行情页来说足够稳定，既不会触发风控，也完全能满足每日采集的时效要求。
2.3 代理策略的取舍
关于代理，很多人一上来就想上代理池，觉得这样才“高级”。实际做行情数据采集时，代理不是必需品。因为这种公开行情页面本身价值密度不算高，数据源对单IP的请求频率限制通常不会特别苛刻，只要频率控制得当，一个干净的住宅IP就够了。
但有两类情况确实需要代理：一是数据源对单IP的日请求次数做了硬上限，比如一天只放行1000次请求，你的采集任务超过这个量就必须换IP；二是你采集的目标数据覆盖多个站点，主IP因为其他站点的原因被连坐限制。我现在这套系统预留了代理接口，但默认走直连，只有出现HTTP 403或429时才动态切换到备用代理。这个“按需启用”的思路可以避免大部分时候都背着代理的额外延迟和失败率。
3. 抓取流程的分层设计：调度、下载、解析、存储
3.1 任务调度层：增量更新与全量补采
调度层是整个系统的“指挥中心”，负责决定“什么时间抓什么页面”。煤炭数据的更新节奏基本是工作日每日更新，个别品种甚至是一周更新一次。设计调度时不能只写死一个cron表达式，因为周末、节假日数据源可能不更新，而每天需要抓的页面数量又会随品种数量变化。
我采用一个最简单的方案：维护一份抓取任务清单，每条任务记录包含任务类型、URL模板、参数列表。调度器每天定时扫描任务清单，哪些任务该在今天执行，就动态生成具体的抓取URL去下载层排队。这样加一个新品种只需要往任务清单里加一行记录，不用改代码。
增量更新的逻辑是：默认每天抓取当日数据；如果发现当日数据为空或者数据源返回了“暂无数据”的提示，就自动回退到前一个工作日进行补采。补采的深度最多回退5个工作日，超过这个深度就触发告警，让负责人决定是否要手动核查数据源是不是改版了。
3.2 下载层：请求重试、超时控制
下载层负责“把页面拿到手”，但它真正的难点不是请求本身，而是怎么处理异常。网络抖动、数据源临时故障、偶发的429限流，这些都是正常现象。我的下载器对所有请求做两层重试：第一层是网络异常重试，比如连接超时、ReadTimeout、ConnectionError，这类错误通常过几秒就能恢复；第二层是HTTP状态码重试，遇到502、503、504这类服务器临时错误时，退避一段时间后再试。
重试必须配合退避策略，否则就是变相给数据源制造压力。我用的是一种指数退避加随机抖动的方式：
def fetch_with_retry(session, url, retries=3): for attempt in range(retries): try: resp = throttled_request(session, url) if resp.status_code == 200: return resp.text elif resp.status_code in (403, 429): raise RateLimitError(f"HTTP {resp.status_code}") else: raise HTTPError(f"Unexpected status: {resp.status_code}") except (RateLimitError, HTTPError, requests.RequestException) as e: wait_time = (2 ** attempt) + random.uniform(0, 1) time.sleep(wait_time) return None这里有个容易被忽略的点：重试时不要直接复用同一个Session，因为有些连接已经被数据源标记了异常状态。我在重试逻辑里每次都会新建Session连接，虽然多了一点握手开销，但成功率明显提升。对于每日几百个页面的采集量，这点开销完全可以接受。
3.3 解析层：表格解析与字段校验
解析层是把HTML转成结构化数据的关键环节。煤炭行情页面大量使用表格，而表格解析最怕的是“列错位”。数据源偶尔会把某些列合并、隐藏，或者在表头新增一列，直接按固定索引取值就会出错。
我用Python的BeautifulSoup加lxml解析器来处理，同时维护一个“表头到字段名”的映射表。解析时先确认页面的表头顺序，再按表头映射到内部字段，而不是写死第几列代表价格：
from bs4 import BeautifulSoup FIELD_MAP = { "日期": "date", "品种": "variety", "价格": "price", "涨跌": "change", "涨跌幅": "change_pct", "单位": "unit", } def parse_price_table(html): soup = BeautifulSoup(html, "lxml") table = soup.find("table") header_row = table.find("tr") headers = [th.get_text(strip=True) for th in header_row.find_all(["th", "td"])] rows = [] for tr in table.find_all("tr")[1:]: cells = [td.get_text(strip=True) for td in tr.find_all("td")] row = {} for idx, header in enumerate(headers): if header in FIELD_MAP and idx < len(cells): row[FIELD_MAP[header]] = cells[idx] rows.append(row) return rows解析之后不能直接入库，要先做字段校验。我加了三个维度的检查：
- 空值检查：核心字段（日期、品种、价格）为空时，该行标记为异常，不写入正式表
- 类型检查：价格字段必须是数值型，遇到“暂无”“--”这类非数值内容，转换为空值并记录日志
- 范围检查：煤炭价格不可能一夜之间从500涨到5000，所以每个品种都配置了合理的波动范围，超出范围的数值会被标记为可疑数据，进入人工复核队列
这三个检查的价格在于：它们不是“防止爬虫出错”，而是“防止脏数据污染下游”。下游分析师拿到的数据必须是干净的，否则一个小数点错误就可能让一份研报的结论全错。
3.4 存储层：去重策略与库表设计
存储层我选择MySQL，主要是为了后续对接BI工具和定时报表方便。表结构的设计要考虑到“同一品种同一日期只能有一条记录”，所以建表时直接加唯一索引：
CREATE TABLE coal_daily_price ( id INT AUTO_INCREMENT PRIMARY KEY, variety VARCHAR(32) NOT NULL, trade_date DATE NOT NULL, price DECIMAL(10,2), change_value DECIMAL(10,2), change_pct DECIMAL(5,2), unit VARCHAR(16), source VARCHAR(64), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_variety_date (variety, trade_date) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;入库时用INSERT ... ON DUPLICATE KEY UPDATE而不是先查再插，这样既保证幂等性，又省掉一次查询：
def upsert_rows(cursor, rows): sql = """ INSERT INTO coal_daily_price (variety, trade_date, price, change_value, change_pct, unit, source) VALUES (%(variety)s, %(trade_date)s, %(price)s, %(change_value)s, %(change_pct)s, %(unit)s, %(source)s) ON DUPLICATE KEY UPDATE price = VALUES(price), change_value = VALUES(change_value), change_pct = VALUES(change_pct), unit = VALUES(unit), source = VALUES(source) """ for row in rows: cursor.execute(sql, row)其实这里有一个设计上的取舍值得说说：为什么用唯一索引做幂等，而不是先DELETE再INSERT？因为先删后插在数据源当天数据延迟更新时会导致历史记录被清空，而我们希望保留“数据源最后一次更新”的状态。用唯一索引加覆盖写的方式，可以保证每天都只保留最新状态，同时磁盘上不会堆积大量重复记录。
4. 核心代码实现：调度、解析与入库的关键片段
4.1 定时调度与任务清单
调度器用APScheduler实现，每天上午9点半和下午5点各触发一次任务扫描。为什么要两个时间点？因为煤炭现货报价有的品种上午出，有的品种下午才更新，分两个批次可以减少“抓不到数据”的概率。
任务清单我直接放在了数据库里，这样加任务不用改代码：
SELECT task_name, url_template, params_json, schedule_type FROM crawl_task_config WHERE is_active = 1 AND next_run_time >= NOW() OR next_run_time IS NULL每次调度触发时，系统扫描所有需要执行的任务，生成具体的URL列表，然后交给下载层去异步抓取。抓取完成后，无论成功还是失败，每条任务都会记录下一次执行时间，同时更新成功次数和失败次数的统计。如果某个任务连续失败超过3天，调度器会把它标记为“异常”，暂停后续自动执行，并把异常信息推送到告警群。
4.2 字段清洗与多表关联
解析出来的原始字段经常不能直接用，比如“价格”字段在页面上显示成“现价 780 元/吨”，或者日期显示成“01-20”而不是完整的“2025-01-20”。我写了一个清洗模块，专门做三件事：提取数值、标准化日期、统一单位。
数值提取用正则表达式解决：
import re def extract_number(text): if not text: return None match = re.search(r"-?\d+(\.\d+)?", text.replace(",", "")) if match: return float(match.group()) return None日期标准化要考虑两种格式：一种是带年份的完整日期，另一种是只有月和日的简称。对于后者，需要根据当前日期推断年份，同时处理“跨年”边界问题，比如12月31日的简称日期到了1月1日，年份应该还是上一年：
from datetime import datetime, timedelta def normalize_date(text, today=None): today = today or datetime.now().date() cleaned = text.strip() try: return datetime.strptime(cleaned, "%Y-%m-%d").date() except ValueError: pass try: m, d = cleaned.split("-") candidate = datetime(today.year, int(m), int(d)).date() if candidate - today > timedelta(days=90): candidate = datetime(today.year - 1, int(m), int(d)).date() return candidate except ValueError: return None这个“跨年推断”的逻辑其实是个很容易踩的坑。如果你不考虑跨年，每年1月初你会发现系统里突然出现一批日期错误的数据，而且这种错很难被普通校验发现，因为日期格式本身是合法的。
4.3 多线程下载与入库性能
下载和解析可以做到并行，但入库必须串行，因为MySQL不支持多线程同时写同一张表还能保证唯一索引不出冲突。我的做法是：下载和解析用ThreadPoolExecutor并行处理，解析结果放进一个内存队列，再单独开一个线程从队列里批量取数据、批量入库。
批量入库的批次大小我调到500条一批，实测比一条一条插快了近10倍。对于煤炭数据这个量级，500条一批完全不会给数据库带来压力。
5. 踩坑实录与排查链路：从403到脏数据的完整复盘
5.1 偶发403：从User-Agent到IP限流
系统上线第三周的某天，运维反馈当天的采集任务成功率为0，所有请求都返回403。我第一反应是UA被识别了，于是赶紧换了新的UA池，结果仍然403。接着查了代理IP的可用性，也正常。最后打开数据源页面，才发现是数据源整体升级了防护策略，对所有非浏览器请求都返回403，连正常访问偶尔也会要求验证。
这个问题的排查链路是这样的：先看HTTP状态码分布，确认是全部失败还是部分失败；如果是全部失败，考虑是不是IP被全局封禁；排除IP因素后，再看是不是UA、Cookie这些请求头出了问题。最终解决方式是增加了Cookie的自动获取流程，也就是先访问一次首页拿到必要的Cookie再带着去访问数据页，模拟真实用户的访问路径。这比单纯伪造UA要有效得多。
5.2 请求频率被限制：并发不是越高越好
另一个印象深刻的坑是：我把并发从5调到10之后，原本跑得好好的任务突然开始大面积超时。查下来发现，数据源对单IP的并发连接数做了限制，超过之后新的连接会被挂起，直到超时。
这个问题的本质是“系统设计时没有考虑目标服务器的承受能力”。数据源本身只是一个普通网站，不是为你准备的高并发API服务，你用太高的并发去访问，人家第一反应不是“服务好你”，而是“这个IP有问题”。最终我把并发调回5，并且在代码里加了自适应限速：每次收到429或503时，自动把并发数减半，等一段时间后再逐步恢复。这个机制上线后，再也没有出现过全链路超时的情况。
5.3 脏数据的隐蔽风险：类型正确但逻辑错误
最坑的数据错误不是格式错误，而是“看起来完全正常、但逻辑上不可能”的数据。有一次，动力煤价格在库里连续三天都是600元/吨不变，而数据源页面上每天数值都有变化。排查发现，解析器把表格中的一列错位了，取到的是“昨日结算”而不是“今日价格”，而恰好那几天“昨日结算”的数值都是一样的，所以入库后看起来像“数据没更新”。
这种问题最容易被忽略，因为数据校验时类型、范围都没问题。从那以后，我加了一道专门的“环比校验”：每个品种每日的价格与前一天做对比，如果变化率超过设定阈值，系统自动把这条记录标为“待复核”，而不是直接覆盖前一日的记录。这个机制帮我抓住了好几次数据解析错位的问题。
6. 上线后的运维细节与扩展思路
6.1 日志、监控与告警
爬虫系统的运维核心不是“跑通”，而是“可观测”。我规定所有关键节点必须打结构化日志，包括请求开始时间、请求结束时间、状态码、解析行数、入库条数、耗时。这些日志统一写入日志文件，同时按天切割，方便检索。
监控分两层：一层是基础设施监控，比如任务是否在跑、进程是否存活；另一层是业务监控，比如当天采集成功率、解析行数是否与前一天有明显波动。我设置了两个告警级别：单任务失败不告警，只记日志；连续失败超过3次或成功率低于80%，才会推送告警消息。
这里有个经验：告警阈值要留有余地，不要一点小波动就告警，否则团队会进入“狼来了”状态，最后真的出问题时没人响应。
6.2 失败回补机制
无论系统多稳定，总有那么几天数据源不配合。我把“失败任务重跑”做成了一个独立的操作入口，只要数据源恢复正常，负责人手动触发一次回补任务，系统就会根据任务清单里记录的最后成功日期，自动补齐所有缺失的数据。回补任务复用正常的下载、解析、入库流程，唯一区别是参数里的日期范围换成多个工作日。
这个设计的价值在于：它把“异常处理”从被动变成了主动。不用在凌晨数据缺失时人工写脚本临时抓数据，而是按一个固定的“补数流程”执行。
6.3 后续扩展：从采集到分析
爬虫只是数据链路的第一公里。数据进库之后，真正给团队创造价值的是分析应用。目前我在库表基础上做了几个简单的视图：品种价格趋势、日环比排名、港口库存与价格联动关系。下一步计划接入一个简单的预测模型，用历史数据和外部事件数据做价格波动区间的参考预测。
如果你也想做类似的项目，我给你一个忠告：不要一开始就追求大而全的分布式爬虫系统，先用最简单的方式跑通数据链路，再逐步加调度、告警、校验这些“耐用性”功能。一个能稳定跑三个月的简单系统，远比一个看起来高级但天天需要维护的复杂系统有价值。
最后分享一个我从这个项目里学到的东西：爬虫项目最重要的不是爬虫本身，而是数据质量和可维护性。把数据源变了怎么办、数据错了怎么发现、任务挂了怎么恢复这三件事想清楚，你的爬虫项目才算真正“精进”了。
