㊗️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～
㊙️本期爬虫难度指数：⭐⭐☆☆☆（基础级）
🉐福利： 一次订阅后，专栏内的所有文章可永久免费看，持续更新中，保底1000+(篇)硬核实战内容。

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。
运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO
欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏👉 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

• ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
• ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
• ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
• ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅专栏👉《Python爬虫实战》👈，一次订阅后，专栏内的所有文章可永久免费阅读，持续更新中。
  
💕订阅后更新会优先推送，按目录学习更高效💯～

0️⃣ 前言（Preface）

这篇文章要爬取 GeoNames 行政区目录页，使用 Python 的 requests + BeautifulSoup + lxml + SQLite 完成采集、解析、清洗、入库与导出，最终得到一份可本地查询的行政区地理字典库。

读完这篇文章，你可以获得：

1. 一套从网页目录页采集 GeoNames 行政区数据的完整工程代码。
2. 一个适合 SQLite 入库的行政区字段设计，包括名称、国家、行政级别、GeoName ID、链接。
3. 一些实际爬虫开发中很容易踩坑的处理经验，比如 robots、频控、重试、去重、断点续跑和解析容错。

我个人比较喜欢 GeoNames 这类数据源。它不像新闻站点那样页面变化很频繁，也不像电商页面那样到处都是动态渲染和风控；它更接近“地理字典库”，字段稳定、关系清楚，适合拿来做数据入门项目，也适合拿来给业务系统做地名标准化、行政区映射、经纬度补全等基础能力。唯一要注意的是：越基础的数据，越不能随便抓完就用，字段定义、数据来源、许可和更新周期都需要认真看一眼。

1️⃣ 摘要（Abstract）

本文围绕 GeoNames 行政区目录页展开，使用 Python 爬取行政区列表和详情链接，抽取名称、国家、行政级别、GeoName ID、来源链接等字段，并将清洗后的结果写入 SQLite，同时支持 CSV 导出。

读完本文你将掌握：

1. 如何判断一个站点适合用静态 HTML 抓取、API 获取，还是需要浏览器渲染。
2. 如何把爬虫拆成 Fetcher、Parser、Storage 三层，降低后期维护成本。
3. 如何为地理字典类数据设计去重、容错、断点续跑和本地查询方案。

本文代码不是为了“炫技”，而是为了可复现、可维护、可扩展。我的建议一直很简单：能用稳定公开接口就优先用接口，确实需要解析页面时，再写克制、低频、可停止的爬虫。

2️⃣ 背景与需求（Why）

2.1 为什么要爬 GeoNames 行政区数据

在很多数据项目里，地理字段看起来只是一个普通维度，但真正落地时经常会遇到这些问题：

• 同一个地方有不同语言名称，例如 Zhejiang、浙江、Chekiang。
• 用户输入的地区不标准，例如 NY、New York、New York State 可能指向不同粒度。
• 数据来自多个系统，不同系统的行政区编码不一致。
• 报表需要按国家、省/州、市/县等行政层级汇总。
• 地图、物流、风控、内容推荐等业务需要把地名转换成标准地理实体。

GeoNames 的价值在于它给很多地理实体提供了稳定的 geonameId。这个 ID 可以作为本地地理字典库的主键或外部参照键。对于数据分析来说，只要把原始地名映射到标准 ID，后面无论做聚合、去重、关联还是维表补全，都会轻松很多。

这篇文章的目标不是抓全量全球数据。全量数据更适合使用官方 dump 文件。本文主要以“行政区目录页”为对象，写一套小而完整的爬虫工程：先能跑通，再能扩展。

2.2 目标站点与目标字段

目标站点是 GeoNames 的公开页面。本文关注行政区相关结果，也就是 GeoNames 里常见的 featureClass=A 类地理实体。实际页面中，行政区可能包含国家、一级行政区、二级行政区、三级行政区等不同粒度。

本次采集字段如下：

中文字段	英文字段	说明
名称	name	行政区名称
国家	country	行政区所属国家或地区名称
国家代码	country_code	ISO 两位代码，能从页面参数或链接中补充
行政级别	admin_level	如 country、ADM1、ADM2、ADM3
原始级别文本	admin_label	页面上的原始说明，如 first-order administrative division
GeoName ID	geoname_id	GeoNames 的唯一 ID
链接	source_url	对应 GeoNames 详情页链接
内容指纹	content_hash	用于辅助判断记录是否变化
抓取时间	fetched_at	数据进入本地库的时间

为什么要保留 admin_label？因为网页上的描述可能比我们归一化后的 ADM1 更细，比如 “seat of a first-order administrative division” 和 “first-order administrative division” 在业务上可能要区别对待。归一化字段方便查询，原始字段方便回溯。

---

3️⃣ 合规与注意事项

爬虫不是“能抓就行”。尤其是公开数据源，更应该尊重对方服务的运行成本。本文只讨论公开页面、公开字段和低频抓取，不涉及登录绕过、付费绕过、验证码绕过、接口破解，也不采集个人敏感信息。

3.1 robots.txt 基本说明

robots.txt 是站点放在根路径下的一份爬虫访问建议文件。它通常会告诉爬虫哪些路径允许访问，哪些路径不建议访问。它不是权限系统，但对技术人员来说，尊重它是一种基本礼貌。

在项目代码里，我们会用 Python 标准库 urllib.robotparser 做一个运行前检查。生产环境里，我更建议把 robots 检查作为采集任务启动前的固定步骤，而不是想起来才看一眼。

3.2 频率控制

本文代码默认会做这些控制：

• 每次请求设置 timeout，避免连接卡死。
• 请求失败后进行有限重试，不无限循环。
• 请求间隔加入随机抖动，避免固定节奏打到服务器。
• 不使用攻击式并发，不追求极限速度。
• 遇到 429 Too Many Requests 时主动退避。

这类地理字典数据并不需要秒级更新。慢一点、稳一点，比跑得快但经常被封更值得。

3.3 不采集敏感信息

本文目标字段都是行政区公开信息：名称、国家、层级、GeoName ID 和链接。这些字段不涉及个人隐私，也不需要登录态。

如果你在实际项目里扩展字段，请保持一个原则：只采集任务必要字段，不采集和目标无关的信息。爬虫越克制，后续数据治理越轻松。

3.4 不绕过登录和付费限制

