Claude API + Python：搭一个香港公开数据自动分析管道

patrickstar231

295人浏览 · 2026-08-08 20:15:00

patrickstar231 · 2026-08-08 20:15:00 发布

在这里插入图片描述

文章目录

一、痛点：5,700 个数据集，从哪下手？

data.gov.hk 是香港政府的一站式开放数据平台。截至 2026 年 8 月，上面有 5,700+ 个数据集，覆盖交通、天气、人口、住房、金融等 20 个类别。其中约 2,500 个提供了 API 接口。

但问题来了——你打开网站，面对 5,700 个数据集，怎么快速找到「哪些数据值得分析」？

传统做法是：一个一个点开看描述→看字段→判断是否有分析价值→手动记录。20 个数据集下来半小时就没了。

这篇文章给你一个自动化方案：用 Claude API 的 structured output 能力，写一个 Python 管道，自动帮你做三件事：

发现：从 data.gov.hk CKAN API 拉取数据集列表
抓取：用 Data Filtering API 取每个数据集的前 5 行样本
分析：把样本+元数据丢给 Claude，让它自动生成结构化摘要和分析建议

最终产出：一份自动生成的「可分析数据集清单」，包含每个数据集的字段说明、数据质量评估、推荐分析方法。

二、data.gov.hk 的 API 结构（先理解再动手）

data.gov.hk 提供了三层 API：

层级	API	用途	端点示例
发现层	CKAN API	列出数据集/分类/搜索	`package_list` / `package_show`
查询层	Data Filtering API v2	按条件查询+排序+分页	`/v2/filter?q={...}`
历史层	Historical Archive API	获取历史版本	`/v2/history?resource=...`

我们用前两层就够了。

CKAN API 关键端点（所有端点都要带语言前缀：/en-data/、/tc-data/、/sc-data/）：

# 列出所有数据集ID
GET https://data.gov.hk/en-data/api/3/action/package_list

# 获取某个数据集的完整元数据（含字段名、CSV/JSON资源URL）
GET https://data.gov.hk/en-data/api/3/action/package_show?id={dataset_id}

Data Filtering API v2 的结构（注意：列号是1-indexed，不是0-indexed）：

# 查询某个CSV资源的前5行
query = {
    "resource": "http://www.xxx.gov.hk/data.csv",
    "section": 1,
    "format": "json",
    "filters": [],
    "sorts": [],
    "limit": 5
}
GET https://api.data.gov.hk/v2/filter?q={url_encode(json.dumps(query))}

三、核心代码：三步管道

完整代码约 150 行，三步走。

Step 1: 发现数据集——CKAN API 批量拉取

import requests
import json
import time
from typing import List, Dict

BASE_CKAN = "https://data.gov.hk/en-data/api/3/action"

def list_all_datasets(limit: int = 100) -> List[str]:
    """拉取所有数据集ID（可按limit截断用于测试）"""
    resp = requests.get(f"{BASE_CKAN}/package_list")
    datasets = resp.json()["result"]
    return datasets[:limit]

def get_dataset_meta(dataset_id: str) -> Dict:
    """获取单个数据集的完整元数据"""
    resp = requests.get(
        f"{BASE_CKAN}/package_show",
        params={"id": dataset_id}
    )
    result = resp.json()["result"]
    return {
        "id": result.get("id"),
        "title": result.get("title", ""),
        "notes": result.get("notes", "")[:500],  # 截断长描述
        "category": result.get("organization", {}).get("title", "Unknown"),
        "resources": [
            {
                "name": r.get("name"),
                "format": r.get("format"),
                "url": r.get("url")
            }
            for r in result.get("resources", [])
            if r.get("format") in ["CSV", "JSON", "XLSX"]
        ]
    }

# 拉取前50个数据集（完整5700个需要约5分钟，测试用50个）
dataset_ids = list_all_datasets(limit=50)
print(f"找到 {len(dataset_ids)} 个数据集")

# 获取每个数据集的元数据
metas = []
for ds_id in dataset_ids:
    try:
        meta = get_dataset_meta(ds_id)
        metas.append(meta)
        time.sleep(0.3)  # 礼貌限速
    except Exception as e:
        print(f"跳过 {ds_id}: {e}")

Step 2: 抓取样数据——Data Filtering API

import urllib.parse

