简介：CrashData 是一份面向数据分析初学者与航空安全研究爱好者的 R 语言数据采集脚本资源，用于从 planecrashinfo.com 逐年抓取飞机失事的历史记录，解决公开事故数据分散、难以批量整理的问题。压缩包共 2 个文件，包含 1 个 R 脚本与 1 个 Markdown 说明文档，整体仅约 1KB，轻量易读。脚本虽采用逐年提取的 hacky 方式，但最终会转储为规整的 data.frame，方便直接进入后续统计与可视化流程。读者可借此获得一份可复用的网页抓取与数据清洗范例，理解如何将非结构化事故页面整理为结构化表格，并在此基础上开展时间趋势、事故类型分布等分析。目前已有 1422 人学习下载，适合具备基础 R 语法、希望练习爬虫与数据框操作的学习者参考。

1. 从公开事故档案里挖数据：CrashData 到底在解决什么问题

很多人第一次听到「从公开航空事故档案站点抓历史数据」，脑子里浮现的是爬虫脚本加一张 CSV 表，觉得没什么技术含量。但真做过的人知道，这类站点的数据是给人看的，不是给机器读的——事故条目按年代分页、字段用自然语言描述、机型名称写法不统一、伤亡数字混在正文段落里。你要把它变成能跑统计、能做可视化、能喂给模型的结构化数据，中间隔着一整套清洗和校验流程。CrashData 这个方向要解决的核心问题就是：把散落在公开事故档案页面里的历史记录，稳定地抽成结构化字段，并且让这套抽取流程可复现、可增量更新。它适合做数据分析、可视化、航空安全研究入门，或者单纯想练一套「非标准网页到结构化数据集」完整链路的人。

我最初接触这个方向，是因为有人想拿历史事故数据做时间序列分析，结果发现手工整理了几十条就崩溃了——字段对不齐、年份格式混乱、同一个事故在不同页面重复出现。后来我把流程拆成「抓取 → 解析 → 归一化 → 去重 → 校验」五步，才算跑通。这篇笔记就按这个顺序讲，重点放在字段抽取规则、参数设置和踩过的坑上，不堆概念。

2. 抓取层怎么搭：分页规律、请求节奏与增量策略

2.1 先摸清站点的分页结构再写代码

公开事故档案站点通常按年代或年份组织页面，常见形式是索引页列出各年份链接，每个年份页再列出该年的事故条目。不要一上来就写递归爬虫，先用浏览器开发者工具或简单的请求把首页 HTML 拉下来，看链接的 href 规律。我一般会先手动访问三到五个年份页，确认 URL 模式是否统一，比如是/year/1972.html还是?year=1972，有没有分页参数。

确认规律后，抓取层只需要做两件事：从索引页提取年份链接列表，再逐个请求年份页。下面是一个最小可用的抓取骨架，用 Python 的 requests 加 BeautifulSoup，重点是请求间隔和重试逻辑。

import time import requests from bs4 import BeautifulSoup HEADERS = { "User-Agent": "Mozilla/5.0 (compatible; CrashDataResearch/1.0)" } BASE_DELAY = 2.0 # 秒，基础请求间隔 def fetch(url, retries=3): for i in range(retries): try: resp = requests.get(url, headers=HEADERS, timeout=15) resp.raise_for_status() resp.encoding = resp.apparent_encoding return resp.text except requests.RequestException as e: wait = BASE_DELAY * (i + 1) print(f"请求失败 {url}，{wait}s 后重试：{e}") time.sleep(wait) return None def get_year_links(index_html): soup = BeautifulSoup(index_html, "html.parser") links = [] for a in soup.find_all("a", href=True): href = a["href"] # 根据实际站点规律调整，这里假设年份页路径含四位数字 if any(str(y) in href for y in range(1900, 2030)): links.append(href) return list(dict.fromkeys(links)) # 去重且保持顺序 if __name__ == "__main__": index_html = fetch("https://example-archive-site/index.html") if index_html: years = get_year_links(index_html) print(f"发现 {len(years)} 个年份链接") for y in years: html = fetch(y) if html: # 后续解析逻辑 pass time.sleep(BASE_DELAY)

