简介：这份资源面向城市经济、社会学与环境科学等领域的研究者及数据分析开发者，提供《中国城市统计年鉴》1985至2024年全套数据的配套可运行源码，帮助解决年鉴数据获取、整理与二次开发的问题。压缩包共3个文件，以inscode工程配置、html页面与gitignore忽略规则为主，整体约5KB，属于轻量级代码包，便于快速部署与二次开发。目前已有183人学习下载，适合需要将年鉴数据接入分析流程或搭建可视化页面的用户参考。资源覆盖地级市与县级市的人口、资源环境、经济发展、科技创新、人民生活、公共服务等指标，并区分原版、截面与面板三类数据形态；源码可辅助完成数据加载、结构解析与结果展示，同时提示1997年起两类城市统计制度差异带来的指标不可比问题，为后续建模与比较研究提供排错思路与目录组织参考。
1. 年鉴数据更新这件事，卡住人的从来不是爬虫
每年《中国城市统计年鉴》出新版，后台总有人问同一类问题：去年的脚本今年跑不通了，字段对不上，城市名多了几个生僻字，Excel 打开乱码。表面看是爬虫问题，实际是数据工程问题——年鉴的表格结构年年在微调，指标口径偶尔换名字，城市行政区划还会撤县设区。你写一个requests.get加pandas.read_html能跑通一年，第二年大概率翻车。这篇笔记面向的是需要把年鉴数据做成可复用、可增量更新管线的从业者，不是教你怎么发一个 HTTP 请求。我会把「可运行源码」拆成四块：目录结构怎么定、解析层怎么写、增量更新怎么判断、脏数据怎么兜底。新手能照着把最小闭环跑起来，熟手能看到字段映射和校验的边界在哪。
2. 先定数据模型：年鉴表格为什么不能直接丢进 DataFrame
2.1 年鉴的三种表格形态与对应的解析策略
《中国城市统计年鉴》的正文表格大致分三类，混在一起处理是第一个大坑。
第一类是横向指标表：第一列是城市名，后面几十列是指标，表头有两行（第一行是大类如「人口」「经济」，第二行是具体指标）。这种表用pandas.read_excel(header=[0,1])能读，但合并单元格会导致第一行出现大量NaN，需要ffill补全。
第二类是纵向指标表：第一列是指标名，后面若干列是年份或城市。这种表转置后才能和第一类对齐。
第三类是分地区汇总表：按省分组，每组一个小计行。小计行如果不剔除，做城市级聚合时会翻倍。
我一般会先写一个探测函数，把每个 sheet 的前 5 行打印出来，人工确认形态，再决定用哪条解析分支。不要试图写一个万能解析器，年鉴的表格形态每年都在变，万能解析器最后会变成一堆if-else的屎山。
import pandas as pd def probe_sheet(path, sheet_name, nrows=5): """探测 sheet 形态：打印前 nrows 行，判断是横向还是纵向表""" df = pd.read_excel(path, sheet_name=sheet_name, header=None, nrows=nrows) print(f"=== sheet: {sheet_name} ===") print(df.to_string()) # 判断依据：第一列是否全为城市名（横向表）或指标名（纵向表） first_col = df.iloc[:, 0].astype(str).tolist() print("first_col sample:", first_col[:3]) return df这段代码的作用是人工辅助判断，不是自动解析。header=None保证原始表头不被 pandas 吃掉，nrows=5控制输出量。跑完之后你看第一列的内容，如果是「北京」「天津」这种，就是横向表；如果是「年末总人口」「GDP」这种，就是纵向表。参数nrows建议设 5 到 8，太少看不出合并单元格结构，太多刷屏。
2.2 统一 schema：把宽表转成长表
不管原始表是横向还是纵向，最终都要落成一张长表：city, year, indicator, value, unit, source_sheet。这样做的好处是增量更新时只需要比对(city, year, indicator)三元组，不用管表格形态怎么变。
def melt_wide_table(df, year, sheet_name): """把横向指标表转成长表 df: 已读取的宽表，第一列为城市名 year: 数据年份 """ df = df.rename(columns={df.columns[0]: "city"}) # 剔除小计行：城市名包含"合计""全省""总计"的 mask = ~df["city"].astype(str).str.contains("合计|全省|总计|全国", na=False) df = df[mask] long_df = df.melt(id_vars=["city"], var_name="indicator", value_name="value") long_df["year"] = year long_df["source_sheet"] = sheet_name # 清洗 value：去掉逗号、空格、破折号 long_df["value"] = ( long_df["value"].astype(str) .str.replace(",", "", regex=False) .str.replace("—", "", regex=False) .str.strip() ) long_df["value"] = pd.to_numeric(long_df["value"], errors="coerce") return long_df.dropna(subset=["value"])逻辑说明：melt是核心，把宽表压成长表。mask那行是血泪经验——年鉴里每个省下面都有「全省合计」行，不剔除的话你做城市均值会被拉高。value清洗里errors="coerce"把无法转数字的变成NaN再丢掉，比直接报错好，因为年鉴里确实有「—」表示缺失。参数id_vars只保留城市列，其余全部 melt，这样新增指标不用改代码。
2.3 城市名标准化：别等到 join 的时候才发现对不上
城市名是最大的玄学。年鉴里可能写「北京市」，你的行政区划表里写「北京」；可能写「襄樊市」，但行政区划已经改成「襄阳市」。我一般维护一张别名映射表，用 CSV 存，格式是raw_name, std_name, note。
def normalize_city(name, alias_map): """城市名标准化：去后缀、查别名表""" name = str(name).strip() # 去掉"市""地区""自治州"等后缀做初步匹配 for suffix in ["市", "地区", "自治州", "盟"]: if name.endswith(suffix): name = name[: -len(suffix)] break return alias_map.get(name, name) # 别名表加载 alias_df = pd.read_csv("city_alias.csv") alias_map = dict(zip(alias_df["raw_name"], alias_df["std_name"]))这段的关键是先做后缀剥离再查表，因为年鉴有时写全称有时写简称。alias_map用 dict 查是 O(1)，几万行数据无压力。注意别名表要人工维护，不要试图用模糊匹配自动对齐，「襄樊」和「襄阳」编辑距离很远，但「南京」和「南充」编辑距离很近，自动匹配必翻车。
3. 增量更新：怎么判断哪些行是新的、哪些是改过的
3.1 用主键做 diff，而不是整表覆盖
增量更新的核心是主键比对。主键是(city, year, indicator)。新数据读进来之后，和库里已有的数据做 outer join，分三种情况：主键存在且值相同 → 跳过；主键存在但值不同 → 更新；主键不存在 → 插入。
def diff_update(old_df, new_df, key_cols=["city", "year", "indicator"]): """增量 diff：返回需要 insert 和 update 的行""" merged = new_df.merge( old_df, on=key_cols, how="left", suffixes=("", "_old") ) # 新增行：old 的 value 为 NaN to_insert = merged[merged["value_old"].isna()].drop(columns=["value_old"]) # 更新行：old 有值但和新值不同 changed = merged[ merged["value_old"].notna() & (merged["value"] != merged["value_old"]) ] to_update = changed[key_cols + ["value"]] return to_insert, to_update逻辑说明：merge的how="left"保证新数据所有行都保留。value_old为NaN说明这条主键在旧库里没有，是新增。value != value_old说明值变了，需要更新。注意浮点数比较要用np.isclose而不是!=，否则 0.1+0.2 这种精度问题会让你误判大量更新。
import numpy as np # 浮点安全比较 changed = merged[ merged["value_old"].notna() & ~np.isclose(merged["value"], merged["value_old"], rtol=1e-6, equal_nan=True) ]参数rtol=1e-6是相对容差，年鉴数据一般保留两位小数，这个容差足够。equal_nan=True处理两边都是 NaN 的情况，避免把缺失值误判为变更。
3.2 版本戳与回滚：给每次更新留后悔药
增量更新最怕的是更新错了没法回滚。我一般会在表里加两列：batch_id和updated_at。每次跑更新生成一个batch_id（用时间戳），更新时不是覆盖，而是插入新版本行，查询时取最新batch_id。
import time def apply_update(conn, to_insert, to_update, batch_id=None): """把 diff 结果写入数据库，保留历史版本""" if batch_id is None: batch_id = str(int(time.time())) to_insert = to_insert.copy() to_insert["batch_id"] = batch_id to_insert["updated_at"] = pd.Timestamp.now() to_insert.to_sql("city_yearbook", conn, if_exists="append", index=False) # 更新：先标记旧版本失效，再插入新版本 for _, row in to_update.iterrows(): conn.execute( "UPDATE city_yearbook SET is_current=0 WHERE city=? AND year=? AND indicator=?", (row["city"], row["year"], row["indicator"]), ) to_update = to_update.copy() to_update["batch_id"] = batch_id to_update["updated_at"] = pd.Timestamp.now() to_update["is_current"] = 1 to_update.to_sql("city_yearbook", conn, if_exists="append", index=False)逻辑说明：is_current字段是查询时的过滤条件，SELECT * FROM city_yearbook WHERE is_current=1拿到最新数据。回滚就是把某个batch_id的is_current置 0，把上一个batch_id置 1。batch_id用时间戳保证单调递增，方便排序。注意to_sql的if_exists="append"，不要用replace，否则历史版本全丢。
3.3 更新日志：出问题时第一个要看的东西
每次更新要落一份日志，记录：batch_id, 新增行数, 更新行数, 跳过行数, 异常行数, 耗时。异常行要单独存一张表，包含原始行内容和失败原因。
def log_update(conn, batch_id, stats, errors_df): """记录更新日志和异常行""" log_df = pd.DataFrame([{ "batch_id": batch_id, "insert_count": stats["insert"], "update_count": stats["update"], "skip_count": stats["skip"], "error_count": len(errors_df), "elapsed_sec": stats["elapsed"], "created_at": pd.Timestamp.now(), }]) log_df.to_sql("update_log", conn, if_exists="append", index=False) if len(errors_df) > 0: errors_df["batch_id"] = batch_id errors_df.to_sql("update_errors", conn, if_exists="append", index=False)这张日志表在排查时价值极高。比如某次更新error_count突然从 0 跳到 200，你就知道年鉴表格结构变了，需要回去看update_errors里的原始行。elapsed_sec用来监控性能退化，如果从 30 秒涨到 300 秒，说明数据量或解析逻辑出了问题。
4. 避坑与排查：年鉴更新里最容易翻车的 5 个地方
4.1 坑一：Excel 读取时表头行数判断错误
现象：读进来的 DataFrame 第一行是「人口」这种大类，第二行才是具体指标，但代码只用了header=0，导致指标名全是「Unnamed: 1」这种。
原因：年鉴表格普遍是双层表头，read_excel默认只读一行。
解决：先用header=None探测，确认表头行数后再用header=[0,1]读取。如果双层表头有合并单元格，读完后用df.columns = ['_'.join([str(a), str(b)]) for a,b in df.columns]拼成单层。
4.2 坑二：城市名里的全角空格和不可见字符
现象：merge的时候明明看着城市名一样，就是 join 不上。
原因：年鉴 PDF 转 Excel 时经常带入全角空格\u3000或零宽字符\u200b。
解决：标准化函数里加一步name = name.replace("\u3000", "").replace("\u200b", "").strip()。更彻底的做法是用unicodedata.normalize("NFKC", name)做全角转半角。
4.3 坑三：数值列里混入「—」「…」「#」
现象：pd.to_numeric报错，或者整列变成 object 类型。
原因：年鉴里缺失值用「—」表示，有些年份用「…」，还有用「#」表示数据暂缺。
解决：errors="coerce"兜底，同时把这些符号统一替换成空字符串。替换列表要写全：["—", "…", "#", "*", "①"]。替换完再to_numeric，剩下的 NaN 就是真正的缺失。
4.4 坑四：增量更新时浮点精度导致全量误判
现象：明明数据没变，to_update却有几千行。
原因：旧数据存的是float32，新数据是float64，比较时0.1 != 0.100000001。
解决：比较前统一astype(float)，并用np.isclose代替!=。如果数据量大，可以先把两边都round(4)再比较，年鉴数据精度到小数点后两位足够。
4.5 坑五：并发写入导致主键冲突
现象：两个更新任务同时跑，数据库里出现重复主键。
原因：to_sql的append不做主键检查。
解决：在数据库层给(city, year, indicator, batch_id)建唯一索引，写入时用INSERT OR IGNORE（SQLite）或ON CONFLICT DO NOTHING（PostgreSQL）。更简单的做法是加文件锁，保证同一时间只有一个更新任务在跑。
import fcntl def acquire_lock(lock_path): """文件锁：防止并发更新""" f = open(lock_path, "w") try: fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) return f except BlockingIOError: raise RuntimeError("另一个更新任务正在运行")fcntl.flock是 Linux 下的文件锁，LOCK_NB表示非阻塞，拿不到锁直接报错而不是傻等。Windows 下用msvcrt.locking，或者干脆用数据库的行锁。
5. 进阶技巧：把年鉴更新做成可调度的管线
5.1 用配置文件驱动解析，而不是硬编码
年鉴每年微调，硬编码 sheet 名和列名意味着每年改代码。我一般把解析规则抽成 YAML：
# config/yearbook_2024.yaml sheets: - name: "人口" type: "wide" header_rows: [0, 1] city_col: 0 skip_patterns: ["合计", "全省", "总计"] - name: "经济" type: "wide" header_rows: [0, 1] city_col: 0 skip_patterns: ["合计", "全省"]解析函数读 YAML 决定用哪条分支。新增年份只需要加一个 YAML 文件，代码不动。header_rows支持列表，因为有些 sheet 是单层表头有些是双层。skip_patterns是正则列表，匹配到的行直接剔除。
5.2 数据校验：用断言把脏数据挡在入库前
更新前跑一组校验，不通过就不入库。校验规则我一般写这几条：
| 校验项 | 规则 | 失败处理 |
|---|---|---|
| 城市名覆盖率 | 标准化后能匹配到行政区划表的比例 > 95% | 告警，人工确认别名表 |
| 数值范围 | GDP 在 0 到 1e6 之间 | 标记异常行，不入库 |
| 主键唯一性 | (city, year, indicator)无重复 | 去重后入库 |
| 年份连续性 | 新增年份 = 最大年份 + 1 | 告警，确认是否补录旧年 |
def validate(df, city_ref, year_range=(2000, 2030)): """入库前校验，返回 (通过的行, 异常的行)""" errors = [] # 城市名覆盖率 matched = df["city"].isin(city_ref["std_name"]).mean() if matched < 0.95: errors.append(f"城市名覆盖率仅 {matched:.2%}") # 数值范围 bad_value = df[(df["indicator"].str.contains("GDP")) & (df["value"] > 1e6)] if len(bad_value) > 0: errors.append(f"GDP 异常行 {len(bad_value)} 条") # 年份范围 bad_year = df[(df["year"] < year_range[0]) | (df["year"] > year_range[1])] if len(bad_year) > 0: errors.append(f"年份越界 {len(bad_year)} 条") return errors校验不通过时不要直接抛异常退出，而是把异常行写进update_errors表，正常行继续入库。这样一次更新不会因为几条脏数据全废。matched < 0.95这个阈值是经验值，低于 95% 说明别名表该更新了。
5.3 调度与监控：让更新自己跑，你只看告警
把更新脚本包成一个 CLI，用argparse接收年份和配置文件路径，然后挂到 cron 或调度平台上。
# 每年 3 月 1 日凌晨 2 点跑更新 0 2 1 3 * /usr/bin/python3 /opt/yearbook/update.py --year 2024 --config /opt/yearbook/config/yearbook_2024.yaml >> /var/log/yearbook.log 2>&1监控看三个指标：update_log里的error_count、elapsed_sec、以及城市名覆盖率。error_count连续两次大于 0 就发告警；elapsed_sec超过历史均值 3 倍发告警；覆盖率低于 95% 发告警。告警渠道用邮件或 webhook 都行，关键是别让更新静默失败。
我自己的习惯是每次年鉴更新后，手动抽 3 个城市 5 个指标，和原始 Excel 对一遍。自动化校验能挡住 90% 的问题，剩下 10% 是口径变化——比如「固定资产投资」某年改成了「固定资产投资额」，这种只能靠人工发现。年鉴数据更新这件事，代码只占一半，另一半是对数据本身的理解。希望帮到你。
本文还有配套的精品资源，点击获取
