一、问题背景：每天点"导出"要疯掉了
我们FAB的MES系统有个功能，但99%的人只用了一种方式——**手动导出**：
1. 打开MES网页
2. 填入查询条件（日期、工站、产品）
3. 点击"导出CSV"
4. 等5分钟下载
5. 打开Excel看看数据
每天要导10次，每次5分钟，一天花50分钟在"等下载"上。
偶尔忙起来忘了导出，第二天分析发现没数据。
**更难受的是**：数据自动化工具（SPC控制图、异常检测）做出来了，但没人天天喂数据给它们。
**其实MES有API接口**——给程序用的"数据取餐口"。直接调用API，3秒拿到数据。
**学完这一篇，你能做到：**
1. 用Python给API发请求，拿到数据
2. 解析JSON格式的数据
3. 把数据存下来，给其他分析工具用
────────────────────────────────────────
二、技术原理：API就是"程序之间的对话"
2.1 什么是API？
你可以把API理解成**自动售货机**：
- 你投币（发请求）→ 售货机吐出饮料（返回数据）
- 你手里不用有人（不需要登录网页）
- 24小时工作（不用等上班时间）
# 最基础的API调用
import requests
# 发请求（就像按售货机的按钮）
url = "http://mes-api.fab.local/api/lots"
response = requests.get(url)
# 看结果（售货机吐出了什么）
print(response.status_code) # 200 = 成功
print(response.json()) # 返回的数据（JSON格式）
就这两行，用数据了。
2.2 认识HTTP状态码
API调用结果好不好，先看状态码。记住这4个就够了：
| 代码 | 意思 | 怎么办 |
|------|------|--------|
| **200** | 成功 ✅ | 拿数据 |
| **401** | 没权限（Token过期） | 重新登录获取Token |
| **404** | 地址错了 | 检查URL拼写 |
| **500** | MES服务器崩了 | 等一会儿再试 |
2.3 JSON格式长什么样
API返回的数据一般是JSON，长得像Python的字典和列表：
# 假设API返回了这样的数据
data = {
"status": "success",
"count": 3,
"data": [
{"lot_id": "FAB-001", "process": "ETCH",
"thickness": 1250.5, "yield": 96.5},
{"lot_id": "FAB-002", "process": "ETCH",
"thickness": 1248.2, "yield": 95.8},
]
}
# 解析：取data字段
lots = data['data']
print(f"拿到了 {len(lots)} 批数据")
for lot in lots:
print(f" {lot['lot_id']}: 厚度={lot['thickness']}, 良率={lot['yield']}%")
**JSON和Python字典的区别**：基本一样。JSON是字符串，`response.json()` 把它转成Python的字典/列表。
────────────────────────────────────────
三、实战案例：三步写出你的API采集器
3.1 第一步：带Token的请求
大多数MES的API需要认证，最常见的方式是**Bearer Token**——相当于你的工牌：
import requests
# 你的Token（从MES IT同事那里拿）
TOKEN = "eyJhbGciOiJIUzI1NiIs..." # 一串很长的字符串
# 把Token放到请求头里
headers = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
}
# 发请求
url = "http://mes-api.fab.local/api/lots"
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
print(f"成功获取 {len(data.get('data', []))} 条数据")
else:
print(f"请求失败: {response.status_code}")
print(f"错误信息: {response.text}")
**为什么这样写？** `headers` 是HTTP请求的"附加信息"，Bearer Token放这里，服务端读到Token就知道你是谁、有什么权限。`response.text` 是原始响应内容，调试的时候打印出来最快找到问题。
3.2 第二步：传参数筛选数据
API一般支持按日期、工站等条件筛选，用 `params` 参数：
# 筛选特定日期和工序的数据
params = {
"date": "2026-01-15", # 日期范围
"process": "ETCH", # 工站
"limit": 100 # 每页最多返回100条
}
response = requests.get(url, headers=headers, params=params)
data = response.json()
lots = data.get('data', [])
total = data.get('total', 0)
print(f"查询条件: 日期={params['date']}, 工站={params['process']}")
print(f"符合条件的数据: {total} 条, 本次返回: {len(lots)} 条")
**为什么这样写？** `params` 会自动拼接到URL后面变成 `/api/lots?date=2026-01-15&process=ETCH&limit=100`。用字典管理参数比直接拼字符串更安全（requests会自动处理特殊字符编码）。
3.3 第三步：自动处理分页
如果数据量大，API不会一次全部返回。需要"翻页"：
def collect_all_lots(date, process, headers):
"""
采集指定日期和工序的全部Lot数据（自动翻页）
参数:
date: 日期字符串 "YYYY-MM-DD"
process: 工序名 "ETCH"
headers: 请求头（含Token）
返回:
list: 所有Lot数据的列表
"""
base_url = "http://mes-api.fab.local/api/lots"
all_lots = []
offset = 0
limit = 100
while True:
# 请求一页
params = {"date": date, "process": process,
"limit": limit, "offset": offset}
response = requests.get(base_url, headers=headers, params=params)
if response.status_code != 200:
print(f"第{offset//limit+1}页请求失败: {response.status_code}")
break
data = response.json()
page_lots = data.get('data', [])
total = data.get('total', 0)
# 把这一页的数据加入总列表
all_lots.extend(page_lots)
print(f"第{offset//limit+1}页: 获取 {len(page_lots)} 条, "
f"累计 {len(all_lots)}/{total} 条")
# 如果已经拿完所有数据，退出循环
offset += limit
if offset >= total:
break
return all_lots
# 使用
headers = {"Authorization": f"Bearer {TOKEN}"}
lots = collect_all_lots("2026-01-15", "ETCH", headers)
print(f"共采集 {len(lots)} 批Lot数据")
**为什么这样写？** 用 `while True` + `offset` 翻页是API采集的标准模式——每次请求一批，用完 `offset += limit` 后继续，直到 `offset >= total` 停止。`all_lots.extend(page_lots)` 比 `append` 更高效，因为 `extend` 直接把列表展开添加。我把整个逻辑封装成一个函数，以后调用就一行 `collect_all_lots("2026-01-15", "ETCH", headers)`，不用重复写翻页逻辑。
3.4 加上异常处理
def safe_collect(date, process, token):
"""
安全的数据采集函数
自动处理：网络超时、认证失败、服务器错误
永不崩溃——出错就打印日志，返回已经采集到的数据
"""
headers = {"Authorization": f"Bearer {token}"}
all_lots = []
try:
all_lots = collect_all_lots(date, process, headers)
except requests.exceptions.Timeout:
print("⚠ 请求超时！网络连接不稳定")
except requests.exceptions.ConnectionError:
print("⚠ 连接失败！请检查网络或MES服务是否正常")
except requests.exceptions.RequestException as e:
print(f"⚠ 请求异常: {e}")
# 即使出错了，也返回已成功采集的部分
print(f"最终采集结果: {len(all_lots)} 条")
return all_lots
# 使用
token = "YOUR_TOKEN_HERE"
data = safe_collect("2026-01-15", "ETCH", token)
# 可以用采集到的数据做其他事了
if data:
yields = [lot.get('yield', 0) for lot in data]
print(f"最低良率: {min(yields):.1f}%, "
f"最高良率: {max(yields):.1f}%, "
f"平均良率: {sum(yields)/len(yields):.1f}%")
**为什么这样写？** `try-except` 包裹整个采集过程，让程序不管遇到什么网络问题都不会崩溃。`requests.exceptions` 下有好几种异常类型，区分捕获方便针对性处理。最关键的是——**出错不要只报错，要返回已有的数据**。采集到99批，哪怕第100批失败了，99批也能用。
────────────────────────────────────────
四、效果对比
| 对比维度 | 手动导出 | API采集 | 提升 |
|---------|---------|--------|------|
| 单次耗时 | 5分钟 | 1-3秒 | **+100倍** |
| 每日操作 | 重复10次手动操作 | 跑一次脚本 | **全自动** |
| 数据完整性 | 容易忘记或漏选条件 | 固定参数、100%覆盖 | **零遗漏** |
| 出错处理 | 手动重新操作 | 自动重试+异常处理 | **省心** |
| 后续扩展 | 再分析要重新导出 | 存了数据直接给分析工具用 | **一次采集多次使用** |
────────────────────────────────────────
五、自己动手
# 练习：这里有一个模拟的API（不需要真的MES）
import requests
# 模拟API（公开的测试接口）
url = "https://jsonplaceholder.typicode.com/posts"
# 练习1：用GET请求获取数据
# 练习2：筛选出userId=1的数据（用params）
# 练习3：把数据存到本地CSV文件
# ✏️ 下面写你的代码
**思考题**：
1. 如果API返回了500错误，你的程序会不会崩溃？if语句和try-except哪个更好？
2. 如果你每天要采集5个不同工序的数据，怎么设计代码不重复？
3. 采集的数据怎么保存？存CSV、JSON、还是SQLite？
────────────────────────────────────────
六、常见误区
| 问题 | 表现 | 解决方法 |
|------|------|---------|
| **忘了加Token** | 一直返回401 | 检查headers里有没有Authorization字段 |
| **URL末尾多了一个/或少了** | 返回404 | 跟API文档核对URL |
| **以为一次能拿完** | 只拿到100条 | 检查有没有分页参数（limit/offset/cursor） |
| **params传了列表** | requests参数编码异常 | 用 `params={'ids': [1,2,3]}` 会自动处理 |
| **没做异常处理** | 网络波动直接崩溃 | 加try-except，即使错也要有返回值 |
| **请求太频繁** | 被MES封IP | 每次请求之间加 `time.sleep(1)` 降速 |
────────────────────────────────────────
> �� **你们MES有API接口吗？你对接时踩过什么坑？评论区聊聊**
> �� **收藏+点赞，后面讲到数据清洗和预处理会用到这里采集的数据** ��
