Python爬虫实战：基于 OpenAlex 抓取机构论文列表，用 requests+SQLite 构建科研分析底表

1. 科研数据采集的真实痛点：为什么你抓的机构论文列表总是不全

做科研机构分析时，最让人头疼的不是画图，而是底表。你想统计某高校近五年的发文量、平均被引、合作网络，第一步就得有一张干净、稳定、能反复更新的论文明细表。很多人第一反应是去抓机构主页的论文列表页，结果发现分页只有前几页、字段残缺、标题还带一堆 HTML 标签，清洗成本比抓取本身还高。

OpenAlex 是当前学术元数据领域比较适合做这类底表的数据源。它把机构、作者、论文、来源都抽象成带唯一 ID 的实体，提供正式的 REST API，支持 search、filter、sort、select、cursor 深分页。你可以把它理解成一个“学术数据的结构化仓库”，而不是一个需要你逐页解析的网站。它适合谁？适合做科研管理、学科评估、机构对标、文献计量分析的同学，也适合想练手 API 采集 + SQLite 落库的 Python 学习者。

我试过直接拿机构名去 filter works，结果返回一堆同名不同校的论文，后来才改成“先 search 拿机构 ID，再按 ID 拉论文列表”的两段式流程，稳定性立刻上来了。这篇就按这个思路，用 requests + SQLite 搭一张可复用的科研分析底表，交付可复制的请求参数、分页与限流配置、建表 SQL 和字段校验脚本。

2. TaoToken 前置：给采集脚本配一个稳定的模型调用入口

采集脚本本身不依赖大模型，但你在做科研分析时，往往需要顺手让模型帮你做字段归一、标题翻译、主题归类，或者把一批论文摘要压缩成机构画像。这时候如果每次都要手动切不同厂商的 API，脚本会变得很碎。TaoToken 提供的是统一的模型调用入口，兼容常见接口格式，适合把“采集 + 分析”串成一条流水线。

它的定位不是替代你的编辑器，也不是让你绕过什么限制，而是把模型调用这件事收敛到一个 Key、一个地址上。你可以在官网了解整体能力，在模型对话页快速试跑，在 Coding Plan 里做长期编码和 Agent 任务，在控制台管理用量，在 API Keys 页面生成密钥，接入文档里能看到具体的请求格式。

对这篇教程来说，TaoToken 的价值在于：当你把 OpenAlex 底表建好之后，可以立刻接一段模型调用，对 paper_title 做主题打标，或者对机构做年度研究热点摘要。采集层用 requests + SQLite，分析层用统一模型入口，两边互不干扰。

3. 可复制配置：请求参数、分页与限流一次写清

3.1 环境与依赖

建议 Python 3.11，依赖只有三个：

pip install requests beautifulsoup4 lxml

项目结构建议这样组织，后面所有代码都按这个路径放：

openalex_institution_papers/ ├── requirements.txt ├── main.py ├── openalex_client.py ├── parser_utils.py ├── storage.py ├── data/ │ └── openalex_papers.db └── output/ └── openalex_institution_papers.csv

3.2 请求层：headers、timeout、重试与 cursor 分页

请求层要解决四件事：headers 怎么配、timeout 给多少、失败怎么重试、分页怎么走。OpenAlex 的 works 列表支持 cursor 深分页，基础 page 分页在结果集较大时会受限，所以正式采集直接走 cursor。