如果某些数据需要登录、订阅或付费，请使用对方提供的正式授权方式。爬虫不应该被用来绕过访问限制。本文所有实现都基于公开可访问页面和普通 HTTP 请求，不包含任何规避手段。

---

4️⃣ 技术选型与整体流程（What/How）

4.1 静态、动态还是 API

常见数据采集方式可以分成三类：

类型	判断方式	典型工具
静态 HTML	右键查看源代码能看到主要数据	requests、BeautifulSoup、lxml
动态渲染	源代码没有数据，浏览器执行 JS 后才出现	Playwright、Selenium
API / JSON	Network 里有结构化接口返回 JSON/XML	requests、httpx

本文按照“静态目录页”来写。GeoNames 页面本身偏传统，很多数据可以直接从 HTML 中解析出来，不需要启动浏览器。启动浏览器会增加运行成本，也会让部署复杂很多。除非页面完全由 JavaScript 渲染，否则我不建议一上来就用 Playwright。

不过有一点要说清楚：GeoNames 同时也提供 Webservice 和数据下载。如果你的目标是生产级全量同步，优先考虑官方数据下载或公开 API；如果你的目标是练习网页解析、构建小型字段库，目录页爬取更适合教学。

4.2 整体流程

本文项目流程如下：

输入国家代码 / 起始目录页
        |
        v
Fetcher 请求页面
        |
        v
Parser 解析列表项与详情链接
        |
        v
可选：请求详情页补字段
        |
        v
Cleaner 清洗名称、国家、行政级别、GeoName ID
        |
        v
Storage 写入 SQLite
        |
        v
Exporter 导出 CSV

更简单地说，就是四个词：

采集 -> 解析 -> 清洗 -> 存储

我会把代码拆成几层：

• fetcher.py：负责 HTTP 请求、headers、timeout、重试、robots 检查。
• parser.py：负责 HTML 解析、列表页解析、详情页解析、字段容错。
• storage.py：负责 SQLite 建表、写入、去重、查询。
• exporter.py：负责导出 CSV。
• runner.py：负责命令行入口和流程编排。

这种拆法看起来文件多一点，但后面维护很舒服。页面结构变了，只改 Parser；数据库结构变了，只改 Storage；需要换成异步请求，也主要动 Fetcher。

4.3 为什么选 requests + bs4 + lxml

本文选择：

• requests：成熟、稳定、调试方便。
• BeautifulSoup：对不完美 HTML 容错好。
• lxml：解析速度快，也能配合 CSS/XPath 思路。
• sqlite3：Python 标准库自带，不需要额外安装数据库。
• csv：标准库导出，便于 Excel、BI 或后续脚本读取。

没有选择 Scrapy 的原因是：本文数据规模不大，Scrapy 会让项目结构更专业，但对初学者来说启动成本稍高。没有选择 Playwright 的原因是：页面不需要浏览器渲染，没必要引入重型依赖。

---

5️⃣ 环境准备与依赖安装

5.1 Python 版本

建议使用 Python 3.10 或更高版本。本文代码在语法上没有使用特别激进的新特性，Python 3.10、3.11、3.12 都可以。

查看版本：

python --version

推荐创建虚拟环境：

python -m venv .venv

# macOS / Linux
source .venv/bin/activate

# Windows PowerShell
.venv\Scripts\Activate.ps1

5.2 安装依赖

创建 requirements.txt：

requests==2.32.3
beautifulsoup4==4.12.3
lxml==5.3.0
python-dotenv==1.0.1

安装：

pip install -r requirements.txt

sqlite3、csv、hashlib、argparse、urllib.robotparser 都是标准库，不需要额外安装。

5.3 推荐项目结构

geonames_admin_spider/
├── README.md
├── requirements.txt
├── .env.example
├── data/
│   ├── .gitkeep
│   ├── geonames_admin.sqlite
│   └── geonames_admin.csv
├── logs/
│   └── .gitkeep
└── geonames_admin/
    ├── __init__.py
    ├── config.py
    ├── fetcher.py
    ├── parser.py
    ├── storage.py
    ├── exporter.py
    └── runner.py

初始化目录：

mkdir -p geonames_admin data logs
touch geonames_admin/__init__.py
touch data/.gitkeep logs/.gitkeep

---

6️⃣ 核心实现：请求层（Fetcher）

请求层的目标很明确：不要让解析层关心网络细节。Parser 只应该拿到 HTML 字符串，不应该知道 headers 怎么配、失败怎么重试、timeout 是多少。

6.1 配置文件

创建 geonames_admin/config.py：

from pathlib import Path

BASE_URL = "https://www.geonames.org"
SEARCH_URL = f"{BASE_URL}/search.html"
ROBOTS_URL = f"{BASE_URL}/robots.txt"

PROJECT_ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = PROJECT_ROOT / "data"
LOG_DIR = PROJECT_ROOT / "logs"

DB_PATH = DATA_DIR / "geonames_admin.sqlite"
CSV_PATH = DATA_DIR / "geonames_admin.csv"

DEFAULT_COUNTRY = "US"
DEFAULT_MAX_PAGES = 3

REQUEST_TIMEOUT = (5, 20)
RETRY_TIMES = 3
BACKOFF_BASE_SECONDS = 1.5

MIN_SLEEP_SECONDS = 1.2
MAX_SLEEP_SECONDS = 3.0

USER_AGENT = (
    "Mozilla/5.0 (compatible; GeoNamesAdminSpider/1.0; "
    "+contact: your_email@example.com)"
)

DEFAULT_HEADERS = {
    "User-Agent": USER_AGENT,
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en,zh-CN;q=0.9,zh;q=0.8",
    "Connection": "keep-alive",
}

这里的 User-Agent 建议换成你自己的项目名和联系邮箱。不要伪装成搜索引擎爬虫，也不要随便复制一个很奇怪的浏览器指纹。爬取公开数据时，清楚说明自己的客户端身份更合适。

6.2 Fetcher 代码

创建 geonames_admin/fetcher.py：

import logging
import random
import time
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urljoin, urlencode
from urllib.robotparser import RobotFileParser

import requests
from requests import Response
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

from .config import (
    BASE_URL,
    DEFAULT_HEADERS,
    REQUEST_TIMEOUT,
    RETRY_TIMES,
    BACKOFF_BASE_SECONDS,
    ROBOTS_URL,
    USER_AGENT,
)

logger = logging.getLogger(__name__)


