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

【摘要】

本文系统梳理FMEA 风险分析领域的核心问题与落地路径。我们FAB的MES系统有个功能，但99%的人只用了一种方式——手动导出：。全文围绕「问题背景：每天点"导出"要疯掉了、打开MES网页、填入查询条件（日期、工站、产品）、点击"导出CSV"、等5分钟下载」逐层展开，给出原理说明、实操步骤与验证方法。核心结论：| 请求太频繁 | 被MES封IP | 每次请求之间加 time.sleep(1) 降速 |。补充说明：FMEA 的定位是事前预防工具：在设计与工艺定稿前系统列举潜在失效模式、后果与原因，据此安排预防与探测措施。事后补做 FMEA 只能当作记录，难以产生实际收益。

【核心要点】

• 问题背景：每天点"导出"要疯掉了：我们FAB的MES系统有个功能，但99%的人只用了一种方式——手动导出：
• 打开Excel看看数据：每天要导10次，每次5分钟，一天花50分钟在"等下载"上。
• FMEA 风险分析：FMEA 的定位是事前预防工具：在设计与工艺定稿前系统列举潜在失效模式、后果与原因，据此安排预防与探测措施。事后补做 FMEA 只能当作记录，难以产生实际收益。
• FMEA 风险分析：评价维度由三部分构成——严重度衡量后果的破坏性，发生度衡量原因出现的可能性，探测度衡量现有控制手段能否在后果发生前发现它。三者含义独立，不能相互替代。

【适用场景】

• 新工艺/新设备导入前的风险评估与预防措施规划。
• 重大质量事故后的系统性风险复盘。
• 客户审核与体系认证中的风险文件准备。
• FMEA 失效模式来源的系统性收集机制。

一、问题背景：每天点"导出"要疯掉了

我们FAB的MES（制造执行系统，Manufacturing Execution System）系统有个功能，但99%的人只用了一种方式——手动导出：

FMEA（失效模式与影响分析，Failure Mode and Effects Analysis） 的有效性取决于失效模式的覆盖度。系统性收集来源包括历史质量记录、客户投诉、设备故障数据与同类产品的共性问题，仅靠团队头脑风暴容易遗漏低频但高严重度的风险。

1. 打开MES网页

FMEA 的定位是事前预防工具：在设计与工艺定稿前系统列举潜在失效模式、后果与原因，据此安排预防与探测措施。事后补做 FMEA 只能当作记录，难以产生实际收益。

2. 填入查询条件（日期、工站、产品）

工艺 FMEA 与设计 FMEA 的视角不同：前者关注工序参数与设备状态导致的失效，后者关注产品结构与功能设计缺陷。两者不能互相替代。

3. 点击"导出CSV"

评价维度由三部分构成——严重度衡量后果的破坏性，发生度衡量原因出现的可能性，探测度衡量现有控制手段能否在后果发生前发现它。三者含义独立，不能相互替代。

4. 等5分钟下载

MES 的实时性要求分层设计：工单状态与报工需要秒级响应，而统计报表与分析可以分钟级甚至小时级。把所有功能都按最高实时性建设会显著推高成本，按需求分层才是合理做法。

5. 打开Excel看看数据

每天要导10次，每次5分钟，一天花50分钟在"等下载"上。

偶尔忙起来忘了导出，第二天分析发现没数据。

更难受的是：数据自动化工具（SPC控制图、异常检测）做出来了，但没人天天喂数据给它们。

其实MES有API接口——给程序用的"数据取餐口"。直接调用API，3秒拿到数据。

学完这一篇，你能做到：

存量 MES 的改造比新建更难：历史数据与既有习惯形成的路径依赖，使得任何变更都需要兼容处理。因此改造通常采用渐进式替换而非一次性切换。

1. 用Python给API发请求，拿到数据

追溯能力依赖数据的完整性而非系统的复杂度。追溯链要求在物料批次、设备、参数、人员、时间五个维度上都留有记录，任何一环缺失都会让追溯在关键节点断裂。

2. 解析JSON格式的数据

MES 的数据模型是整个系统的地基：物料、设备、工序、工单、批次五类主数据的编码规则与关联关系一旦确定，后续所有功能都建立在其上。模型设计不当会在系统运行一段时间后集中爆发为数据不一致问题。