# openalex_client.py import os import re import time from typing import Dict, Any, List, Iterator, Optional import requests from bs4 import BeautifulSoup from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def extract_short_id(openalex_url_or_id: str) -> str: if not openalex_url_or_id: return "" text = openalex_url_or_id.strip() match = re.search(r'([IWASTFP]\d+)$', text, flags=re.IGNORECASE) return match.group(1).upper() if match else text.upper() class OpenAlexClient: def __init__( self, api_key: Optional[str] = None, timeout: int = 30, per_page: int = 100, sleep_seconds: float = 0.25, max_retries: int = 5, ) -> None: self.api_base = "https://api.openalex.org" self.web_base = "https://openalex.org" self.explore_base = "https://explore.openalex.org" self.api_key = (api_key or os.getenv("OPENALEX_API_KEY", "")).strip() self.timeout = timeout self.per_page = per_page self.sleep_seconds = sleep_seconds self.max_retries = max_retries self.session = self._build_session() def _build_session(self) -> requests.Session: session = requests.Session() retry = Retry( total=3, connect=3, read=3, backoff_factor=0.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset(["GET"]), raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=10, pool_maxsize=10) session.mount("https://", adapter) session.mount("http://", adapter) session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/123.0.0.0 Safari/537.36 " "OpenAlexInstitutionCrawler/1.0" ), "Accept": "application/json, text/html;q=0.9, */*;q=0.8", "Referer": "https://explore.openalex.org/", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Connection": "keep-alive", }) return session def _request( self, url: str, params: Optional[Dict[str, Any]] = None, expect_json: bool = True, attach_api_key: bool = True, ) -> Any: params = dict(params or {}) if attach_api_key and self.api_key: params.setdefault("api_key", self.api_key) last_error = None for attempt in range(self.max_retries): try: response = self.session.get(url, params=params, timeout=self.timeout) if response.status_code == 429: sleep_time = min(2 ** attempt, 30) print(f"[WARN] hit 429, backoff {sleep_time}s -> {url}") time.sleep(sleep_time) continue if response.status_code in (500, 502, 503, 504): sleep_time = min(2 ** attempt, 20) print(f"[WARN] server error {response.status_code}, retry in {sleep_time}s") time.sleep(sleep_time) continue response.raise_for_status() return response.json() if expect_json else response.text except requests.RequestException as exc: last_error = exc if attempt == self.max_retries - 1: break sleep_time = min(2 ** attempt, 20) print(f"[WARN] request failed: {exc}, retry in {sleep_time}s") time.sleep(sleep_time) raise RuntimeError(f"Request failed after retries: {url}, last_error={last_error}") def search_institutions(self, keyword: str, limit: int = 10) -> List[Dict[str, Any]]: url = f"{self.api_base}/institutions" params = { "search": keyword, "per_page": min(limit, 100), "select": "id,display_name,country_code,type,works_count,cited_by_count,homepage_url", } data = self._request(url, params=params, expect_json=True, attach_api_key=True) return data.get("results", []) def get_institution(self, institution_id: str) -> Dict[str, Any]: short_id = extract_short_id(institution_id) url = f"{self.api_base}/institutions/{short_id}" return self._request(url, expect_json=True, attach_api_key=True) def fetch_institution_web_snapshot(self, institution_id: str) -> Dict[str, str]: short_id = extract_short_id(institution_id).lower() candidates = [ f"{self.web_base}/institutions/{short_id}", f"{self.explore_base}/institutions/{short_id}", ] for url in candidates: try: html = self._request(url, expect_json=False, attach_api_key=False) soup = BeautifulSoup(html, "lxml") title = soup.title.get_text(strip=True) if soup.title else "" meta_desc = "" tag = soup.find("meta", attrs={"name": "description"}) if tag and tag.get("content"): meta_desc = tag["content"].strip() if title or meta_desc: return {"web_url": url, "html_title": title, "meta_description": meta_desc} except Exception: continue return {"web_url": candidates[0], "html_title": "", "meta_description": ""} def iter_works_by_institution( self, institution_id: str, year_start: Optional[int] = None, year_end: Optional[int] = None, max_pages: Optional[int] = None, ) -> Iterator[Dict[str, Any]]: short_id = extract_short_id(institution_id) filter_parts = [f"authorships.institutions.id:{short_id}"] if year_start is not None and year_end is not None: filter_parts.append(f"publication_year:{year_start}-{year_end}") elif year_start is not None: filter_parts.append(f"publication_year:>{year_start - 1}") elif year_end is not None: filter_parts.append(f"publication_year:<{year_end + 1}") cursor = "*" page_count = 0 while True: page_count += 1 params = { "filter": ",".join(filter_parts), "sort": "publication_date:desc", "per_page": self.per_page, "cursor": cursor, "select": "id,title,display_name,publication_year,cited_by_count,authorships", } data = self._request( f"{self.api_base}/works", params=params, expect_json=True, attach_api_key=True, ) results = data.get("results", []) meta = data.get("meta", {}) print( f"[INFO] works page={page_count}, " f"count_in_page={len(results)}, next_cursor={bool(meta.get('next_cursor'))}" ) if not results: break for item in results: yield item cursor = meta.get("next_cursor") if not cursor: break if max_pages is not None and page_count >= max_pages: print(f"[INFO] reached max_pages={max_pages}, stop early.") break time.sleep(self.sleep_seconds)