@dataclass
class FetchResult:
    url: str
    status_code: int
    text: str


class RobotsChecker:
    """
    简单 robots.txt 检查器。

    注意：
    1. robots.txt 是爬虫访问建议，不是身份认证系统。
    2. 生产环境建议把 robots 检查作为任务启动前的固定步骤。
    3. 如果 robots.txt 暂时不可访问，本文默认给出警告并继续低频请求。
       如果你的合规要求更严格，可以改成直接拒绝抓取。
    """

    def __init__(self, robots_url: str = ROBOTS_URL, user_agent: str = USER_AGENT):
        self.robots_url = robots_url
        self.user_agent = user_agent
        self.parser = RobotFileParser()
        self.loaded = False

    def load(self) -> None:
        try:
            self.parser.set_url(self.robots_url)
            self.parser.read()
            self.loaded = True
            logger.info("robots.txt loaded: %s", self.robots_url)
        except Exception as exc:
            self.loaded = False
            logger.warning("failed to load robots.txt: %s", exc)

    def can_fetch(self, url: str) -> bool:
        if not self.loaded:
            self.load()

        if not self.loaded:
            logger.warning("robots.txt unavailable, continue with low-frequency mode: %s", url)
            return True

        allowed = self.parser.can_fetch(self.user_agent, url)
        if not allowed:
            logger.warning("robots.txt disallows fetching: %s", url)
        return allowed


class GeoNamesFetcher:
    def __init__(self, min_sleep: float = 1.2, max_sleep: float = 3.0):
        self.session = requests.Session()
        self.session.headers.update(DEFAULT_HEADERS)
        self.min_sleep = min_sleep
        self.max_sleep = max_sleep
        self.robots = RobotsChecker()

        retry = Retry(
            total=RETRY_TIMES,
            connect=RETRY_TIMES,
            read=RETRY_TIMES,
            status=RETRY_TIMES,
            backoff_factor=BACKOFF_BASE_SECONDS,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset(["GET"]),
            raise_on_status=False,
        )

        adapter = HTTPAdapter(max_retries=retry, pool_connections=5, pool_maxsize=5)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def sleep(self) -> None:
        delay = random.uniform(self.min_sleep, self.max_sleep)
        time.sleep(delay)

    def build_search_url(
        self,
        country: str,
        feature_class: str = "A",
        q: str = "",
        extra_params: Optional[dict] = None,
    ) -> str:
        params = {
            "q": q,
            "country": country,
            "featureClass": feature_class,
        }
        if extra_params:
            params.update(extra_params)

        return f"{BASE_URL}/search.html?{urlencode(params)}"

    def get(self, url: str, referer: Optional[str] = None) -> FetchResult:
        absolute_url = urljoin(BASE_URL, url)

        if not self.robots.can_fetch(absolute_url):
            raise PermissionError(f"robots.txt disallows fetching: {absolute_url}")

        headers = {}
        if referer:
            headers["Referer"] = referer

        logger.info("GET %s", absolute_url)
        response: Response = self.session.get(
            absolute_url,
            headers=headers,
            timeout=REQUEST_TIMEOUT,
        )

        if response.status_code == 403:
            raise RuntimeError(
                "HTTP 403 Forbidden. Please slow down, check headers, "
                "or stop crawling if the site does not welcome automated access."
            )

        if response.status_code == 429:
            raise RuntimeError(
                "HTTP 429 Too Many Requests. Please reduce frequency and retry later."
            )

        response.raise_for_status()

        if not response.encoding or response.encoding.lower() == "iso-8859-1":
            response.encoding = response.apparent_encoding or "utf-8"

        return FetchResult(
            url=absolute_url,
            status_code=response.status_code,
            text=response.text,
        )

这段代码里有几个细节值得注意。

第一，timeout 用的是二元组 (5, 20)，分别表示连接超时和读取超时。很多新手只写一个 timeout=30，也能用，但不够清晰。

第二，重试只针对 GET，而且只对网络波动或服务器临时错误重试。403、404 这类错误不能靠重试解决，应该停下来检查。

第三，sleep 放在 Fetcher 里统一处理。不要在各个业务函数里到处 time.sleep()，后面不好调。

---

7️⃣ 核心实现：解析层（Parser）

解析层要解决四件事：

1. 从列表页拿到行政区详情链接。
2. 从链接里提取 geoname_id。
3. 从列表页或详情页抽取名称、国家、行政级别。
4. 字段缺失时尽量容错，不让整个任务崩掉。

GeoNames 页面属于传统 HTML，适合用 BeautifulSoup。为了让代码更耐用，我不会写特别死的选择器，而是采用“结构选择器 + 正则兜底”的方式。

7.1 数据模型

创建 geonames_admin/parser.py：

import hashlib
import re
from dataclasses import dataclass, asdict
from typing import Iterable, Optional
from urllib.parse import urljoin, urlparse, parse_qs

from bs4 import BeautifulSoup, Tag

from .config import BASE_URL


GEONAME_LINK_RE = re.compile(r"/(?P<geoname_id>\d{3,})/[^/]+\.html")
ADMIN_LEVEL_PATTERNS = [
    ("ADM1", re.compile(r"\bfirst-order administrative division\b", re.I)),
    ("ADM2", re.compile(r"\bsecond-order administrative division\b", re.I)),
    ("ADM3", re.compile(r"\bthird-order administrative division\b", re.I)),
    ("ADM4", re.compile(r"\bfourth-order administrative division\b", re.I)),
    ("ADM5", re.compile(r"\bfifth-order administrative division\b", re.I)),
    ("country", re.compile(r"\bcountry\b|\bindependent political entity\b", re.I)),
    ("area", re.compile(r"\bdependent political entity\b|\bpolitical entity\b", re.I)),
]


@dataclass
class AdminRecord:
    name: str
    country: str
    country_code: str
    admin_level: str
    admin_label: str
    geoname_id: int
    source_url: str
    content_hash: str

    def to_dict(self) -> dict:
        return asdict(self)


def clean_text(value: str) -> str:
    if not value:
        return ""
    return re.sub(r"\s+", " ", value).strip()


def make_hash(*parts: str) -> str:
    raw = "||".join(part or "" for part in parts)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def extract_geoname_id(url: str) -> Optional[int]:
    match = GEONAME_LINK_RE.search(url)
    if not match:
        return None
    return int(match.group("geoname_id"))