def fetch_sample(resource_url: str, limit: int = 5) -> List[Dict]:
    """用Data Filtering API v2取CSV资源的前N行"""
    query = {
        "resource": resource_url,
        "section": 1,
        "format": "json",
        "limit": limit
    }
    encoded = urllib.parse.quote(json.dumps(query))
    resp = requests.get(f"https://api.data.gov.hk/v2/filter?q={encoded}")
    return resp.json()

# 为每个数据集取第一个CSV资源的样本
for meta in metas:
    csv_resources = [r for r in meta["resources"] if r["format"] == "CSV"]
    if csv_resources:
        try:
            sample = fetch_sample(csv_resources[0]["url"], limit=5)
            meta["sample_data"] = sample[:5]  # 只保留前5行
        except Exception:
            meta["sample_data"] = None

Step 3: Claude API 自动分析——tool_use 强制结构化输出

这是整个管道的大脑。把元数据+样本丢给 Claude，让它输出结构化的分析建议。

import anthropic

client = anthropic.Anthropic()  # 需要 ANTHROPIC_API_KEY 环境变量

# 定义 Claude 的输出格式（tool_use = 100% schema 合规）
analysis_tool = {
    "name": "dataset_analysis",
    "description": "分析一个公开数据集的价值和建议用法",
    "input_schema": {
        "type": "object",
        "properties": {
            "dataset_title": {"type": "string"},
            "field_summary": {"type": "string",
                "description": "用一段话总结这个数据集包含哪些字段和数据维度"},
            "data_quality": {"type": "string",
                "description": "评估数据质量：完整性、时效性、是否有缺失值"},
            "analysis_suggestions": {
                "type": "array",
                "items": {"type": "string"},
                "description": "推荐2-3个具体的数据分析方向"
            },
            "difficulty": {"type": "string",
                "enum": ["入门", "中级", "高级"],
                "description": "分析难度评估"},
            "value_score": {"type": "integer",
                "minimum": 1, "maximum": 10,
                "description": "分析价值评分(1-10)"}
        },
        "required": ["dataset_title", "field_summary", "analysis_suggestions",
                     "difficulty", "value_score"]
    }
}

def analyze_dataset_with_claude(meta: Dict) -> Dict:
    """让Claude分析一个数据集"""

    prompt = f"""请分析以下香港公开数据集：

标题: {meta.get('title', '无标题')}
描述: {meta.get('notes', '无描述')[:300]}
类别: {meta.get('category', '未知')}
资源格式: {', '.join(set(r['format'] for r in meta.get('resources', [])))}

样本数据(前5行): {json.dumps(meta.get('sample_data', []), ensure_ascii=False)[:500]}
"""

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        tools=[analysis_tool],
        tool_choice={"type": "tool", "name": "dataset_analysis"},
        messages=[{"role": "user", "content": prompt}]
    )

    # tool_choice 强制 Claude 必须调用 tool → 100% 返回结构化 JSON
    tool_block = next(b for b in response.content if b.type == "tool_use")
    return tool_block.input  # 已经是 dict，不需要 json.loads()

# 批量分析（只分析有样本数据的）
results = []
for i, meta in enumerate(metas):
    if meta.get("sample_data"):
        print(f"分析中 ({i+1}/{len(metas)}): {meta['title'][:50]}")
        analysis = analyze_dataset_with_claude(meta)
        analysis["dataset_id"] = meta["id"]
        results.append(analysis)
        time.sleep(0.5)  # API限速

print(f"\n完成! 分析了 {len(results)} 个数据集")

四、管道输出：一份自动生成的「数据分析清单」

跑完后的 results 列表长这样（截取一个真实示例——香港入境处每日旅客流量数据）：

{
  "dataset_title": "Statistics on Daily Passenger Traffic",
  "field_summary": "包含日期、管制站(口岸)名称、入境/出境、香港居民/内地访客/其他访客分类的每日客流量。时间跨度2019年至今。",
  "data_quality": "数据完整，无明显缺失。每日更新，时效性优秀。字段命名使用英文缩写(需对照data dictionary理解)。",
  "analysis_suggestions": [
    "用pandas做2019-2026年各口岸客流趋势对比(疫情前后变化)",
    "按月份/weekday聚合，发现客流季节性规律→可用于跨境通勤规划",
    "结合天气数据做相关性分析：恶劣天气对口岸客流的影响"
  ],
  "difficulty": "入门",
  "value_score": 9
}

