Python自动化客服监控：基于七鱼API与飞书机器人的实时告警与报告系统

1. 项目概述：从手动救火到自动化巡航的客服运营革命

如果你也负责过游戏、电商或者任何有在线客服团队的运营工作，下面这个场景你一定不陌生：下午三点，你刚开完一个冗长的会议，打开客服后台一看，排队人数已经悄悄爬升到了50+，聊天群里用户抱怨声一片，而你的客服主管可能还在处理其他事务，对此一无所知。到了月底，为了统计客服的绩效、响应率、工单处理量，你又得熬夜从后台导出几十个Excel表格，手动合并、计算，一折腾就是一整晚。这种“事后发现，手动统计”的模式，不仅消耗大量人力，更让服务质量的控制永远慢半拍。

今天要分享的，就是我用Python为网易七鱼客服系统搭建的一套全自动监控与报告推送系统。它不是什么高深莫测的黑科技，而是用最朴素的脚本，解决了最实际的运营痛点：实时排队监控、自动小时报推送、工单处理跟踪。核心目标就一个：把运营人员从重复、低效的“数据搬运工”角色中解放出来，让系统自动盯盘、自动报警、自动出报告，让人力聚焦在更有价值的策略分析和问题解决上。整个系统基于七鱼的开放API，通过飞书或企业微信机器人推送消息，部署简单，效果立竿见影。接下来，我会从设计思路、技术细节到避坑经验，完整拆解这个项目的实现过程。

2. 核心需求解析与方案选型

在动手写代码之前，明确我们要解决的具体问题至关重要。盲目堆砌功能只会制造出难以维护的“屎山”。我的需求来源于一个真实的游戏私服运营团队，他们的客服使用网易七鱼，痛点非常典型。

2.1 四大核心运营痛点

1. 排队黑洞：客服排队人数是服务质量的晴雨表。但后台数据不会主动喊你，运营不可能24小时盯着后台。经常出现的情况是，某个客服临时离开或某个热门活动突然开始，排队人数在半小时内从个位数暴涨到几十人，等有人发现时，用户等待时间可能已经超过10分钟，体验严重受损。
2. 报告体力活：管理层需要定期（如每小时）了解客服团队的服务概况，包括在线人数、排队情况、接入率、平均等待时间等。过去需要运营人员定点去后台截图、录数据、做Excel、再发到群里。这个过程枯燥、易错，且严重依赖人的自觉性。
3. 工单管理盲区：七鱼的工单系统记录了用户复杂问题的处理流程。但哪些工单超时未处理？哪个客服今天还没处理过工单？这些关键的管理洞察，如果没有自动化统计，就埋没在大量的列表信息中。
4. 绩效统计噩梦：月底计算客服的KPI，如响应率、满意度、处理量，需要从不同维度的报表中提取数据，进行复杂的交叉计算。手动操作不仅耗时数天，而且极易出错，引发团队内部对公平性的质疑。

2.2 为什么选择“API + 机器人推送”的技术栈？

面对这些需求，解决方案有很多。可以直接用七鱼后台的仪表盘（但无法主动告警），也可以购买昂贵的第三方SaaS监控工具（可能过度定制或不符合国内IM习惯）。我选择自建Python脚本，基于以下考量：

• 成本与可控性：七鱼提供了完备的开放API，免费且功能强大。Python脚本部署在自有服务器或云函数上，几乎零额外成本。所有逻辑完全自定义，后续增减功能、调整阈值都非常灵活。
• 无缝融入工作流：团队日常沟通都在飞书或企业微信上进行。将告警和报告直接推送到群聊，符合大家的使用习惯，信息触达率最高，实现了“人找信息”到“信息找人”的转变。
• 技术栈轻量成熟：Requests库处理HTTP请求简单可靠，飞书/企微的机器人Webhook配置更是傻瓜式。整个技术栈没有复杂依赖，学习成本和维护成本都极低。
• 与现有系统解耦：这套系统独立于七鱼客服工作台运行，不会影响客服人员的正常操作。它只是一个“监控眼”和“广播站”，架构上的解耦带来了部署的便捷性和系统的稳定性。

注意：在方案设计初期，我曾考虑过是否要做一个带Web界面的仪表盘。但仔细评估后放弃了，因为我们的核心需求是“及时感知”和“定期同步”，而不是“实时可视化”。一个推送到IM的简短消息卡片，在移动端和PC端的阅读体验和及时性上，反而优于需要主动打开网页的仪表盘。工具永远服务于场景。

3. 七鱼API深度对接与数据封装

