AirGate/backend/utils/feishu.py

"""飞书自建应用通知（复用 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"):
    """发送飞书卡片消息到配置的接收人（非阻塞）

    webhook_url 参数保留兼容性但不再使用，改为从 GlobalConfig 读取飞书应用配置。
    """
    def _send():
        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': [
                    {
                        'tag': 'div',
                        'text': {'tag': 'lark_md', 'content': content},
                    }
                ],
            }

            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)