这里有几个参数值得单独说。per_page最大 100，别贪多；sleep_seconds默认 0.25 秒，是给单机采集留的礼貌间隔；max_retries控制 429 和 5xx 的退避次数。API Key 通过环境变量注入，不写死在代码里：

# Linux / macOS export OPENALEX_API_KEY="your_api_key_here" # Windows PowerShell $env:OPENALEX_API_KEY="your_api_key_here"

3.3 解析层：字段映射与容错

解析层只做一件事：把 API 返回的 JSON 变成底表记录。字段映射如下：

# parser_utils.py import re from typing import Dict, Any, List, Optional def normalize_name(text: str) -> str: if not text: return "" text = text.strip().lower() return re.sub(r"[\W_]+", "", text) def choose_best_institution(candidates: List[Dict[str, Any]], query: str) -> Optional[Dict[str, Any]]: if not candidates: return None q = normalize_name(query) exact = [x for x in candidates if normalize_name(x.get("display_name", "")) == q] if exact: return sorted(exact, key=lambda x: x.get("works_count", 0), reverse=True)[0] contains = [x for x in candidates if q and q in normalize_name(x.get("display_name", ""))] if contains: return sorted(contains, key=lambda x: x.get("works_count", 0), reverse=True)[0] return sorted(candidates, key=lambda x: x.get("works_count", 0), reverse=True)[0] def extract_short_id(openalex_url_or_id: str) -> str: if not openalex_url_or_id: return "" text = openalex_url_or_id.strip() match = re.search(r"([IWASTFP]\d+)$", text, flags=re.IGNORECASE) return match.group(1).upper() if match else text.upper() def work_to_record(institution_name: str, item: Dict[str, Any]) -> Optional[Dict[str, Any]]: openalex_raw = item.get("id", "") openalex_id = extract_short_id(openalex_raw) if not openalex_id: return None authorships = item.get("authorships") or [] title = item.get("display_name") or item.get("title") or "" return { "institution_name": institution_name, "paper_title": title.strip(), "year": item.get("publication_year"), "author_count": len(authorships), "cited_by_count": item.get("cited_by_count", 0) or 0, "openalex_id": openalex_id, "openalex_url": openalex_raw, }

缺字段是常态，不是异常。title 和 display_name 二选一，authorships 缺失按空数组处理，publication_year 缺失记 None，cited_by_count 缺失默认 0，id 缺失直接跳过，避免脏主键进库。

3.4 存储层：建表 SQL 与 UPSERT 去重

SQLite 建表用openalex_id做唯一键，天然支持 UPSERT，重复跑不会产生重复行。

# storage.py import csv import os import sqlite3 from datetime import datetime from typing import Iterable, Dict, Any, List, Tuple def ensure_parent_dir(path: str) -> None: folder = os.path.dirname(path) if folder: os.makedirs(folder, exist_ok=True) def init_db(db_path: str) -> None: ensure_parent_dir(db_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ CREATE TABLE IF NOT EXISTS papers ( id INTEGER PRIMARY KEY AUTOINCREMENT, institution_name TEXT NOT NULL, paper_title TEXT, year INTEGER, author_count INTEGER, cited_by_count INTEGER, openalex_id TEXT NOT NULL UNIQUE, openalex_url TEXT, crawled_at TEXT NOT NULL ) """ ) cur.execute("CREATE INDEX IF NOT EXISTS idx_papers_year ON papers(year)") cur.execute("CREATE INDEX IF NOT EXISTS idx_papers_inst ON papers(institution_name)") cur.execute("CREATE INDEX IF NOT EXISTS idx_papers_cited ON papers(cited_by_count)") conn.commit() conn.close() def upsert_papers(db_path: str, rows: Iterable[Dict[str, Any]]) -> int: conn = sqlite3.connect(db_path) cur = conn.cursor() now = datetime.utcnow().isoformat(timespec="seconds") affected = 0 for row in rows: cur.execute( """ INSERT INTO papers ( institution_name, paper_title, year, author_count, cited_by_count, openalex_id, openalex_url, crawled_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(openalex_id) DO UPDATE SET institution_name = excluded.institution_name, paper_title = excluded.paper_title, year = excluded.year, author_count = excluded.author_count, cited_by_count = excluded.cited_by_count, openalex_url = excluded.openalex_url, crawled_at = excluded.crawled_at """, ( row.get("institution_name"), row.get("paper_title"), row.get("year"), row.get("author_count"), row.get("cited_by_count"), row.get("openalex_id"), row.get("openalex_url"), now, ), ) affected += 1 conn.commit() conn.close() return affected def export_csv(db_path: str, csv_path: str) -> None: ensure_parent_dir(csv_path) conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ SELECT institution_name, paper_title, year, author_count, cited_by_count, openalex_id FROM papers ORDER BY year DESC NULLS LAST, cited_by_count DESC """ ) rows = cur.fetchall() conn.close() with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow([ "institution_name", "paper_title", "year", "author_count", "cited_by_count", "openalex_id", ]) writer.writerows(rows) def fetch_preview(db_path: str, limit: int = 5) -> List[Tuple]: conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( """ SELECT institution_name, paper_title, year, author_count, cited_by_count, openalex_id FROM papers ORDER BY year DESC NULLS LAST, cited_by_count DESC LIMIT ? """, (limit,), ) rows = cur.fetchall() conn.close() return rows

