zyc/AirGate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: switch feishu alerts from Webhook to App (private message)

Browse Source 
- Replace Webhook with App ID + App Secret + mobile number
- Reuse AirDrama's feishu app (send private card messages)
- Add test button in system settings
- Add test-feishu API endpoint
- Default monitor interval changed to 60 seconds
- Token caching for feishu tenant_access_token

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
seaislee1209 2026-03-29 21:56:01 +08:00
parent 6b3a0bda34
commit 9e81717e08
7 changed files with 249 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
backend/apps/monitor/migrations/0011_globalconfig_feishu_app_id_and_more.py Normal file
View File

@ -0,0 +1,38 @@
# Generated by Django 4.2.21 on 2026-03-29 13:55
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('monitor', '0010_iamuser_deny_policy_exempt'),
]
operations = [
migrations.AddField(
model_name='globalconfig',
name='feishu_app_id',
field=models.CharField(blank=True, max_length=200, verbose_name='飞书 App ID'),
),
migrations.AddField(
model_name='globalconfig',
name='feishu_app_secret',
field=models.CharField(blank=True, max_length=200, verbose_name='飞书 App Secret'),
),
migrations.AlterField(
model_name='globalconfig',
name='feishu_alert_mobiles',
field=models.CharField(blank=True, help_text='接收告警的飞书用户手机号，多个用逗号分隔', max_length=500, verbose_name='飞书通知手机号（逗号分隔）'),
),
migrations.AlterField(
model_name='globalconfig',
name='feishu_webhook_url',
field=models.URLField(blank=True, max_length=500, verbose_name='飞书 Webhook URL（已弃用）'),
),
migrations.AlterField(
model_name='globalconfig',
name='monitor_interval_seconds',
field=models.IntegerField(default=60, verbose_name='监控间隔（秒）'),
),
]

10
backend/apps/monitor/models.py
View File

@ -184,9 +184,13 @@ class GlobalConfig(models.Model):
help_text='如 [50, 80, 90]')
default_project_policies = models.JSONField('添加项目时自动授权的策略', default=list, blank=True,
help_text='如 ["ArkFullAccess", "TOSFullAccess"]')
monitor_interval_seconds = models.IntegerField('监控间隔（秒）', default=3600)
feishu_webhook_url = models.URLField('飞书 Webhook URL', max_length=500, blank=True)
feishu_alert_mobiles = models.CharField('飞书通知手机号（逗号分隔）', max_length=500, blank=True)
monitor_interval_seconds = models.IntegerField('监控间隔（秒）', default=60)
feishu_app_id = models.CharField('飞书 App ID', max_length=200, blank=True)
feishu_app_secret = models.CharField('飞书 App Secret', max_length=200, blank=True)
feishu_alert_mobiles = models.CharField('飞书通知手机号（逗号分隔）', max_length=500, blank=True,
help_text='接收告警的飞书用户手机号，多个用逗号分隔')
# 保留 webhook 字段兼容性，但不再使用
feishu_webhook_url = models.URLField('飞书 Webhook URL（已弃用）', max_length=500, blank=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:

3
backend/apps/monitor/serializers.py
View File

@ -121,7 +121,8 @@ class GlobalConfigSerializer(serializers.ModelSerializer):
'default_alert_thresholds',
'default_project_policies',
'monitor_interval_seconds',
'feishu_webhook_url', 'feishu_alert_mobiles',
'feishu_app_id', 'feishu_app_secret',
'feishu_alert_mobiles',
'updated_at',
]
read_only_fields = ['updated_at']

1
backend/apps/monitor/urls.py
View File

