接口自动化巡检实战：用 Python 一口气盯住 6 个业务接口

一、写在前面接口这东西平时安安静静一旦挂了往往还是用户先发现的。与其被动救火不如写个脚本定时去“点名”谁掉线、谁变慢、谁返回了异常码一眼就能看明白。前段时间接手一个内网数据平台它对外提供 6 个业务接口——每日实况、人工预报、周预报、月预报外加两类风险预报模型数据。它们的共同点很整齐都要带同一个 key 鉴权都走 GET返回体是统一的 {code, message, data} 结构。这意味着巡检逻辑本身并不复杂难的是把它做得“能长期跑、出问题能喊人”。我把要做的事情拆成了三条一条命令批量跑完全部接口不用一个个点结果落盘成 JSON方便归档也方便以后接别的处理流程有接口挂了就自动推企业微信群别让我天天盯着看。下面按“公共模块 → 单接口调试 → 批量巡检 → 生产插件”的顺序把整个过程捋一遍。二、先看看目录结构项目本身很小几个文件各司其职后面加接口也只需要往清单里补一行。接口巡检/├── config.py # 公共配置地址、密钥加载、请求助手├── key.txt # 接口密钥单独存放不入库├── case_1_daily_live.py # 接口1每日实况数据├── case_2_manual_forecast.py # 接口2人工预报数据├── case_3_week_forecast.py # 接口3周预报数据├── case_4_month_forecast.py # 接口4月预报数据├── case_5_week_risk.py # 接口5周风险预报模型数据├── case_6_month_risk.py # 接口6月风险预报模型数据├── run_all.py # 批量巡检入口└── output/ # 巡检报告输出目录└── ops_RISKXM_20260814102905.json之所以把公共部分单独抽成 config.py是因为 6 个接口的请求参数高度重复——地址、密钥、超时、结果打印。如果每个脚本都抄一遍改一处地址就要改六个文件这种亏我以前吃过不想再吃第二次。图 1 项目模块与数据流向三、公共请求模块把重复的东西收口config.py 负责三件事读取密钥、拼公共参数、发请求并顺带记下耗时。把耗时也一起返回是因为排查“接口是不是变慢了”时这个数字很关键。# -*- coding: utf-8 -*-数据接口巡检 —— 公共配置与请求助手import osimport jsonimport timeimport requests# 基础配置 # 服务地址内网。地址变更时只改这一处。BASE_URL http://10.0.0.10:48080/risk-api/# key.txt 与本文件同目录_HERE os.path.dirname(os.path.abspath(__file__))KEY_FILE os.path.join(_HERE, key.txt)TIMEOUT 30def load_key() - str:从 key.txt 读取密钥兼容 keyxxx 与裸 xxx 两种写法。if not os.path.exists(KEY_FILE):raise FileNotFoundError(f未找到密钥文件: {KEY_FILE})with open(KEY_FILE, r, encodingutf-8) as f:content f.read().strip()if in content:content content.split(, 1)[1].strip()return contentdef build_params(extra: dict None) - dict:构造公共请求参数始终携带 key。params {key: load_key()}if extra:params.update(extra)return paramsdef api_get(path: str, params: dict None, timeout: int TIMEOUT) - dict:发送 GET 请求并返回解析后的 JSON。url BASE_URL.rstrip(/) / path.lstrip(/)start time.time()resp requests.get(url, paramsparams, timeouttimeout)elapsed round(time.time() - start, 3)try:data resp.json()except ValueError:data {raw_text: resp.text}data[_elapsed_sec] elapseddata[_status_code] resp.status_codedata[_url] resp.urlreturn datadef print_result(title: str, result: dict):打印接口测试结果方便单接口调试时肉眼确认。print( * 80)print(f[{title}])print(fHTTP状态码: {result.get(_status_code)} 耗时: {result.get(_elapsed_sec)}s)print(f请求URL: {result.get(_url)})print(f业务code: {result.get(code)} message: {result.get(message) or result.get(msg)})print(json.dumps(result, ensure_asciiFalse, indent2))这里有个细节值得说一下接口返回体里的提示字段有的接口叫 message有的叫 msg所以打印时用了 result.get(message) or result.get(msg) 兜一下避免显示成 None 让人误判。四、单接口脚本方便单独调试批量跑之前先保证每个接口能单独调通。每个 case 脚本只做一件事——带上参数请求一次然后打印结果。以接口 1 为例# -*- coding: utf-8 -*-接口1获取每日实况数据 /getDailyLiveDataimport sysimport ossys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))from config import api_get, build_params, print_resultPATH /getDailyLiveDataTEST_PARAMS {time: 2026-07-22, # 时间条件格式 YYYY-MM-DDperiod: day, # 周期按日查询stationIdList: 2, # 站点 ID 列表}def main():params build_params(TEST_PARAMS)result api_get(PATH, paramsparams)print_result(接口1 获取每日实况数据, result)if __name__ __main__:main()其余 5 个脚本结构完全一致区别只在 PATH 和 TEST_PARAMS。其中周预报、月预报、两类风险模型这几个接口支持按年/周月查历史不传参数时默认返回当前周期所以调试阶段参数可以先留空。五、批量巡检一条命令跑完所有接口单接口调通后把接口清单抽出来用一个循环挨个请求最后汇总成一张表。核心就是下面这段# -*- coding: utf-8 -*-批量巡检依次调用接口 1~6并汇总结果。import sysimport osimport jsonimport tracebacksys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))from config import api_get, build_paramsINTERFACES [(1.获取每日实况数据, /getDailyLiveData,{time: 2026-07-22, period: day, stationIdList: 2}),(2.人工预报数据, /getManualForecastData,{forecastDate: 2026-07-22 00:00:00}),(3.周预报数据, /getWeekForecastData, {}),(4.月预报数据, /getMonthForecastData, {}),(5.周风险预报模型数据, /getWeekRiskData, {}),(6.月风险预报模型数据, /getMonthRiskData, {}),]def main():summary []for name, path, extra in INTERFACES:try:result api_get(path, paramsbuild_params(extra))code result.get(code)summary.append({接口: name,路径: path,HTTP状态码: result.get(_status_code),业务code: code,message: result.get(message),耗时(s): result.get(_elapsed_sec),状态: 成功 if code 200 else 失败,})except Exception as e:traceback.print_exc()summary.append({接口: name, 路径: path,状态: 异常, message: str(e)})print(json.dumps(summary, ensure_asciiFalse, indent2))if __name__ __main__:main()注意那个 try/except——单个接口超时或异常绝不能把整个巡检带崩。所以异常也照样记一条状态标记成“异常”最后一起汇报。六、生产版做成带告警的巡检插件调试脚本自己用没问题但要放到服务器上定时跑就得再往前一步参数从外面传进来、结果写文件、出问题推企微。改造后的插件长这样# -*- coding: utf-8 -*-接口巡检监控插件批量巡检 JSON 报告 企微告警import jsonimport osimport timefrom datetime import datetimeimport requestsDEFAULT_INTERFACES [{name: 1.获取每日实况数据, path: /getDailyLiveData,params: {time: 2026-07-22, period: day, stationIdList: 2}},{name: 2.人工预报数据, path: /getManualForecastData,params: {forecastDate: 2026-07-22 00:00:00}},{name: 3.周预报数据, path: /getWeekForecastData, params: {}},{name: 4.月预报数据, path: /getMonthForecastData, params: {}},{name: 5.周风险预报模型数据, path: /getWeekRiskData, params: {}},{name: 6.月风险预报模型数据, path: /getMonthRiskData, params: {}},]def run(param: dict, record_id: str, last_message: str, save_dir: str):project_name str(param[project_name])project_name_en str(param.get(project_name_en) or project_name)key str(param[key])base_url str(param[base_url]).rstrip(/) # 去掉末尾 /, 避免 // 导致 404timeout int(param.get(timeout, 30))verbose int(param.get(verbose, 0)) # 0仅统计 1统计明细wecom_webhook str(param.get(wecom_webhook, )).strip()interfaces param.get(interfaces, DEFAULT_INTERFACES)if not os.path.exists(save_dir):os.makedirs(save_dir)summary, details [], {}for item in interfaces:name, path item[name], str(item[path])req_params {key: key}req_params.update(item.get(params, {}) or {})url f{base_url}/{path.lstrip(/)}rec {接口: name, 路径: path, HTTP状态码: None,业务code: None, message: None, 耗时(s): None, 状态: 异常}start time.time()try:resp requests.get(url, paramsreq_params, timeouttimeout)elapsed round(time.time() - start, 3)try:data resp.json()except ValueError:data {raw_text: resp.text}data[_elapsed_sec] elapseddata[_status_code] resp.status_codedata[_url] resp.urlcode data.get(code)ok (resp.status_code 200 and code 200)rec.update({HTTP状态码: resp.status_code,业务code: code,message: data.get(message) or data.get(msg),耗时(s): elapsed,状态: 成功 if ok else 失败,})if verbose:details[name] dataexcept Exception as e:rec[message] str(e)rec[耗时(s)] round(time.time() - start, 3)summary.append(rec)now_tag datetime.now().strftime(%Y%m%d%H%M%S)output {project_name: project_name, timestamp: now_tag, summary: summary}if verbose:output[details] detailsif wecom_webhook:output[webhook] push_wecom(wecom_webhook, project_name, now_tag, summary, timeout)json_path os.path.join(save_dir, fops_{project_name_en}_{now_tag}.json)with open(json_path, w, encodingutf-8) as f:json.dump(output, f, ensure_asciiFalse, indent2)return json.dumps({status: 1, recordId: record_id,file_urls: [{success: True, file_url: json_path}]},ensure_asciiFalse, indent2)告警推送单独抽了一个函数拼成企业微信 markdown 消息成功的接口标绿、失败的标红一眼就能看出问题在哪def push_wecom(webhook: str, project_name: str, now_tag: str,summary: list, timeout: int) - dict:把巡检结果推送到企业微信群机器人。ok_count sum(1 for s in summary if s[状态] 成功)fail_count len(summary) - ok_countlines [f## [{project_name}] 接口巡检报告,f 时间: {now_tag[:4]}-{now_tag[4:6]}-{now_tag[6:8]} f{now_tag[8:10]}:{now_tag[10:12]}:{now_tag[12:14]},f 接口总数: {len(summary)} 成功: font color\info\{ok_count}/fontf 失败: font color\warning\{fail_count}/font,]for s in summary:status_txt (font color\info\成功/font if s[状态] 成功else font color\warning\失败/font)lines.append(f- {s[接口]} {status_txt} HTTP{s[HTTP状态码]} fcode{s[业务code]} {s[message] or } 耗时{s[耗时(s)]}s)if fail_count 0:lines.append(\nfont color\warning\存在失败接口, 请及时处理!/font)msg_body {msgtype: markdown, markdown: {content: \n.join(lines)}}resp requests.post(webhook, jsonmsg_body, timeouttimeout)data resp.json()if data.get(errcode) ! 0:return {success: False, message: f企微推送失败: {data.get(errmsg)}}return {success: True, message: f已推送企微, errcode{data.get(errcode)}}调用时把参数塞进一个字典就行verbose 设成 0 只输出统计设成 1 会把每个接口的完整返回也一起存进 JSONparam {project_name: 风险监测项目,project_name_en: RISKXM,key: YOUR_API_KEY, # 实际密钥请放入配置文件base_url: http://10.0.0.10:48080/risk-api/,timeout: 30,verbose: 0,wecom_webhook: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key****,}图 2 接口自动化巡检执行流程七、跑起来看看结果执行一次巡检output 目录下会生成一份带时间戳的报告。这份 JSON 的结构很直白summary 里是每个接口的明细{project_name: 风险监测项目,timestamp: 20260814102905,summary: [{接口: 1.获取每日实况数据,路径: /getDailyLiveData,HTTP状态码: 200,业务code: 200,message: 成功,耗时(s): 0.066,状态: 成功},{接口: 5.周风险预报模型数据,路径: /getWeekRiskData,HTTP状态码: 200,业务code: 200,message: 成功,耗时(s): 0.309,状态: 成功}],webhook: {success: true,message: 已推送企微, errcode0}}图 3 巡检报告 JSON 结构从耗时上看6 个接口都在毫秒级返回其中“周风险预报模型数据”明显慢一些接近 0.3 秒——因为它背后要做模型计算慢一点也算合理。把每次的耗时存下来日积月累就能看出接口有没有“悄悄变慢”。图 4 各接口单次调用响应耗时对比八、几个踩过的坑做这个小工具的过程里有几个坑印象比较深记下来给后来人省点时间地址末尾的斜杠。base_url 带一个 /路径又以 / 开头拼出来就变成 //服务器直接 404。统一用 rstrip(/) lstrip(/) 处理最省心。提示字段不统一。有的接口返回 message有的返回 msg取值时两个都要兜一下否则日志里一片 None。异常必须兜住。巡检脚本最忌讳“一个接口挂整轮全崩”。每个接口单独 try异常也记录成一条结果剩下的接口照跑不误。密钥别写死在代码里。单独放 key.txt代码里读文件真要进版本库记得把 key.txt 加进 .gitignore。企微 markdown 的颜色标签要成对闭合。info 是绿色、warning 是红色写错了消息会原样显示标签很尴尬。九、小结整个工具加起来不到 200 行却解决了一个挺实际的问题接口有没有挂、有没有变慢不用再靠人肉一个个点。回头再看思路其实就三步——先把公共逻辑收口再让单接口能独立调试最后套一层调度和告警。接口清单抽出来之后以后新增接口只需要在列表里加一行剩下的它自己会跑。如果你也在维护一堆零散的接口不妨照着搭一个花不了多少时间但省下来的心是实打实的。