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.csv3.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, flagsre.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( total3, connect3, read3, backoff_factor0.5, status_forcelist(429, 500, 502, 503, 504), allowed_methodsfrozenset([GET]), raise_on_statusFalse, ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize10) 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;q0.9, */*;q0.8, Referer: https://explore.openalex.org/, Accept-Language: zh-CN,zh;q0.9,en;q0.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, paramsparams, timeoutself.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(fRequest 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, paramsparams, expect_jsonTrue, attach_api_keyTrue) 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_jsonTrue, attach_api_keyTrue) 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_jsonFalse, attach_api_keyFalse) soup BeautifulSoup(html, lxml) title soup.title.get_text(stripTrue) 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 [fauthorships.institutions.id:{short_id}] if year_start is not None and year_end is not None: filter_parts.append(fpublication_year:{year_start}-{year_end}) elif year_start is not None: filter_parts.append(fpublication_year:{year_start - 1}) elif year_end is not None: filter_parts.append(fpublication_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, paramsparams, expect_jsonTrue, attach_api_keyTrue, ) results data.get(results, []) meta data.get(meta, {}) print( f[INFO] works page{page_count}, fcount_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_KEYyour_api_key_here # Windows PowerShell $env:OPENALEX_API_KEYyour_api_key_here3.3 解析层字段映射与容错解析层只做一件事把 API 返回的 JSON 变成底表记录。字段映射如下字段类型示例值说明institution_nameTEXTStanford University机构名paper_titleTEXTA large-scale study论文标题yearINTEGER2024发表年份author_countINTEGER7作者数量cited_by_countINTEGER128引用数openalex_idTEXTW2741809807论文 OpenAlex 短 IDopenalex_urlTEXThttps://openalex.org/W2741809807原始 URLcrawled_atTEXT2026-04-01T10:00:00抓取时间# 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, keylambda x: x.get(works_count, 0), reverseTrue)[0] contains [x for x in candidates if q and q in normalize_name(x.get(display_name, ))] if contains: return sorted(contains, keylambda x: x.get(works_count, 0), reverseTrue)[0] return sorted(candidates, keylambda x: x.get(works_count, 0), reverseTrue)[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, flagsre.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 缺失记 Nonecited_by_count 缺失默认 0id 缺失直接跳过避免脏主键进库。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_okTrue) 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(timespecseconds) 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, encodingutf-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 rows3.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(descriptionOpenAlex institution papers crawler) parser.add_argument(--institution-query, typestr, default, helpinstitution search keyword) parser.add_argument(--institution-id, typestr, default, helpOpenAlex institution ID) parser.add_argument(--year-start, typeint, defaultNone, helppublication year start) parser.add_argument(--year-end, typeint, defaultNone, helppublication year end) parser.add_argument(--max-pages, typeint, defaultNone, helplimit cursor pages for testing) parser.add_argument(--db-path, typestr, defaultdata/openalex_papers.db) parser.add_argument(--csv-path, typestr, defaultoutput/openalex_institution_papers.csv) parser.add_argument(--search-limit, typeint, default10) 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, start1): print( f {idx}. {item.get(display_name)} | {item.get(id)} | fcountry{item.get(country_code)} | type{item.get(type)} | fworks_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, limitargs.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)} | ftitle{web_snapshot.get(html_title)!r} ) rows [] for item in client.iter_works_by_institution( institution_idinstitution_id, year_startargs.year_start, year_endargs.year_end, max_pagesargs.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, limit5) 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 24.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(ftotal{total}, empty_id{empty_id}, bad_year{bad_year}, bad_cited{bad_cited}) conn.close()正常结果应该是empty_id0, bad_year0, bad_cited0。如果 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。分页抓不全多半是还在用?page1page2基础分页在结果集较大时有上限正式采集直接改 cursor。filter 不工作往往是把机构名直接塞进 filter正确流程是/institutions?searchxxx拿 ID再/works?filterauthorships.institutions.id:I...。6. 语义一致 CTA把采集和分析接起来底表建好之后下一步通常是让模型帮你做主题打标、机构画像或年度热点摘要。这时候你需要一个稳定的模型调用入口而不是每次手动切厂商。你可以先在模型对话页试跑一段论文标题归类确认效果后到 API Keys 页面生成密钥再按接入文档把调用写进脚本。如果这条流水线要长期跑比如每天定时采集 自动生成机构简报可以在 Coding Plan 里统一管理编码和 Agent 任务把采集、清洗、分析串成一条可维护的链路。采集层用 requests SQLite 保证数据可复现分析层用统一入口保证调用可收敛两边各司其职这张科研分析底表才算真正立住。 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.csv3.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, flagsre.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( total3, connect3, read3, backoff_factor0.5, status_forcelist(429, 500, 502, 503, 504), allowed_methodsfrozenset([GET]), raise_on_statusFalse, ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize10) 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;q0.9, */*;q0.8, Referer: https://explore.openalex.org/, Accept-Language: zh-CN,zh;q0.9,en;q0.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, paramsparams, timeoutself.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(fRequest 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, paramsparams, expect_jsonTrue, attach_api_keyTrue) 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_jsonTrue, attach_api_keyTrue) 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_jsonFalse, attach_api_keyFalse) soup BeautifulSoup(html, lxml) title soup.title.get_text(stripTrue) 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 [fauthorships.institutions.id:{short_id}] if year_start is not None and year_end is not None: filter_parts.append(fpublication_year:{year_start}-{year_end}) elif year_start is not None: filter_parts.append(fpublication_year:{year_start - 1}) elif year_end is not None: filter_parts.append(fpublication_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, paramsparams, expect_jsonTrue, attach_api_keyTrue, ) results data.get(results, []) meta data.get(meta, {}) print( f[INFO] works page{page_count}, fcount_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_KEYyour_api_key_here # Windows PowerShell $env:OPENALEX_API_KEYyour_api_key_here3.3 解析层字段映射与容错解析层只做一件事把 API 返回的 JSON 变成底表记录。字段映射如下字段类型示例值说明institution_nameTEXTStanford University机构名paper_titleTEXTA large-scale study论文标题yearINTEGER2024发表年份author_countINTEGER7作者数量cited_by_countINTEGER128引用数openalex_idTEXTW2741809807论文 OpenAlex 短 IDopenalex_urlTEXThttps://openalex.org/W2741809807原始 URLcrawled_atTEXT2026-04-01T10:00:00抓取时间# 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, keylambda x: x.get(works_count, 0), reverseTrue)[0] contains [x for x in candidates if q and q in normalize_name(x.get(display_name, ))] if contains: return sorted(contains, keylambda x: x.get(works_count, 0), reverseTrue)[0] return sorted(candidates, keylambda x: x.get(works_count, 0), reverseTrue)[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, flagsre.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 缺失记 Nonecited_by_count 缺失默认 0id 缺失直接跳过避免脏主键进库。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_okTrue) 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(timespecseconds) 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, encodingutf-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 rows3.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(descriptionOpenAlex institution papers crawler) parser.add_argument(--institution-query, typestr, default, helpinstitution search keyword) parser.add_argument(--institution-id, typestr, default, helpOpenAlex institution ID) parser.add_argument(--year-start, typeint, defaultNone, helppublication year start) parser.add_argument(--year-end, typeint, defaultNone, helppublication year end) parser.add_argument(--max-pages, typeint, defaultNone, helplimit cursor pages for testing) parser.add_argument(--db-path, typestr, defaultdata/openalex_papers.db) parser.add_argument(--csv-path, typestr, defaultoutput/openalex_institution_papers.csv) parser.add_argument(--search-limit, typeint, default10) 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, start1): print( f {idx}. {item.get(display_name)} | {item.get(id)} | fcountry{item.get(country_code)} | type{item.get(type)} | fworks_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, limitargs.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)} | ftitle{web_snapshot.get(html_title)!r} ) rows [] for item in client.iter_works_by_institution( institution_idinstitution_id, year_startargs.year_start, year_endargs.year_end, max_pagesargs.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, limit5) 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 24.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(ftotal{total}, empty_id{empty_id}, bad_year{bad_year}, bad_cited{bad_cited}) conn.close()正常结果应该是empty_id0, bad_year0, bad_cited0。如果 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。分页抓不全多半是还在用?page1page2基础分页在结果集较大时有上限正式采集直接改 cursor。filter 不工作往往是把机构名直接塞进 filter正确流程是/institutions?searchxxx拿 ID再/works?filterauthorships.institutions.id:I...。6. 语义一致 CTA把采集和分析接起来底表建好之后下一步通常是让模型帮你做主题打标、机构画像或年度热点摘要。这时候你需要一个稳定的模型调用入口而不是每次手动切厂商。你可以先在模型对话页试跑一段论文标题归类确认效果后到 API Keys 页面生成密钥再按接入文档把调用写进脚本。如果这条流水线要长期跑比如每天定时采集 自动生成机构简报可以在 Coding Plan 里统一管理编码和 Agent 任务把采集、清洗、分析串成一条可维护的链路。采集层用 requests SQLite 保证数据可复现分析层用统一入口保证调用可收敛两边各司其职这张科研分析底表才算真正立住。