def normalize_admin_level(text: str) -> tuple[str, str]:
    cleaned = clean_text(text)

    for level, pattern in ADMIN_LEVEL_PATTERNS:
        if pattern.search(cleaned):
            label_match = pattern.search(cleaned)
            label = label_match.group(0) if label_match else cleaned
            return level, label.lower()

    return "", ""


def extract_country_code_from_url(url: str) -> str:
    parsed = urlparse(url)
    query = parse_qs(parsed.query)
    country_values = query.get("country") or query.get("countryCode")
    if country_values:
        return country_values[0].upper()

    path_parts = [p for p in parsed.path.split("/") if p]
    for index, part in enumerate(path_parts):
        if part.lower() == "countries" and index + 1 < len(path_parts):
            maybe_code = path_parts[index + 1]
            if len(maybe_code) == 2:
                return maybe_code.upper()

    return ""

这里先定义一个 AdminRecord，它就是入库前的数据形状。字段能在这一层确定，就不要拖到 Storage 里再拼凑。Storage 应该只关心存储，不关心网页长什么样。

7.2 列表页解析

继续在 parser.py 中加入列表页解析逻辑：

def find_main_geoname_link(row: Tag) -> tuple[str, str, Optional[int]]:
    """
    从一行结果中找出 GeoNames 详情页链接。

    返回：
    - name
    - absolute_url
    - geoname_id
    """
    candidates = []

    for a in row.find_all("a", href=True):
        href = a.get("href", "")
        match = GEONAME_LINK_RE.search(href)
        if not match:
            continue

        name = clean_text(a.get_text(" ", strip=True))
        if not name:
            continue

        absolute_url = urljoin(BASE_URL, href)
        geoname_id = int(match.group("geoname_id"))
        candidates.append((name, absolute_url, geoname_id))

    if not candidates:
        return "", "", None

    # 通常一行里第一个 geonameId 链接就是主实体链接
    return candidates[0]


def extract_country_from_row(row: Tag, default_country_code: str = "") -> tuple[str, str]:
    """
    尝试从行里提取国家名称和国家代码。

    页面结构可能变化，所以这里采用多种兜底：
    1. 优先找 /countries/ 或带 country= 的链接。
    2. 找不到则返回默认 country code。
    """
    for a in row.find_all("a", href=True):
        href = a.get("href", "")
        text = clean_text(a.get_text(" ", strip=True))

        if not text:
            continue

        if "/countries/" in href or "country=" in href:
            code = extract_country_code_from_url(href) or default_country_code
            return text, code

    return "", default_country_code


def row_looks_like_admin(row_text: str) -> bool:
    normalized = row_text.lower()

    keywords = [
        "administrative division",
        "independent political entity",
        "dependent political entity",
        "country",
    ]

    return any(keyword in normalized for keyword in keywords)


def parse_admin_row(row: Tag, default_country_code: str = "") -> Optional[AdminRecord]:
    row_text = clean_text(row.get_text(" ", strip=True))

    if not row_text:
        return None

    name, source_url, geoname_id = find_main_geoname_link(row)
    if not name or not source_url or not geoname_id:
        return None

    if not row_looks_like_admin(row_text):
        return None

    country, country_code = extract_country_from_row(row, default_country_code)
    admin_level, admin_label = normalize_admin_level(row_text)

    if not admin_level:
        # 有些页面会显示 feature code，却不显示完整英文描述。
        # 这种情况下先保留记录，后面可以靠详情页补字段。
        admin_level = ""
        admin_label = ""

    content_hash = make_hash(
        str(geoname_id),
        name,
        country,
        country_code,
        admin_level,
        admin_label,
        source_url,
    )

    return AdminRecord(
        name=name,
        country=country,
        country_code=country_code,
        admin_level=admin_level,
        admin_label=admin_label,
        geoname_id=geoname_id,
        source_url=source_url,
        content_hash=content_hash,
    )


def parse_list_page(html: str, default_country_code: str = "") -> list[AdminRecord]:
    soup = BeautifulSoup(html, "lxml")

    records: list[AdminRecord] = []

    # 优先解析表格行；如果页面结构变化，只要链接仍在行内，仍有机会解析出来。
    for row in soup.find_all("tr"):
        record = parse_admin_row(row, default_country_code=default_country_code)
        if record:
            records.append(record)

    # 如果没有表格行，退化为从页面整体寻找 geoname 链接。
    # 这种兜底不如表格准确，但能帮助我们发现页面结构变化。
    if not records:
        for a in soup.find_all("a", href=True):
            href = a.get("href", "")
            match = GEONAME_LINK_RE.search(href)
            if not match:
                continue

            name = clean_text(a.get_text(" ", strip=True))
            if not name:
                continue

            source_url = urljoin(BASE_URL, href)
            geoname_id = int(match.group("geoname_id"))
            content_hash = make_hash(str(geoname_id), name, source_url)

            records.append(
                AdminRecord(
                    name=name,
                    country="",
                    country_code=default_country_code,
                    admin_level="",
                    admin_label="",
                    geoname_id=geoname_id,
                    source_url=source_url,
                    content_hash=content_hash,
                )
            )

    return deduplicate_records(records)


def deduplicate_records(records: Iterable[AdminRecord]) -> list[AdminRecord]:
    seen: set[int] = set()
    unique: list[AdminRecord] = []

    for record in records:
        if record.geoname_id in seen:
            continue
        seen.add(record.geoname_id)
        unique.append(record)

    return unique

这一段解析逻辑不是假设页面永远不变，而是假设页面大概率保留 GeoNames 详情链接。只要链接格式仍然是带 geonameId 的详情页，我们就能把最关键的 ID 抽出来。

7.3 下一页链接解析

目录页通常会有 “next” 链接。不要自己猜分页参数，最好直接解析页面里的下一页链接。

继续添加：

def parse_next_page_url(html: str) -> str:
    soup = BeautifulSoup(html, "lxml")

    for a in soup.find_all("a", href=True):
        text = clean_text(a.get_text(" ", strip=True)).lower()
        href = a.get("href", "")

        if text in {"next", "next >", ">", "下一页"}:
            return urljoin(BASE_URL, href)

    # 兜底：有些页面可能只显示 next >
    for a in soup.find_all("a", href=True):
        text = clean_text(a.get_text(" ", strip=True)).lower()
        if "next" in text:
            return urljoin(BASE_URL, a["href"])

    return ""

