企业官网建设流程全解析

简介：这是一份面向知识产权研究、数据分析与法律服务场景的Python专利爬虫项目源码，适合已掌握Python基础、希望了解真实爬虫工程结构的中级学习者。项目以中介者模式协调请求调度，通过延时控制、并发限制等方式降低对目标站点的访问压力，并围绕查询、页面采集、条目收集、进度控制、Excel导出与等待引擎等模块划分职责，形成从检索到结果落地的完整链路，可用于专利元数据收集与后续的趋势分析、竞争监控等实践。压缩包共56个文件，以23个py源码与22个pyc编译文件为主，另有8个xml配置及说明文档、可执行文件等，整体约17.36MB，源码与配置分离，便于按层阅读。已有5499人浏览学习。读者可从中获取分层清晰的工程目录、请求节流与异常处理的实现思路、数据实体与枚举的建模方式，以及将抓取结果导出为表格的工具脚本，作为爬虫设计与合规采集的参考。

1. 专利数据抓取不是 requests 一把梭：先把下载链路拆成四段

写专利爬虫的人，第一版代码几乎都是同一个样子：拼一个检索 URL，for 循环翻页，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/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6", "Referer": referer, "Connection": "keep-alive", }) # 连接池：pool_connections 是保留的连接池数量，pool_maxsize 是单池最大连接数 # max_retries 设 0，重试交给我们自己的退避函数，方便记录失败原因 adapter = HTTPAdapter(pool_connections=4, pool_maxsize=8, max_retries=0) s.mount("https://", adapter) s.mount("http://", adapter) return s

逻辑上就三件事：统一请求头、复用连接、把重试权收回来。参数上，pool_connections和pool_maxsize不用开大，单站点并发压到 2～4 时，8 个池位足够，开太大反而容易在短时间内打出密集请求。max_retries=0是有意为之——urllib3 自带的重试对 429 的处理不够透明，你拿不到"第几次重试、因为什么失败"，排错时就是个黑匣子。timeout建议用元组(连接超时, 读取超时)，例如(5, 20)：连接 5 秒还不通说明网络或站点侧有问题，读取给 20 秒是因为专利详情页偶尔会带较大的附图，单值 timeout 会把这两类问题糊在一起。

