用 Scrapy 爬取百家姓与姓氏源流数据：多源采集、数据清洗与结构化存储实战

一、前言

姓氏是中国文化的重要载体，每个姓氏背后都承载着几千年的家族源流、地域文化和历史名人。对于起名类产品来说，姓氏数据是核心基础数据之一 —— 了解一个姓氏的起源、郡望、堂号、历史名人，才能为用户提供更有文化内涵的名字推荐。笔者在开发529宝宝起名网时，发现单一数据源的姓氏数据存在覆盖不全、信息陈旧、字段缺失等问题，于是决定通过多源数据采集构建一套完整的姓氏知识库。

百家姓相关的数据在网络上分布广泛，但质量参差不齐。有的网站只收录了前 100 个姓氏，有的起源描述过于简略，有的缺少郡望堂号信息。要构建高质量的姓氏知识库，需要从多个数据源采集，然后进行清洗、去重、融合和校验。

本文将从零开始，完整记录用 Scrapy 框架爬取百家姓与姓氏源流数据的全过程，涵盖：

• 多源数据对比与采集目标设计
• Scrapy 爬虫架构与中间件设计
• 多源数据采集实现（列表页 + 详情页 + 名人数据）
• 数据清洗、去重与多源融合
• 结构化存储与数据质量校验
• 效果验证与性能优化

二、数据来源与采集目标

2.1 候选数据源对比

表格

综合评估后，我们选择某百家姓查询站作为主数据源，某姓氏文化网作为补充源，百度百科用于交叉验证。本文以主数据源的爬取过程为例进行讲解。

2.2 采集字段设计

表格

2.3 数据量预估

表格

三、Scrapy 爬虫架构设计

3.1 项目结构

baijiaxing_crawler/ ├── scrapy.cfg ├── baijiaxing_crawler/ │ ├── __init__.py │ ├── items.py # 数据结构定义 │ ├── middlewares.py # 中间件（UA轮换、代理、重试） │ ├── pipelines.py # 数据管道（清洗、去重、存储） │ ├── settings.py # 配置文件 │ └── spiders/ │ ├── __init__.py │ ├── surname_list.py # 姓氏列表页爬虫 │ ├── surname_detail.py # 姓氏详情页爬虫 │ └── famous_people.py # 历史名人爬虫 ├── data/ │ ├── raw/ # 原始数据 │ └── cleaned/ # 清洗后数据 └── logs/ # 日志文件

3.2 Items 定义

import scrapy class SurnameItem(scrapy.Item): """姓氏基础信息""" surname = scrapy.Field() pinyin = scrapy.Field() population = scrapy.Field() ratio = scrapy.Field() rank = scrapy.Field() origin = scrapy.Field() origin_type = scrapy.Field() junwang = scrapy.Field() tanghao = scrapy.Field() main_provinces = scrapy.Field() migration_history = scrapy.Field() source_url = scrapy.Field() crawled_at = scrapy.Field() class FamousPeopleItem(scrapy.Item): """历史名人""" surname = scrapy.Field() name = scrapy.Field() dynasty = scrapy.Field() title = scrapy.Field() achievement = scrapy.Field() source_url = scrapy.Field()

3.3 中间件设计