这段代码里BASE_DELAY是关键参数。设太小容易触发限流，设太大抓全量数据要很久。我的经验是公开档案类站点 1.5 到 3 秒比较稳，如果遇到 429 或 503，就把间隔翻倍并加指数退避。retries不要超过 5 次，否则一个坏链接会拖慢整批任务。apparent_encoding是为了处理老页面可能用的非 UTF-8 编码，不设的话中文或特殊字符会乱码。

2.2 增量抓取：用本地状态文件避免重复请求

全量抓一次可能几百到上千个页面，但后续更新只需要抓新增或变更的年份页。我一般会在本地维护一个 JSON 状态文件，记录每个 URL 的最后抓取时间和内容哈希。下次运行时先比对，哈希没变就跳过解析。

import json import hashlib import os STATE_FILE = "crawl_state.json" def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) return {} def save_state(state): with open(STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def content_hash(text): return hashlib.md5(text.encode("utf-8")).hexdigest() def should_fetch(url, html, state): h = content_hash(html) old = state.get(url) if old and old.get("hash") == h: return False, h return True, h

状态文件里存url、hash、fetched_at三个字段就够了。注意哈希是对 HTML 正文算的，不是对 URL。有些站点会在页面底部加动态时间戳，导致每次哈希都变，这种情况要先剔除动态区域再算哈希，否则增量策略失效。

2.3 请求头与合规边界

请求头里带一个能表明用途的 User-Agent 是基本礼貌，也方便对方在日志里识别。不要伪造浏览器完整指纹去绕反爬，公开档案类站点一般对低频访问是宽容的。如果站点有 robots.txt，先读一遍，确认允许抓取的路径范围。我一般会把并发控制在 1 到 2，不做多线程轰炸，宁可慢一点也不要给站点造成压力。这不是技术限制，是长期可用的前提。

3. 解析层：把自然语言事故描述拆成结构化字段

3.1 先定义字段表，再写解析规则

解析最容易翻车的地方是「边看页面边写正则」，写到后面字段互相冲突。正确顺序是先定字段表，再针对每个字段写抽取规则。历史事故数据我一般会抽这些字段：

字段表定好后，每个字段单独写抽取函数，互不干扰。这样某个字段规则改了，不会影响其他字段。

3.2 日期和伤亡数字的抽取：正则加后处理

日期在事故描述里写法极多，常见的有January 1, 1972、1 Jan 1972、1972-01-01。我一般先用一组正则覆盖主要格式，再统一转成 ISO 日期。伤亡数字更麻烦，有的写all 100 on board died，有的写killing 45 of 50，需要区分「死亡数」和「总人数」。

import re from datetime import datetime MONTH_MAP = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12 } def parse_date(text): # 格式一：January 1, 1972 m = re.search(r"([A-Za-z]+)\s+(\d{1,2}),\s*(\d{4})", text) if m: mon = MONTH_MAP.get(m.group(1)[:3].lower()) if mon: return f"{m.group(3)}-{mon:02d}-{int(m.group(2)):02d}" # 格式二：1 Jan 1972 m = re.search(r"(\d{1,2})\s+([A-Za-z]+)\s+(\d{4})", text) if m: mon = MONTH_MAP.get(m.group(2)[:3].lower()) if mon: return f"{m.group(3)}-{mon:02d}-{int(m.group(1)):02d}" # 格式三：1972-01-01 m = re.search(r"(\d{4})-(\d{2})-(\d{2})", text) if m: return m.group(0) return None def parse_fatalities(text): # 匹配 "killing 45" / "45 killed" / "all 100 died" patterns = [ r"killing\s+(\d+)", r"(\d+)\s+(?:people\s+)?killed", r"all\s+(\d+)\s+(?:on board\s+)?died", r"(\d+)\s+fatalities" ] for p in patterns: m = re.search(p, text, re.IGNORECASE) if m: return int(m.group(1)) return None

parse_date里月份映射只取前三个字母转小写，能覆盖Jan、January、JAN等写法。parse_fatalities的四个模式按优先级排列，先匹配更具体的killing N，再匹配泛化的N killed。注意不要用\d+直接抓所有数字，否则会把年份、航班号也抓进来。如果一段描述里同时出现多个数字，宁可返回 None 也不要猜，留空比填错好。

