这个系列案例写到今天已经攒了小二十个。今天要聊的这一个是适用范围很广的组合玩法把爬虫和文本润色接口串成一条自动流水线——定时抓一批原始文本交给润色接口清洗改写最后输出成可以直接用的干净文案。说直白点就是让Python每天自动帮你完成“采集初稿 - 技术性润色 - 结构化归档”这三件事省掉大量复制粘贴和人工修改时间。无论你是正在做内容采集加工的开发者还是需要批量整理素材的运营这套思路都值得参考。下面我把完整案例拆开讲清楚。1. 案例整体设计与技术选型1.1 这个案例到底要解决什么问题先说说我为什么要写这个“文本润色接口”的案例。前面几期爬虫案例大多停留在“把网页数据抓下来存成文件”这一步数据是有了但直接用还很费劲网页正文里全是广告噪音、导航文字、无意义换行句子表述也参差不齐。很多做内容运营或产品文案的朋友真正的需求不是抓数据而是拿到“能直接用的文字”。这个案例的目标非常明确做一个每日自动运行的爬虫任务把指定页面的正文抓下来调用文本润色接口进行规范化处理最后输出一份适合直接发布的干净文案。整个过程不需要人工打开网页、复制正文、再跑去编辑器里反复修改全部交给脚本在后台完成。适合谁来参考首先是刚接触爬虫、想从“抓取”进阶到“加工”的Python学习者其次是那些需要批量处理文本素材的从业者比如新媒体运营、文案编辑、数据采集分析岗。哪怕你不写爬虫只看“如何封装一个稳定的HTTP接口调用层”里面也有不少通用经验可以带走。1.2 技术栈选型与理由这个案例的技术选型我刻意保持“最小够用”原则不引入重型框架。核心依赖只有三个requests负责网络请求BeautifulSoup负责解析HTMLAPScheduler负责定时调度。为什么不用Scrapy因为每日案例的核心是跑通“采集-加工-落地”的完整链路Scrapy在并发和框架规范性上更强但学习曲线也更高对轻量任务来说有些杀鸡用牛刀。什么时候该换Scrapy当你的采集目标数量多到需要分布式、需要中间件管道、需要增量抓取的时候再换也不迟。文本润色接口这块我选择统一封装成HTTP接口来调用。理由很直接生产环境里的文本处理能力可能是Python进程内函数也可能是独立部署的算法服务还可能是团队内部统一的NLP平台。用HTTP接口封装能把“调用方”和“实现方”彻底解耦换后端模型服务只需要改一个URL爬虫代码完全不用动。这也是从业者做系统集成的常见思路。1.3 目录结构与代码骨架工程结构我习惯按职责拆成五个文件单文件跑起来容易但后续维护会很难受。spider_daily/ ├── config.py # 全局配置目标URL、接口地址、运行时间 ├── spider.py # 采集模块负责抓取和解析正文 ├── polish.py # 润色模块封装文本润色接口的请求逻辑 ├── scheduler.py # 调度入口定时任务与参数传递 ├── runner.py # 单次执行入口供手动运行或子进程调用 ├── logs/ # 运行日志目录 └── output/ # 润色结果输出目录这样的分层逻辑很清晰spider.py只负责“把原文拿回来”polish.py只负责“把文本变干净”scheduler.py和runner.py负责“什么时候跑、怎么组合”。真出了问题时你只需要检查对应的模块就能定位不需要在几百行的大杂烩里翻来翻去。2. 核心模块拆解爬虫、润色接口与调度2.1 爬虫模块的实现方案爬虫模块的核心任务不是“抓得多”而是“抓得准”。在写spider.py时我首选用CSS选择器而不是正则表达式来提取正文因为HTML结构会有各种不规则嵌套正则表达式处理起来容易失控而BeautifulSoup的选择器能直接定位到目标节点。# spider.py import requests from bs4 import BeautifulSoup DEFAULT_HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 } def fetch_html(url: str, timeout: int 10) - str: 抓取页面HTML返回文本内容。 resp requests.get(url, headersDEFAULT_HEADERS, timeouttimeout) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text def extract_content(html: str, title_selector: str, content_selector: str) - dict: 根据CSS选择器解析标题和正文返回结构化数据。 soup BeautifulSoup(html, html.parser) title_node soup.select_one(title_selector) content_node soup.select_one(content_selector) if title_node is None or content_node is None: raise ValueError(页面结构变化未找到指定的标题或正文节点) title title_node.get_text(stripTrue) content content_node.get_text(\n, stripTrue) return {title: title, content: content}这里有两个关键细节。第一个是resp.encoding resp.apparent_encoding很多网站页面没声明编码或声明错误不修正这行代码的话解析出来全是一堆乱码后面润色接口处理乱码文本也没有意义。第二个是raise_for_status()遇到404、500必须直接抛出异常而不是带着错误页面继续往下跑否则润色接口会把404页面里的文字当成正文处理输出一堆废话。另外一个值得注意的点是解析策略要“宽容”。实际目标页面改版太常见了所以我通常会在解析不到内容时抛出带节点信息的异常这样日志里能直接看到是哪一步选择器失效方便快速调整。2.2 润色接口的封装细节封装HTTP接口调用最重要的不是把请求发出去而是把“各种异常情况”都处理干净。我在polish.py里设计的接口请求逻辑包括超时控制、状态码判断、JSON解析失败处理、业务错误码识别。# polish.py import logging import time import requests logger logging.getLogger(__name__) class TextPolishClient: 文本润色接口客户端统一封装请求、重试与结果解析。 def __init__(self, api_url: str, timeout: int 30, max_retries: int 3): self.api_url api_url self.timeout timeout self.max_retries max_retries def polish(self, text: str, style: str general) - str: 调用润色接口处理文本返回润色后的内容。 payload {text: text, style: style} for attempt in range(1, self.max_retries 1): try: resp requests.post( self.api_url, jsonpayload, timeoutself.timeout, headers{Content-Type: application/json}, ) resp.raise_for_status() data resp.json() if data.get(code) ! 0: raise RuntimeError(f业务错误: {data.get(message)}) return data[data][text] except requests.exceptions.Timeout as exc: logger.warning(润色接口超时第 %s 次重试, attempt) except requests.exceptions.RequestException as exc: logger.warning(润色接口请求异常: %s, exc) except (ValueError, KeyError, RuntimeError) as exc: logger.error(润色接口响应异常: %s, exc) break if attempt self.max_retries: time.sleep(2 * attempt) raise RuntimeError(润色接口重试多次后仍失败)为什么要单独维护一个客户端类而不是写个简单函数因为在实际项目里一个接口的调用方可能有多个入口有可能是定时任务有可能是手动命令行也有可能是后面要接入的Web界面。把请求逻辑封装成类实例化后统一复用重试参数和超时参数都能在启动时注入这比每个调用方各写一段requests.post干净得多也方便做单元测试。重试策略我采用指数退避第一次失败等2秒第二次失败等4秒。这比固定间隔重试更温和能给不稳定接口多一些恢复时间也不至于给服务端造成太大压力。2.3 定时调度模块定时调度我选了APScheduler的BlockingScheduler。之所以不用系统自带的crontab是因为定时任务通常还需要“启动前准备一些运行上下文”比如读取配置、初始化日志、组装爬虫和润色客户端。用Python进程统一管理逻辑更内聚部署到Windows和Linux都能跑。# scheduler.py import logging import subprocess import sys from datetime import datetime from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger import config logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, ) logger logging.getLogger(__name__) def run_daily_task(): 每日任务执行函数通过独立子进程运行 runner.py。 logger.info(定时任务触发启动爬虫润色流水线) proc subprocess.run( [sys.executable, runner.py], capture_outputTrue, textTrue, timeout600, ) logger.info(子进程退出码: %s, proc.returncode) if proc.stdout: logger.info(子进程输出: %s, proc.stdout.strip()) if proc.returncode ! 0: logger.error(子进程错误: %s, proc.stderr.strip()) def main(): scheduler BlockingScheduler() trigger CronTrigger( hourconfig.RUN_HOUR, minuteconfig.RUN_MINUTE, day_of_weekconfig.RUN_DAY_OF_WEEK, ) scheduler.add_job(run_daily_task, trigger, iddaily_polish_task, misfire_grace_time3600, coalesceTrue) logger.info(定时任务已启动运行时间: %s:%s, 星期: %s, config.RUN_HOUR, config.RUN_MINUTE, config.RUN_DAY_OF_WEEK) scheduler.start() if __name__ __main__: main()这里有两个参数值得单独解释。misfire_grace_time3600表示如果任务因为某些原因比如电脑休眠错过了原定执行时间在一小时之内补跑都算有效超过一小时就放弃。coalesceTrue表示同一时刻只执行一次不会把多次错过的任务积压在一起连环触发。这两个参数组合使用能大幅减少“定时任务该跑没跑”的困扰。2.4 参数传递与脚本联调日常开发里经常遇到“A脚本需要让B脚本知道跑哪个URL”的情况。这个案例里我也加入了参数传递的完整实现用两个层面解决一个是命令行参数一个是子进程调用。先在runner.py里用argparse接收外部传入的参数这样手动运行时可以灵活指定URL和输出文件# runner.py import argparse import json import logging from datetime import datetime import config from polish import TextPolishClient from spider import fetch_html, extract_content def parse_args(): parser argparse.ArgumentParser(description每日爬虫润色任务) parser.add_argument(--url, defaultconfig.TARGET_URL, help目标页面地址) parser.add_argument(--output, defaultNone, help输出文件路径) parser.add_argument(--style, defaultconfig.POLISH_STYLE, help润色风格) return parser.parse_args() def main(): args parse_args() logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(config.LOG_FILE, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(__name__) logger.info(开始抓取页面: %s, args.url) html fetch_html(args.url) article extract_content(html, config.TITLE_SELECTOR, config.CONTENT_SELECTOR) logger.info(抓取成功标题: %s, article[title]) client TextPolishClient(config.POLISH_API_URL) polished_text client.polish(article[content], styleargs.style) filename args.output or foutput/{datetime.now():%Y%m%d_%H%M%S}.md with open(filename, w, encodingutf-8) as f: f.write(f# {article[title]}\n\n{polished_text}\n) logger.info(润色完成结果已保存: %s, filename) if __name__ __main__: main()在scheduler.py的子进程调用部分如果后续需要给runner.py传参只需修改subprocess.run的参数列表比如[sys.executable, runner.py, --style, news]。这种方式的好处是定时任务进程和实际执行任务进程相互隔离即使其中一次的爬虫代码出现严重异常导致进程崩溃也不会拖垮整个调度器。3. 实操全程跑通一个完整的润色流水线3.1 环境准备与依赖安装复现这个案例前先把依赖装好建议用虚拟环境隔离别直接装到全局。python -m venv venv source venv/bin/activate # Windows 下执行 venv\Scripts\activate pip install requests beautifulsoup4 apscheduler因为案例要调用文本润色接口但大家本地未必有现成的服务我建议先写一个简易的Mock服务端用Flask把接口模拟出来这样整条链路完全可以在本地跑通不受外部环境影响。# mini_server.py from flask import Flask, request, jsonify app Flask(__name__) app.post(/rewrite) def rewrite(): data request.get_json() text data.get(text, ) style data.get(style, general) # 这里只做一个示例清理多余空白并加一行注释 cleaned \n.join(line.strip() for line in text.splitlines() if line.strip()) result f【{style}润色】\n cleaned \n\n—— 本段内容已由本地润色服务处理 return jsonify({code: 0, data: {text: result}}) if __name__ __main__: app.run(host127.0.0.1, port9000)先启动Mock服务再跑主流程就能用最小的成本验证整个流水线的逻辑是否正确。真实项目中这个Mock服务可以替换成团队内部统一的文本处理平台调用方式保持不变。3.2 构造润色请求与重试机制在实际配置请求参数时我踩过一个坑请求体里直接传超长文本时有的接口网关会对Body大小有限制。所以我在TextPolishClient里加入了文本长度检查和建议分段处理的日志提示。这里给出一份参考的config.py# config.py TARGET_URL file:///path/to/demo.html TITLE_SELECTOR h1 CONTENT_SELECTOR article POLISH_API_URL http://127.0.0.1:9000/rewrite POLISH_STYLE general RUN_HOUR 9 RUN_MINUTE 30 RUN_DAY_OF_WEEK mon-fri LOG_FILE logs/spider_daily.log关于重试机制有一点容易被忽略不是所有异常都值得重试。接口返回400这种参数级错误说明请求本身有问题重试一万次也是同样的结果这时候应该直接抛异常并记录日志只有Timeout、ConnectionError这类网络层异常或服务端5xx错误才值得走重试流程。所以我上面的代码里遇到ValueError、KeyError这类响应解析错误时用的是break退出循环而不是继续重试。3.3 轮询式运行与结果归档整个流水线跑起来之后输出格式我建议采用Markdown文件按日期命名天然适合沉淀成内容库。手动执行一次的效果如下python runner.py --url ./demo.html --style news执行输出大致是2025-06-25 09:30:01 - INFO - 开始抓取页面: ./demo.html 2025-06-25 09:30:02 - INFO - 抓取成功标题: 产品发布新闻稿 2025-06-25 09:30:03 - INFO - 润色完成结果已保存: output/20250625_093003.md打开生成的Markdown文件就能看到已经处理过的正文。如果定时调度器在线每天到点就会自动执行这套流程不用再手工干预。3.4 接口异常时的降级处理处理接口异常这块我在生产环境中吃过教训把润色接口当作“一定可靠”的组件结果接口那边模型服务升级重启流水线整个挂掉连原始文本都没保住。后来我在流程里加入了“原始内容兜底”机制调用润色接口失败时不做硬性失败而是把抓取到的原始正文原样保存并在日志里标记一条POLISH_SKIPPED。这样做的好处很明显任务目标从“必须产出完美文案”降级为“至少保留原始素材”第二天接口恢复后还能手动补跑润色。自动化任务设计里“降级优先”是很实用的一个思想宁可少做一步不能让整条链路崩溃导致数据丢失。4. 常见问题与排查技巧实录4.1 页面编码乱码问题我写爬虫最常遇到的第一个坑就是乱码症状是抓回来的文本里到处是é、“这类诡异字符或者中文直接变成问号。排查分两步先检查页面响应头里的charset声明再看resp.apparent_encoding得到的值。两者不一致时以实际解析出的编码为准。在fetch_html里同时设置resp.encoding resp.apparent_encoding能解决绝大多数问题。个别页面会声明charsetgb2312但实际是gbk这种情况下直接用resp.encoding gbk硬编码更可靠。4.2 接口请求超时与连接重置润色接口通常比普通API更耗时因为模型需要完整读完文本再生成结果。把默认超时设为5秒几乎一定会翻车。我建议把timeout设为30秒以上并且把“长文本请求”视作预期内的正常耗时。连接重置问题则多半和网络代理或服务端连接数限制有关排查时可以先用curl手动发一下同样的请求排除是不是自己代码里的问题。问题速查表现象可能原因处理方式中文乱码页面编码识别错误手动设置resp.encoding为实测编码接口不断超时超时设置过短调高timeout加入指数退避重试定时任务不触发电脑休眠错过执行点配misfire_grace_time和coalesce同一天重复产出多个调度器实例并存用文件锁或确保只启动一个调度进程抓取返回403缺少合理的UA补充User-Agent并适当降低抓取频率4.3 定时任务漏跑与重复执行的坑漏跑的原因除了电脑休眠还有个隐蔽场景是时区问题。APScheduler默认使用本机时区如果代码里混用了CronTrigger和带tzinfo的时间对象可能导致任务延后或提前执行。建议统一在调度器初始化时显式指定时区别依赖系统默认值。重复执行通常是部署了多个进程引起的。比如你一边用scheduler.py做定时调度一边又手动跑了runner.py就会生成两条重复记录。我的方案是在runner.py开头加一个基于fcntl的文件锁来防止重复实例或者至少保证生产环境里只有一个调度入口。4.4 目标页面结构变化与接口限流的应对页面结构变化是长期跑爬虫无法避免的问题。我的处理方式是给爬虫模块加“结构自检”日志每次解析完成后检查正文长度是否低于阈值或者是否出现页面里常见的“无权限”关键词一旦异常立刻告警。这样即使页面悄悄改版了也不是等到任务完全失败才发现而是在数据质量明显下降的第一时间得到提醒。接口限流方面在润色客户端里加入min_interval参数控制连续两次请求之间的最小间隔。比如处理大量文本时每调用一次润色接口本地至少等待1秒再发下一个请求用温和的节奏避免被服务端限流封禁。这个策略在对接任何HTTP服务时都通用。4.5 长期运行的一些私人心得这个案例做到现在我自己总结了一套运行习惯日志一定要同时写文件和标准输出只写文件会导致出问题时排查特别不方便只写标准输出又会丢历史记录输出目录按月建子目录避免积年累月后一个文件夹里几千个文件没法找每周看一眼日志统计如果连续三天都出现重试说明接口或页面大概率出状态了早点介入永远比事后补救省事。做自动化任务稳定比功能多更重要。把“不出错”“不静默失败”“数据不丢”这三条守住这套流水线就能长期稳定地跑下去。