3.5 入口文件：把三段串起来

# main.py import argparse from typing import List, Dict, Any from openalex_client import OpenAlexClient from parser_utils import choose_best_institution, work_to_record from storage import init_db, upsert_papers, export_csv, fetch_preview def parse_args(): parser = argparse.ArgumentParser(description="OpenAlex institution papers crawler") parser.add_argument("--institution-query", type=str, default="", help="institution search keyword") parser.add_argument("--institution-id", type=str, default="", help="OpenAlex institution ID") parser.add_argument("--year-start", type=int, default=None, help="publication year start") parser.add_argument("--year-end", type=int, default=None, help="publication year end") parser.add_argument("--max-pages", type=int, default=None, help="limit cursor pages for testing") parser.add_argument("--db-path", type=str, default="data/openalex_papers.db") parser.add_argument("--csv-path", type=str, default="output/openalex_institution_papers.csv") parser.add_argument("--search-limit", type=int, default=10) return parser.parse_args() def print_candidates(candidates: List[Dict[str, Any]]) -> None: if not candidates: print("[INFO] no institution candidates found.") return print("\n[INFO] institution candidates:") for idx, item in enumerate(candidates, start=1): print( f" {idx}. {item.get('display_name')} | {item.get('id')} | " f"country={item.get('country_code')} | type={item.get('type')} | " f"works_count={item.get('works_count')}" ) def main(): args = parse_args() if not args.institution_query and not args.institution_id: raise ValueError("you must provide --institution-query or --institution-id") client = OpenAlexClient() init_db(args.db_path) if args.institution_id: institution = client.get_institution(args.institution_id) else: candidates = client.search_institutions(args.institution_query, limit=args.search_limit) print_candidates(candidates) institution = choose_best_institution(candidates, args.institution_query) if not institution: raise RuntimeError("failed to resolve institution by query") institution_id = institution["id"] institution_name = institution.get("display_name", "") print(f"\n[INFO] selected institution: {institution_name} | {institution_id}") web_snapshot = client.fetch_institution_web_snapshot(institution_id) print( f"[INFO] web snapshot: url={web_snapshot.get('web_url')} | " f"title={web_snapshot.get('html_title')!r}" ) rows = [] for item in client.iter_works_by_institution( institution_id=institution_id, year_start=args.year_start, year_end=args.year_end, max_pages=args.max_pages, ): row = work_to_record(institution_name, item) if row: rows.append(row) print(f"[INFO] parsed rows: {len(rows)}") affected = upsert_papers(args.db_path, rows) print(f"[INFO] upserted rows: {affected}") export_csv(args.db_path, args.csv_path) print(f"[INFO] csv exported to: {args.csv_path}") preview = fetch_preview(args.db_path, limit=5) print("\n[INFO] preview top 5 rows:") for x in preview: print(x) if __name__ == "__main__": main()

4. 验证请求与成功结果：跑起来看到什么

4.1 三种启动方式

按机构关键词启动，适合不确定机构 ID 的情况：

python main.py --institution-query "Stanford University" --year-start 2022 --year-end 2025

按机构 ID 启动，适合已经确认过 ID 的批量任务：