3.3 机型名称归一化：用映射表而不是模糊匹配

机型名称是解析层最耗时的部分。同一机型可能有Boeing 707、B707、707、Boeing 707-300等多种写法。我一般会建一张映射表，把常见别名映射到标准名，而不是用编辑距离做模糊匹配——模糊匹配在机型这种短文本上误判率很高，707和727编辑距离很近但完全是两种飞机。

AIRCRAFT_ALIAS = { "b707": "Boeing 707", "boeing 707": "Boeing 707", "707": "Boeing 707", "b727": "Boeing 727", "boeing 727": "Boeing 727", "727": "Boeing 727", "dc9": "McDonnell Douglas DC-9", "dc-9": "McDonnell Douglas DC-9", # 按实际数据逐步补充 } def normalize_aircraft(raw): if not raw: return None key = raw.strip().lower().replace("-", "").replace(" ", "") # 先查无空格无连字符的键 for alias, standard in AIRCRAFT_ALIAS.items(): if alias.replace("-", "").replace(" ", "") == key: return standard return raw.strip() # 未命中则保留原文，后续人工补映射

映射表不要一次写全，先跑一批数据，把未命中的机型名收集起来，人工确认后再补进去。这样迭代两三轮，覆盖率就能到 90% 以上。未命中的保留原文，不要强行归到某个类别，否则会污染数据。

4. 清洗与去重：同一事故出现三次该怎么合并

4.1 去重键的选择：日期加地点加运营方

历史事故数据里重复条目很常见，可能因为同一事故在不同年份页被重复收录，或者索引页和详情页各出现一次。去重不能只用日期，因为同一天可能有多起事故；也不能只用运营方，因为同一运营方事故很多。我一般用「日期 + 地点前若干字符 + 运营方」拼成一个去重键，再做一次人工抽查。

def dedup_key(record): date = record.get("date") or "" loc = (record.get("location") or "")[:20].lower().strip() op = (record.get("operator") or "").lower().strip() return f"{date}|{loc}|{op}" def dedup(records): seen = {} for r in records: key = dedup_key(r) if key not in seen: seen[key] = r else: # 保留字段更完整的记录 old = seen[key] if count_non_empty(r) > count_non_empty(old): seen[key] = r return list(seen.values()) def count_non_empty(record): return sum(1 for v in record.values() if v not in (None, "", []))

dedup_key里地点只取前 20 个字符，是因为同一地点在不同条目里可能有细微拼写差异，取前缀能提高合并率。保留字段更完整的记录，是为了避免去重后反而丢了信息。去重后一定要抽样检查，我遇到过把两起同一天同一运营方但不同地点的事故误合并的情况，后来把地点前缀从 10 个字符加到 20 个字符才解决。

4.2 缺失值处理：留空、标记、还是剔除

结构化数据里缺失值有三种处理方式：留空、填默认值、剔除整条。我的原则是：关键字段（日期、运营方）缺失就剔除，非关键字段（注册号、总人数）缺失就留空。不要用 0 填充缺失的伤亡数字，0 和「未知」在统计上是完全不同的含义。如果要做聚合分析，在分析层再用dropna或fillna处理，不要在清洗层就填掉。

4.3 输出格式与校验

清洗完的数据我一般输出两份：一份 CSV 方便人工查看，一份 JSONL 方便程序读取。输出前跑一遍校验：日期格式是否统一、伤亡数字是否非负、去重后条数是否合理。校验不通过就停下来看日志，不要带着脏数据往下走。

import csv import json def validate(record): errors = [] if record.get("date") and not re.match(r"\d{4}-\d{2}-\d{2}", record["date"]): errors.append(f"日期格式异常：{record['date']}") for f in ("fatalities", "occupants"): v = record.get(f) if v is not None and (not isinstance(v, int) or v < 0): errors.append(f"{f} 异常：{v}") return errors def export(records, csv_path, jsonl_path): with open(csv_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=records[0].keys()) writer.writeheader() writer.writerows(records) with open(jsonl_path, "w", encoding="utf-8") as f: for r in records: f.write(json.dumps(r, ensure_ascii=False) + "\n")

5. 避坑与排查：那些让我返工三次的问题

5.1 编码问题导致中文或特殊字符变问号

现象：抓下来的页面里，遇难者姓名或地点中的特殊字符变成?或乱码。原因：老页面可能用ISO-8859-1或Windows-1252编码，而 requests 默认按UTF-8解码。解决：用resp.apparent_encoding让 chardet 自动判断，或者手动指定resp.encoding = "ISO-8859-1"。如果还不行，用resp.content.decode("latin-1")再处理。

5.2 正则贪婪匹配把整段都吞了

现象：summary字段抽出来包含了下一条事故的内容。原因：用了.*而不是.*?，或者没有限定结束边界。解决：把贪婪匹配改成非贪婪，并加上明确的结束标记，比如re.search(r"Summary:(.*?)(?=Date:|$)", text, re.DOTALL)。写完正则先用几条样本测，不要直接跑全量。

5.3 增量抓取时哈希一直变导致重复解析

现象：每次运行都重新解析所有页面，增量策略形同虚设。原因：页面里有动态时间戳或随机广告位，导致 HTML 哈希每次都不同。解决：在算哈希前先剔除动态区域，比如用 BeautifulSoup 删掉script、style和已知的动态容器，再对剩余正文算哈希。

5.4 机型映射表越写越乱

现象：映射表里出现同一个标准名对应多个别名，后来自己都分不清哪个是准的。原因：边抓边加，没有统一维护。解决：把映射表单独放一个 JSON 文件，按标准名分组，别名放数组里。每次新增别名前先搜一遍是否已存在，避免重复和冲突。

5.5 请求间隔设太小被限流

现象：抓了几十个页面后开始返回 429，后续全部失败。原因：BASE_DELAY设成了 0.5 秒，且没有退避逻辑。解决：基础间隔调到 2 秒以上，遇到 429 时把间隔翻倍并暂停一段时间。如果站点有Retry-After响应头，优先按它给的时间等待。

6. 进阶技巧：用校验规则反推解析质量

数据清洗做到后面，你会发现真正决定质量的不是抓取速度，而是校验规则的密度。我后来养成了一个习惯：每加一个字段，就同时加一条校验规则，用校验失败率来反推解析规则哪里薄弱。比如fatalities大于occupants的记录，大概率是抽取时把两个数字搞混了；日期早于 1900 年或晚于当前年份的，大概率是正则匹配到了错误位置。

下面这段校验代码会输出每个字段的异常率和几条样本，方便快速定位问题：

def audit(records): total = len(records) stats = {} for field in ("date", "fatalities", "occupants", "aircraft_type"): bad = [r for r in records if not r.get(field)] stats[field] = { "missing_rate": len(bad) / total if total else 0, "samples": [r.get("summary", "")[:60] for r in bad[:3]] } # 交叉校验：死亡数大于总人数 cross_bad = [ r for r in records if r.get("fatalities") and r.get("occupants") and r["fatalities"] > r["occupants"] ] stats["fatalities_gt_occupants"] = { "count": len(cross_bad), "samples": [r.get("summary", "")[:60] for r in cross_bad[:3]] } return stats

跑完audit后，如果某个字段缺失率超过 30%，就回去看解析规则是不是覆盖不够；如果交叉校验异常条数超过 5%，就重点检查伤亡数字的抽取逻辑。这套方法比盲目跑全量再人工翻看效率高得多。

还有一个实用技巧：把每次抓取和解析的中间结果都落盘，不要只存最终 CSV。我一般会存raw_html/、parsed.jsonl、cleaned.csv三层。这样当解析规则改了，不需要重新抓页面，直接从raw_html/重跑解析就行。这个习惯帮我省了大量重复请求的时间，也让排查问题时能快速定位是抓取层还是解析层出的错。

最后说一个我踩过的坑：不要试图一次把所有年份都抓完再解析。正确做法是按年份分批，抓一年、解析一年、校验一年，确认没问题再进下一年。全量抓完再解析，一旦解析规则有系统性错误，返工成本极高。分批推进虽然看起来慢，但整体返工次数少得多。希望帮到你。

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