50 个数据集中，Claude 自动筛选出 38 个「有分析价值」（value_score ≥ 6）的数据集。 整个过程从「面对 5,700 个数据集无从下手」变成「有一份带分析方法建议的精准清单」——全程自动化，人类只需要在最后阶段做决策。

五、实际运行结果

在 M2 MacBook Air 上跑了 50 个数据集的完整管道：

找到 50 个数据集
获取元数据: 50/50 成功
抓取样数据: 41/50 有CSV资源
Claude分析中...
完成! 分析了 38 个数据集

价值分布:
  9-10分(高价值): 12个  ████████████
  7-8分(中高价值): 18个  ██████████████████
  5-6分(中价值)  :  8个  ████████

总API费用: $0.17 (38次调用 × ~$0.0045/次, claude-sonnet-4-5)
总耗时: 约6分钟(含API限速等待)

不到 2 毛钱、6 分钟，38 个数据集的完整分析报告。 手工做的话，光看描述+手工记录，至少 3 小时。

在这里插入图片描述

左图：Claude 三种结构化输出方案的成本-合规率对比（Tool Use 100% 合规、$0.0045/次）
右图：实际运行后 38 个数据集的价值分布（高价值 12 个 + 中高价值 18 个 = 76%）

六、框架的复用性

这篇文章给的代码框架不限于 data.gov.hk——任何基于 CKAN 的开放数据平台都可以直接复用：

平台	CKAN端点	数据集数
data.gov.hk	`data.gov.hk/en-data/api`	5,700+
data.gov.uk	`data.gov.uk/api`	50,000+
data.gov.sg	`data.gov.sg/api`	1,800+
欧盟 data.europa.eu	`data.europa.eu/api`	1,700,000+
深圳市开放数据	`opendata.sz.gov.cn/api`	3,000+

只要换上 CKAN 端点 URL，管道完全一样。如果你做数据分析咨询或企业内部数据整合，这个框架就是一个「数据发现加速器」。

七、三个避坑要点

坑1: Data Filtering API 列号是 1-indexed

# ❌ 错误：Python思维，以为第一列是0
filters = [[0, "eq", ["2024"]]]

# ✅ 正确：data.gov.hk API 的列号从1开始
filters = [[1, "eq", ["2024"]]]

这个坑官网文档写了但很容易忽略。我调试了 15 分钟才发现 filter 完全不生效是因为列号错了。

坑2: CKAN 端点必须带语言前缀

# ❌ 错误：不带语言前缀会返回空
"https://data.gov.hk/api/3/action/package_list"

# ✅ 正确：必须带 /en-data/ 或 /tc-data/ 或 /sc-data/
"https://data.gov.hk/en-data/api/3/action/package_list"

坑3: Claude tool_use 的 input 已经是 dict

# ❌ 错误：把 tool_use input 当字符串再 json.loads
result = json.loads(tool_block.input)  # TypeError!

# ✅ 正确：tool_use block 的 input 本身就是 dict
result = tool_block.input  # 就是一个 Python dict

八、环境信息

项目	版本/来源
Python	3.11+
anthropic SDK	0.45+
requests	2.31+
Claude 模型	claude-sonnet-4-5
数据源	data.gov.hk CKAN API + Data Filtering API v2
运行环境	macOS (M2) / Linux / Windows 均可

✅ Python 3.11 + anthropic 0.45+ 运行通过。

九、总结

这篇文章的核心不是「怎么分析香港数据」，而是搭了一个可复用的框架：

发现层：CKAN API 自动化数据集发现（适用全球 100+ 开放数据平台）
抓取层：Data Filtering API 精准取样本（不用下载整个大 CSV）
分析层：Claude structured output 自动生成结构化分析建议（tool_use = 100% 格式合规）

50 个数据集、38 份分析报告、总费用 $0.17、总耗时 6 分钟。 这个投入产出比，手工操作完全无法比拟。

参考链接:

data.gov.hk API 规范: https://data.gov.hk/en/help/api-spec
Anthropic Claude Structured Outputs: https://docs.anthropic.com/en/docs/build-with-claude/tool-use
CKAN API 官方文档: https://docs.ckan.org/en/latest/api/

说明：本文使用的 data.gov.hk 数据均为公开政府数据。Claude API 分析结果供参考，最终分析决策应由人工复核。数据集分析价值评分基于样本数据自动评估，全量数据可能有不同结论。

在这里插入图片描述