先说我前几天遇到的一个真实场景为了给一个小型传感器项目凑图形化编程环境我去下载某品牌的MPU6050可视化扩展库压缩包解压出来一堆.xml文件——积木定义、语言包、模块配置全在里面。我顺手用 Requests 写了个下载脚本结果第一次跑就撞上了那句有名的报错exceeded retry limit, last status: 429 too many requests。这时候才意识到之前我一直把 XML 抓取当成“requests.get 正则匹配”这种简单事压根没想过还要处理限流、重试策略、非 XML 响应、命名空间解析这一堆问题。这篇文章想跟你分享的就是我后来整理的一套相对健壮的流水线方案Requests 负责 HTTP 抓取urllib3.Retry 负责重试与退避lxml 负责解析与摘要提取。适合正在写爬虫、做数据采集、或者需要批量处理 XML 配置文件的读者参考内容偏实战代码可以直接抄走改改就能用。1. 你手里抓的到底是什么 XML三类文件和三种“不干净”的形态1.1 从 MPU6050 扩展库说起XML 不全是“文档”很多人一提到 XML脑子里浮现的是带book、title这种语义化标签的文档型 XML。但真实世界里XML 更多时候是被当作“配置格式”或“数据交换格式”在用。像 Mixly 扩展库里的blocks.xml、zh_CN.xml本质是图形化积木块的描述文件标签名可能很随意甚至大量信息都堆在属性里而不是文本节点里。我在解析这类文件时总结的经验是动手解析之前先打开文件看前 10 行确认两件事——第一根节点是什么第二关键信息到底在标签文本里还是在属性里。只看不用脑子想象十次有八次能避免后面踩坑。MPU6050 那个库里的扩展文件很多关键信息就是block typempu6050_begin这种属性形式你要是只盯着文本节点抠什么都拿不到。1.2 三种常见的“脏 XML”形态我在实际抓取中遇到的“脏 XML”基本能归纳成三类每一种的应对方式都不一样。第一类是“没有标签的裸 XML”。严格说它可能根本不是 XML而是一个文本文件或者不完整的 XML 片段。热词里有人搜“xml格式文件没有标签怎么办”说的就是这种情况。处理思路不是强行解析而是先判断内容格式如果它只是一堆用换行或空格分隔的文本就直接当纯文本清洗别硬套etree.fromstring。第二类是“被命名空间污染的 XML”。很多 XML 头部带xmlns默认命名空间比如包结构文件里常见的?xml version1.0 encodingutf-8 standaloneyes? types xmlnshttp://schemas.openxmlformats.org/package/2006/content-types。这种情况下你用//default这种路径匹配会直接扑空因为 lxml 不认裸路径名。这个坑我后面单独细说。第三类是“伪装成 XML 的错误响应”。比如服务器返了 HTTP 400但Content-Type还是text/xml打开响应体一看是错误码文本。这种问题光靠解析器是救不回来的必须在 HTTP 层拦截。还有一类顺便提一下弹幕 XML 和 PNG/XML 转换文件也偶发出现。弹幕 XML 通常是d p时间,颜色,...弹幕内容/d这种结构信息在属性里而所谓的 PNG 转 XML更多是把图片元数据序列化成 XML。这类文件的共同点是结构来源于某个特定软件不解析属性就拿不到有效信息。2. urllib3.Retry 配置为什么你的“重试”根本救不了 4292.1 默认 Retry 不重试 429这是最大的盲区先说结论如果你在代码里只写了requests.get(url)没有显式配置适配器和重试策略那么 urllib3 默认的重试只在“连接层面”生效对 HTTP 429、500 这种状态码基本不闻不问。所以一旦服务器限流你看到的就是那句exceeded retry limit, last status: 429 too many requests。这句话看着像是“重试超过了限制”其实认真读一下它更像是 urllib3 在告诉你你已经重试到上限了但最后一次状态码依然是 429。如果默认情况下根本没重试过几次这个报错会误导你让你以为重试策略生效了其实它压根没处理 429 这种状态码只是把连接重试也用光了。那为什么很多人不配置状态码重试原因是 urllib3.Retry 的默认行为只对连接错误和部分幂等请求做重试状态码里它默认只处理 413、429、503 这几个不对实际默认值里status_forcelist是空元组最多根据raise_on_status抛异常。也就是说你要想让它对 429 有反应必须手动把 429 加进status_forcelist。这是最容易被忽略的一层。2.2 正确配置status_forcelist、allowed_methods、backoff_factor先给出一份基础可用的配置from urllib3.util import Retry from requests.adapters import HTTPAdapter retry_config Retry( total5, connect3, read3, status3, backoff_factor0.5, status_forcelist[429, 500, 502, 503, 504], allowed_methodsfrozenset([GET, POST]), raise_on_statusTrue, ) adapter HTTPAdapter(max_retriesretry_config) session requests.Session() session.mount(https://, adapter) session.mount(http://, adapter)逐个参数说一下这些参数背后都是有讲究的。total5是重试总次数上限超过之后不管还有没有单独的 connect/read/status 配额直接放弃。connect3是连接阶段的重试次数主要解决“连不上、DNS 失败、连接被重置”这类问题。read3是读取阶段的重试次数解决“连接建立了但读到一半断了”这种场景。status3是遇到特定状态码后的重试次数这个数值建议比 total 小否则 total 会先兜底。backoff_factor0.5直接决定退避等待时间。urllib3 的计算公式是sleep_time backoff_factor * (2 ** (retry_count - 1))。也就是说第一次重试前等 0.5 秒第二次等 1 秒第三次等 2 秒指数往上翻。这个策略比“固定等 1 秒”要友好因为限流往往和请求频率强相关指数退避可以帮你快速脱离“风暴区”。status_forcelist[429, 500, 502, 503, 504]是状态码重试的白名单。429 太明显了服务器明确告诉你不许再来了你应该等退避结束再试500 是服务器内部错误也许是临时故障502 和 504 通常是网关层出问题经常过几秒就恢复。有人也会把 403 加进去但我不建议——403 往往是权限或反爬策略重试也大概率没用反而浪费配额。allowed_methods在老版本里叫method_whitelist这是个容易踩版本坑的地方。默认情况下 urllib3 只允许重试幂等的 HTTP 方法比如 GET、HEAD、OPTIONS、PUT、DELETEPOST 不在里面。因为 POST 不是幂等的重试可能导致重复提交数据。但做爬虫时 POST 请求太常见了如果你确定目标接口的 POST 是安全且可重入的那就显式加进allowed_methods。2.3 进阶响应头 Retry-After 才是服务器给你的“准话”指数退避是通用策略但有些服务器会在 429 响应头里带Retry-After字段明确告诉你“你需要在多少秒之后再试”或者“你需要在哪个 UTC 时间之后再试”。这时候再靠 backoff_factor 算出的指数退避就不够聪明了。我当时写了一个自定义 Retry 子类重写了get_backoff_time让它优先读取响应头里的Retry-Afterfrom urllib3.util.retry import Retry import time from datetime import datetime, timezone class RespectRetryAfterRetry(Retry): def get_backoff_time(self): # 优先读取服务器返回的 Retry-After if self._resolve_retry_after is not None: retry_after getattr(self, _retry_after, None) if retry_after is not None: try: return float(retry_after) except (TypeError, ValueError): # 有可能是 HTTP 日期格式先尝试按秒处理 # 解析失败就回退到默认指数退避 return super().get_backoff_time() return super().get_backoff_time()实际上 urllib3 较新版本已经内置了对Retry-After的部分支持它有一个Retry-After相关的解析逻辑只不过默认行为仍然以backoff_factor为准。我的建议是如果你的目标服务器明确会返回Retry-After优先以这个字段为准如果目标服务器不返回那就老老实实用指数退避。这个子类要生效还有一个容易被忽略的细节HTTPAdapter.max_retries需要传入的是Retry实例而不是普通的 int。如果你只传一个数字requests 会帮你生成一个简单的 Retry但很多高级配置就丢了。2.4 挂载到 requests.Session 的正确姿势为什么建议用Session而不是直接requests.get因为 Session 会自动管理连接池TCP 连接可以复用。配合HTTPAdapter之后每次重试都会复用连接池里的连接避免重复握手。另外如果你有多个目标域名可以分别挂载不同的适配器比如给 A 站点配严格的重试策略给 B 站点配宽松的重试策略。挂载还有一个顺序问题。session.mount(https://, adapter)里的前缀串是“匹配前缀”不是协议限定。你写https://只匹配 HTTPS 请求写https://api.example.com/只匹配这个域名下的请求。多个挂载重叠时requests 会按最长的前缀优先匹配。所以你可以先挂一个全局宽松策略再给某个重点域名挂一个严格策略请求时会自动选到更长前缀那个不用自己在逻辑里分支。3. lxml 解析的正确打开方式XPath、命名空间与容错3.1 为什么选 lxml 而不是正则或 BeautifulSoup解析 XML可选方案其实不少。正则表达式能应对“一次性提取”的场景但 XML 的嵌套结构复杂起来正则写出来比代码本身还难维护而且一旦标签里出现属性顺序变化、换行、注释正则就会崩。BeautifulSoup 确实容错性很强但它更多是为 HTML 设计的对于严格 XML 的命名空间处理反而不如 lxml 干净。lxml 基于 libxml2底层是 C 实现的解析速度比纯 Python 解析快一个量级而且 XPath 支持完整适合做流水线里需要反复查询的环节。我见过很多初学者用正则去抠 XML 里的文本一旦换行缩进变了就失效然后写一堆.replace()去补救最后代码变成一坨浆糊。有这个时间不如直接把文档解析成 Element 树之后想查什么都是 O(路径长度) 的事。3.2 命名空间是 XML 解析的“头号杀手”先学会无视它前面提到的types xmlnshttp://schemas.openxmlformats.org/package/2006/content-types这种默认命名空间是 XML 解析翻车的最常见原因。你内心的预期是“用//default或者//types一定能找到节点”而实际运行结果是返回空列表。原因在于lxml 认为这个命名空间下的标签全名其实是{http://schemas.openxmlformats.org/package/2006/content-types}types裸的types匹配不到。有两种常规解法。第一种是注册命名空间前缀from lxml import etree root etree.fromstring(xml_bytes) # bytes # 注册前缀之后就能用 pkg:types 来匹配 ns {pkg: http://schemas.openxmlformats.org/package/2006/content-types} root.xpath(//pkg:types, namespacesns)第二种更省事的办法是用local-name()函数直接忽略命名空间只看标签的本地名# 不管命名空间是什么只要本地名是 types 就匹配 root.xpath(//*[local-name()types])我用得最多的是第二种尤其在一个流水线会处理多个不同来源的 XML 文件时每个文件的命名空间不一致你不可能预先注册一堆前缀。用local-name()虽然性能上略差一点但对于文档量不大的场景完全感受不到差别。3.3 遇到“没有标签”的 XML 怎么办itertext 和容错解析回到热词里那个“xml格式文件没有标签怎么办”。真实情况里有些文件压根不是一个合法 XML 文档它就是一堆文本被零散标签包裹着。这时候如果你直接etree.fromstring大概率抛XMLSyntaxError。我的处理策略分两层。第一层先判断响应体开头是不是?xml或者是否有根标签没有就直接按纯文本清洗。第二层如果确实是一个“标签不完整但大体上是 XML 风格”的文件可以用lxml.html.fromstring做容错解析。这个函数是解析 HTML 用的HTML 本身就允许标签不闭合所以它对“半残 XML”的容忍度特别高。解析之后要提取所有文本最顺手的 API 是itertext()from lxml import html root html.fromstring(xml_content) # 提取所有文本节点并过滤掉空串/纯空白 all_text [t.strip() for t in root.itertext() if t.strip()]itertext()会递归遍历整棵树的文本节点不管标签嵌套多深都能把文本捞出来。这个 API 对于“我不知道我关心哪个标签但我想要这个文档里所有文字内容”的场景非常顺手。提取之后配合正则或简单的replace清洗空白就能得到一份可读的纯文本内容。3.4 解析错误的现场恢复XMLSyntaxError 与非 XML 响应解析阶段最常见的报错有两个XMLSyntaxError和“响应内容是 JSON/HTML/纯文本 但 Content-Type 声称 text/xml”。对于XMLSyntaxError我的建议是不要在 lxml 层做太多容错而是在进入解析器之前就拦截。检查响应头、检查响应体前几十个字节判断它到底是不是 XML。可以用一个简单的防护函数def looks_like_xml(content: bytes) - bool: stripped content.lstrip() return stripped.startswith(b?xml) or ( stripped.startswith(b) and in content[:500] )注意我前面说的Content-Type: text/xml只是一个提示它完全可能是错的。如果服务器返回 400即使它声明自己是 text/xml响应体也可能是错误说明文本。所以状态码检查必须在内容解析之前完成别等把响应体交给解析器才反应过来。遇到热词里那种non-xml response from server. response code: 400的情况本质上不是解析问题而是“请求本身没成功”。重试策略会帮你在 429/5xx 上重试但 400 一般不会重试也不该重试因为多半是请求参数有问题。这时候要把状态码、响应体前 200 个字符记进日志方便排查。4. 摘要生成不是简单的字符串截断4.1 摘要的输入与输出结构化优先原则标题里说了“抓取与摘要流水线”前两部分解决了抓取和解析接下来是摘要。很多人做摘要是直接content[:200]这个做法有三个问题第一面对 XML截出来的可能是一堆标签名和属性没有阅读价值第二中文场景下如果按字节截断可能把 UTF-8 多字节字符截出乱码第三XML 里不同字段的重要性差别很大标题、描述、正文应该有不同的优先级。我采取的思路是先抽取结构化字段再按优先级拼接摘要。XPath 就是干这个的——从解析好的 Element 树上把title、description、keywords、link这些关键节点全部提取出来形成一个小字典最后再根据字典生成可读摘要。4.2 关键信息抽取标题、描述、链接与文本统计代码示意def extract_meta(root): meta {} def find_text_by_local_name(node, local_name): nodes node.xpath(f//*[local-name(){local_name}]) if not nodes: return None # 取第一个节点的纯文本 return .join(nodes[0].itertext()).strip() meta[title] ( find_text_by_local_name(root, title) or find_text_by_local_name(root, name) or ) meta[description] ( find_text_by_local_name(root, description) or find_text_by_local_name(root, summary) or ) meta[link] find_text_by_local_name(root, link) or return meta这里有个细节find_text_by_local_name里我用了itertext()而不是nodes[0].text。因为目标标签可能嵌套子标签如titleHello bWorld/b/title只用.text会丢掉World用itertext()加join才能把所有层级文本捞干净。实际场景里我还遇到过同一个标签出现多次的情况比如播客 XML 里有多个item每个 item 里都有title。这时候就需要xpath(//*[local-name()item])先拿到所有条目再逐个提取子字段。摘要流水线的输出结构可以是“元信息 条目列表”这样后续不管是写文件还是做搜索索引都方便。4.3 清洗与截断中文场景下的字符处理抽取完信息之后清洗是必不可少的一步。常见脏内容包括连续空白、控制字符、脚本标签片段、注释残留。清洗时要注意顺序先合并空白再去掉特殊字符import re def clean_text(text: str) - str: if not text: return # 合并空白字符 text re.sub(r\s, , text) # 去掉控制字符 text .join(ch for ch in text if ch.isprintable() or ch in \n\t) # 去掉常见的脚本/样式残留 text re.sub(r\s*script.*?\s*/\s*script\s*, , text, flagsre.I | re.S) text re.sub(r\s*style.*?\s*/\s*style\s*, , text, flagsre.I | re.S) return text.strip()截断这一步如果你需要在 Python 里按“字符数”截断直接用切片text[:max_chars]不要先encode(utf-8)再截字节。Python 字符串切片天然按字符码点处理不会把一个中文字符劈成两半。这是中文内容流水线里特别容易踩的暗坑尤其是响应体拿到手是 bytes你decode完了又随手按 bytes 切片就会输出带的乱码。摘要长度怎么定我一般分两档短摘要 80 字以内用于列表展示长摘要 300 字以内用于详情页。如果标题和描述加起来还不到长摘要阈值就把正文字段body_text通过itertext()提取切一段补齐。摘要生成完之后最好存成纯文本不要带任何 XML 标签这样下游消费最安全。5. 一个可以抄作业的完整流水线5.1 三段式结构fetch → parse → summarize整个流水线我按“输入-处理-输出”分成三段每段只干一件事逻辑清晰还好调试。fetch段负责拿原始字节流这一段的产出是(url, status_code, content_bytes)。parse段负责把字节流变成 Element 树并做格式校验。summarize段负责从 Element 树里抽取元信息、生成摘要文本。三段之间用异常连接任何一段失败都直接抛出带上下文的异常而不是继续往下传半成品。这样做的好处是你可以在fetch层独立排查限流问题在parse层独立排查解析问题在summarize层独立排查字段问题。不会出现“我明明只是摘要写错了却要重新抓一遍数据”的尴尬。5.2 完整代码实现import requests import logging from urllib3.util import Retry from requests.adapters import HTTPAdapter from lxml import etree, html from dataclasses import dataclass, field from typing import Optional logging.basicConfig(levellogging.INFO, format%(asctime)s | %(levelname)s | %(message)s) logger logging.getLogger(__name__) def build_session( total5, connect3, status3, backoff_factor0.5, status_forcelistNone, ) - requests.Session: status_forcelist status_forcelist or [429, 500, 502, 503, 504] retry_config Retry( totaltotal, connectconnect, readstatus, statusstatus, backoff_factorbackoff_factor, status_forceliststatus_forcelist, allowed_methodsfrozenset([GET, POST]), raise_on_statusTrue, ) session requests.Session() adapter HTTPAdapter(max_retriesretry_config, pool_connections10, pool_maxsize20) session.mount(https://, adapter) session.mount(http://, adapter) return session dataclass class FetchedDocument: url: str status_code: int content: bytes content_type: str dataclass class DocumentSummary: url: str title: str description: str link: str body_text: str meta: dict field(default_factorydict) property def short_summary(self) - str: if self.title: base self.title else: base self.body_text[:80] return clean_text(base)[:120] property def long_summary(self) - str: parts [] if self.title: parts.append(self.title) if self.description: parts.append(self.description) if self.body_text: parts.append(self.body_text) text .join(parts) return clean_text(text)[:300] def fetch_document(session: requests.Session, url: str, timeout: tuple (5, 15)) - FetchedDocument: resp session.get(url, timeouttimeout) if resp.status_code 400: raise RuntimeError( ffetch failed: {resp.status_code}, content preview: {resp.content[:200]!r} ) content_type resp.headers.get(Content-Type, ) return FetchedDocument(urlurl, status_coderesp.status_code, contentresp.content, content_typecontent_type) def looks_like_xml(content: bytes) - bool: stripped content.lstrip() return stripped.startswith(b?xml) or ( stripped.startswith(b) and b in content[:500] ) def parse_to_root(content: bytes, tolerant: bool True): if not looks_like_xml(content): if tolerant: # 容错用 HTML 解析器接管 return html.fromstring(content.decode(utf-8, errorsreplace)) raise ValueError(content does not look like XML) try: return etree.fromstring(content) except etree.XMLSyntaxError: if tolerant: return html.fromstring(content.decode(utf-8, errorsreplace)) raise def find_text_by_local_name(node, local_name: str) - str: nodes node.xpath(f//*[local-name(){local_name}]) if not nodes: return return .join(nodes[0].itertext()).strip() def clean_text(text: str) - str: import re if not text: return text re.sub(r\s, , text) text .join(ch for ch in text if ch.isprintable() or ch in \n\t) text re.sub(r\s*script.*?\s*/\s*script\s*, , text, flagsre.I | re.S) text re.sub(r\s*style.*?\s*/\s*style\s*, , text, flagsre.I | re.S) return text.strip() def summarize_document(doc: FetchedDocument) - DocumentSummary: root parse_to_root(doc.content) summary DocumentSummary(urldoc.url) summary.title clean_text(find_text_by_local_name(root, title) or find_text_by_local_name(root, name)) summary.description clean_text( find_text_by_local_name(root, description) or find_text_by_local_name(root, summary) ) summary.link clean_text(find_text_by_local_name(root, link)) all_text [t.strip() for t in root.itertext() if t.strip()] summary.body_text clean_text( .join(all_text)) summary.meta { title: summary.title, description: summary.description, link: summary.link, length: len(summary.body_text), } return summary def run_pipeline(session: requests.Session, url: str) - DocumentSummary: doc fetch_document(session, url) summary summarize_document(doc) return summary5.3 实测抓一个公开的 XML 数据源为了验证流水线好不好用我找了一个公开的站点地图sitemap.xml来做测试。这种文件通常包含几百个 URL 条目条目标签是urlloc...结构规整正适合验证解析和摘要能力。测试环境是普通家用网络主要观察两个点一是连续抓取多个 URL 时会不会触发 429二是解析大文件时内存占用和耗时情况。代码运行方式很简单python pipeline_test.py https://example.com/sitemap.xml实测下来第一次抓取很顺利lxml 解析一个 10MB 左右的 sitemap 大约耗时 300ms 左右内存表现也算平稳。但因为短时间内连续请求了同一个域名的多个路径到第四次请求开始出现 429。这次因为我们的 Retry 配置已经生效程序自动退避等待日志里能看到重试记录最后第 6 次请求成功返回。整个过程的日志长这样INFO | status429, retrying in 0.50s... INFO | status429, retrying in 1.00s... INFO | status200, content-typeapplication/xml如果没有这个配置第二次请求估计就直接抛异常了。测试结果说明这套配置对真实的限流场景是管用的指数退避的节奏也比较合理不会因为等太久拖垮整个采集任务。5.4 常见问题速查表现象原因对策exceeded retry limit, last status: 429重试策略没配置或没把 429 加入status_forcelist用status_forcelist[429, 500, 502, 503, 504]日志里一直没看到重试记录挂载 adapter 失败或者请求没有走 Session检查session.mount前缀是否匹配请求协议XMLSyntaxError出现在解析阶段响应体不是合法 XML先检查looks_like_xml再考虑容错解析响应头是 text/xml但响应体却是 JSON服务器 Content-Type 错误或反爬干预以内容为准不迷信 Content-TypeXPath 返回空列表默认命名空间导致的路径不匹配用//*[local-name()tag]中文摘要出现乱码按字节截断了字符串直接用 Python 切片按字符截断重复抓同一个 URL 被永久封禁请求频率过高或缺少 Cookie/UA降低频率、加随机延时、考虑代理池注意合法合规6. 写在最后的实践经验与建议这套流水线我已经用在好几个真实项目里了从站点地图抓取到扩展库配置解析代码几乎都是同一套骨架。回头总结有几个调整是后来才加进来的但每个都值得做。第一重试策略里的status次数不要设太大。我最初设过status10结果有一次服务器持续 503程序硬是磨了快一分钟才放弃比直接失败还折磨人。后来我统一改成status3配合指数退避总等待时间在 20 秒左右体感最合理。第二解析阶段一定要保留原始响应体的快照。我很早之前踩过一个坑解析失败时只打印了报错没存响应体结果想排查问题时数据已经丢了。后来我在FetchedDocument里永远保留content原始字节失败时直接把前 500 字节写进日志或本地文件定位问题快得多。第三如果并发抓取的 URL 数量比较大建议在 Session 外面再套一层信号量限流比如threading.Semaphore(2)保证同一时刻最多两个请求在飞。这个限制能大幅降低触发 429 的概率重试配置退而成为兜底方案而不是第一道防线。别看这个细节小实际效果很悬殊。最后分享一个调试小技巧抓回来的 XML 如果解析不干净先别急着改代码把前 200 字节打印出来用眼睛看一遍。大多数解析问题都是内容本身比你想象的更“脏”看一眼就明白了。流水线做出来之后别忘了给它加一个简单的日志开关把每次抓取的状态码、耗时、摘要长度记下来长期跑下来你会很感激这份日志。