这里有一个经验：分页不要过度依赖自己拼参数。很多站点会在分页 URL 中加入内部参数，自己拼很容易漏掉。页面给你的下一页链接，通常比你猜出来的更可靠。

7.4 详情页解析

列表页往往已经能拿到大部分字段，但详情页可以用来补字段。比如列表页只有名称和链接，详情页里可能出现更完整的国家、层级、别名、坐标等信息。

本文只抽取本次需要的字段。

继续添加：

def parse_detail_page(html: str, source_url: str) -> dict:
    """
    详情页字段补全。

    返回 dict，而不是 AdminRecord，是因为详情页可能只补充部分字段。
    """
    soup = BeautifulSoup(html, "lxml")
    page_text = clean_text(soup.get_text(" ", strip=True))

    geoname_id = extract_geoname_id(source_url)
    if not geoname_id:
        match = re.search(r"GeoName\s*ID\s*[:#]?\s*(\d+)", page_text, re.I)
        geoname_id = int(match.group(1)) if match else None

    title = ""
    for selector in ["h1", "h2", "title"]:
        node = soup.select_one(selector)
        if node:
            title = clean_text(node.get_text(" ", strip=True))
            break

    name = title
    if " - " in name:
        name = name.split(" - ", 1)[0].strip()

    country = ""
    country_code = ""

    for a in soup.find_all("a", href=True):
        href = a.get("href", "")
        text = clean_text(a.get_text(" ", strip=True))

        if not text:
            continue

        if "/countries/" in href or "country=" in href:
            country = text
            country_code = extract_country_code_from_url(href)
            break

    admin_level, admin_label = normalize_admin_level(page_text)

    return {
        "geoname_id": geoname_id,
        "name": name,
        "country": country,
        "country_code": country_code,
        "admin_level": admin_level,
        "admin_label": admin_label,
    }


def merge_detail(record: AdminRecord, detail: dict) -> AdminRecord:
    """
    用详情页字段补全列表页字段。
    列表页已有值时优先保留列表页；详情页只补空字段。
    """
    name = record.name or detail.get("name", "")
    country = record.country or detail.get("country", "")
    country_code = record.country_code or detail.get("country_code", "")
    admin_level = record.admin_level or detail.get("admin_level", "")
    admin_label = record.admin_label or detail.get("admin_label", "")

    content_hash = make_hash(
        str(record.geoname_id),
        name,
        country,
        country_code,
        admin_level,
        admin_label,
        record.source_url,
    )

    return AdminRecord(
        name=name,
        country=country,
        country_code=country_code,
        admin_level=admin_level,
        admin_label=admin_label,
        geoname_id=record.geoname_id,
        source_url=record.source_url,
        content_hash=content_hash,
    )

详情页解析一定要做成“补全”，不要做成“覆盖”。列表页里看到的字段有时更接近搜索结果语境，详情页里字段可能更复杂。直接覆盖会引入不必要的不稳定。

7.5 缺失字段怎么办

我的处理原则是：

• geoname_id 和 source_url 缺失：丢弃，因为无法去重和追踪。
• name 缺失：丢弃，因为没有业务意义。
• country 缺失：保留，后续可用详情页或国家参数补充。
• admin_level 缺失：保留，但打日志，后续人工检查。
• admin_label 缺失：保留，作为非关键字段。

对于爬虫来说，容错不是“什么都吞掉”，而是区分关键字段和非关键字段。关键字段缺失会污染库，非关键字段缺失可以留给后处理。

---

8️⃣ 数据存储与导出（Storage）

本文使用 SQLite 起步。SQLite 很适合这种地理字典类数据：

• 不需要单独部署数据库。
• 单文件可复制、可备份、可随项目携带。
• 支持索引、唯一约束、事务。
• 后续迁移到 MySQL 或 PostgreSQL 也不难。

8.1 字段映射表

字段名	SQLite 类型	示例值	说明
geoname_id	INTEGER	6252001	GeoNames 主键
name	TEXT	United States	行政区名称
country	TEXT	United States	国家名称
country_code	TEXT	US	国家代码
admin_level	TEXT	country / ADM1	归一化行政级别
admin_label	TEXT	first-order administrative division	页面原始级别描述
source_url	TEXT	GeoNames 详情页	数据来源链接
content_hash	TEXT	sha256	内容指纹
created_at	TEXT	2026-06-09T10:30:00	首次入库时间
updated_at	TEXT	2026-06-09T10:30:00	最近更新时间

8.2 SQLite 存储实现

创建 geonames_admin/storage.py：

import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

from .parser import AdminRecord


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


class SQLiteStorage:
    def __init__(self, db_path: Path):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

    @contextmanager
    def connect(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def init_db(self) -> None:
        with self.connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS geonames_admin (
                    geoname_id INTEGER PRIMARY KEY,
                    name TEXT NOT NULL,
                    country TEXT,
                    country_code TEXT,
                    admin_level TEXT,
                    admin_label TEXT,
                    source_url TEXT NOT NULL UNIQUE,
                    content_hash TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );
                """
            )

            conn.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_geonames_admin_country_code
                ON geonames_admin(country_code);
                """
            )

            conn.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_geonames_admin_admin_level
                ON geonames_admin(admin_level);
                """
            )

            conn.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_geonames_admin_name
                ON geonames_admin(name);
                """
            )

    def upsert_many(self, records: Iterable[AdminRecord]) -> int:
        now = utc_now_iso()
        rows = [
            (
                record.geoname_id,
                record.name,
                record.country,
                record.country_code,
                record.admin_level,
                record.admin_label,
                record.source_url,
                record.content_hash,
                now,
                now,
            )
            for record in records
        ]

        if not rows:
            return 0

        with self.connect() as conn:
            conn.executemany(
                """
                INSERT INTO geonames_admin (
                    geoname_id,
                    name,
                    country,
                    country_code,
                    admin_level,
                    admin_label,
                    source_url,
                    content_hash,
                    created_at,
                    updated_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(geoname_id) DO UPDATE SET
                    name = excluded.name,
                    country = excluded.country,
                    country_code = excluded.country_code,
                    admin_level = excluded.admin_level,
                    admin_label = excluded.admin_label,
                    source_url = excluded.source_url,
                    content_hash = excluded.content_hash,
                    updated_at = excluded.updated_at
                WHERE geonames_admin.content_hash != excluded.content_hash;
                """,
                rows,
            )

        return len(rows)

    def count(self) -> int:
        with self.connect() as conn:
            row = conn.execute("SELECT COUNT(*) AS total FROM geonames_admin").fetchone()
            return int(row["total"])

    def sample(self, limit: int = 5) -> list[dict]:
        with self.connect() as conn:
            rows = conn.execute(
                """
                SELECT
                    geoname_id,
                    name,
                    country,
                    country_code,
                    admin_level,
                    admin_label,
                    source_url
                FROM geonames_admin
                ORDER BY geoname_id
                LIMIT ?;
                """,
                (limit,),
            ).fetchall()

            return [dict(row) for row in rows]

