Python爬虫实战：地名志数据采集与SQLite存储-育师

1. 项目背景与核心价值

地名志这类地方志文献通常包含行政区划沿革、地名由来、地理特征等珍贵数据，但往往以PDF或网页形式存在，难以直接分析利用。去年我在做一个历史文化研究项目时，需要批量分析3000多个地名的时空分布特征，手动复制粘贴显然不现实。这就是Python爬虫+结构化存储的典型应用场景。

通过爬取地名志目录页，我们可以获得完整的地名列表及其层级关系。比如省级地名志通常按"省-市-县-乡镇"四级结构编排，这种半结构化数据非常适合用Python提取后存入SQLite数据库。与MySQL等重型数据库相比，SQLite的轻量特性特别适合存储百万级以下的地名数据，查询效率也能满足大多数研究需求。

2. 技术方案设计

2.1 整体架构设计

项目采用经典的三层架构：

数据采集层：Requests+BeautifulSoup组合
数据处理层：Pandas进行数据清洗
数据存储层：SQLite3持久化存储

选择这个方案主要考虑：

Requests比urllib3更人性化的API设计
BeautifulSoup在解析不规则HTML时比lxml更容错
SQLite无需服务端，适合单机研究项目

2.2 关键代码结构

├── main.py # 主流程控制 ├── crawler.py # 爬虫核心模块 ├── processor.py # 数据处理模块 ├── db.py # 数据库操作 └── config.py # 配置文件

3. 爬虫实现细节

3.1 网页请求与反爬策略

使用Requests时需要特别注意：

设置随机User-Agent（准备10个以上常用浏览器UA）
控制请求频率（建议2-5秒/次）
处理HTTP 429状态码

headers = { 'User-Agent': random.choice(user_agent_list), 'Accept-Language': 'zh-CN,zh;q=0.9' } def safe_request(url): try: time.sleep(random.uniform(2, 5)) response = requests.get(url, headers=headers) response.raise_for_status() return response except Exception as e: print(f"请求失败: {e}") return None

3.2 HTML解析技巧

地名志目录页通常有两种结构：

树形目录（ul/li嵌套）
表格形式（table标签）

对于复杂嵌套结构，BeautifulSoup的find_all()配合CSS选择器是首选：

def parse_tree(html): soup = BeautifulSoup(html, 'html.parser') regions = [] # 省一级 for prov in soup.select('div.province > h2'): province = prov.text.strip() # 市一级 for city in prov.find_next_sibling('ul').select('li.city'): city_name = city.text.strip() regions.append({ 'name': city_name, 'type': 'city', 'parent': province }) return regions

4. 数据结构化处理

4.1 数据清洗要点

原始数据常见问题：

多余空格和特殊字符（\xa0等）
非标准行政区划名称
层级关系错乱

使用Pandas进行清洗：

def clean_data(df): # 去除特殊字符 df['name'] = df['name'].str.replace(r'[\s\xa0]+', ' ', regex=True) # 标准化类型 type_mapping = { '市': 'city', '县': 'county', '镇': 'town' } df['type'] = df['type'].map(type_mapping) return df

4.2 数据结构设计

SQLite表结构设计建议：

CREATE TABLE regions ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, type TEXT CHECK(type IN ('province', 'city', 'county', 'town')), parent_id INTEGER, source_url TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (parent_id) REFERENCES regions(id) );

5. SQLite数据库操作

5.1 高效写入方案

批量插入使用executemany()比单条插入快10倍以上：

def batch_insert(conn, data): sql = """INSERT INTO regions (name, type, parent_id, source_url) VALUES (?, ?, ?, ?)""" conn.executemany(sql, data) conn.commit()

5.2 查询优化技巧

建立合适的索引能大幅提升查询速度：

CREATE INDEX idx_region_name ON regions(name); CREATE INDEX idx_region_type ON regions(type); CREATE INDEX idx_region_parent ON regions(parent_id);

6. 完整示例代码

import sqlite3 from bs4 import BeautifulSoup import requests import pandas as pd import time import random # 配置部分 user_agent_list = [...] base_url = "http://example.com/dimingzhi" def main(): # 初始化数据库 conn = sqlite3.connect('regions.db') init_db(conn) # 爬取数据 html = safe_request(base_url).text regions = parse_tree(html) # 数据处理 df = pd.DataFrame(regions) clean_df = clean_data(df) # 存储数据 data_tuples = [tuple(x) for x in clean_df.to_records(index=False)] batch_insert(conn, data_tuples) conn.close() if __name__ == "__main__": main()

7. 常见问题与解决方案

7.1 反爬机制突破

问题：网站返回403状态码解决方案：

使用requests.Session()保持会话
添加Referer头
考虑使用selenium模拟浏览器

7.2 数据不一致处理

问题：某些下级区域没有对应上级解决方案：

def validate_hierarchy(df): # 建立父子关系映射 parent_map = {row['id']: row['parent_id'] for row in df.itertuples()} # 验证所有parent_id都存在 for _, row in df.iterrows(): if row['parent_id'] and row['parent_id'] not in parent_map: print(f"警告：无效父ID {row['parent_id']}") # 自动修复为上一级有效父ID row['parent_id'] = find_valid_parent(row['type'])

8. 高级技巧与优化

8.1 增量爬取策略

记录已爬取URL避免重复：

def get_unvisited_links(conn): visited = set(row[0] for row in conn.execute("SELECT url FROM visited_urls")) all_links = extract_links(current_page) return [link for link in all_links if link not in visited]

8.2 分布式爬虫扩展

使用Redis作为任务队列：

import redis r = redis.Redis() def distribute_tasks(): while True: url = r.lpop('task_queue') if not url: break process_page(url)

9. 可视化与分析拓展

存储到SQLite后可以方便地进行数据分析：

import matplotlib.pyplot as plt def analyze_data(conn): df = pd.read_sql(""" SELECT type, COUNT(*) as count FROM regions GROUP BY type """, conn) df.plot(kind='bar', x='type', y='count') plt.title('行政区划类型分布') plt.show()