提示：动手前先看目标站点的 robots 与使用条款，把单域名并发控制在 1～2，翻页之间留 1～2 秒间隔。这不是客套，是为了让后面几万条能跑完。

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 条"还以为是站点改了版。稳妥做法是先手动请求一次size=100，比对返回条数和响应里的总数，确认上限再定。终止条件也别只写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, params=None, tries=4, base=1.6): last = None for i in range(tries): try: r = session.get(url, params=params, timeout=(5, 20)) if r.status_code in (429, 503): # 优先尊重服务端给的 Retry-After，没有就用退避值 wait = float(r.headers.get("Retry-After", base ** i)) raise RuntimeError(f"throttled {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(f"give up after {tries} tries: {last}")

base=1.6意味着等待序列大致是 1.6s、2.56s、4.1s，加上抖动后不会形成固定节奏。tries=4是经验值：超过四次还失败，基本不是抖动问题，而是参数错了、会话过期了、或者这个公开号的详情页本身就不存在，继续重试只是浪费时间。日志里把params一并打出来很关键——排错时你能一眼看出是第几页开始出问题，而不是面对一堆一模一样的 URL 干瞪眼。

3. 解析层：列表页与详情页的字段分工

3.1 列表页只取骨架字段，权威数据留给详情页

新手常犯的错是想在列表页一次抓全：标题、申请人、摘要、发明人、法律状态全都要。结果列表页的摘要被截断成 80 个字，申请人字段只给第一申请人，法律状态干脆不渲染。正确分工是：列表页只负责三件事——拿到公开号、拿到详情页链接、拿到一个能判断是否需要继续抓的粗筛字段（公开日、主分类号）。其余字段全部进详情页取。

这个分工还有个实际好处：详情页请求是可以失败重试的，而列表页请求一旦失败，你丢失的是整页 20 条记录的线索。所以列表页的响应我一般先原样落盘成 HTML 快照，再离线解析。跑批量任务时这个习惯能救命——站点改版之后，你不用重新翻五百页，直接拿快照重放解析逻辑就行。

3.2 选择器外置成配置，兜底用正则

专利页面的 DOM 结构不算复杂，但字段名和 class 命名各站差异很大，硬编码在函数里，换一个站点就要全文替换。把 XPath 抽成字典，解析函数只负责按配置取值：

from lxml import html LIST_XP = { "pub_no": './/td[@data-col="pubno"]/text()', "title": './/td[@data-col="title"]//text()', "applicant": './/td[@data-col="applicant"]/text()', "pub_date": './/td[@data-col="pubdate"]/text()', "detail": './/td[@data-col="title"]/a/@href', } def parse_list(page_html: str, base_url: str): tree = html.fromstring(page_html) rows = [] for tr in tree.xpath('//table[@id="result-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 status='new'就只处理没解析过的，WHERE status='parsed'就只补下载。索引建在status上，几万条数据量下这个索引能让续跑查询从全表扫描降到毫秒级。

4. 下载层：PDF 流式落盘、断点续传与文件头校验

4.1 流式下载与 .part 临时文件

直接用r.content然后open(...,'wb').write()在专利 PDF 上是危险动作。专利全文 PDF 常见几 MB 到几十 MB，图片多的能到上百 MB，一次性读进内存，跑几十个并发就把内存吃满了。更麻烦的是中断：进程被 Ctrl+C 或者网络断了，磁盘上留下一个半截文件，下次重跑你以为已经下好了，实际是个损坏的 PDF。

正确做法是stream=True分块写，并且先写.part，校验通过再改名：

from pathlib import Path def download_pdf(session, url: str, dest: Path, chunk_size: int = 64 * 1024) -> Path: dest.parent.mkdir(parents=True, exist_ok=True) part = dest.with_name(dest.name + ".part") headers = {} if part.exists() and part.stat().st_size > 0: # 断点续传：从已有字节数开始请求 headers["Range"] = f"bytes={part.stat().st_size}-" with session.get(url, headers=headers, stream=True, 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(f"unexpected 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_size=64*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}.pdf"

safe那行的字符过滤是最后一道保险。即使上游norm_pub_no出了 bug 漏进来带/的字符串，这里也会被剥掉，不会造成路径穿越。

4.3 三层校验：状态码、Content-Type、文件头

只检查r.status_code == 200是不够的。专利站点在会话过期、请求频率过高、或者该专利"全文未公开"时，经常返回 200 + 一个 HTML 提示页。这个页面被存成.pdf，体积通常是 15～30KB，肉眼看不出来，等你批量用 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 都没用，因为你换不掉那个已经失效的会话状态。

解决是两级：一是每抓 25～30 条主动重建一次 Session，重新走一遍检索首页把上下文刷新；二是在get_with_backoff里对 403 做特殊处理，不要盲目重试，而是抛一个SessionExpired异常，由外层捕获后重建会话再续跑。重试次数只给一次，因为重复用失效会话请求，只会让站点把你的行为特征记得更清楚。

5.2 现象：翻到第 10 页，数据开始和第 1 页重复

跑了一千多条，去重之后只剩四百条，日志里第 10 页的首条公开号和第 1 页一模一样。

原因是排序参数没固定，或者站点对深翻页做了截断——很多检索系统出于性能考虑，只允许翻到前 N 页（常见是 100 页或 50 页），超过之后不报错，直接把你送回第一页。这个时候你的页码在涨，结果集却在原地打转。

解决分两步：先在请求里显式带上固定的排序参数，排除"相关度排序导致结果漂移"这个因素；再在翻页循环里加一条硬判断——记录上一页的首条公开号，如果当前页首条与之前任意一页相同，立刻终止并打印告警。真要拿全量数据，正确做法是缩小检索范围（按年份、按分类号切片），让每个切片的结果集都落在站点允许的翻页深度以内，而不是硬翻。

5.3 现象：PDF 下下来只有 20KB，打开是空白页

文件大小异常统一，都在 15～30KB，PDF 阅读器打开报"文件已损坏"。

原因是详情页的 PDF 链接不是直链，而是一个跳转接口。你请求它，服务端返回 200，Content-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_workers=workers) 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 status='done', pdf_path=?, updated_at=datetime('now') " "WHERE pub_no=?", (str(path), pub_no)) except Exception as e: conn.execute("UPDATE patent SET status='fail', err_msg=?, updated_at=datetime('now') " "WHERE pub_no=?", (str(e)[:200], pub_no)) conn.commit()

workers=3是刻意压低的。专利站点对并发的容忍度普遍不高，3 个线程配 1 秒间隔，实测吞吐比 10 个线程还高，因为后者触发限流后大量时间耗在退避上。这里的worker函数内部要自己建 Session——虽然requests.Session在 CPython 下共享连接池多数时候能跑，但 Cookie 是被多线程共同改写的，会话过期重建时会互相踩，每线程一个 Session 更干净。落库那步必须串行，SQLite 的写锁在并发下会抛database is locked，把commit放在主线程是省事的做法；数据量再大就换成写队列。

校验这块我固定跑三个抽样，成本很低但能提前发现问题：

第二项里页数校验值得单独做一次，因为文件头对但内容截断的 PDF 是存在的（续传时服务端返回的 206 范围算错了）。用pypdf读一下len(reader.pages)，能抓出这类隐蔽损坏。

血泪经验是：不要等跑完一万条再校验。每批 500 条结束就跑一次上面三项，问题能在十分钟内暴露，而不是第二天早上发现整批 PDF 全是提示页。我现在的习惯是脚本跑起来之后，第一批done出现就立刻手工打开三个 PDF 看一眼——文件大小、页数、正文首页的公开号对不对得上。这三眼看下去，能挡掉后面百分之八十的后悔药。希望帮到你。

本文还有配套的精品资源，点击获取