一、前言姓氏是中国文化的重要载体每个姓氏背后都承载着几千年的家族源流、地域文化和历史名人。对于起名类产品来说姓氏数据是核心基础数据之一 —— 了解一个姓氏的起源、郡望、堂号、历史名人才能为用户提供更有文化内涵的名字推荐。笔者在开发529宝宝起名网时发现单一数据源的姓氏数据存在覆盖不全、信息陈旧、字段缺失等问题于是决定通过多源数据采集构建一套完整的姓氏知识库。百家姓相关的数据在网络上分布广泛但质量参差不齐。有的网站只收录了前 100 个姓氏有的起源描述过于简略有的缺少郡望堂号信息。要构建高质量的姓氏知识库需要从多个数据源采集然后进行清洗、去重、融合和校验。本文将从零开始完整记录用 Scrapy 框架爬取百家姓与姓氏源流数据的全过程涵盖多源数据对比与采集目标设计Scrapy 爬虫架构与中间件设计多源数据采集实现列表页 详情页 名人数据数据清洗、去重与多源融合结构化存储与数据质量校验效果验证与性能优化二、数据来源与采集目标2.1 候选数据源对比表格数据源覆盖姓氏数信息丰富度反爬强度数据质量适用性某百家姓查询站504高起源 郡望 名人低中主数据源某姓氏文化网300高源流详细中高补充源某国学网站1000低仅起源简述低中覆盖补充百度百科5000高但格式不统一高高交叉验证维基百科2000中中高交叉验证综合评估后我们选择某百家姓查询站作为主数据源某姓氏文化网作为补充源百度百科用于交叉验证。本文以主数据源的爬取过程为例进行讲解。2.2 采集字段设计表格字段名类型说明必要性surnamestr姓氏必须pinyinstr姓氏拼音必须populationint人口数必须ratiofloat占全国人口比例必须rankint全国排名必须origintext姓氏起源必须origin_typestr起源类型以国为氏 / 以邑为氏等可选junwangstr郡望可选tanghaostr堂号可选famous_peoplejson历史名人列表可选main_provincesstr主要分布省份可选migration_historytext迁徙历史可选source_urlstr数据来源 URL必须crawled_atdatetime采集时间自动2.3 数据量预估表格数据类型预估数量单条大小总大小姓氏基础信息500约 2KB约 1MB姓氏起源详情500约 5KB约 2.5MB历史名人3000约 1KB约 3MB郡望堂号800约 0.5KB约 0.4MB合计4800-约 7MB三、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 response3.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( urlitem[source_url], callbackself.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(urlurljoin(response.url, next_page), callbackself.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 None4.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, surnamesNone, *args, **kwargs): super().__init__(*args, **kwargs) # 支持传入姓氏列表默认爬取TOP100姓氏 self.surnames surnames or [王, 李, 张, 刘, 陈] def start_requests(self): 生成起始请求 for surname in self.surnames: url fhttps://www.example-baijiaxing.com/famous/{surname}.html yield scrapy.Request( urlurl, callbackself.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( urlurljoin(response.url, detail_url), callbackself.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( urlurljoin(response.url, next_page), callbackself.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, encodingutf-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, onsurname, howouter, suffixes(, f_{source_name}), indicatorTrue, ) # 标记数据来源 merged[data_sources] merged.apply( lambda row: self._update_sources(row, source_name), axis1 ) # 字段优先级主数据源优先缺失字段用其他数据源补充 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], axis1, ) # 数值字段取平均值人口、比例 for field in [population, ratio]: source_field f{field}_{source_name} if source_field in merged.columns: merged[field] merged[[field, source_field]].mean(axis1) # 清理合并产生的临时列 drop_cols [c for c in merged.columns if c.endswith(f_{source_name})] merged merged.drop(columnsdrop_cols [_merge], errorsignore) 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( hostcrawler.settings.get(MYSQL_HOST), portcrawler.settings.get(MYSQL_PORT, 3306), usercrawler.settings.get(MYSQL_USER), passwordcrawler.settings.get(MYSQL_PASSWORD), databasecrawler.settings.get(MYSQL_DATABASE), ) def open_spider(self, spider): 爬虫启动时创建连接和表 self.conn pymysql.connect( hostself.host, portself.port, userself.user, passwordself.password, databaseself.database, charsetutf8mb4, ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) 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 pinyinVALUES(pinyin), populationVALUES(population), ratioVALUES(ratio), originVALUES(origin), junwangVALUES(junwang), tanghaoVALUES(tanghao), main_provincesVALUES(main_provinces), migration_historyVALUES(migration_history), source_urlVALUES(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 dynastyVALUES(dynasty), titleVALUES(title), achievementVALUES(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, ascendingFalse) 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, onsurname, suffixes(_crawled, _reference), howinner, ) # 人口数据差异 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 采集结果统计表格数据类型目标数量实际采集成功率说明姓氏基础信息50449898.8%6 个生僻姓氏页面 404姓氏起源详情50448596.2%19 个姓氏起源描述为空郡望堂号50441281.7%小姓氏郡望信息缺失历史名人30003256108.5%超出预期大姓氏名人多主要分布省份50445690.5%小姓氏分布数据不全7.2 数据质量报告表格检查项结果说明姓氏字段完整率100%所有记录都有姓氏人口数据完整率98.5%7 条人口数据缺失起源描述完整率96.2%19 条起源为空郡望完整率81.7%小姓氏缺失较多重复记录率0.3%去重后已清理与参考数据准确率94.2%人口数据差异小于 10%排名一致性99.2%4 条排名与人口不一致7.3 爬虫性能测试表格指标数值说明总采集时长约 2 小时 30 分钟500 个姓氏详情页 3000 名人平均请求间隔2.5 秒含随机延迟并发数4单域名并发限制请求成功率97.8%失败后自动重试 3 次平均页面下载时间1.2 秒含网络延迟数据处理速度约 50 条 / 分钟含解析、清洗、存储内存占用约 150MBScrapy 运行时生成原始数据约 8MBJSON 格式八、优化方向与注意事项8.1 踩过的坑动态渲染内容部分姓氏详情页的起源描述通过 AJAX 加载Scrapy 无法直接获取。解决方案是分析 AJAX 接口直接调用或使用 Selenium/Playwright 渲染。反爬升级采集到约 200 个姓氏时目标站点增加了验证码。解决方案是降低并发数、增加延迟、使用代理 IP 池必要时暂停采集。数据格式不统一不同页面的人口数据格式差异大“9540 万”95,400,000“约 9500 万”需要编写灵活的解析函数处理多种格式。异体字和多音字部分姓氏有多种写法如 “邱” 和 “丘”、“肖” 和 “萧”需要建立映射表进行标准化否则会导致重复数据。增量更新全量采集耗时较长实际运行中应支持增量更新 —— 只采集新增或变更的姓氏通过对比采集时间和内容哈希判断是否需要更新。8.2 法律合规注意事项爬取前务必阅读目标站点的robots.txt和用户协议控制爬取频率避免对目标服务器造成压力爬取的数据仅供个人 / 内部研究使用不得二次分发姓氏文化数据属于公有领域但网站的排版和注释可能受版权保护名人信息涉及个人隐私已故历史名人信息可公开使用在世名人需谨慎8.3 后续优化方向多源自动融合目前融合需要手动指定主数据源后续可实现基于数据质量评分的自动融合知识图谱构建将姓氏、起源、郡望、名人、地域等实体构建为知识图谱支持更复杂的查询定时增量采集配置定时任务每周自动采集新增和变更的数据数据可视化构建姓氏数据可视化大屏展示姓氏分布、人口趋势、名人分布等NLP 增强使用 NLP 技术从起源描述中自动提取起源类型、关键人物、时间节点等结构化信息九、总结本文完整记录了用 Scrapy 爬取百家姓与姓氏源流数据的全过程核心要点如下多源采集是数据质量的保障单一数据源存在覆盖不全、信息陈旧等问题通过主数据源 补充源 交叉验证的多源采集策略数据完整率从单源的 75% 提升到 95% 以上。Scrapy 架构设计要兼顾效率与稳健随机 UA、代理池、指数退避重试、并发控制是应对反爬的标准配置合理的中间件设计能大幅提升采集成功率。数据清洗和标准化不可省略人口格式不统一、异体字、多音字、HTML 残留等问题必须在清洗阶段处理否则会导致后续分析出错。多源融合需要明确优先级以主数据源为基础缺失字段用其他数据源补充数值字段取平均值文本字段取更长更详细的版本这样的融合策略能兼顾数据完整性和准确性。数据质量校验是最后一道防线完整性、唯一性、一致性、准确性四个维度的校验能及时发现采集和清洗过程中的问题确保入库数据质量。通过这套爬虫系统我们构建了包含 500 姓氏、3000 历史名人的结构化姓氏知识库为起名系统的姓氏文化展示和个性化推荐提供了坚实的数据支撑。如果本文对你有帮助欢迎点赞收藏有问题评论区交流。 用 Scrapy 爬取百家姓与姓氏源流数据：多源采集、数据清洗与结构化存储实战
一、前言姓氏是中国文化的重要载体每个姓氏背后都承载着几千年的家族源流、地域文化和历史名人。对于起名类产品来说姓氏数据是核心基础数据之一 —— 了解一个姓氏的起源、郡望、堂号、历史名人才能为用户提供更有文化内涵的名字推荐。笔者在开发529宝宝起名网时发现单一数据源的姓氏数据存在覆盖不全、信息陈旧、字段缺失等问题于是决定通过多源数据采集构建一套完整的姓氏知识库。百家姓相关的数据在网络上分布广泛但质量参差不齐。有的网站只收录了前 100 个姓氏有的起源描述过于简略有的缺少郡望堂号信息。要构建高质量的姓氏知识库需要从多个数据源采集然后进行清洗、去重、融合和校验。本文将从零开始完整记录用 Scrapy 框架爬取百家姓与姓氏源流数据的全过程涵盖多源数据对比与采集目标设计Scrapy 爬虫架构与中间件设计多源数据采集实现列表页 详情页 名人数据数据清洗、去重与多源融合结构化存储与数据质量校验效果验证与性能优化二、数据来源与采集目标2.1 候选数据源对比表格数据源覆盖姓氏数信息丰富度反爬强度数据质量适用性某百家姓查询站504高起源 郡望 名人低中主数据源某姓氏文化网300高源流详细中高补充源某国学网站1000低仅起源简述低中覆盖补充百度百科5000高但格式不统一高高交叉验证维基百科2000中中高交叉验证综合评估后我们选择某百家姓查询站作为主数据源某姓氏文化网作为补充源百度百科用于交叉验证。本文以主数据源的爬取过程为例进行讲解。2.2 采集字段设计表格字段名类型说明必要性surnamestr姓氏必须pinyinstr姓氏拼音必须populationint人口数必须ratiofloat占全国人口比例必须rankint全国排名必须origintext姓氏起源必须origin_typestr起源类型以国为氏 / 以邑为氏等可选junwangstr郡望可选tanghaostr堂号可选famous_peoplejson历史名人列表可选main_provincesstr主要分布省份可选migration_historytext迁徙历史可选source_urlstr数据来源 URL必须crawled_atdatetime采集时间自动2.3 数据量预估表格数据类型预估数量单条大小总大小姓氏基础信息500约 2KB约 1MB姓氏起源详情500约 5KB约 2.5MB历史名人3000约 1KB约 3MB郡望堂号800约 0.5KB约 0.4MB合计4800-约 7MB三、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 response3.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( urlitem[source_url], callbackself.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(urlurljoin(response.url, next_page), callbackself.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 None4.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, surnamesNone, *args, **kwargs): super().__init__(*args, **kwargs) # 支持传入姓氏列表默认爬取TOP100姓氏 self.surnames surnames or [王, 李, 张, 刘, 陈] def start_requests(self): 生成起始请求 for surname in self.surnames: url fhttps://www.example-baijiaxing.com/famous/{surname}.html yield scrapy.Request( urlurl, callbackself.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( urlurljoin(response.url, detail_url), callbackself.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( urlurljoin(response.url, next_page), callbackself.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, encodingutf-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, onsurname, howouter, suffixes(, f_{source_name}), indicatorTrue, ) # 标记数据来源 merged[data_sources] merged.apply( lambda row: self._update_sources(row, source_name), axis1 ) # 字段优先级主数据源优先缺失字段用其他数据源补充 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], axis1, ) # 数值字段取平均值人口、比例 for field in [population, ratio]: source_field f{field}_{source_name} if source_field in merged.columns: merged[field] merged[[field, source_field]].mean(axis1) # 清理合并产生的临时列 drop_cols [c for c in merged.columns if c.endswith(f_{source_name})] merged merged.drop(columnsdrop_cols [_merge], errorsignore) 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( hostcrawler.settings.get(MYSQL_HOST), portcrawler.settings.get(MYSQL_PORT, 3306), usercrawler.settings.get(MYSQL_USER), passwordcrawler.settings.get(MYSQL_PASSWORD), databasecrawler.settings.get(MYSQL_DATABASE), ) def open_spider(self, spider): 爬虫启动时创建连接和表 self.conn pymysql.connect( hostself.host, portself.port, userself.user, passwordself.password, databaseself.database, charsetutf8mb4, ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; ) 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 pinyinVALUES(pinyin), populationVALUES(population), ratioVALUES(ratio), originVALUES(origin), junwangVALUES(junwang), tanghaoVALUES(tanghao), main_provincesVALUES(main_provinces), migration_historyVALUES(migration_history), source_urlVALUES(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 dynastyVALUES(dynasty), titleVALUES(title), achievementVALUES(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, ascendingFalse) 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, onsurname, suffixes(_crawled, _reference), howinner, ) # 人口数据差异 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 采集结果统计表格数据类型目标数量实际采集成功率说明姓氏基础信息50449898.8%6 个生僻姓氏页面 404姓氏起源详情50448596.2%19 个姓氏起源描述为空郡望堂号50441281.7%小姓氏郡望信息缺失历史名人30003256108.5%超出预期大姓氏名人多主要分布省份50445690.5%小姓氏分布数据不全7.2 数据质量报告表格检查项结果说明姓氏字段完整率100%所有记录都有姓氏人口数据完整率98.5%7 条人口数据缺失起源描述完整率96.2%19 条起源为空郡望完整率81.7%小姓氏缺失较多重复记录率0.3%去重后已清理与参考数据准确率94.2%人口数据差异小于 10%排名一致性99.2%4 条排名与人口不一致7.3 爬虫性能测试表格指标数值说明总采集时长约 2 小时 30 分钟500 个姓氏详情页 3000 名人平均请求间隔2.5 秒含随机延迟并发数4单域名并发限制请求成功率97.8%失败后自动重试 3 次平均页面下载时间1.2 秒含网络延迟数据处理速度约 50 条 / 分钟含解析、清洗、存储内存占用约 150MBScrapy 运行时生成原始数据约 8MBJSON 格式八、优化方向与注意事项8.1 踩过的坑动态渲染内容部分姓氏详情页的起源描述通过 AJAX 加载Scrapy 无法直接获取。解决方案是分析 AJAX 接口直接调用或使用 Selenium/Playwright 渲染。反爬升级采集到约 200 个姓氏时目标站点增加了验证码。解决方案是降低并发数、增加延迟、使用代理 IP 池必要时暂停采集。数据格式不统一不同页面的人口数据格式差异大“9540 万”95,400,000“约 9500 万”需要编写灵活的解析函数处理多种格式。异体字和多音字部分姓氏有多种写法如 “邱” 和 “丘”、“肖” 和 “萧”需要建立映射表进行标准化否则会导致重复数据。增量更新全量采集耗时较长实际运行中应支持增量更新 —— 只采集新增或变更的姓氏通过对比采集时间和内容哈希判断是否需要更新。8.2 法律合规注意事项爬取前务必阅读目标站点的robots.txt和用户协议控制爬取频率避免对目标服务器造成压力爬取的数据仅供个人 / 内部研究使用不得二次分发姓氏文化数据属于公有领域但网站的排版和注释可能受版权保护名人信息涉及个人隐私已故历史名人信息可公开使用在世名人需谨慎8.3 后续优化方向多源自动融合目前融合需要手动指定主数据源后续可实现基于数据质量评分的自动融合知识图谱构建将姓氏、起源、郡望、名人、地域等实体构建为知识图谱支持更复杂的查询定时增量采集配置定时任务每周自动采集新增和变更的数据数据可视化构建姓氏数据可视化大屏展示姓氏分布、人口趋势、名人分布等NLP 增强使用 NLP 技术从起源描述中自动提取起源类型、关键人物、时间节点等结构化信息九、总结本文完整记录了用 Scrapy 爬取百家姓与姓氏源流数据的全过程核心要点如下多源采集是数据质量的保障单一数据源存在覆盖不全、信息陈旧等问题通过主数据源 补充源 交叉验证的多源采集策略数据完整率从单源的 75% 提升到 95% 以上。Scrapy 架构设计要兼顾效率与稳健随机 UA、代理池、指数退避重试、并发控制是应对反爬的标准配置合理的中间件设计能大幅提升采集成功率。数据清洗和标准化不可省略人口格式不统一、异体字、多音字、HTML 残留等问题必须在清洗阶段处理否则会导致后续分析出错。多源融合需要明确优先级以主数据源为基础缺失字段用其他数据源补充数值字段取平均值文本字段取更长更详细的版本这样的融合策略能兼顾数据完整性和准确性。数据质量校验是最后一道防线完整性、唯一性、一致性、准确性四个维度的校验能及时发现采集和清洗过程中的问题确保入库数据质量。通过这套爬虫系统我们构建了包含 500 姓氏、3000 历史名人的结构化姓氏知识库为起名系统的姓氏文化展示和个性化推荐提供了坚实的数据支撑。如果本文对你有帮助欢迎点赞收藏有问题评论区交流。