一切自动化的基础是数据。网易七鱼开放平台提供了丰富的API，我们的首要任务就是学会如何安全、高效地调用它们，并把返回的复杂JSON数据封装成我们业务逻辑里容易理解的Python对象。

3.1 获取API凭证与核心接口梳理

首先，你需要登录网易七鱼开放平台（通常在你的七鱼管理后台可以找到入口），创建一个“自建应用”。创建成功后，你会获得一对至关重要的凭证：App Key和App Secret。这相当于系统的用户名和密码，务必妥善保管，不要硬编码在脚本里，更不要上传到公开的代码仓库。

我们的系统主要依赖以下几个核心接口：

1. 获取会话概要：这是排队监控的基石。接口返回当前所有会话的排队数、等待中会话数、在线客服数等关键实时指标。
2. 获取客服工作量：用于小时报中的客服排名。可以按时间维度查询每个客服的接待量、会话时长等。
3. 获取工单列表：用于工单监控。可以通过筛选条件获取特定状态（如“待处理”、“处理中”、“已完结”）的工单，并能看到负责人、创建时间、最后更新时间等。
4. 获取满意度评价：用于绩效分析。可以拉取指定时间段内用户对客服服务的评价数据。

3.2 构建稳健的API请求模块

直接在每个脚本里写HTTP请求代码是灾难的开始。我们必须封装一个统一的、健壮的请求模块。这个模块需要处理以下几个关键问题：

• 认证：七鱼API通常使用App Key和App Secret生成签名，或者直接使用access_token。我们需要一个函数来自动处理令牌的获取与刷新。
• 错误重试：网络请求可能失败。对于获取核心监控数据的请求，必须加入重试机制（例如，最多重试3次，每次间隔2秒）。
• 速率限制：了解并遵守七鱼API的调用频率限制，避免请求被禁。
• 日志记录：所有API调用，无论成功失败，都应记录日志，方便后期排查问题。

下面是一个高度简化的qiyu_api.py模块核心部分示例，展示了如何封装一个带错误处理和认证的请求函数：

import requests import time import hashlib import hmac import base64 from urllib.parse import urlencode import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class QiyuAPI: def __init__(self, app_key, app_secret, base_url='https://qiyukf.com'): self.app_key = app_key self.app_secret = app_secret self.base_url = base_url self.session = requests.Session() # 可以在这里初始化一些公共headers def _generate_signature(self, params): """生成七鱼API要求的签名。""" # 1. 将参数按键名升序排序 sorted_params = sorted(params.items(), key=lambda x: x[0]) # 2. 拼接成 key1=value1&key2=value2 的格式 param_str = '&'.join([f'{k}={v}' for k, v in sorted_params]) # 3. 将app_secret作为密钥，对参数字符串进行HMAC-SHA1加密 sign = hmac.new(self.app_secret.encode('utf-8'), param_str.encode('utf-8'), hashlib.sha1).digest() # 4. 将加密结果用Base64编码 signature = base64.b64encode(sign).decode('utf-8') return signature def call_api(self, endpoint, method='GET', params=None, data=None, max_retries=3): """调用七鱼API的通用方法，包含重试机制。""" url = f"{self.base_url}{endpoint}" # 将app_key加入请求参数 all_params = params or {} all_params['appKey'] = self.app_key all_params['time'] = str(int(time.time() * 1000)) # 当前时间戳，毫秒 # 生成签名并加入参数 all_params['sign'] = self._generate_signature(all_params) for attempt in range(max_retries): try: logger.info(f"尝试调用API: {url}, 参数: {all_params}") if method.upper() == 'GET': response = self.session.get(url, params=all_params, timeout=10) else: # POST # 注意：有些七鱼API是POST，且参数在body中，签名方式可能不同，这里仅为示例 response = self.session.post(url, json=data, params=all_params, timeout=10) response.raise_for_status() # 如果状态码不是200，抛出HTTPError result = response.json() # 检查七鱼API返回的业务码（例如，code=200表示成功） if result.get('code') != 200: logger.error(f"API业务错误: {result.get('message')}") # 这里可以根据不同的业务码决定是否重试 return None return result.get('data') # 返回数据部分 except requests.exceptions.RequestException as e: logger.warning(f"API请求失败 (尝试 {attempt + 1}/{max_retries}): {e}") if attempt < max_retries - 1: time.sleep(2 ** attempt) # 指数退避 else: logger.error(f"API请求重试{max_retries}次后仍失败: {url}") return None except ValueError as e: logger.error(f"解析JSON响应失败: {e}, 响应内容: {response.text}") return None # 具体的业务方法封装 def get_session_summary(self): """获取当前会话概要，包含排队数等。""" endpoint = '/openapi/v2/session/summary' # 示例端点，需替换为真实端点 data = self.call_api(endpoint) # 对返回数据进行清洗和格式化 if data: # 例如，将等待用户数、在线客服数提取出来 return { 'waiting_count': data.get('waitingSessionCount', 0), 'online_staff_count': data.get('onlineStaffCount', 0), 'avg_waiting_seconds': data.get('avgWaitingSeconds', 0) } return None def get_workload_report(self, start_time, end_time): """获取客服工作量报告。""" endpoint = '/openapi/v2/staff/workload' params = {'startTime': start_time, 'endTime': end_time} return self.call_api(endpoint, params=params)

