如果你负责过一条业务线的网页采集大概率经历过这样的场景脚本早上还在正常跑下午就断了。登录服务器一看日志里只有一行IndexError根本不知道数据是从哪个环节走丢的要是赶上页面改版选择器一夜之间全部失效业务方追着你问“今天数据怎么还没更新”。我做过不少采集工具从最初的单文件 requests 脚本到后来交给业务方长期运行的采集服务踩过的坑不在少数。这次用一个真实案例——基于 Playwright 抓取 CSDN 资讯频道的文章列表和详情正文把整个工程打磨成“配置化 日志 重试 定时任务 自动详情正文”的完整方案。这套设计和写十行脚本是完全不同的思路核心就一句话让规则可配置让异常可追踪让恢复可自动化。适合谁看如果你的采集任务还停留在“写死选择器、try except 堆一身、打日志靠 print”那这篇值得花二十分钟过一遍。文章会从架构设计、配置结构、日志体系、重试策略、定时调度讲到常见问题的排查套路给出的代码都是我在实际项目里跑过的可以直接抄作业。1. 整体架构设计配置化为什么是企业级采集的底线1.1 脚本式爬虫为什么总在关键时刻挂掉先聊聊多数人第一版采集器的写法。打开一个熟悉的网页右键复制 XPathrequests 一发正则或者 BeautifulSoup 一解析数据入库就算完事。这种脚本在“一次性取数”的场景完全没问题但一旦进入“企业级采集”——连续全每天运行、无人值守——问题就会被放大好几倍。我总结下来脚本式方案主要有四类硬伤规则与代码耦合。选择器一旦变化改的是 Python 代码等于每次页面改版都要走一遍开发、测试、发布流程效率极低。没有完善的日志链路。你无从知道这一轮采集跑了多少页、失败在哪个 URL、失败原因是什么。出了问题只能靠肉眼翻输出运气成分很大。没有重试与补偿机制。网络抖动、服务端 5xx、页面偶发加载不完整任何一次失败都可能让整轮采集中断而没人知道。手工触发缺少调度。按时段自动跑这件事经常被当成“后面再说”的最后一环到最后往往干脆没做。这些都是我在实际项目里反复吃过的亏。业务方提需求催得急快速用脚本顶了一版结果运维成本全落在自己身上。后来我把思路整个转过来把采集当成一个可观测、可恢复、可配置的工程系统来做而不是一写一跑的脚本。1.2 配置驱动 三层分离方案蓝图要解决上面四个问题我在架构层面把采集器拆成三层调度层负责按 cron 表达式到点触发跑完一轮自动退出采集层负责用 Playwright 打开浏览器、访问列表页、解析详情链接、进入详情页抓正文存储层负责将解析结果落盘或者写库同时把本轮统计信息任务 ID、成功数、失败数输出到日志。这三层在代码上彼此独立但用一块“配置”串起来。配置里不只有站点入口 URL还包括列表选择器、详情页标题选择器、正文选择器、等待时间、重试次数、限速间隔这些参数。业务方或者我自己想调整页面规则时改配置文件即可不用碰主流程代码。为什么一定要做成配置化核心原因是网站改版频率超出你的预期。特别是内容站点前端组件一升级class 名就可能换一轮。把选择器收敛到一个 YAML 或 JSON 文件里遇到改版改一行配置、重启一下进程或者让进程定时 reload 配置整个采集器就又能用了。这种规则与逻辑的解耦看起来简单但少了它任何采集方案都会变成“勤恳修代码的循环”。同样重要的还有“日志即数据”。我把每一条采集记录都带上任务 ID、URL、耗时、状态码轮次结束再输出一条 summary。这给后续做监控、告警、失败重跑留足了抓手。日志不只是给人 debug 用的更是给系统做自愈用的——后面第 2 章我会详细展开。1.3 为什么选 Playwright 而不是 Requests / Selenium / Scrapy技术选型上我最早用 requests lxml后来换成 Selenium再之后才稳定到 Playwright。简单对比一下requests轻量但没法处理 JS 动态渲染的页面。CSDN 资讯这类频道页虽然以服务端渲染为主但详情页里存在懒加载图片、异步渲染的模块你用 requests 能拿到 HTML却拿不到浏览器完整执行后的状态遇到需要点击“展开更多”的交互场景就非常尴尬。Selenium能渲染 JS但生态偏老WebDriver 管理麻烦元素等待机制远不如 Playwright 的 auto-wait 智能。Scrapy这是完整的爬虫框架功能全面但学习曲线陡异步调度和中间件体系比较重。如果你需要的是一个轻量、可插拔的企业级采集组件Scrapy 反而有些过度设计。Playwright浏览器实例管理简单自带 selector 自动等待、导航重试、多标签页支持、录制器、Trace Viewer。对“采集点击交互较多的页面”和“编写可维护的采集代码”都特别友好。选型最终回归到一个核心判断我们的重点不是“能不能拿到数据”而是“在页面频繁变化、运行环境不稳定时这套系统能不能自己撑住”。Playwright 在这一点的表现明显优于老方案这也是我在企业项目里逐步收敛到它的原因。2. 配置结构与日志体系把规则和观测先立起来2.1 配置结构设计选择器、限速、重试全收进 YAML我习惯用 YAML 做采集配置。相比 Python 脚本里的 dictYAML 可读性更好也能写注释对不熟悉代码的同学更友好。下面是我常用的基础结构你可以按需扩展global: headless: true screenshot_dir: ./screenshots sites: csdn_news: name: CSDN 资讯频道 entry_url: https://www.csdn.net/nav/news list: link_selector: a[href*/article/details/] pagination: type: scroll # 资讯页是滚动加载 scroll_times: 3 scroll_wait_ms: 2000 article: title_selector: h1.title-article content_selector: #content_views date_selector: .time remove_selectors: - script - style - iframe - .recommend-box retry: max_attempts: 5 base_delay_s: 0.5 max_delay_s: 10 limits: wait_timeout_ms: 30000 min_interval_ms: 1000创建配置时不要照着 CSDN 当前的 DOM 直接抄务必花 5 分钟看看实际页面结构。打开 DevTools 可以发现CSDN 文章标题通常是h1.title-article正文容器是#content_views但列表页卡片的 class 名变化频繁建议用a[href*/article/details/]这种“稳定特征”来选链接而不是依赖一长串拼接出来的 class 名。选择器的具体技巧在第 3 章还会延伸。这里先把配置文件的骨架立起来——配置文件是整个采集器的“规则中枢”主流程代码不依赖任何具体站点名称和选择器。2.2 配置加载与校验改规则不用动代码配置加载模块不能只做一个yaml.safe_load()就完事。企业级场景里配置很可能是运维从配置中心拉下来的修改后进程不一定能感知到。所以我会给配置模块加上三样东西文件监听用watchdog监听 YAML 文件变化发现修改自动 reloadschema 校验必填字段比如entry_url、content_selector缺失时直接拒绝启动避免错误配置带病运行类型转换timeout_ms转成 Playwright 需要的毫秒数scroll_times转成整型保证下游拿到的类型干净。一个最简配置管理器是这样的import yaml from pathlib import Path class ConfigManager: def __init__(self, config_path: str): self.path Path(config_path) self.data self._load() def _load(self): with open(self.path, r, encodingutf-8) as f: cfg yaml.safe_load(f) self._validate(cfg) return cfg def _validate(self, cfg): sites cfg.get(sites, {}) if not sites: raise ValueError(sites 不能为空) for name, site in sites.items(): for field in (entry_url, article): if field not in site: raise ValueError(f站点 {name} 缺少字段 {field}) def reload(self): self.data self._load()这里有个工程细节采集器运行时应该从同一个配置快照生成 job而不是每次取数时直接读文件否则一轮采集中途配置变更会导致同一轮里规则不一致。我的做法是reload 后只对“下一轮”生效当前执行中的轮次继续用旧快照。这有点像数据库的事务隔离虽然简单但能避免不少脏数据问题。2.3 结构化日志从 print 调试升级到 JSON 输出单机脚本可以 print 排错企业级采集必须用结构化日志。我通常用 Python 内置 logging 自定义 JSONFormatter每行日志就是一个 JSON 对象包含时间戳、级别、logger 名、任务 ID、URL、耗时等字段。这样日志可以直接灌入 ELK、Loki 或 ClickHouse做后续的可视化和告警。给个可直接用的实现import logging import json from datetime import datetime class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) - str: payload { time: datetime.now().isoformat(), level: record.levelname, logger: record.name, msg: record.getMessage(), } for key in (task_id, url, elapsed_ms, attempt): if hasattr(record, key): payload[key] getattr(record, key) return json.dumps(payload, ensure_asciiFalse) def setup_logger(name: str collector, levellogging.INFO): logger logging.getLogger(name) logger.setLevel(level) handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler) return logger打日志的时候通过logger.info(msg, extra{task_id: task_id, url: url, elapsed_ms: elapsed})带上上下文。这样一旦出现“某个 URL 超时”日志里能看到是哪个任务、哪一轮、耗时多久排查效率直接翻倍。使用过程中我会给每个轮次生成一个唯一 task_id比如 uuid4().hex 前八位从轮次开始、每个 URL 抓取、到轮次结束总结全部带上这个 id。这个概念非常关键——没有 task_id 的日志就是一条没法串联的碎片看起来每条都正常但一到问题现场就什么关联信息都拿不到。3. 采集核心实现重试、页面交互与详情正文提取3.1 三层重试机制导航、等待、内容验证各司其职网络环境不是百分百可靠的页面渲染也不是。我在项目里实现了一套“三层重试”分别处理三种不同的问题第一层导航重试page.goto()抛出 Navigation timeout、Target closed、net::ERR_CONNECTION_RESET时重新发起导航请求。第二层元素等待重试page.wait_for_selector()超时说明页面结构可能变了或者资源加载太慢间隔一段时间后重试。第三层内容验证重试即便页面加载完成、选择器也出现了正文内容也可能为空懒加载没触发、接口返回了空数据。这时把当前页面的正文文本长度作为校验指标不达标就刷新重试。一个融合了三层逻辑的函数长这样import asyncio import random import logging logger logging.getLogger(collector) async def fetch_with_retry(page, url, retry_cfg: dict) - bool: max_attempts retry_cfg[max_attempts] for attempt in range(1, max_attempts 1): try: await page.goto(url, wait_untildomcontentloaded, timeout30000) await page.wait_for_selector(#content_views, timeout15000) text_len await page.eval_on_selector( #content_views, el el.innerText.length ) if text_len 20: raise ValueError(正文内容过短视为无效) return True except Exception as exc: delay min( retry_cfg[base_delay_s] * (2 ** (attempt - 1)), retry_cfg[max_delay_s], ) delay random.uniform(0, 0.5) logger.warning( 第 %s 次抓取失败: %s, attempt, exc, extra{url: url, attempt: attempt} ) if attempt max_attempts: await asyncio.sleep(delay) return False指数退避的细节值得说一下base_delay_s * 2^(attempt-1)让第一次失败后等 0.5 秒第二次 1 秒第三次 2 秒依次翻倍但封顶到max_delay_s。最后再加上 0~0.5 秒的随机抖动避免多个任务在同一时刻一起重试给服务端造成瞬时压力。这套策略是从分布式系统的重试算法直接搬过来的用在采集场景一样管用。另外wait_until我习惯写成domcontentloaded而不是networkidle。很多页面因为长连接或轮询永远等不到 networkidle用networkidle看着稳妥实际经常把等待时间直接拉爆。3.2 少用 sleep多用 Playwright 的自动等待新手经常踩的坑是写page.wait_for_timeout(5000)硬等。这有两个问题一是时间浪费在空等上页面早就加载完了还在那儿死等二是页面异常时等多久都是白等问题照样复现不了。Playwright 对 selector 的操作自带 auto-waitpage.click(selector)会持续轮询直到元素可操作找不到就等能操作了就立刻执行。所以在做点击、输入这类交互时完全没必要在点击前手写 sleep。我的习惯是优先用wait_for_selector加条件判断不要裸 sleep。唯一允许 sleep 的场景是模拟用户滚动加载for _ in range(scroll_times): await page.mouse.wheel(0, 1500) await page.wait_for_timeout(scroll_wait_ms)滚动次数有限每轮滚动后等待指定时间让懒加载的数据有充足时间渲染到位。这里的scroll_wait_ms也是配置项一般设 1500~3000 毫秒比较合适太长会拖慢整轮采集太短又可能触发不了加载。3.3 详情正文提取从“拿到 HTML”到“拿到干净文本”CSDN 详情页正文在#content_views容器里。用 Playwright 拿到它之后我建议先执行一次 DOM 清理去掉脚本、样式、iframe、推荐模块这类噪声再取 inner_text。这样拿到的文本干净后续落地到数据库或做 NLP 分析都省事。实现如下async def extract_content(page) - str: container await page.query_selector(#content_views) if not container: return await container.evaluate( (el) { el.querySelectorAll(script, style, iframe, .recommend-box, .toolbar) .forEach(node node.remove()); } ) text await container.inner_text() return text.strip()这里有个性能细节container.evaluate直接在页面上下文里执行 JS操作的是浏览器 DOM效率比通过 Python 逐个remove()高很多。这也是 Playwright 的一个优势——你可以在页面上下文里直接跑脚本逻辑不需要中间走协议。如果文章正文是分页加载的比如有“下一页”按钮那就要在配置里加一个“翻页按钮 selector”每次抓完当前页正文后点击下一页继续拼接。CSDN 资讯文章一般单页展示这个能力做成可选就行但架构上要预留位置。4. 定时任务与部署落地让采集器自己按点开跑4.1 为什么用 APScheduler 而不是 crontab定时触发最简单的方案是 crontab。但我在实际项目里不推荐纯 crontab原因有二crontab 的最小粒度是分钟不适合“每 30 秒采集一次”这种高频场景进程由系统拉起生命周期不好控制失败退出没有回调日志也分散很难和采集器的日志体系整合。所以我用 APScheduler。它支持 cron 触发器和 interval 触发器跑在应用进程内部异常、超时、日志都可以收拢到现有体系里。部署时用BlockingScheduler就能满足大部分需求不需要额外引入消息队列。一个直接可用的调度示例from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger import asyncio import uuid logger logging.getLogger(collector) def run_task(): task_id uuid.uuid4().hex[:8] logger.info(轮次开始, extra{task_id: task_id}) asyncio.run(collect_all_sites(task_id)) logger.info(轮次结束, extra{task_id: task_id}) scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job( run_task, CronTrigger(hour*/2, minute5), idcsdn_news_regular, ) scheduler.start()timezone参数千万别漏。服务器时区默认是 UTC如果 cron 表达式写hour8但没指定时区实际触发时间会差 8 小时。这个坑我见过不止一次也不是第一次有人在生产环境里踩。4.2 进程模型与防重复执行的关键细节APScheduler 在单进程里能保证同一个 job 串行执行但如果你部署时用了 Gunicorn 多 worker或者运维把采集脚本做成了“每 10 分钟拉起一次”的外层调度就可能出现两个采集进程同时在跑导致重复入库。防重复执行的措施有三类在应用内部维护一个 running 标记job 执行开始时置 True结束置 False第二次触发发现已在跑就直接跳过用 Redis 的SET NX EX命令做分布式锁适合多实例部署在数据库层面给“来源 URL 抓取时间”加唯一索引作为最后一道兜底。单机单进程的场景第一类最简单is_running False def safe_run_task(): global is_running if is_running: logger.info(上一个任务仍在运行本轮跳过) return is_running True try: run_task() finally: is_running False注意一定要把is_running False放进 finally否则中途异常会导致后续所有轮次永远被标记为 running。这是最容易忽略但后果很严重的细节——我见过一次线上事故就是异常没走到复位逻辑整个采集服务静默停了三天。4.3 浏览器实例的复用与释放Playwright 启动浏览器比较重每轮任务都重新 launch 会浪费时间。我的做法是任务开头启动一次 browser整个采集循环都在这个上下文里跑每个站点单独开一个 context独立 cookie、独立 UA任务结束统一关掉 browser避免内存泄漏。形如这样async def collect_all_sites(task_id): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) try: for site_name, cfg in CONFIG[sites].items(): await collect_site(browser, site_name, cfg) finally: await browser.close()不要在一个 browser 里无限新建 context 而不关闭时间一长连接数会爆炸。每个站点跑完立即context.close()这是最基本的资源纪律。5. 踩坑记录安装、超时、选择器失效与兜底预案5.1 Playwright 安装与环境问题速查我在不同环境里部署 Playwright 遇到过几类问题先列出来帮你排雷浏览器下载失败pip install playwright安装的是 Python 包真正需要的是playwright install chromium。命令行下载浏览器时企业内网经常因为证书或代理问题失败多试几次或者设置PLAYWRIGHT_DOWNLOAD_HOST指向内网镜像。npx playwright install卡住npx 尝试下载的是 Node 端的 Playwright 包跟你 Python 项目没关系。Python 项目里直接用playwright install chromium即可。无头模式被检测采集 CSDN 这种常规内容站通常不严格但如果你遇到滑块验证或“浏览器环境异常”提示优先把headlessFalse打开跑一次看看再考虑在 launch 参数中加--disable-blink-featuresAutomationControlled或换浏览器指纹。这里多说一句这类操作属于对抗反爬建议先评估你是否真的需要绕过对方机制很多站点只需要你把合法 cookie 设置进 context 就行。5.2 CSDN 页面结构与选择器变更怎么办CSDN 资讯页的列表结构随版本迭代调整过多次。我遇到过两次一次是列表容器的 class 名从blog-list-box变成blog-list-card另一次是文章链接从/news/[id]变成/article/details/[id]。处理思路有三个层级不要完全依赖 class 名优先用带语义的标签属性比如a[href*/article/details/]这个特征比任何 class 都稳定配置支持多个备用 selector代码按顺序尝试一个失效自动换下一个每轮任务把命中数量记录下来如果命中数为 0立刻告警“选择器可能失效”而不是默默空跑。第三点尤其重要。采集器最怕的不是报错而是“看起来跑得好好的实际上一个数据都没抓到”。没有命数统计这种情况可能要到你手动查库才发现那时已经白白空跑了好几天。5.3 超时、元素状态与正文为空的排查套路定位元素报strict mode violation页面上有多个元素匹配同一 selectorPlaywright 默认要求唯一。要么改用page.locator(selector).first()要么把 selector 写得更精确。报Element is not attached to the DOM这通常发生在你获取元素之后、页面刷新或跳转前后你还在引用旧元素句柄。避免做法是先把需要的数据全部取下来保存到 Python 变量再去点击跳转不要跨页面状态持握手柄。超时时间的选择导航超时 30 秒等待 selector 超时 15~30 秒。如果页面平均加载只有 3 秒超时设太长反而会让失败轮次拖着不结束任务积压越来越多。建议先做一次 20~30 条 URL 的小规模采样统计正常耗时再定超时上限。详情页加载完成后正文为空十有八九是懒加载。内容在页面底部才出现而你取的时机太早。解决办法很简单先滚到页面底部await page.evaluate(window.scrollTo(0, document.body.scrollHeight)) await page.wait_for_timeout(1500)乱码问题一般出现在编码判断错误。Playwright 渲染后的 DOM 已经按页面编码解析好了通常不需要再处理 charset。如果你用 requests 拿原始 HTML就必须注意响应头里的 charset——但这种场景在 Playwright 方案里基本不存在这算是一个额外的省心点。5.4 兜底预案失败截图与任务隔离最后给一个我踩坑总结出来的兜底方案采集任务必须在极端失败时留下可视化证据。我习惯在每轮任务结束时如果发现有 URL 抓取失败超过阈值就自动把当前页面截图存到screenshots/目录文件名带 task_id 和时间戳。if error: shot_path f./screenshots/fail_{task_id}_{time.time()}.png await page.screenshot(pathshot_path, full_pageTrue) logger.info(失败截图已保存, extra{path: shot_path})这张截图能直接告诉你页面渲染成了什么样子——是验证码拦截、白屏还是把用户登录弹窗挡住了。很多东西光看日志是猜不出来的但一张图什么都说明白了。任务隔离指的是一个站点的故障不要拖垮整个调度。比如 CSDN 站点连续三小时超时这时候应该标记该站点“暂时搁置”跳过它继续跑其他站点等下一轮再恢复。我在调度层加了个简单的熔断计数连续失败 N 次就把该站点挂起 30 分钟期满自动恢复。这个机制让整个采集系统即便在局部故障时也能维持全局的可用性。我个人在实际操作中的体会是把采集做“重”一点短期看好像多花了时间长期看反而省事。配置、日志、重试、调度的基本功扎实了后面遇到页面改版、服务器波动、数据质量问题都有一整套应对路径而不是每次都在上线后熬夜救火。最后再分享一个小技巧项目里写一个dry_run.py启动时只抓 1 个列表页和 1 个详情页把标题、正文长度、链接数打印出来配置改完先跑一遍 dry-run 再进定时任务。这个两分钟的检查能挡住百分之八十的低级问题尤其是选择器写错和配置格式不对这两类。企业级采集拼的不是某一招多厉害而是每个环节都不出低级事故这比什么都重要。