import random import time from scrapy import signals from scrapy.downloadermiddlewares.retry import RetryMiddleware from scrapy.utils.response import response_status_message class RandomUserAgentMiddleware: """随机User-Agent中间件""" USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", ] def process_request(self, request, spider): request.headers["User-Agent"] = random.choice(self.USER_AGENTS) class ProxyMiddleware: """代理IP中间件""" def __init__(self, proxy_list): self.proxy_list = proxy_list self.proxy_index = 0 @classmethod def from_crawler(cls, crawler): return cls(crawler.settings.getlist("PROXY_LIST")) def process_request(self, request, spider): if self.proxy_list: proxy = self.proxy_list[self.proxy_index % len(self.proxy_list)] request.meta["proxy"] = proxy self.proxy_index += 1 class CustomRetryMiddleware(RetryMiddleware): """自定义重试中间件（增加指数退避）""" def process_response(self, request, response, spider): if response.status in self.retry_http_codes: reason = response_status_message(response.status) retry_times = request.meta.get("retry_times", 0) # 指数退避：1s, 2s, 4s, 8s time.sleep(min(2 ** retry_times, 10)) return self._retry(request, reason, spider) or response return response

3.4 配置文件

# settings.py BOT_NAME = "baijiaxing_crawler" SPIDER_MODULES = ["baijiaxing_crawler.spiders"] NEWSPIDER_MODULE = "baijiaxing_crawler.spiders" # 并发控制 CONCURRENT_REQUESTS = 8 CONCURRENT_REQUESTS_PER_DOMAIN = 4 CONCURRENT_REQUESTS_PER_IP = 4 # 下载延迟 DOWNLOAD_DELAY = 2 RANDOMIZE_DOWNLOAD_DELAY = True # 中间件 DOWNLOADER_MIDDLEWARES = { "baijiaxing_crawler.middlewares.RandomUserAgentMiddleware": 400, "baijiaxing_crawler.middlewares.ProxyMiddleware": 410, "baijiaxing_crawler.middlewares.CustomRetryMiddleware": 550, } # 管道 ITEM_PIPELINES = { "baijiaxing_crawler.pipelines.DataCleaningPipeline": 300, "baijiaxing_crawler.pipelines.DuplicatePipeline": 400, "baijiaxing_crawler.pipelines.JsonExportPipeline": 500, "baijiaxing_crawler.pipelines.MySQLPipeline": 600, } # 重试设置 RETRY_ENABLED = True RETRY_TIMES = 3 RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429] # 日志 LOG_LEVEL = "INFO" LOG_FILE = "logs/crawler.log" # 代理列表（实际使用时填入真实代理） PROXY_LIST = [] # MySQL配置 MYSQL_HOST = "localhost" MYSQL_PORT = 3306 MYSQL_USER = "root" MYSQL_PASSWORD = "password" MYSQL_DATABASE = "surname_db"

该 Scrapy 爬虫架构已应用于 在线起名工具 的姓氏知识库建设，支持多源数据采集、自动去重和增量更新。

四、多源数据采集实现

4.1 姓氏列表页爬虫

import scrapy from urllib.parse import urljoin from baijiaxing_crawler.items import SurnameItem class SurnameListSpider(scrapy.Spider): """姓氏列表页爬虫：采集所有姓氏的基础信息""" name = "surname_list" allowed_domains = ["www.example-baijiaxing.com"] start_urls = ["https://www.example-baijiaxing.com/surname/list.html"] def parse(self, response): """解析姓氏列表页""" # 解析姓氏列表 for row in response.css(".surname-table tr"): surname = row.css(".surname-name::text").get() if not surname: continue item = SurnameItem() item["surname"] = surname.strip() item["pinyin"] = row.css(".surname-pinyin::text").get("").strip() item["population"] = self._parse_population( row.css(".surname-population::text").get("") ) item["ratio"] = self._parse_ratio( row.css(".surname-ratio::text").get("") ) item["rank"] = self._parse_int( row.css(".surname-rank::text").get("") ) # 详情页链接 detail_url = row.css(".surname-name a::attr(href)").get() if detail_url: item["source_url"] = urljoin(response.url, detail_url) # 继续爬取详情页 yield scrapy.Request( url=item["source_url"], callback=self.parse_detail, meta={"item": item}, ) else: item["source_url"] = response.url yield item # 翻页 next_page = response.css(".pagination .next a::attr(href)").get() if next_page: yield scrapy.Request(url=urljoin(response.url, next_page), callback=self.parse) def parse_detail(self, response): """解析姓氏详情页""" item = response.meta["item"] # 起源 item["origin"] = response.css(".origin-content::text").get("").strip() # 起源类型 origin_type = response.css(".origin-type::text").get("") item["origin_type"] = origin_type.strip() if origin_type else "" # 郡望 item["junwang"] = ", ".join( response.css(".junwang-list li::text").getall() ).strip() # 堂号 item["tanghao"] = ", ".join( response.css(".tanghao-list li::text").getall() ).strip() # 主要分布省份 item["main_provinces"] = ", ".join( response.css(".province-list li::text").getall() ).strip() # 迁徙历史 item["migration_history"] = response.css( ".migration-content::text" ).get("").strip() yield item @staticmethod def _parse_population(text): """解析人口数（如'95,400,000'或'9540万'）""" import re text = text.strip().replace(",", "") if "万" in text: num = re.findall(r"[\d.]+", text) return int(float(num[0]) * 10000) if num else None num = re.findall(r"\d+", text) return int(num[0]) if num else None @staticmethod def _parse_ratio(text): """解析比例（如'7.25%'或'0.0725'）""" import re text = text.strip() if "%" in text: num = re.findall(r"[\d.]+", text) return float(num[0]) / 100 if num else None num = re.findall(r"[\d.]+", text) return float(num[0]) if num else None @staticmethod def _parse_int(text): """解析整数""" import re num = re.findall(r"\d+", text) return int(num[0]) if num else None

4.2 历史名人爬虫

import scrapy from urllib.parse import urljoin from baijiaxing_crawler.items import FamousPeopleItem class FamousPeopleSpider(scrapy.Spider): """历史名人爬虫：按姓氏采集历史名人""" name = "famous_people" allowed_domains = ["www.example-baijiaxing.com"] def __init__(self, surnames=None, *args, **kwargs): super().__init__(*args, **kwargs) # 支持传入姓氏列表，默认爬取TOP100姓氏 self.surnames = surnames or ["王", "李", "张", "刘", "陈"] def start_requests(self): """生成起始请求""" for surname in self.surnames: url = f"https://www.example-baijiaxing.com/famous/{surname}.html" yield scrapy.Request( url=url, callback=self.parse, meta={"surname": surname}, ) def parse(self, response): """解析名人列表页""" surname = response.meta["surname"] for person in response.css(".famous-list .person-item"): item = FamousPeopleItem() item["surname"] = surname item["name"] = person.css(".person-name::text").get("").strip() item["dynasty"] = person.css(".person-dynasty::text").get("").strip() item["title"] = person.css(".person-title::text").get("").strip() item["achievement"] = person.css( ".person-achievement::text" ).get("").strip() item["source_url"] = response.url # 名人详情页（可选，获取更详细的成就描述） detail_url = person.css(".person-name a::attr(href)").get() if detail_url: yield scrapy.Request( url=urljoin(response.url, detail_url), callback=self.parse_person_detail, meta={"item": item}, ) else: yield item # 翻页 next_page = response.css(".pagination .next a::attr(href)").get() if next_page: yield scrapy.Request( url=urljoin(response.url, next_page), callback=self.parse, meta={"surname": surname}, ) def parse_person_detail(self, response): """解析名人详情页""" item = response.meta["item"] # 获取更详细的成就描述 detail = response.css(".person-detail::text").get("") if detail and len(detail) > len(item["achievement"]): item["achievement"] = detail.strip() yield item

读者可前往 在线起名工具 查看姓氏源流数据，每个姓氏都有起源、郡望、堂号和历史名人的完整信息。

五、数据清洗与多源融合

5.1 数据清洗管道

import re import hashlib from datetime import datetime from itemadapter import ItemAdapter class DataCleaningPipeline: """数据清洗管道""" def process_item(self, item, spider): adapter = ItemAdapter(item) # 1. 去除HTML标签和多余空白 for field in adapter.field_names(): value = adapter.get(field) if isinstance(value, str): # 去除HTML标签 value = re.sub(r"<[^>]+>", "", value) # 去除多余空白 value = re.sub(r"\s+", " ", value).strip() # 去除特殊字符 value = value.replace("\u3000", " ").replace("\xa0", " ") adapter[field] = value # 2. 姓氏标准化（去除异体字、统一写法） if "surname" in adapter.field_names(): surname = adapter["surname"] # 常见异体字映射 variant_map = {"邱": "丘", "肖": "萧", "付": "傅", "代": "戴"} adapter["surname"] = variant_map.get(surname, surname) # 3. 人口数据标准化 if "population" in adapter.field_names(): pop = adapter["population"] if isinstance(pop, str): pop = self._parse_population(pop) adapter["population"] = pop # 4. 比例数据标准化 if "ratio" in adapter.field_names(): ratio = adapter["ratio"] if isinstance(ratio, str): ratio = self._parse_ratio(ratio) adapter["ratio"] = ratio # 5. 添加采集时间 adapter["crawled_at"] = datetime.now().isoformat() return item @staticmethod def _parse_population(text): """解析人口数""" text = str(text).strip().replace(",", "") if "万" in text: num = re.findall(r"[\d.]+", text) return int(float(num[0]) * 10000) if num else None num = re.findall(r"\d+", text) return int(num[0]) if num else None @staticmethod def _parse_ratio(text): """解析比例""" text = str(text).strip() if "%" in text: num = re.findall(r"[\d.]+", text) return float(num[0]) / 100 if num else None num = re.findall(r"[\d.]+", text) return float(num[0]) if num else None class DuplicatePipeline: """去重管道""" def __init__(self): self.seen = set() def process_item(self, item, spider): adapter = ItemAdapter(item) # 生成唯一键（姓氏+来源URL的MD5） key_content = f"{adapter.get('surname', '')}|{adapter.get('name', '')}|{adapter.get('source_url', '')}" key = hashlib.md5(key_content.encode("utf-8")).hexdigest() if key in self.seen: raise DropItem(f"重复数据: {key_content}") self.seen.add(key) return item def close_spider(self, spider): spider.logger.info(f"去重管道共处理 {len(self.seen)} 条唯一数据")

5.2 多源数据融合

import pandas as pd import json from typing import List, Dict class DataFusion: """多源数据融合""" def __init__(self): self.sources = {} def load_source(self, name: str, file_path: str): """加载一个数据源""" with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) self.sources[name] = pd.DataFrame(data) print(f"加载数据源 {name}: {len(self.sources[name])} 条记录") def fuse(self, primary_source: str, other_sources: List[str]) -> pd.DataFrame: """以主数据源为基础，融合其他数据源""" if primary_source not in self.sources: raise ValueError(f"主数据源 {primary_source} 未加载") df = self.sources[primary_source].copy() df["data_sources"] = primary_source for source_name in other_sources: if source_name not in self.sources: print(f"警告：数据源 {source_name} 未加载，跳过") continue other_df = self.sources[source_name] df = self._merge_source(df, other_df, source_name) return df def _merge_source(self, base_df: pd.DataFrame, other_df: pd.DataFrame, source_name: str) -> pd.DataFrame: """合并单个数据源""" # 以姓氏为键进行合并 merged = base_df.merge( other_df, on="surname", how="outer", suffixes=("", f"_{source_name}"), indicator=True, ) # 标记数据来源 merged["data_sources"] = merged.apply( lambda row: self._update_sources(row, source_name), axis=1 ) # 字段优先级：主数据源优先，缺失字段用其他数据源补充 fill_fields = ["origin", "junwang", "tanghao", "famous_people", "main_provinces", "migration_history"] for field in fill_fields: source_field = f"{field}_{source_name}" if source_field in merged.columns: merged[field] = merged[field].fillna(merged[source_field]) # 如果主数据源字段为空，用其他数据源填充 merged[field] = merged.apply( lambda row: row[source_field] if (pd.isna(row[field]) or row[field] == "") and pd.notna(row[source_field]) else row[field], axis=1, ) # 数值字段取平均值（人口、比例） for field in ["population", "ratio"]: source_field = f"{field}_{source_name}" if source_field in merged.columns: merged[field] = merged[[field, source_field]].mean(axis=1) # 清理合并产生的临时列 drop_cols = [c for c in merged.columns if c.endswith(f"_{source_name}")] merged = merged.drop(columns=drop_cols + ["_merge"], errors="ignore") return merged @staticmethod def _update_sources(row, source_name): """更新数据来源标记""" sources = row.get("data_sources", "") if row.get("_merge") in ("both", "right_only"): if source_name not in str(sources): sources = f"{sources},{source_name}".strip(",") return sources

六、结构化存储与数据质量校验

6.1 MySQL 存储管道

import pymysql from itemadapter import ItemAdapter from datetime import datetime class MySQLPipeline: """MySQL存储管道""" def __init__(self, host, port, user, password, database): self.host = host self.port = port self.user = user self.password = password self.database = database self.conn = None self.cursor = None @classmethod def from_crawler(cls, crawler): return cls( host=crawler.settings.get("MYSQL_HOST"), port=crawler.settings.get("MYSQL_PORT", 3306), user=crawler.settings.get("MYSQL_USER"), password=crawler.settings.get("MYSQL_PASSWORD"), database=crawler.settings.get("MYSQL_DATABASE"), ) def open_spider(self, spider): """爬虫启动时创建连接和表""" self.conn = pymysql.connect( host=self.host, port=self.port, user=self.user, password=self.password, database=self.database, charset="utf8mb4", ) self.cursor = self.conn.cursor() self._create_tables() def _create_tables(self): """创建数据表""" self.cursor.execute(""" CREATE TABLE IF NOT EXISTS surnames ( id INT AUTO_INCREMENT PRIMARY KEY, surname VARCHAR(10) NOT NULL UNIQUE, pinyin VARCHAR(50), population BIGINT, ratio DECIMAL(10,6), rank INT, origin TEXT, origin_type VARCHAR(50), junwang VARCHAR(500), tanghao VARCHAR(500), main_provinces VARCHAR(500), migration_history TEXT, source_url VARCHAR(500), data_sources VARCHAR(100), crawled_at DATETIME, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_surname (surname), INDEX idx_rank (rank), INDEX idx_population (population) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """) self.cursor.execute(""" CREATE TABLE IF NOT EXISTS famous_people ( id INT AUTO_INCREMENT PRIMARY KEY, surname VARCHAR(10) NOT NULL, name VARCHAR(50) NOT NULL, dynasty VARCHAR(50), title VARCHAR(100), achievement TEXT, source_url VARCHAR(500), crawled_at DATETIME, UNIQUE KEY unique_person (surname, name), INDEX idx_surname (surname), INDEX idx_dynasty (dynasty) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """) self.conn.commit() def process_item(self, item, spider): adapter = ItemAdapter(item) if "origin" in adapter.field_names(): # 姓氏数据 self._upsert_surname(adapter) elif "achievement" in adapter.field_names(): # 名人数据 self._upsert_famous_people(adapter) return item def _upsert_surname(self, adapter): """插入或更新姓氏数据""" sql = """ INSERT INTO surnames (surname, pinyin, population, ratio, `rank`, origin, origin_type, junwang, tanghao, main_provinces, migration_history, source_url, crawled_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE pinyin=VALUES(pinyin), population=VALUES(population), ratio=VALUES(ratio), origin=VALUES(origin), junwang=VALUES(junwang), tanghao=VALUES(tanghao), main_provinces=VALUES(main_provinces), migration_history=VALUES(migration_history), source_url=VALUES(source_url) """ self.cursor.execute(sql, ( adapter.get("surname"), adapter.get("pinyin"), adapter.get("population"), adapter.get("ratio"), adapter.get("rank"), adapter.get("origin"), adapter.get("origin_type"), adapter.get("junwang"), adapter.get("tanghao"), adapter.get("main_provinces"), adapter.get("migration_history"), adapter.get("source_url"), adapter.get("crawled_at", datetime.now()), )) self.conn.commit() def _upsert_famous_people(self, adapter): """插入或更新名人数据""" sql = """ INSERT INTO famous_people (surname, name, dynasty, title, achievement, source_url, crawled_at) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE dynasty=VALUES(dynasty), title=VALUES(title), achievement=VALUES(achievement) """ self.cursor.execute(sql, ( adapter.get("surname"), adapter.get("name"), adapter.get("dynasty"), adapter.get("title"), adapter.get("achievement"), adapter.get("source_url"), adapter.get("crawled_at", datetime.now()), )) self.conn.commit() def close_spider(self, spider): """爬虫关闭时释放连接""" if self.cursor: self.cursor.close() if self.conn: self.conn.close()

6.2 数据质量校验

import pandas as pd class DataQualityChecker: """数据质量校验""" def __init__(self, df: pd.DataFrame): self.df = df self.report = {} def check_completeness(self): """检查完整性（缺失值）""" required_fields = ["surname", "population", "rank", "origin"] missing = {} for field in required_fields: if field in self.df.columns: missing_count = self.df[field].isna().sum() + (self.df[field] == "").sum() missing[field] = { "missing_count": int(missing_count), "missing_rate": round(missing_count / len(self.df) * 100, 2), } self.report["completeness"] = missing return missing def check_uniqueness(self): """检查唯一性（重复值）""" dup_count = self.df["surname"].duplicated().sum() self.report["uniqueness"] = { "duplicate_count": int(dup_count), "duplicate_rate": round(dup_count / len(self.df) * 100, 2), } return self.report["uniqueness"] def check_consistency(self): """检查一致性（数据逻辑）""" issues = [] # 排名应该连续且唯一 if "rank" in self.df.columns: ranks = self.df["rank"].dropna().sort_values() if len(ranks) > 0: expected = range(1, len(ranks) + 1) if list(ranks) != list(expected): issues.append("排名不连续或有重复") # 人口排名应该与人口数一致 if "population" in self.df.columns and "rank" in self.df.columns: sorted_by_pop = self.df.sort_values("population", ascending=False) if not sorted_by_pop["rank"].is_monotonic_increasing: issues.append("人口数与排名不一致") # 比例应该在0-1之间 if "ratio" in self.df.columns: invalid_ratio = self.df[ (self.df["ratio"] < 0) | (self.df["ratio"] > 1) ] if len(invalid_ratio) > 0: issues.append(f"有{len(invalid_ratio)}条比例数据超出0-1范围") self.report["consistency"] = issues return issues def check_accuracy(self, reference_df: pd.DataFrame): """检查准确性（与参考数据对比）""" merged = self.df.merge( reference_df, on="surname", suffixes=("_crawled", "_reference"), how="inner", ) # 人口数据差异 if "population_crawled" in merged.columns: pop_diff = ( abs(merged["population_crawled"] - merged["population_reference"]) / merged["population_reference"] ) accuracy = (pop_diff < 0.1).mean() * 100 # 差异小于10%视为准确 self.report["accuracy"] = { "population_accuracy": round(accuracy, 2), "sample_count": len(merged), } return self.report.get("accuracy", {}) def generate_report(self) -> str: """生成数据质量报告""" report_lines = ["# 数据质量报告", ""] report_lines.append(f"## 总记录数：{len(self.df)}") report_lines.append("") if "completeness" in self.report: report_lines.append("## 完整性检查") for field, info in self.report["completeness"].items(): report_lines.append( f"- {field}：缺失 {info['missing_count']} 条（{info['missing_rate']}%）" ) report_lines.append("") if "uniqueness" in self.report: report_lines.append("## 唯一性检查") report_lines.append( f"- 重复记录：{self.report['uniqueness']['duplicate_count']} 条" ) report_lines.append("") if "consistency" in self.report: report_lines.append("## 一致性检查") if self.report["consistency"]: for issue in self.report["consistency"]: report_lines.append(f"- {issue}") else: report_lines.append("- 未发现一致性问题") report_lines.append("") return "\n".join(report_lines)

上述采集和存储结果可在 在线起名工具 中验证，姓氏知识库已覆盖 500 + 姓氏、3000 + 历史名人，数据完整率超过 95%。

七、效果验证与性能测试

7.1 采集结果统计

表格

7.2 数据质量报告

表格

7.3 爬虫性能测试

表格

八、优化方向与注意事项

8.1 踩过的坑

1. 动态渲染内容：部分姓氏详情页的起源描述通过 AJAX 加载，Scrapy 无法直接获取。解决方案是分析 AJAX 接口直接调用，或使用 Selenium/Playwright 渲染。
2. 反爬升级：采集到约 200 个姓氏时，目标站点增加了验证码。解决方案是降低并发数、增加延迟、使用代理 IP 池，必要时暂停采集。
3. 数据格式不统一：不同页面的人口数据格式差异大（“9540 万”"95,400,000"“约 9500 万”），需要编写灵活的解析函数处理多种格式。
4. 异体字和多音字：部分姓氏有多种写法（如 “邱” 和 “丘”、“肖” 和 “萧”），需要建立映射表进行标准化，否则会导致重复数据。
5. 增量更新：全量采集耗时较长，实际运行中应支持增量更新 —— 只采集新增或变更的姓氏，通过对比采集时间和内容哈希判断是否需要更新。

8.2 法律合规注意事项

• 爬取前务必阅读目标站点的robots.txt和用户协议
• 控制爬取频率，避免对目标服务器造成压力
• 爬取的数据仅供个人 / 内部研究使用，不得二次分发
• 姓氏文化数据属于公有领域，但网站的排版和注释可能受版权保护
• 名人信息涉及个人隐私，已故历史名人信息可公开使用，在世名人需谨慎

8.3 后续优化方向

• 多源自动融合：目前融合需要手动指定主数据源，后续可实现基于数据质量评分的自动融合
• 知识图谱构建：将姓氏、起源、郡望、名人、地域等实体构建为知识图谱，支持更复杂的查询
• 定时增量采集：配置定时任务，每周自动采集新增和变更的数据
• 数据可视化：构建姓氏数据可视化大屏，展示姓氏分布、人口趋势、名人分布等
• NLP 增强：使用 NLP 技术从起源描述中自动提取起源类型、关键人物、时间节点等结构化信息

九、总结

本文完整记录了用 Scrapy 爬取百家姓与姓氏源流数据的全过程，核心要点如下：

1. 多源采集是数据质量的保障：单一数据源存在覆盖不全、信息陈旧等问题，通过主数据源 + 补充源 + 交叉验证的多源采集策略，数据完整率从单源的 75% 提升到 95% 以上。
2. Scrapy 架构设计要兼顾效率与稳健：随机 UA、代理池、指数退避重试、并发控制是应对反爬的标准配置，合理的中间件设计能大幅提升采集成功率。
3. 数据清洗和标准化不可省略：人口格式不统一、异体字、多音字、HTML 残留等问题必须在清洗阶段处理，否则会导致后续分析出错。
4. 多源融合需要明确优先级：以主数据源为基础，缺失字段用其他数据源补充，数值字段取平均值，文本字段取更长更详细的版本，这样的融合策略能兼顾数据完整性和准确性。
5. 数据质量校验是最后一道防线：完整性、唯一性、一致性、准确性四个维度的校验，能及时发现采集和清洗过程中的问题，确保入库数据质量。

通过这套爬虫系统，我们构建了包含 500 + 姓氏、3000 + 历史名人的结构化姓氏知识库，为起名系统的姓氏文化展示和个性化推荐提供了坚实的数据支撑。

如果本文对你有帮助，欢迎点赞收藏，有问题评论区交流。