简介这是一份面向知识产权研究、数据分析与法律服务场景的Python专利爬虫项目源码适合已掌握Python基础、希望了解真实爬虫工程结构的中级学习者。项目以中介者模式协调请求调度通过延时控制、并发限制等方式降低对目标站点的访问压力并围绕查询、页面采集、条目收集、进度控制、Excel导出与等待引擎等模块划分职责形成从检索到结果落地的完整链路可用于专利元数据收集与后续的趋势分析、竞争监控等实践。压缩包共56个文件以23个py源码与22个pyc编译文件为主另有8个xml配置及说明文档、可执行文件等整体约17.36MB源码与配置分离便于按层阅读。已有5499人浏览学习。读者可从中获取分层清晰的工程目录、请求节流与异常处理的实现思路、数据实体与枚举的建模方式以及将抓取结果导出为表格的工具脚本作为爬虫设计与合规采集的参考。1. 专利数据抓取不是 requests 一把梭先把下载链路拆成四段写专利爬虫的人第一版代码几乎都是同一个样子拼一个检索 URLfor 循环翻页BeautifulSoup 抠标题再 requests.get 把 PDF 存下来。跑前三十条很顺第三十一条开始返回 403或者 PDF 存下来只有 20KB打开是一片乱码。python 专利爬虫下载这件事难点从来不在发请求而在于请求、解析、落库、下载这四段能不能各自独立重试。这份资源包就是按这四段拆的请求层管会话与退避解析层管字段抽取落库层用公开号去重下载层做流式落盘和文件头校验。适合需要批量拿到专利著录项和全文 PDF 的从业者也适合刚学爬虫、想找一个字段结构清晰、反爬不算变态的真实站点练手的人。下面按能复现的顺序拆开讲。2. 请求层Session、分页参数与退避节奏怎么定2.1 先建 Session再谈抓取很多人图省事每抓一条就requests.get(url)一次。这种写法在专利这类需要翻十几页、点进几十个详情页的场景里代价很直接每次都要重新做 TCP 握手和 TLS 协商服务端看到的是几十个互不相关的短连接行为特征比正常浏览器刺眼得多。更实际的问题是 Cookie 丢不住——不少检索站点第一次访问会下发一个会话标识后续分页和详情页都靠它关联你的检索上下文短连接写法下这个上下文反复重建翻到后面会出现页码对了但结果集变了的玄学现象。常见做法是开头建一个Session让连接池、Cookie、默认请求头都挂在上面import requests from requests.adapters import HTTPAdapter def build_session(referer: str) - requests.Session: s requests.Session() # 请求头按真实浏览器补齐缺 Accept-Language 时部分站点会返回简版页面 s.headers.update({ User-Agent: (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36), Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.6, Referer: referer, Connection: keep-alive, }) # 连接池pool_connections 是保留的连接池数量pool_maxsize 是单池最大连接数 # max_retries 设 0重试交给我们自己的退避函数方便记录失败原因 adapter HTTPAdapter(pool_connections4, pool_maxsize8, max_retries0) s.mount(https://, adapter) s.mount(http://, adapter) return s逻辑上就三件事统一请求头、复用连接、把重试权收回来。参数上pool_connections和pool_maxsize不用开大单站点并发压到 24 时8 个池位足够开太大反而容易在短时间内打出密集请求。max_retries0是有意为之——urllib3 自带的重试对 429 的处理不够透明你拿不到第几次重试、因为什么失败排错时就是个黑匣子。timeout建议用元组(连接超时, 读取超时)例如(5, 20)连接 5 秒还不通说明网络或站点侧有问题读取给 20 秒是因为专利详情页偶尔会带较大的附图单值 timeout 会把这两类问题糊在一起。提示动手前先看目标站点的 robots 与使用条款把单域名并发控制在 12翻页之间留 12 秒间隔。这不是客套是为了让后面几万条能跑完。2.2 分页参数写成生成器终止条件别只看空列表专利检索站点的翻页参数通常就三四个关键词、页码、每页条数、排序方式。排序字段值得单独说一句——默认相关度排序下同一关键词两次检索的结果顺序可能不同你按页码往下翻很容易重复抓到同一批记录同时漏掉另一些。我一般固定成公开日倒序参数值常见是pd_desc这类保证翻页过程中的结果集相对静止。把翻页写成生成器好处是页码逻辑和抓取逻辑分离中途要改成从第 37 页续跑只改一个入参def iter_page_params(keyword: str, page_size: int 20, start_page: int 1, max_pages: int 50): 按公开日倒序翻页page_size 超过站点上限会被静默截断先用 20 试 for page in range(start_page, max_pages 1): yield { q: keyword, page: page, size: page_size, sort: pd_desc, }page_size不要一上来就设 100。很多站点对size有硬上限超了不报错直接按 20 返回你在日志里看到每页都是 20 条还以为是站点改了版。稳妥做法是先手动请求一次size100比对返回条数和响应里的总数确认上限再定。终止条件也别只写if not items: break——部分站点翻到末页之后不返回空列表而是把你重定向回第一页于是循环永远不结束、数据开始循环重复。可靠的终止条件是三选一响应里带的总数除以每页条数、页面上下一页按钮是否还存在、连续两页的首条公开号是否相同。第三种最土但最不容易翻车。2.3 退避与超时把偶尔 429变成可观测事件专利站点的限流通常不是直接封 IP而是给你一串 429 或 503混着几次正常响应。这时候裸写time.sleep(1)是不够的固定间隔等于给服务端一个稳定的请求节奏反而更容易被识别退避太激进又会把整批任务拖成几小时。我一般用指数退避加随机抖动并且把每次重试的原因记下来import time, random, logging log logging.getLogger(fetcher) def get_with_backoff(session, url, paramsNone, tries4, base1.6): last None for i in range(tries): try: r session.get(url, paramsparams, timeout(5, 20)) if r.status_code in (429, 503): # 优先尊重服务端给的 Retry-After没有就用退避值 wait float(r.headers.get(Retry-After, base ** i)) raise RuntimeError(fthrottled {r.status_code} wait{wait:.1f}s) r.raise_for_status() return r except Exception as e: last e if i tries - 1: break wait base ** i random.uniform(0, 0.6) log.warning(retry %d/%d url%s params%s err%s, i 1, tries, url, params, e) time.sleep(wait) raise RuntimeError(fgive up after {tries} tries: {last})base1.6意味着等待序列大致是 1.6s、2.56s、4.1s加上抖动后不会形成固定节奏。tries4是经验值超过四次还失败基本不是抖动问题而是参数错了、会话过期了、或者这个公开号的详情页本身就不存在继续重试只是浪费时间。日志里把params一并打出来很关键——排错时你能一眼看出是第几页开始出问题而不是面对一堆一模一样的 URL 干瞪眼。3. 解析层列表页与详情页的字段分工3.1 列表页只取骨架字段权威数据留给详情页新手常犯的错是想在列表页一次抓全标题、申请人、摘要、发明人、法律状态全都要。结果列表页的摘要被截断成 80 个字申请人字段只给第一申请人法律状态干脆不渲染。正确分工是列表页只负责三件事——拿到公开号、拿到详情页链接、拿到一个能判断是否需要继续抓的粗筛字段公开日、主分类号。其余字段全部进详情页取。这个分工还有个实际好处详情页请求是可以失败重试的而列表页请求一旦失败你丢失的是整页 20 条记录的线索。所以列表页的响应我一般先原样落盘成 HTML 快照再离线解析。跑批量任务时这个习惯能救命——站点改版之后你不用重新翻五百页直接拿快照重放解析逻辑就行。3.2 选择器外置成配置兜底用正则专利页面的 DOM 结构不算复杂但字段名和 class 命名各站差异很大硬编码在函数里换一个站点就要全文替换。把 XPath 抽成字典解析函数只负责按配置取值from lxml import html LIST_XP { pub_no: .//td[data-colpubno]/text(), title: .//td[data-coltitle]//text(), applicant: .//td[data-colapplicant]/text(), pub_date: .//td[data-colpubdate]/text(), detail: .//td[data-coltitle]/a/href, } def parse_list(page_html: str, base_url: str): tree html.fromstring(page_html) rows [] for tr in tree.xpath(//table[idresult-list]//tbody/tr): item {} for key, xp in LIST_XP.items(): got tr.xpath(xp) # 取第一个非空文本多文本节点比如标题里嵌了 em 高亮要拼起来 item[key] .join(t.strip() for t in got if t.strip()) if got else None if not item.get(pub_no): continue # 公开号是主键缺了这条就没意义 item[detail] base_url.rstrip(/) / item[detail].lstrip(/) rows.append(item) return rows几点说明//text()会把标题里em高亮标签拆成多个文本节点所以必须.join而不是取[0]否则标题永远只有前半截这是最容易被忽略的一处。if not item.get(pub_no)是必要的过滤——列表页里常混着广告位或推荐专利的行这些行没有公开号进库会污染主键。公开号本身要做规范化因为同一个号在不同页面可能写成CN 123456789 A、CN123456789A、cn123456789a。统一成两位国家代码 数字 一位字母的大写无空格形式是后面去重和文件命名的前提。兜底逻辑用一个正则就够import re PUB_NO_RE re.compile(r([A-Z]{2})\s*(\d{6,12})\s*([A-Z]\d?)?, re.I) def norm_pub_no(raw: str) - str | None: if not raw: return None m PUB_NO_RE.search(raw.replace( , )) if not m: return None country, digits, kind m.group(1).upper(), m.group(2), (m.group(3) or ).upper() return f{country}{digits}{kind}\d{6,12}的下界给 6 是因为部分国家的公开号数字位较短上界 12 足够覆盖现有格式。kind那一段用[A-Z]\d?是为了兼容A、B1、U这类文献种类代码。这个正则只是兜底正常路径还是走 XPath正则用于处理那些结构异常的边角记录。3.3 落库用公开号做主键状态位驱动流程拿到数据之后不要直接写 CSV。CSV 的问题是没法做增量你跑到第 8000 条中断重跑时不知道哪些已经下过 PDF。用 SQLite 建一张表公开号做主键加一个状态字段整个流程就变成可续跑的CREATE TABLE IF NOT EXISTS patent ( pub_no TEXT PRIMARY KEY, title TEXT NOT NULL, applicant TEXT, pub_date TEXT, detail_url TEXT, pdf_path TEXT, status TEXT NOT NULL DEFAULT new, -- new / parsed / done / fail err_msg TEXT, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_status ON patent(status);写入用 UPSERT语义是新记录插入老记录只更新标题这类易变字段已经下好的 PDF 路径不能被覆盖INSERT INTO patent(pub_no, title, applicant, pub_date, detail_url, updated_at) VALUES(?, ?, ?, ?, ?, datetime(now)) ON CONFLICT(pub_no) DO UPDATE SET title excluded.title, applicant COALESCE(excluded.applicant, patent.applicant), pdf_path COALESCE(patent.pdf_path, excluded.pdf_path), updated_at excluded.updated_at;COALESCE(patent.pdf_path, excluded.pdf_path)这个写法是关键先取旧值旧值为空才用新值等价于已有成果不覆盖。状态字段把整条链路切成四段重跑时WHERE statusnew就只处理没解析过的WHERE statusparsed就只补下载。索引建在status上几万条数据量下这个索引能让续跑查询从全表扫描降到毫秒级。4. 下载层PDF 流式落盘、断点续传与文件头校验4.1 流式下载与 .part 临时文件直接用r.content然后open(...,wb).write()在专利 PDF 上是危险动作。专利全文 PDF 常见几 MB 到几十 MB图片多的能到上百 MB一次性读进内存跑几十个并发就把内存吃满了。更麻烦的是中断进程被 CtrlC 或者网络断了磁盘上留下一个半截文件下次重跑你以为已经下好了实际是个损坏的 PDF。正确做法是streamTrue分块写并且先写.part校验通过再改名from pathlib import Path def download_pdf(session, url: str, dest: Path, chunk_size: int 64 * 1024) - Path: dest.parent.mkdir(parentsTrue, exist_okTrue) part dest.with_name(dest.name .part) headers {} if part.exists() and part.stat().st_size 0: # 断点续传从已有字节数开始请求 headers[Range] fbytes{part.stat().st_size}- with session.get(url, headersheaders, streamTrue, timeout(5, 60)) as r: if r.status_code 416: # Range 超出文件长度说明已经下完 part.replace(dest) return dest r.raise_for_status() ctype r.headers.get(Content-Type, ).lower() if pdf not in ctype and octet-stream not in ctype: raise ValueError(funexpected content-type: {ctype}) # 206 表示续传成功追加写否则是完整响应从头写 mode ab if r.status_code 206 else wb if mode wb and part.exists(): part.unlink() with open(part, mode) as f: for block in r.iter_content(chunk_size): if block: f.write(block) with open(part, rb) as f: head f.read(5) if head ! b%PDF-: part.unlink() raise ValueError(bad pdf header, likely an error page) part.replace(dest) return dest逻辑说明Range请求头只在.part存在且非空时加服务端返回 206 就走追加返回 200 说明它不支持续传只能整份重下——这两种情况在代码里靠r.status_code分流不能想当然认为加了 Range 就一定续传。chunk_size64*1024是折中值太小系统调用次数多太大内存峰值高64KB 在几 MB 的文件上表现最均衡。读取超时给到 60 秒是因为最后几个 chunk 有时会卡在服务端的输出缓冲上20 秒容易误判成失败。4.2 目录与命名用公开号不要用标题用标题做文件名是另一个高频翻车点。专利标题里常有/、、、在 Linux 上:合法但在 Windows 上非法/在任何系统上都会把路径切断更麻烦的是同名标题极多一种数据处理方法这种标题一抓一大把互相覆盖。命名规范我一般定成三层data/{公开日年份}/{公开日月份}/{规范公开号}.pdf。年份月份两级目录是为了避免单目录下堆十几万个文件——某些文件系统在单目录条目过万之后ls和随机读取都会明显变慢。文件名只用norm_pub_no处理过的公开号字符集限制在[A-Z0-9]天然没有非法字符。def build_dest(root: Path, pub_no: str, pub_date: str) - Path: # pub_date 形如 2024-03-15取年月建目录缺失时归到 unknown year, month (pub_date[:4], pub_date[5:7]) if len(pub_date) 7 else (unknown, unknown) safe .join(ch for ch in pub_no.upper() if ch.isalnum()) return root / year / month / f{safe}.pdfsafe那行的字符过滤是最后一道保险。即使上游norm_pub_no出了 bug 漏进来带/的字符串这里也会被剥掉不会造成路径穿越。4.3 三层校验状态码、Content-Type、文件头只检查r.status_code 200是不够的。专利站点在会话过期、请求频率过高、或者该专利全文未公开时经常返回 200 一个 HTML 提示页。这个页面被存成.pdf体积通常是 1530KB肉眼看不出来等你批量用 PDF 解析库读的时候才集体报错。三层校验的顺序是有讲究的状态码最便宜先做Content-Type次之能过滤掉绝大多数 HTML 伪装文件头%PDF-最贵要读磁盘但最可靠因为有些站点返回的Content-Type是application/octet-stream光看响应头判断不了。三层都过不了的文件直接删掉并把状态标成fail、把err_msg记下来比留一堆坏文件强。实践中fail里有相当一部分是该专利确实没有全文 PDF这类不需要重试人工看一眼err_msg就能批量清理。5. 排查与避坑五条翻车记录5.1 现象前 30 条正常第 31 条开始全 403现象很规律——不管换什么关键词都是抓满三十条左右开始报 403等十几分钟又恢复。原因不是 IP 被封而是会话里的检索上下文过期了。多数检索站点会给每个会话分配一个短时效的检索标识翻页和详情页请求都要带上它超过时限或者翻页次数过多服务端认为这个会话异常直接返回 403。这时候换 IP、换 UA 都没用因为你换不掉那个已经失效的会话状态。解决是两级一是每抓 2530 条主动重建一次 Session重新走一遍检索首页把上下文刷新二是在get_with_backoff里对 403 做特殊处理不要盲目重试而是抛一个SessionExpired异常由外层捕获后重建会话再续跑。重试次数只给一次因为重复用失效会话请求只会让站点把你的行为特征记得更清楚。5.2 现象翻到第 10 页数据开始和第 1 页重复跑了一千多条去重之后只剩四百条日志里第 10 页的首条公开号和第 1 页一模一样。原因是排序参数没固定或者站点对深翻页做了截断——很多检索系统出于性能考虑只允许翻到前 N 页常见是 100 页或 50 页超过之后不报错直接把你送回第一页。这个时候你的页码在涨结果集却在原地打转。解决分两步先在请求里显式带上固定的排序参数排除相关度排序导致结果漂移这个因素再在翻页循环里加一条硬判断——记录上一页的首条公开号如果当前页首条与之前任意一页相同立刻终止并打印告警。真要拿全量数据正确做法是缩小检索范围按年份、按分类号切片让每个切片的结果集都落在站点允许的翻页深度以内而不是硬翻。5.3 现象PDF 下下来只有 20KB打开是空白页文件大小异常统一都在 1530KBPDF 阅读器打开报文件已损坏。原因是详情页的 PDF 链接不是直链而是一个跳转接口。你请求它服务端返回 200Content-Type是text/html内容是正在跳转或者请登录后下载。代码里没校验类型直接写盘就得到一堆 HTML。解决办法就是第 4.3 节那三层校验特别是文件头判断。另外要看一下这个跳转接口需不需要带上详情页的Referer——有一部分站点的下载接口会校验来源页缺Referer就给你返回提示页而不是文件。补上Referer: 详情页URL之后同一批链接的下载成功率能从六成提到九成以上。5.4 现象文件名报OSError: Invalid argument整批任务中断在本地 Windows 上跑没问题换到服务器上跑跑到某个公开号就抛异常退出。原因是标题里带了全角冒号或者斜杠被直接拼进了文件名。Windows 上非法字符是\ / : * ? |Linux 上只有/和\0非法所以同一份代码在两个平台上表现不同这种问题在本地测不出来。解决办法是根本不用标题做文件名见 4.2并且在写盘前统一跑一遍字符白名单过滤。另外整个下载循环里单条记录的异常必须被try/except包住——一条记录失败不该让整批任务退出把它标成fail记进err_msg跑完再看失败清单这才是批量任务该有的容错粒度。5.5 现象中断后重跑几千条全部重下跑到一半网络断了重跑脚本明明磁盘上已经有 PDF 了还是从头下了一遍白等两小时。原因是下载前的判断写成了文件存在就跳过而中断留下的半截文件也满足存在于是被当成已完成或者反过来判断条件写得太严路径拼错一个字符所有文件都判为不存在。稳妥的做法是把是否存在改成是否有效文件存在且大于 10KB 且文件头是%PDF-三个条件同时成立才跳过。同时.part临时文件的存在本身就代表下过但没完成续传逻辑从它的大小接着要字节而不是从零开始。这套判断我一般单独封成一个is_valid_pdf(path)函数下载前和校验脚本里共用一份避免两处逻辑不一致。6. 增量抓取与抽样校验让第二次跑比第一次快十倍第一次跑全量是最慢的之后每次跑都应该是增量。判断依据不是文件时间而是库里pub_date的最大值——只请求比这个日期更新的记录def pending_rows(conn, limit: int 500): sql SELECT pub_no, detail_url FROM patent WHERE status new ORDER BY pub_date DESC LIMIT ? return conn.execute(sql, (limit,)).fetchall() def run_batch(session, conn, worker, workers: int 3): from concurrent.futures import ThreadPoolExecutor, as_completed rows pending_rows(conn) with ThreadPoolExecutor(max_workersworkers) as ex: futs {ex.submit(worker, pub_no, url): pub_no for pub_no, url in rows} for fut in as_completed(futs): pub_no futs[fut] try: path fut.result() conn.execute(UPDATE patent SET statusdone, pdf_path?, updated_atdatetime(now) WHERE pub_no?, (str(path), pub_no)) except Exception as e: conn.execute(UPDATE patent SET statusfail, err_msg?, updated_atdatetime(now) WHERE pub_no?, (str(e)[:200], pub_no)) conn.commit()workers3是刻意压低的。专利站点对并发的容忍度普遍不高3 个线程配 1 秒间隔实测吞吐比 10 个线程还高因为后者触发限流后大量时间耗在退避上。这里的worker函数内部要自己建 Session——虽然requests.Session在 CPython 下共享连接池多数时候能跑但 Cookie 是被多线程共同改写的会话过期重建时会互相踩每线程一个 Session 更干净。落库那步必须串行SQLite 的写锁在并发下会抛database is locked把commit放在主线程是省事的做法数据量再大就换成写队列。校验这块我固定跑三个抽样成本很低但能提前发现问题抽样项做法预期结果总数一致性检索页显示的总条数与库内COUNT(*)比对差异在个位数以内受深翻页截断影响PDF 有效性随机抽 50 条读文件头 用 PDF 库取页数全部%PDF-页数 0字段完整性抽 20 条比对详情页原文标题、申请人、公开日三项无空值、无截断第二项里页数校验值得单独做一次因为文件头对但内容截断的 PDF 是存在的续传时服务端返回的 206 范围算错了。用pypdf读一下len(reader.pages)能抓出这类隐蔽损坏。血泪经验是不要等跑完一万条再校验。每批 500 条结束就跑一次上面三项问题能在十分钟内暴露而不是第二天早上发现整批 PDF 全是提示页。我现在的习惯是脚本跑起来之后第一批done出现就立刻手工打开三个 PDF 看一眼——文件大小、页数、正文首页的公开号对不对得上。这三眼看下去能挡掉后面百分之八十的后悔药。希望帮到你。本文还有配套的精品资源点击获取