实操心得：七鱼不同版本的API签名方式可能有差异。务必仔细阅读对应版本的官方文档。我在初期就曾因为用了旧版的签名算法，调试了半天。建议将签名函数单独测试，用官方提供的示例参数验证通过后再集成。

4. 核心功能模块实现详解

有了稳定的数据获取层，我们就可以构建上层的业务功能了。整个系统由三个核心脚本构成，分别对应三个核心需求。

4.1 模块一：实时排队监控与分级告警 (minor_queue_alert.py)

这个脚本的目标是定时检查排队人数，并在超过阈值时通过IM机器人发出告警。

核心逻辑流程：

1. 调用API：使用封装好的QiyuAPI.get_session_summary()获取当前排队人数(waiting_count)和在线客服数(online_staff_count)。
2. 应用告警规则：这是策略的核心。我设定了两级告警：
   • 警告级：排队人数 > 20。此时在飞书群里发送一个橙色（或黄色）的警告卡片，提示“排队人数较多，请关注”。
   • 严重级：排队人数 > 50且在线客服数 < 2（这个条件很重要，如果客服都在线只是忙，告警级别可以不同）。此时发送红色告警卡片，并@相关责任人。
3. 构建并发送消息：将告警信息组织成飞书/企微机器人支持的格式（通常是JSON），通过Webhook发送。
4. 避免告警风暴：如果排队人数一直在阈值之上，我们不能每分钟都发一条告警。常见的做法是引入“恢复检测”和“静默期”。例如，只有状态从“正常”变为“告警”时才发消息，或者每小时只发一次持续告警的提醒。

飞书消息卡片示例代码片段：

import json import requests def send_feishu_alert(waiting_count, online_staff, level='warning'): webhook_url = "YOUR_FEISHU_WEBHOOK_URL" if level == 'critical': title = "🚨 七鱼客服排队严重告警" color = "red" content = f"**当前排队人数：{waiting_count}人**\n在线客服：{online_staff}人\n排队人数已超过50人，且在线客服不足，请立即处理！<at user_id=\"ou_xxxxxx\">负责人</at>" else: # warning title = "⚠️ 七鱼客服排队警告" color = "orange" content = f"**当前排队人数：{waiting_count}人**\n在线客服：{online_staff}人\n排队人数较多，建议关注。" card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": title}, "template": color }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", "content": content } }, { "tag": "action", "actions": [{ "tag": "button", "text": {"tag": "plain_text", "content": "📊 查看七鱼后台"}, "type": "primary", "url": "https://qiyukf.com/your_admin_url" }] } ] } } try: resp = requests.post(webhook_url, json=card, timeout=5) resp.raise_for_status() print(f"飞书告警发送成功: {level}") except Exception as e: print(f"飞书告警发送失败: {e}")

4.2 模块二：自动化小时报推送 (qiuqiu_minor_hourly.py)

小时报的目标是在工作时段（如10:00-19:00）的每个整点后，自动推送过去一小时的服务数据概览。

核心逻辑流程：

1. 确定时间范围：计算上一个整小时的时间段（如现在是15:05，则拉取14:00-15:00的数据）。
2. 聚合多维度数据：
   • 从get_session_summary（或历史统计接口）获取该时间段的平均排队人数、平均等待时间、接入率。
   • 从get_workload_report获取该时间段内客服工作量排名（按接待会话数排序）。
   • 从满意度接口获取该时间段的平均满意度。
3. 生成报告文本：将上述数据组织成清晰易读的格式。使用Markdown或飞书卡片可以做得非常美观。
4. 定时推送：通过Cron或类似工具，在每小时的第X分钟（如03分）触发脚本执行。

报告内容设计示例：

【七鱼客服小时报】14:00 - 15:00 ----------------------------- 📈 核心指标 • 平均排队人数：12人 • 平均等待时间：86秒 • 会话接入率：94.7% • 平均满意度：4.8星 🏆 客服工作量TOP5 1. 客服-小明：接待 45 会话 2. 客服-小红：接待 38 会话 3. 客服-小刚：接待 35 会话 4. 客服-小美：接待 32 会话 5. 客服-小强：接待 28 会话 💡 运营建议 当前服务压力适中，整体服务质量优秀。继续保持！