这里用了两个去重策略：

1. geoname_id 作为主键，防止同一实体重复入库。
2. source_url 作为唯一字段，防止同一链接因为 ID 解析异常造成重复。

content_hash 用来判断内容是否变化。如果内容没有变化，updated_at 不会被无意义刷新。这对后续做增量同步很有用。

8.3 CSV 导出

创建 geonames_admin/exporter.py：

import csv
import sqlite3
from pathlib import Path


def export_sqlite_to_csv(db_path: Path, csv_path: Path) -> int:
    csv_path = Path(csv_path)
    csv_path.parent.mkdir(parents=True, exist_ok=True)

    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row

    try:
        rows = conn.execute(
            """
            SELECT
                geoname_id,
                name,
                country,
                country_code,
                admin_level,
                admin_label,
                source_url,
                created_at,
                updated_at
            FROM geonames_admin
            ORDER BY country_code, admin_level, name;
            """
        ).fetchall()

        with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
            writer = csv.writer(f)
            writer.writerow(
                [
                    "geoname_id",
                    "name",
                    "country",
                    "country_code",
                    "admin_level",
                    "admin_label",
                    "source_url",
                    "created_at",
                    "updated_at",
                ]
            )

            for row in rows:
                writer.writerow([row[col] for col in row.keys()])

        return len(rows)
    finally:
        conn.close()

这里导出使用 utf-8-sig，主要是为了兼容一些办公软件打开 CSV 时的中文编码识别。纯程序读取时，utf-8 就够了；给人打开时，utf-8-sig 更省心。

---

9️⃣ 运行方式与结果展示

9.1 命令行入口

创建 geonames_admin/runner.py：

import argparse
import logging
from pathlib import Path

from .config import (
    CSV_PATH,
    DB_PATH,
    DEFAULT_COUNTRY,
    DEFAULT_MAX_PAGES,
    MAX_SLEEP_SECONDS,
    MIN_SLEEP_SECONDS,
)
from .exporter import export_sqlite_to_csv
from .fetcher import GeoNamesFetcher
from .parser import parse_detail_page, parse_list_page, parse_next_page_url, merge_detail
from .storage import SQLiteStorage


def setup_logging(verbose: bool = False) -> None:
    level = logging.DEBUG if verbose else logging.INFO

    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    )


def crawl(
    country: str,
    max_pages: int,
    db_path: Path,
    csv_path: Path,
    fetch_detail: bool,
    export_csv: bool,
) -> None:
    fetcher = GeoNamesFetcher(
        min_sleep=MIN_SLEEP_SECONDS,
        max_sleep=MAX_SLEEP_SECONDS,
    )

    storage = SQLiteStorage(db_path)
    storage.init_db()

    current_url = fetcher.build_search_url(
        country=country,
        feature_class="A",
        q="",
    )

    previous_url = ""
    total_parsed = 0
    total_saved = 0

    for page_no in range(1, max_pages + 1):
        logging.info("fetch list page %s/%s: %s", page_no, max_pages, current_url)

        result = fetcher.get(current_url, referer=previous_url or None)
        records = parse_list_page(result.text, default_country_code=country)

        logging.info("parsed %s records from list page", len(records))

        if fetch_detail:
            enriched = []
            for record in records:
                try:
                    fetcher.sleep()
                    detail_result = fetcher.get(record.source_url, referer=result.url)
                    detail = parse_detail_page(detail_result.text, record.source_url)
                    enriched.append(merge_detail(record, detail))
                except Exception as exc:
                    logging.warning(
                        "detail fetch failed, keep list record. geoname_id=%s error=%s",
                        record.geoname_id,
                        exc,
                    )
                    enriched.append(record)
            records = enriched

        saved = storage.upsert_many(records)
        total_parsed += len(records)
        total_saved += saved

        logging.info("saved %s records, db total=%s", saved, storage.count())

        next_url = parse_next_page_url(result.text)
        if not next_url:
            logging.info("no next page found, stop")
            break

        previous_url = result.url
        current_url = next_url
        fetcher.sleep()

    if export_csv:
        exported = export_sqlite_to_csv(db_path, csv_path)
        logging.info("exported %s rows to %s", exported, csv_path)

    logging.info(
        "done. total_parsed=%s total_saved=%s db=%s",
        total_parsed,
        total_saved,
        db_path,
    )

    for item in storage.sample(limit=5):
        logging.info("sample row: %s", item)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Crawl GeoNames administrative directory pages into SQLite."
    )

    parser.add_argument(
        "--country",
        default=DEFAULT_COUNTRY,
        help="ISO country code, e.g. US, CH, JP, DE.",
    )

    parser.add_argument(
        "--max-pages",
        type=int,
        default=DEFAULT_MAX_PAGES,
        help="Maximum list pages to crawl.",
    )

    parser.add_argument(
        "--db",
        default=str(DB_PATH),
        help="SQLite database path.",
    )

    parser.add_argument(
        "--csv",
        default=str(CSV_PATH),
        help="CSV export path.",
    )

    parser.add_argument(
        "--detail",
        action="store_true",
        help="Fetch detail pages to enrich missing fields.",
    )

    parser.add_argument(
        "--no-export",
        action="store_true",
        help="Do not export CSV after crawling.",
    )

    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Enable debug logging.",
    )

    return parser.parse_args()


def main() -> None:
    args = parse_args()
    setup_logging(verbose=args.verbose)

    crawl(
        country=args.country.upper(),
        max_pages=args.max_pages,
        db_path=Path(args.db),
        csv_path=Path(args.csv),
        fetch_detail=args.detail,
        export_csv=not args.no_export,
    )


if __name__ == "__main__":
    main()

9.2 启动命令

在项目根目录执行：

python -m geonames_admin.runner --country US --max-pages 2

