这个系列案例写到今天,已经攒了小二十个。今天要聊的这一个是适用范围很广的组合玩法:把爬虫和文本润色接口串成一条自动流水线——定时抓一批原始文本,交给润色接口清洗改写,最后输出成可以直接用的干净文案。说直白点,就是让Python每天自动帮你完成“采集初稿 -> 技术性润色 -> 结构化归档”这三件事,省掉大量复制粘贴和人工修改时间。无论你是正在做内容采集加工的开发者,还是需要批量整理素材的运营,这套思路都值得参考。下面我把完整案例拆开讲清楚。
1. 案例整体设计与技术选型
1.1 这个案例到底要解决什么问题
先说说我为什么要写这个“文本润色接口”的案例。前面几期爬虫案例大多停留在“把网页数据抓下来存成文件”这一步,数据是有了,但直接用还很费劲:网页正文里全是广告噪音、导航文字、无意义换行,句子表述也参差不齐。很多做内容运营或产品文案的朋友,真正的需求不是抓数据,而是拿到“能直接用的文字”。
这个案例的目标非常明确:做一个每日自动运行的爬虫任务,把指定页面的正文抓下来,调用文本润色接口进行规范化处理,最后输出一份适合直接发布的干净文案。整个过程不需要人工打开网页、复制正文、再跑去编辑器里反复修改,全部交给脚本在后台完成。
适合谁来参考?首先是刚接触爬虫、想从“抓取”进阶到“加工”的Python学习者;其次是那些需要批量处理文本素材的从业者,比如新媒体运营、文案编辑、数据采集分析岗。哪怕你不写爬虫,只看“如何封装一个稳定的HTTP接口调用层”,里面也有不少通用经验可以带走。
1.2 技术栈选型与理由
这个案例的技术选型我刻意保持“最小够用”原则,不引入重型框架。核心依赖只有三个:requests负责网络请求,BeautifulSoup负责解析HTML,APScheduler负责定时调度。为什么不用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, headers=DEFAULT_HEADERS, timeout=timeout) 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(strip=True) content = content_node.get_text("\n", strip=True) 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, json=payload, timeout=self.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( level=logging.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_output=True, text=True, timeout=600, ) 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( hour=config.RUN_HOUR, minute=config.RUN_MINUTE, day_of_week=config.RUN_DAY_OF_WEEK, ) scheduler.add_job(run_daily_task, trigger, id="daily_polish_task", misfire_grace_time=3600, coalesce=True) logger.info("定时任务已启动,运行时间: %s:%s, 星期: %s", config.RUN_HOUR, config.RUN_MINUTE, config.RUN_DAY_OF_WEEK) scheduler.start() if __name__ == "__main__": main()这里有两个参数值得单独解释。misfire_grace_time=3600表示如果任务因为某些原因(比如电脑休眠)错过了原定执行时间,在一小时之内补跑都算有效,超过一小时就放弃。coalesce=True表示同一时刻只执行一次,不会把多次错过的任务积压在一起连环触发。这两个参数组合使用,能大幅减少“定时任务该跑没跑”的困扰。
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", default=config.TARGET_URL, help="目标页面地址") parser.add_argument("--output", default=None, help="输出文件路径") parser.add_argument("--style", default=config.POLISH_STYLE, help="润色风格") return parser.parse_args() def main(): args = parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(config.LOG_FILE, encoding="utf-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"], style=args.style) filename = args.output or f"output/{datetime.now():%Y%m%d_%H%M%S}.md" with open(filename, "w", encoding="utf-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(host="127.0.0.1", port=9000)先启动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能解决绝大多数问题。个别页面会声明charset=gb2312但实际是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 长期运行的一些私人心得
这个案例做到现在,我自己总结了一套运行习惯:日志一定要同时写文件和标准输出,只写文件会导致出问题时排查特别不方便,只写标准输出又会丢历史记录;输出目录按月建子目录,避免积年累月后一个文件夹里几千个文件没法找;每周看一眼日志统计,如果连续三天都出现重试,说明接口或页面大概率出状态了,早点介入永远比事后补救省事。做自动化任务,稳定比功能多更重要。把“不出错”“不静默失败”“数据不丢”这三条守住,这套流水线就能长期稳定地跑下去。