注意：小时报的推送时间要避开整点（如设置在第3分钟），因为整点可能是系统数据结算的高峰期，API响应可能变慢或数据不准确。预留几分钟的缓冲期更稳妥。

4.3 模块三：工单处理监控与绩效统计 (ticket_hourly_report.py)

这个脚本关注的是工单生命周期和客服的工单处理情况。

核心逻辑流程：

1. 获取工单列表：调用工单列表接口，筛选出状态为“待处理”和“处理中”的工单。特别关注createTime和deadline（如果有），用于判断是否超时。
2. 统计关键指标：
   • 待处理工单总数：让管理者一目了然。
   • 即将超时/已超时工单列表：列出工单ID、标题、负责人、超时时间，这是需要立即跟进的重点。
   • 客服工单处理量统计：统计每个客服名下“已完结”的工单数量，用于绩效参考。
   • 零处理客服提醒：如果某个客服当天（或当班）尚未处理任何工单，可以温和提醒（注意方式，避免在公开群内点名批评）。
3. 生成并推送报告：将统计结果格式化为报告，在每天固定的时间点（如下午17点）推送，方便下班前同步进度。

一个实用的技巧：工单监控脚本可以和小时报脚本合并，作为小时报的一个附加部分。也可以独立运行，频率调整为每2-4小时一次，避免信息过载。

5. 部署、调度与生产环境考量

脚本写好了，怎么让它7x24小时稳定运行呢？这里涉及到部署环境、定时调度和稳定性保障。

5.1 部署环境选择

• 云服务器（推荐）：购买一台最基础的Linux云服务器（如1核1G）。优点是控制力强，可以安装任何需要的环境，方便调试和查看日志。成本每月约30-50元。
• Serverless云函数：如果你追求极致的成本和免运维，可以使用腾讯云SCF、阿里云FC或华为云FunctionGraph。将脚本打包部署，并配置定时触发器。优点是按量计费，几乎免费。缺点是对本地文件操作、复杂依赖的支持有时不如虚拟机灵活，调试稍麻烦。
• 本地电脑（不推荐）：电脑关机服务就中断，不适合生产环境。

5.2 定时任务调度：Cron的精准使用

在Linux服务器上，Cron是定时任务的不二之选。配置时要注意几个细节：

# 编辑当前用户的cron任务 crontab -e # 示例：每天工作时段（10-19点），每30分钟执行一次排队监控 */30 10-19 * * * cd /home/user/qiyu-automation && /usr/bin/python3 /home/user/qiyu-automation/minor_queue_alert.py >> /home/user/qiyu-automation/cron.log 2>&1 # 示例：每天工作时段，每小时第3分钟执行小时报推送 3 10-19 * * * cd /home/user/qiyu-automation && /usr/bin/python3 /home/user/qiyu-automation/qiuqiu_minor_hourly.py >> /home/user/qiyu-automation/cron.log 2>&1 # 示例：每天下午5点执行工单报告 0 17 * * * cd /home/user/qiyu-automation && /usr/bin/python3 /home/user/qiyu-automation/ticket_hourly_report.py >> /home/user/qiyu-automation/cron.log 2>&1

关键解释：

• cd /path/to/project：确保脚本在正确的目录下运行，能正确找到配置文件或模块。
• /usr/bin/python3：使用绝对路径指定Python解释器，避免因环境变量问题找不到命令。
• >> /path/to/cron.log 2>&1：将脚本的标准输出和错误输出都重定向追加到日志文件。这是最重要的调试手段！没有它，脚本静默失败了你都不知道。

5.3 稳定性与防重复执行机制

当脚本被定时触发时，必须考虑一个情况：如果上一次执行还没结束（比如API响应慢），下一次执行又开始了怎么办？这可能导致重复告警、数据错乱。

解决方案：使用文件锁（Lock File）

import os import sys import time def acquire_lock(lockfile='/tmp/qiyu_hourly_report.lock'): """尝试获取文件锁，防止脚本重复执行。""" import fcntl try: lock_fd = os.open(lockfile, os.O_CREAT | os.O_RDWR) # 尝试获取非阻塞排他锁 fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # 获取成功，将PID写入文件，方便排查 os.write(lock_fd, str(os.getpid()).encode()) return lock_fd except (IOError, OSError): # 获取锁失败，说明已有实例在运行 print(f"另一个实例正在运行，退出。锁文件: {lockfile}") sys.exit(0) def release_lock(lock_fd, lockfile='/tmp/qiyu_hourly_report.lock'): """释放文件锁。""" import fcntl fcntl.flock(lock_fd, fcntl.LOCK_UN) os.close(lock_fd) try: os.remove(lockfile) except OSError: pass # 在脚本主函数开始处调用 if __name__ == '__main__': lock_fd = acquire_lock() try: # 你的主业务逻辑 here main() finally: release_lock(lock_fd)