如果希望请求详情页补充字段：

python -m geonames_admin.runner --country US --max-pages 2 --detail

如果你只是想入库，不想导出 CSV：

python -m geonames_admin.runner --country US --max-pages 2 --no-export

如果需要看更详细日志：

python -m geonames_admin.runner --country US --max-pages 2 --verbose

9.3 输出位置

默认输出：

data/geonames_admin.sqlite
data/geonames_admin.csv

SQLite 表名：

geonames_admin

9.4 查询 SQLite

可以用 Python 查询：

import sqlite3

conn = sqlite3.connect("data/geonames_admin.sqlite")
conn.row_factory = sqlite3.Row

rows = conn.execute(
    """
    SELECT geoname_id, name, country, country_code, admin_level, source_url
    FROM geonames_admin
    ORDER BY country_code, admin_level, name
    LIMIT 10;
    """
).fetchall()

for row in rows:
    print(dict(row))

conn.close()

也可以用命令行：

sqlite3 data/geonames_admin.sqlite

进入 SQLite 后执行：

.headers on
.mode column

SELECT geoname_id, name, country_code, admin_level
FROM geonames_admin
LIMIT 10;

9.5 示例结果

下面是结果格式示例，实际数据以你运行时页面返回为准：

geoname_id	name	country	country_code	admin_level	source_url
6252001	United States	United States	US	country	GeoNames 详情页
5332921	California	United States	US	ADM1	GeoNames 详情页
5128638	New York	United States	US	ADM1	GeoNames 详情页
4736286	Texas	United States	US	ADM1	GeoNames 详情页
4142224	Delaware	United States	US	ADM1	GeoNames 详情页

如果你用的是 --country CH、--country DE 或其他国家代码，样例会变成对应国家下的行政区数据。

---

🔟 常见问题与排错

10.1 遇到 403 怎么办

403 Forbidden 表示服务器拒绝当前请求。常见原因包括：

• User-Agent 过于异常。
• 请求频率过高。
• 请求路径不适合自动化访问。
• 站点不欢迎当前类型的自动化请求。

处理方式：

1. 先停止任务，不要继续重试。
2. 检查 robots 和站点说明。
3. 降低频率，确认是否误伤。
4. 设置清晰、真实的 User-Agent。
5. 如果仍然不允许访问，就不要继续抓取。

不要把 403 当成“技术挑战”。很多时候，停下来是更专业的选择。

10.2 遇到 429 怎么办

429 Too Many Requests 明确表示请求太多。处理方式：

• 增大 MIN_SLEEP_SECONDS 和 MAX_SLEEP_SECONDS。
• 减少 max_pages。
• 不抓详情页，先只抓列表页。
• 改成定时慢速同步。
• 生产环境可以记录游标，第二天继续跑。

比如把配置改成：

MIN_SLEEP_SECONDS = 5.0
MAX_SLEEP_SECONDS = 10.0

不要用代理池硬顶。这个项目是地理字典数据，不值得用激进方式采集。

10.3 HTML 抓到空壳怎么办

如果 requests 抓到的 HTML 里没有你在浏览器看到的数据，通常有三种情况：

1. 页面由 JavaScript 动态渲染。
2. 数据来自隐藏接口。
3. 服务器按请求头返回不同内容。

排查步骤：

• 在浏览器中右键查看网页源代码，看源代码里有没有目标字段。
• 打开开发者工具 Network，看是否有 JSON 或 XML 接口。
• 打印 response.text[:1000]，确认是否拿到了错误页。
• 检查状态码、编码和跳转历史。

如果确实是动态渲染，再考虑 Playwright。不要一开始就上浏览器，否则项目会变重。

10.4 解析报错怎么办

解析报错通常是选择器写得太死。本文代码用的是相对宽松的方式：先找表格行，再找符合 GeoNames 详情页格式的链接。

如果页面改版，可以先保存一份 HTML：

from pathlib import Path

Path("debug_page.html").write_text(result.text, encoding="utf-8")

然后本地打开 debug_page.html，重新观察结构。调 Parser 时，最好不要每改一次代码就请求一次线上页面。把 HTML 保存下来，本地反复调，会更友好。

10.5 编码或乱码如何处理

本文 Fetcher 中有这段处理：

if not response.encoding or response.encoding.lower() == "iso-8859-1":
    response.encoding = response.apparent_encoding or "utf-8"

另外 CSV 使用：

encoding="utf-8-sig"

一般可以解决大部分中文或特殊字符打开乱码的问题。

如果你后续要处理多语言地名，建议全链路统一 UTF-8：

• SQLite 文本默认可存 UTF-8。
• Python 源码使用 UTF-8。
• CSV 导出使用 UTF-8 或 UTF-8-SIG。
• 日志文件也指定 UTF-8。

10.6 为什么没有抓全量

全量 GeoNames 数据量很大，目录页分页抓取并不是最合适的方式。全量同步更适合下载官方 dump，然后按 feature_class='A' 过滤行政区。本文目标是讲网页目录页解析、字段入库和爬虫工程结构，所以选择了页面采集这个切口。

实际生产中，我会这样选：

• 小范围、低频、教学：目录页爬取。
• 批量同步、稳定生产：官方 dump。
• 按条件实时查询：官方 Webservice。
• 前端页面必须渲染后才有数据：Playwright。

---

1️⃣1️⃣ 进阶优化

11.1 并发

本文不建议上来就并发。如果你确认目标站点允许低强度自动访问，并且确实需要提升速度，可以考虑线程池。

一个温和的线程池示例：

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_one_detail(fetcher, record):
    result = fetcher.get(record.source_url)
    return record, result.text

def fetch_details_gently(fetcher, records, max_workers=2):
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(fetch_one_detail, fetcher, r) for r in records]

        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as exc:
                print("detail failed:", exc)

    return results

注意这里 max_workers=2 已经够了。对这种数据源来说，线程池的目的不是压榨服务器，而是避免单个慢请求拖住整个任务。

11.2 asyncio

如果你熟悉异步，可以换成 httpx.AsyncClient。但我不建议初版就写异步。异步代码的调试成本更高，而且很容易不小心把并发开大。

对于本文场景，异步不是必要项。先把字段抽准、入库稳定、日志完善，比追求速度更重要。

11.3 断点续跑

断点续跑有两种常见方式。

第一种是基于 URL 游标：