python main.py --institution-id I97018004 --year-start 2022 --year-end 2025

本地联调只抓前 2 页，避免调试时把额度跑光：

python main.py --institution-query "Stanford University" --year-start 2024 --max-pages 2

4.2 成功时的输出形态

运行后你会看到候选机构列表、选中的机构、网页快照信息、每页抓取条数、解析行数、入库行数，最后是前 5 行预览。预览格式类似：

[INFO] preview top 5 rows: ('Stanford University', '<paper_title_1>', 2025, 8, 37, 'W1234567890') ('Stanford University', '<paper_title_2>', 2025, 5, 19, 'W1234567891') ('Stanford University', '<paper_title_3>', 2024, 11, 203, 'W1234567892') ('Stanford University', '<paper_title_4>', 2024, 6, 44, 'W1234567893') ('Stanford University', '<paper_title_5>', 2024, 9, 12, 'W1234567894')

这里没有贴真实抓到的论文标题，因为 OpenAlex 的数据会随索引更新而变化，静态文章里写死的“真实结果”很快就会过期。你按上面的命令现场跑，拿到的就是当下最准的底表。

4.3 字段校验脚本

入库后建议跑一段校验，确认没有空主键、年份异常、引用数为负：

import sqlite3 conn = sqlite3.connect("data/openalex_papers.db") cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM papers") total = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM papers WHERE openalex_id IS NULL OR openalex_id = ''") empty_id = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM papers WHERE year IS NOT NULL AND (year < 1900 OR year > 2100)") bad_year = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM papers WHERE cited_by_count < 0") bad_cited = cur.fetchone()[0] print(f"total={total}, empty_id={empty_id}, bad_year={bad_year}, bad_cited={bad_cited}") conn.close()

正常结果应该是empty_id=0, bad_year=0, bad_cited=0。如果 empty_id 不为 0，说明解析层的主键过滤没生效，回去检查work_to_record里的extract_short_id。

5. 本篇常见错排查

5.1 403 和 429 怎么区分处理

403 先查三件事：headers 是否缺失、API Key 是否带上、请求路径是否写错。429 通常是请求太密或额度触顶，先降速，再确认 Key 可用，必要时查一下额度状态接口。大批量任务不要硬刚 REST API，改走分批和离线方案。

5.2 HTML 抓到空壳怎么办

OpenAlex 的部分网页页承担前端展示职责，服务端返回的 HTML 未必包含完整业务数据。所以网页层只做三件事：拿 title、拿 meta description、记录可回查 URL。真正的论文明细一律从 API 来，不要为了“混合抓取”四个字把主数据逻辑绑死在网页解析上。

5.3 解析报错怎么定位

最常见的报错是KeyError和TypeError: object of type 'NoneType' has no len()。处理原则：dict 用 get，数组字段默认 []，数值字段默认 0，主键缺失跳过该条。生产上更可取的是记录异常条数、保存失败样本、继续跑后续记录。

5.4 编码乱码怎么处理

这个项目里真正容易出编码问题的地方不是 API，而是导出。SQLite 用 Python 默认 Unicode 即可，CSV 导出用utf-8-sig，这样在 Windows Excel 里打开更省心。

5.5 搜错机构、分页抓不全、filter 不工作

搜错机构通常是因为机构名歧义，解决办法是加大--search-limit，打印候选后手工指定--institution-id。分页抓不全多半是还在用?page=1&page=2，基础分页在结果集较大时有上限，正式采集直接改 cursor。filter 不工作往往是把机构名直接塞进 filter，正确流程是/institutions?search=xxx拿 ID，再/works?filter=authorships.institutions.id:I...。

6. 语义一致 CTA：把采集和分析接起来

底表建好之后，下一步通常是让模型帮你做主题打标、机构画像或年度热点摘要。这时候你需要一个稳定的模型调用入口，而不是每次手动切厂商。你可以先在模型对话页试跑一段论文标题归类，确认效果后，到 API Keys 页面生成密钥，再按接入文档把调用写进脚本。如果这条流水线要长期跑，比如每天定时采集 + 自动生成机构简报，可以在 Coding Plan 里统一管理编码和 Agent 任务，把采集、清洗、分析串成一条可维护的链路。

采集层用 requests + SQLite 保证数据可复现，分析层用统一入口保证调用可收敛，两边各司其职，这张科研分析底表才算真正立住。