19.API数据采集：用requests自动拉取MES数据

一、问题背景每天点导出要疯掉了我们FAB的MES系统有个功能但99%的人只用了一种方式——**手动导出**1. 打开MES网页2. 填入查询条件日期、工站、产品3. 点击导出CSV4. 等5分钟下载5. 打开Excel看看数据每天要导10次每次5分钟一天花50分钟在等下载上。偶尔忙起来忘了导出第二天分析发现没数据。**更难受的是**数据自动化工具SPC控制图、异常检测做出来了但没人天天喂数据给它们。**其实MES有API接口**——给程序用的数据取餐口。直接调用API3秒拿到数据。**学完这一篇你能做到**1. 用Python给API发请求拿到数据2. 解析JSON格式的数据3. 把数据存下来给其他分析工具用────────────────────────────────────────二、技术原理API就是程序之间的对话2.1 什么是API你可以把API理解成**自动售货机**- 你投币发请求→ 售货机吐出饮料返回数据- 你手里不用有人不需要登录网页- 24小时工作不用等上班时间# 最基础的API调用import requests# 发请求就像按售货机的按钮url http://mes-api.fab.local/api/lotsresponse 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: fBearer {TOKEN},Content-Type: application/json}# 发请求url http://mes-api.fab.local/api/lotsresponse requests.get(url, headersheaders)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, headersheaders, paramsparams)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?date2026-01-15processETCHlimit100。用字典管理参数比直接拼字符串更安全requests会自动处理特殊字符编码。3.3 第三步自动处理分页如果数据量大API不会一次全部返回。需要翻页def collect_all_lots(date, process, headers):采集指定日期和工序的全部Lot数据自动翻页参数:date: 日期字符串 YYYY-MM-DDprocess: 工序名 ETCHheaders: 请求头含Token返回:list: 所有Lot数据的列表base_url http://mes-api.fab.local/api/lotsall_lots []offset 0limit 100while True:# 请求一页params {date: date, process: process,limit: limit, offset: offset}response requests.get(base_url, headersheaders, paramsparams)if response.status_code ! 200:print(f第{offset//limit1}页请求失败: {response.status_code})breakdata response.json()page_lots data.get(data, [])total data.get(total, 0)# 把这一页的数据加入总列表all_lots.extend(page_lots)print(f第{offset//limit1}页: 获取 {len(page_lots)} 条, f累计 {len(all_lots)}/{total} 条)# 如果已经拿完所有数据退出循环offset limitif offset total:breakreturn all_lots# 使用headers {Authorization: fBearer {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: fBearer {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_HEREdata 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不需要真的MESimport requests# 模拟API公开的测试接口url https://jsonplaceholder.typicode.com/posts# 练习1用GET请求获取数据# 练习2筛选出userId1的数据用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接口吗你对接时踩过什么坑评论区聊聊** **收藏点赞后面讲到数据清洗和预处理会用到这里采集的数据**