CREATE TABLE IF NOT EXISTS crawl_state (
    task_name TEXT PRIMARY KEY,
    last_url TEXT,
    updated_at TEXT NOT NULL
);

每抓完一页就保存下一页 URL。任务中断后，从 last_url 继续。

第二种是基于已抓集合：

CREATE TABLE IF NOT EXISTS fetched_url (
    url TEXT PRIMARY KEY,
    status_code INTEGER,
    fetched_at TEXT NOT NULL
);

每次请求前先判断 URL 是否已经抓过。这个方案更适合详情页很多的场景。

对于本文项目，geoname_id 本身就是天然去重键。即使任务中断后从第一页重新跑，也不会插入重复记录，只是会多请求一些页面。数据量不大时，这种方式已经够用。

11.4 日志与监控

生产环境至少记录这些指标：

• 请求总数。
• 成功请求数。
• 失败请求数。
• 解析记录数。
• 入库记录数。
• 详情页失败数。
• 403 / 429 次数。
• 单页解析耗时。

可以先用日志实现：

logging.info(
    "metrics page=%s parsed=%s saved=%s db_total=%s",
    page_no,
    len(records),
    saved,
    storage.count(),
)

后面再接入 Prometheus、Grafana 或任务平台。

11.5 定时任务

Linux 上可以用 cron 做一个低频更新任务：

crontab -e

加入：

0 3 * * 1 cd /path/to/geonames_admin_spider && /path/to/.venv/bin/python -m geonames_admin.runner --country US --max-pages 3 >> logs/cron.log 2>&1

这表示每周一凌晨 3 点跑一次。对于地理字典库来说，周级更新通常已经足够。

如果任务越来越复杂，可以考虑 Airflow、Prefect 或 Dagster。我的建议是：先不要过早上平台。一个脚本能稳定解决的问题，就先用脚本。

11.6 使用官方 dump 做生产级补充

如果你后续想从页面采集升级为本地全量库，可以下载 GeoNames 的国家文件或全量文件，然后按 feature class 过滤。

伪代码大概是：

import csv
import zipfile
from pathlib import Path

GEONAME_COLUMNS = [
    "geonameid",
    "name",
    "asciiname",
    "alternatenames",
    "latitude",
    "longitude",
    "feature_class",
    "feature_code",
    "country_code",
    "cc2",
    "admin1_code",
    "admin2_code",
    "admin3_code",
    "admin4_code",
    "population",
    "elevation",
    "dem",
    "timezone",
    "modification_date",
]

def iter_admin_rows(zip_path: Path):
    with zipfile.ZipFile(zip_path) as zf:
        txt_names = [name for name in zf.namelist() if name.endswith(".txt")]
        if not txt_names:
            return

        with zf.open(txt_names[0], "r") as raw:
            text = (line.decode("utf-8").rstrip("\n") for line in raw)

            reader = csv.DictReader(
                text,
                fieldnames=GEONAME_COLUMNS,
                delimiter="\t",
            )

            for row in reader:
                if row["feature_class"] == "A":
                    yield row

这个方向更适合大规模、长期维护的地理库。页面爬取适合补充、验证和教学；dump 文件适合全量同步。

1️⃣2️⃣ 总结与延伸阅读

这篇文章完成了一套 GeoNames 行政区目录页爬虫，从工程角度拆成了请求层、解析层、存储层和导出层。最终我们可以把名称、国家、行政级别、GeoName ID、链接等字段写入 SQLite，并导出 CSV 供后续分析使用。

复盘一下完成的事情：

1. 明确了目标字段和数据用途。
2. 说明了 robots、频率控制、敏感信息和访问限制等合规注意事项。
3. 选择了适合静态页面的技术栈：requests + BeautifulSoup + lxml。
4. 实现了带 headers、timeout、重试、退避和 robots 检查的 Fetcher。
5. 实现了列表页、下一页和详情页解析。
6. 实现了 SQLite 建表、唯一约束、内容 hash 和 upsert。
7. 提供了命令行运行方式和 CSV 导出方式。
8. 补充了 403、429、空壳 HTML、解析报错和乱码处理思路。

下一步可以继续做这些扩展：

• 用 Scrapy 重写，获得更完整的调度、去重、日志和中间件能力。
• 用官方 dump 文件构建全量 GeoNames 行政区库。
• 加入国家表、层级关系表，构建行政区树。
• 将 SQLite 迁移到 PostgreSQL，并用 PostGIS 扩展空间查询。
• 接入定时任务，每周或每天做增量更新。
• 给地理字典库封装一个 FastAPI 查询服务，供内部系统使用。

最后说一句实际感受：爬虫工程最难的地方往往不是“把页面抓下来”，而是让数据长期稳定、可解释、可回溯。GeoNames 这种地理数据源很适合练这个基本功。不要急着写并发，也不要急着堆框架；先把字段、去重、日志和合规做扎实，后面的扩展会自然很多。

🌟 文末

好啦～以上就是本期的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！

小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～
三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥

✅ 专栏持续更新中｜建议收藏 + 订阅

墙裂推荐订阅专栏 👉 《Python爬虫实战》，本专栏秉承着以“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一期内容都做到：

✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）

📣 想系统提升的小伙伴：强烈建议先订阅专栏 《Python爬虫实战》，再按目录大纲顺序学习，效率十倍上升～

✅ 互动征集

想让我把【某站点/某反爬/某验证码/某分布式方案】等写成某期实战？

评论区留言告诉我你的需求，我会优先安排实现(更新)哒~

---

⭐️ 若喜欢我，就请关注我叭～（更新不迷路）
⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力）
⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）

---

✅ 免责声明

本文爬虫思路、相关技术和代码仅用于学习参考，对阅读本文后的进行爬虫行为的用户本作者不承担任何法律责任。

使用或者参考本项目即表示您已阅读并同意以下条款：

• 合法使用： 不得将本项目用于任何违法、违规或侵犯他人权益的行为，包括但不限于网络攻击、诈骗、绕过身份验证、未经授权的数据抓取等。
• 风险自负： 任何因使用本项目而产生的法律责任、技术风险或经济损失，由使用者自行承担，项目作者不承担任何形式的责任。
• 禁止滥用： 不得将本项目用于违法牟利、黑产活动或其他不当商业用途。
• 使用或者参考本项目即视为同意上述条款,即 “谁使用，谁负责” 。如不同意，请立即停止使用并删除本项目。！！！