3. 把数据存下来，给其他分析工具用

────────────────────────────────────────

传统做法把三项相乘得到 RPN（风险优先数，Risk Priority Number） 并据此排序，但 RPN 存在明显缺陷：不同组合可能得出相同数值，风险性质却完全不同；且数值本身缺少量纲意义。新版标准因此引入措施优先级（AP）分级。

二、技术原理：API就是"程序之间的对话"

2.1 什么是API？

你可以把API理解成自动售货机：

• 你投币（发请求）→ 售货机吐出饮料（返回数据）

• 你手里不用有人（不需要登录网页）

• 24小时工作（不用等上班时间）

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的字典和列表：

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},

]

}

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的字典/列表。

MES 的失败原因里，技术问题通常排在组织问题之后。工艺路线不稳定、职责边界不清、编码体系混乱这三类前置问题未解决时，系统上线只会把混乱数字化。

三、实战案例：三步写出你的API采集器

3.1 第一步：带Token的请求

大多数MES的API需要认证，最常见的方式是Bearer Token——相当于你的工牌：

import requests

TOKEN = "eyJhbGciOiJIUzI1NiIs..." # 一串很长的字符串

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)

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（批次，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

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%覆盖	零遗漏

出错处理	手动重新操作	自动重试+异常处理	省心

后续扩展	再分析要重新导出	存了数据直接给分析工具用	一次采集多次使用

工业数据采集的第一难点是协议异构：不同年代、不同厂商的设备使用不同协议（Modbus、Profibus、OPC UA、专有协议等）。网关层的协议适配能力决定了采集方案的可行性上限。

五、自己动手

import requests

url = "https://jsonplaceholder.typicode.com/posts"

思考题：

1. 如果API返回了500错误，你的程序会不会崩溃？if语句和try-except哪个更好？

FMEA 的实际输出应是一份带责任人与完成时间的措施清单，而不是一张评分表。评分只是排序手段，措施落地才是目的。

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接口吗？你对接时踩过什么坑？评论区聊聊

📚 收藏+点赞，后面讲到数据清洗和预处理会用到这里采集的数据 👆

---

工艺 FMEA 应聚焦可控变量：工序参数、设备状态、材料批次、操作方式与环境条件。把不可控因素列入 FMEA 只会增加条目而无法产出措施。

【常见坑】

• 为什么这样写？ try-except 包裹整个采集过程，让程序不管遇到什么网络问题都不会崩溃。requests.exceptions 下有好几种异常类型，区分捕获方便针对性处理。最关键的是——出错不要只报错，要返回已有的数据。采集到99批，哪怕第100批失败了，99批也能用。
• | 数据完整性 | 容易忘记或漏选条件 | 固定参数、100%覆盖 | 零遗漏 |
• FMEA 由一个人闭门完成。跨职能参与是有效性的前提，缺少设备、工艺、质量任一方都会留下盲区。
• 把 FMEA 做成一次性交付文件，工艺变更后不更新。失效模式集合随工艺演进变化，过期 FMEA 会给出虚假的安全感。
• 只做 RPN 排序不看严重度独立性。高严重度项即使发生度低，也应保留在优先处理清单中。

常见问题（FAQ）

Q：RPN 排序够不够用？

A：作为初步排序手段可用，但不足以支撑决策。RPN 相同却风险性质不同的情况很常见，且它无法体现「严重度极高的低概率事件」这类必须优先处理的项。实践中的做法是把措施优先级分级与严重度的独立门槛结合使用。

Q：FMEA 什么时候做最有效？

A：设计与工艺定稿之前。此时修改成本最低，措施可以并入正式方案。工艺定型后补做，多数措施只能沦为检测手段，无法从源头消除风险。

Q：如何避免 FMEA 流于形式？

A：三个要点：跨职能团队参与、每项措施明确责任人与期限、把 FMEA 与变更管理流程绑定。只要工艺发生变更就触发 FMEA 复查，文件才不会过期。

Q：预防措施和探测措施怎么权衡？

A：原则是能预防的不靠探测。预防措施从源头降低失效发生概率，收益是持续的；探测措施只能提高发现概率，一旦探测失效则风险直接传递到下游。因此在措施优先级评定中，严重度高且难以探测的项应优先安排预防型措施，探测型措施作为补充。