这个机制确保了同一时间只有一个脚本实例在运行。记得在脚本异常退出时，也要在finally块中释放锁。

6. 常见问题排查与优化经验

在实际运行中，你肯定会遇到各种各样的问题。下面是我踩过坑后总结出来的“避坑指南”。

6.1 API调用失败与数据异常

6.2 消息推送失败

• Webhook地址失效：飞书/企微的机器人Webhook地址如果泄露被他人调用，管理员可能会重置它。定期检查机器人是否还在群里，消息是否还能发送。
• 消息格式错误：飞书卡片和企微机器人的消息格式是特定的JSON。一个多余的逗号、一个错误的字段名都会导致推送失败。务必使用json.dumps()打印出你要发送的消息体，并先通过简单的curl命令或Postman手动测试。
• 频率限制：IM机器人通常有发送频率限制。避免在短时间内发送大量消息。我们的监控频率（30分钟/次）远低于限制，一般没问题。

6.3 性能与日志优化

• 日志分级：不要只用print。使用Python内置的logging模块，设置不同的级别（INFO, WARNING, ERROR）。将调试信息、常规运行信息、错误信息分别记录，方便排查。
• 日志轮转：Cron日志和脚本自身日志会越来越大。使用logrotate工具配置日志轮转策略，避免磁盘被撑满。
• 异常捕获要具体：不要只用except Exception。针对网络超时、JSON解析错误、密钥错误等不同类型的异常，进行分别捕获和处理，给出更明确的错误提示。

6.4 一个真实的踩坑案例：时区问题

我们的服务器设置在UTC时区，而业务数据需要按北京时间（UTC+8）来统计。在编写小时报脚本时，我直接用了datetime.now()来生成start_time和end_time，结果推送的报告时间范围总是对不上。

解决方案：在所有时间处理逻辑中，显式地指定时区。

from datetime import datetime, timedelta import pytz # 需要安装 pytz 包 def get_bj_hour_range(hour_offset=1): """获取北京时间的上一个整小时时间段。""" bj_tz = pytz.timezone('Asia/Shanghai') # 获取当前北京时间 now_bj = datetime.now(bj_tz) # 计算上一个整小时 end_hour = now_bj.replace(minute=0, second=0, microsecond=0) start_hour = end_hour - timedelta(hours=hour_offset) # 格式化为API需要的毫秒时间戳 start_ts = int(start_hour.timestamp() * 1000) end_ts = int(end_hour.timestamp() * 1000) return start_ts, end_ts, start_hour.strftime('%H:%M'), end_hour.strftime('%H:%M')

这个细节问题让我排查了近一个小时，教训就是：在处理任何与时间相关的业务逻辑时，第一时间就要明确时区。

7. 扩展思路与高级玩法

基础系统稳定运行后，你可以考虑以下扩展方向，让它变得更智能、更强大。

1. 多维度根因分析：当排队告警触发时，脚本可以自动分析：是某个特定客服组的排队激增？还是整体在线人数下降？亦或是同时段会话涌入量异常？通过调用更多细分API，在告警消息中附带初步分析，能极大提升处理效率。
2. 与运维监控系统集成：将七鱼的关键指标（如平均等待时间、满意度）通过Prometheus之类的监控系统暴露出来，并配置Grafana看板。这样就能在一个统一的平台上监控服务器性能、应用状态和客服服务质量。
3. 自动化处理建议：基于历史数据训练简单的模型（或设定规则），在报告末尾给出自动化建议。例如：“过去一周，每周三下午3点排队人数都会上升，建议在该时段增加一名备用客服。”
4. 搭建简易管理面板：使用Flask或Streamlit快速搭建一个内部网页，可以手动触发报告、查看历史告警记录、临时调整告警阈值等，提升运营同学的使用体验。

这套系统上线后，最直接的反馈就是运营群里的“救火”消息变少了，大家对客服状态的感知从“盲人摸象”变成了“全局透视”。技术的目的从来不是炫技，而是实实在在地提升效率、解决问题。希望这份详细的拆解，能帮助你构建起属于自己的自动化运营工具，把时间花在更有创造力的事情上。