@ -45,6 +45,7 @@ urlpatterns = [
# Global config
path('config/', views.global_config_view),
path('config/test-feishu/', views.test_feishu_view),
# Alerts
path('alerts/', views.alert_list_view),

22
backend/apps/monitor/views.py
View File

@ -1267,6 +1267,28 @@ def global_config_view(request):
return Response(serializer.data)
@api_view(['POST'])
def test_feishu_view(request):
"""测试飞书通知"""
config = GlobalConfig.get_solo()
app_id = config.feishu_app_id
app_secret = config.feishu_app_secret
mobile = request.data.get('mobile', '') or (config.feishu_alert_mobiles or '').split(',')[0].strip()
if not app_id or not app_secret:
return Response({'message': '请先配置飞书 App ID 和 App Secret'},
status=status.HTTP_400_BAD_REQUEST)
if not mobile:
return Response({'message': '请填写接收人手机号'},
status=status.HTTP_400_BAD_REQUEST)
from utils.feishu import send_feishu_test
success, msg = send_feishu_test(app_id, app_secret, mobile)
if success:
return Response({'message': msg})
return Response({'message': msg}, status=status.HTTP_400_BAD_REQUEST)
# ==================== Alerts ====================
@api_view(['GET'])

171
backend/utils/feishu.py
View File

@ -1,42 +1,171 @@
"""飞书机器人通知"""
"""飞书自建应用通知（复用 AirDrama 的飞书应用，发私信卡片）"""
import json
import logging
import threading
import requests
logger = logging.getLogger(__name__)
# Token 缓存
_token_cache = {'token': '', 'expires': 0}
def _get_tenant_access_token(app_id: str, app_secret: str) -> str:
"""获取飞书 tenant_access_token（带简单缓存）"""
import time
if _token_cache['token'] and _token_cache['expires'] > time.time():
return _token_cache['token']
resp = requests.post(
'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
json={'app_id': app_id, 'app_secret': app_secret},
timeout=5,
)
data = resp.json()
if data.get('code') != 0:
raise RuntimeError(f'Feishu token error: {data}')
token = data['tenant_access_token']
expire = data.get('expire', 7200)
_token_cache['token'] = token
_token_cache['expires'] = time.time() + expire - 60 # 提前60秒过期
return token
def _get_open_id_by_mobile(token: str, mobile: str) -> str:
"""通过手机号查询飞书 open_id"""
resp = requests.post(
'https://open.feishu.cn/open-apis/contact/v3/users/batch_get_id',
headers={'Authorization': f'Bearer {token}'},
json={'mobiles': [mobile]},
timeout=5,
)
data = resp.json()
if data.get('code') != 0:
raise RuntimeError(f'Feishu user lookup error: {data}')
user_list = data.get('data', {}).get('user_list', [])
if user_list and user_list[0].get('user_id'):
return user_list[0]['user_id']
return None
def send_feishu_alert(webhook_url: str, title: str, content: str,
template: str = "red"):
"""发送飞书卡片消息（非阻塞）"""
if not webhook_url:
logger.warning(f"飞书 Webhook 未配置，跳过通知: {title}")
return
"""发送飞书卡片消息到配置的接收人（非阻塞）
webhook_url 参数保留兼容性但不再使用，改为从 GlobalConfig 读取飞书应用配置。
"""
def _send():
payload = {
"msg_type": "interactive",
"card": {
"config": {"wide_screen_mode": True},
"header": {
"title": {"tag": "plain_text", "content": title},
"template": template,
try:
from apps.monitor.models import GlobalConfig
config = GlobalConfig.get_solo()
app_id = config.feishu_app_id
app_secret = config.feishu_app_secret
mobiles_str = config.feishu_alert_mobiles
if not app_id or not app_secret:
logger.warning(f"飞书应用未配置，跳过通知: {title}")
return
if not mobiles_str:
logger.warning(f"飞书告警手机号未配置，跳过通知: {title}")
return
mobiles = [m.strip() for m in mobiles_str.split(',') if m.strip()]
if not mobiles:
return
token = _get_tenant_access_token(app_id, app_secret)
card = {
'config': {'wide_screen_mode': True},
'header': {
'title': {'tag': 'plain_text', 'content': title},
'template': template,
},
"elements": [
'elements': [
{
"tag": "div",
"text": {"tag": "lark_md", "content": content},
'tag': 'div',
'text': {'tag': 'lark_md', 'content': content},
}
],
},
}
try:
resp = requests.post(webhook_url, json=payload, timeout=10)
resp.raise_for_status()
logger.info(f"飞书通知已发送: {title}")
}
for mobile in mobiles:
try:
open_id = _get_open_id_by_mobile(token, mobile)
if not open_id:
logger.warning(f'未找到手机号 {mobile} 对应的飞书用户')
continue
resp = requests.post(
'https://open.feishu.cn/open-apis/im/v1/messages',
headers={'Authorization': f'Bearer {token}'},
params={'receive_id_type': 'open_id'},
json={
'receive_id': open_id,
'msg_type': 'interactive',
'content': json.dumps(card, ensure_ascii=False),
},
timeout=5,
)
data = resp.json()
if data.get('code') != 0:
logger.error(f'飞书发送失败 {mobile}: {data}')
else:
logger.info(f'飞书通知已发送 {mobile}: {title}')
except Exception as e:
logger.error(f'飞书通知错误 {mobile}: {e}')
except Exception as e:
logger.error(f"飞书通知发送失败: {e}")
thread = threading.Thread(target=_send, daemon=True)
thread.start()
def send_feishu_test(app_id: str, app_secret: str, mobile: str):
"""发送测试消息。Returns (success, message)。"""
try:
token = _get_tenant_access_token(app_id, app_secret)
open_id = _get_open_id_by_mobile(token, mobile)
if not open_id:
return False, f'未找到手机号 {mobile} 对应的飞书用户'
card = {
'config': {'wide_screen_mode': True},
'header': {
'title': {'tag': 'plain_text', 'content': 'AirGate 告警测试'},
'template': 'blue',
},
'elements': [
{
'tag': 'div',
'text': {
'tag': 'lark_md',
'content': '这是一条测试消息，说明飞书告警通道配置正常。',
},
},
],
}
resp = requests.post(
'https://open.feishu.cn/open-apis/im/v1/messages',
headers={'Authorization': f'Bearer {token}'},
params={'receive_id_type': 'open_id'},
json={
'receive_id': open_id,
'msg_type': 'interactive',
'content': json.dumps(card, ensure_ascii=False),
},
timeout=5,
)
data = resp.json()
if data.get('code') != 0:
return False, f'发送失败: {data.get("msg", "")}'
return True, '测试消息已发送'
except Exception as e:
return False, str(e)

33
frontend/src/views/settings/SettingsView.vue
View File

@ -23,14 +23,25 @@
<el-form-item label="监控间隔(秒)">
<el-input-number v-model="config.monitor_interval_seconds" :min="60" :step="60" />
</el-form-item>
<el-form-item label="飞书 Webhook URL">
<el-input v-model="config.feishu_webhook_url" placeholder="https://open.feishu.cn/open-apis/bot/v2/hook/..." />
<el-divider content-position="left">飞书通知</el-divider>
<el-form-item label="飞书 App ID">
<el-input v-model="config.feishu_app_id" placeholder="cli_xxxxxxxx" />
</el-form-item>
<el-form-item label="飞书通知手机号">
<el-input v-model="config.feishu_alert_mobiles" placeholder="手机号1,手机号2" />
<el-form-item label="飞书 App Secret">
<el-input v-model="config.feishu_app_secret" type="password" show-password
placeholder="飞书自建应用的密钥" />
</el-form-item>
<el-form-item label="告警接收手机号">
<el-input v-model="config.feishu_alert_mobiles" placeholder="手机号，多个用逗号分隔" />
<div style="font-size:12px;color:#999;margin-top:4px;">
填写飞书用户的手机号，告警会以私信卡片发送
</div>
</el-form-item>
<el-form-item>
<el-button type="primary" @click="saveConfig" :loading="savingConfig">保存配置</el-button>
<el-button @click="testFeishu" :loading="testingFeishu" style="margin-left:12px;">
测试飞书通知
</el-button>
</el-form-item>
</el-form>
</el-card>
@ -146,6 +157,20 @@ async function saveConfig() {
}
}
const testingFeishu = ref(false)
async function testFeishu() {
testingFeishu.value = true
try {
const { data } = await api.post('/api/v1/config/test-feishu/')
ElMessage.success(data.message)
} catch (e) {
ElMessage.error(e.response?.data?.message || '测试失败')
} finally {
testingFeishu.value = false
}
}
async function loadAccounts() {
loadingAccounts.value = true
try {