Q：FMEA 多久需要更新一次？

A：没有固定周期，触发条件是「变化」：工艺参数调整、设备更换或大修、材料或供应商变更、产品设计变更、以及发生了新型失效。建立与变更管理绑定的复查机制，比按固定周期更新更有效，也更容易执行。

Q：MES 和 ERP 到底谁管什么？

A：简单区分：ERP 管「要不要做、要多少、成本多少」，面向计划与财务；MES 管「怎么做、做得怎么样」，面向现场执行与过程数据。两者的交界通常在工单下达与完工回报，边界设计不清就会出现重复录入与口径冲突。

【总结】

FMEA 风险分析的难点往往不在单点技术，而在于把原理、数据与现场验证串成闭环。| 请求太频繁 | 被MES封IP | 每次请求之间加 time.sleep(1) 降速 |。评价维度由三部分构成——严重度衡量后果的破坏性，发生度衡量原因出现的可能性，探测度衡量现有控制手段能否在后果发生前发现它。三者含义独立，不能相互替代。建议先用小范围试点验证有效性，再逐步扩大适用范围，并保留完整的数据与判断记录以便复盘。

术语速查

• AP（Action Priority，措施优先级）：按严重度、发生度、探测度的组合直接划分高/中/低优先级，替代单纯依赖 RPN 排序。　工程意义：解决了 RPN 相同但风险性质完全不同的问题。
• MES（Manufacturing Execution System，制造执行系统）：位于计划层与控制层之间，负责工单执行、过程追溯、数据采集与质量管控的信息系统。　工程意义：是打通 ERP 计划与车间现场的中间层。
• Lot（Lot，批次）：作为生产、追溯与质量判定基本单位的一组产品。　工程意义：批次粒度设计要在追溯精度与管理成本之间取平衡。
• SPC（Statistical Process Control，统计过程控制）：用控制图等统计方法对过程进行实时监控，区分随机波动与异常波动的管理方法。　工程意义：是把「事后检验」变为「过程预防」的核心工具。
• FMEA（Failure Mode and Effects Analysis，失效模式与影响分析）：系统识别潜在失效模式、评估其后果与原因，并优先处理高风险项的分析方法。　工程意义：是从设计端预防问题的核心工具。
• RPN（Risk Priority Number，风险优先数）：严重度 S、发生度 O、可探测度 D 三者的乘积，用于对风险排序。　工程意义：但 RPN 存在重复数值与量纲问题，新版标准已引入 AP 分级。
• WIP（Work In Process，在制品）：处于生产流程中尚未完工的产品或批次。　工程意义：WIP 金额与停留时间过高意味着流程存在瓶颈或积压。
• ISA-95（ISA-95 / IEC 62264，企业控制系统集成标准）：定义企业层到现场层五级功能模型（L0~L4）与接口的国际标准。　工程意义：MES 功能边界与集成接口设计的基本依据。
• SCADA（Supervisory Control and Data Acquisition，数据采集与监视控制系统）：面向现场设备的数据采集与集中监控系统。　工程意义：常作为 MES 与设备之间的数据通道。
• OPC（Optical Proximity Correction，光学邻近效应校正）：在掩模版图形上预先做几何补偿，使经过光学衍射与工艺效应后在晶圆上得到目标图形的技术。　工程意义：28nm 以下节点的必需手段，直接决定线宽一致性。
• MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）：轻量级发布订阅协议，适合低带宽、不稳定的工业现场数据传输。　工程意义：适合高频采集数据的边缘汇聚。

相关阅读

• FAB数据采集自动化：MES API/SECS-GEM/OPC UA怎么选
• 半导体制造MES数据采集实战：Python+MQTT实现设备数据秒级采集
• 半导体FAB设备物联网实战：用MQTT搭建设备数据采集架构
• MES与ERP集成：工单/物料/成本的数据打通

📚 同栏目延伸阅读：时序预测模型LSTM：设备状态提前30分钟预警、报表自动化：每天自动生成FAB日报、时间序列分析:预测FAB生产趋势、半导体产业全景：从沙子到芯片的完整产业链