SmartUp/backend/app/utils/dingtalk.py

"""DingTalk webhook signing and message formatting (ported from monitor script)."""
import base64
import hashlib
import hmac
import json
import time
from typing import Any
from urllib.parse import quote_plus


def dingtalk_signed_url(webhook_url: str, secret: str) -> str:
    timestamp = str(int(time.time() * 1000))
    string_to_sign = f"{timestamp}\n{secret}".encode("utf-8")
    digest = hmac.new(secret.encode("utf-8"), string_to_sign, hashlib.sha256).digest()
    sign = quote_plus(base64.b64encode(digest).decode("utf-8"))
    sep = "&" if "?" in webhook_url else "?"
    return f"{webhook_url}{sep}timestamp={timestamp}&sign={sign}"


def format_dingtalk_rate_changed(upstream_name: str, changed_at: str, changes: list[dict[str, Any]]) -> dict[str, Any]:
    lines = [
        f"### 📊 {upstream_name} 分组倍率变更",
        "",
        f"- **时间**：{changed_at}",
        f"- **变化数量**：{len(changes)}",
        "",
    ]
    for ch in changes:
        name = ch.get("group_name") or ch.get("group_id") or "unknown"
        platform = ch.get("platform") or "-"
        old = ch.get("old_rate")
        new = ch.get("new_rate")
        lines.append(f"- `{name}` ({platform})：`{old}` → `{new}`")
    return {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{upstream_name} 分组倍率变更",
            "text": "\n".join(lines),
        },
    }


def format_dingtalk_website_rate_changed(
    website_name: str,
    target_group_name: str,
    changed_at: str,
    old_rate: Any,
    new_rate: Any,
) -> dict[str, Any]:
    group_name = target_group_name or "unknown"
    lines = [
        f"### 网站倍率变更：{website_name}",
        "",
        f"- 时间：{changed_at}",
        f"- 分组：`{group_name}`",
        f"- 倍率：`{old_rate}` -> `{new_rate}`",
    ]
    return {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{website_name} 网站倍率变更",
            "text": "\n".join(lines),
        },
    }


def format_dingtalk_priority_changed(
    website_name: str, upstream_name: str, changed_at: str,
    updates: list[dict],
) -> dict[str, Any]:
    success = sum(1 for u in updates if u.get("status") == "success")
    failed = sum(1 for u in updates if u.get("status") == "failed")
    skipped = sum(1 for u in updates if u.get("status") == "skipped")
    lines = [
        f"### 🔄 {website_name} 账号优先级变更",
        "",
        f"- **触发上游**：{upstream_name}",
        f"- **时间**：{changed_at}",
        f"- **摘要**：{success} 更新 / {failed} 失败 / {skipped} 跳过",
        "",
    ]

    # 按目标分组聚合
    from collections import defaultdict
    groups = defaultdict(list)
    for u in updates:
        tg_name = u.get("target_group_name") or u.get("target_group_id") or u.get("group_id") or "未知目标分组"
        groups[tg_name].append(u)

    for tg_name, items in groups.items():
        lines.append(f"目标分组：{tg_name}")
        for u in items:
            emoji = {"success": "✅", "failed": "❌", "skipped": "⏭️"}.get(u.get("status", ""), "➖")
            acc_name = u.get("account_name") or u.get("account_id") or "未知账号"
            acc_id = u.get("account_id") or "未知ID"
            
            old_p = u.get("old_priority")
            old_p_str = "未知" if old_p in (None, "", "未知") else str(old_p)
            
            new_p = u.get("new_priority")
            new_p_str = "未知" if new_p in (None, "", "未知") else str(new_p)
            
            src_group = u.get("source_group_name") or u.get("source_group_id") or "未知来源"
            
            rate = u.get("source_rate")
            rate_str = str(rate) if rate is not None else "—"
            
            line = f"- {emoji} {acc_name}（账号ID {acc_id}）：old={old_p_str} → new={new_p_str}，来源分组={src_group}，倍率={rate_str}"
            if u.get("status") == "failed" and u.get("message"):
                line += f"，原因={u.get('message')}"
            lines.append(line)
        lines.append("")

    while lines and not lines[-1]:
        lines.pop()

    return {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{website_name} 账号优先级变更",
            "text": "\n".join(lines),
        },
    }


def format_dingtalk_balance_low(
    upstream_name: str, balance: float, threshold: float, changed_at: str
) -> dict[str, Any]:
    lines = [
        f"### ⚠️ {upstream_name} 余额不足",
        "",
        f"- **当前余额**：{balance:.2f}",
        f"- **告警阈值**：{threshold:.2f}",
        f"- **时间**：{changed_at}",
    ]
    return {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{upstream_name} 余额不足",
            "text": "\n".join(lines),
        },
    }


def format_dingtalk_status(upstream_name: str, event: str, changed_at: str, error: str = "") -> dict[str, Any]:
    emoji = "🔴" if event == "upstream_unhealthy" else "🟢"
    label = "服务异常" if event == "upstream_unhealthy" else "服务恢复"
    lines = [
        f"### {emoji} {upstream_name} {label}",
        "",
        f"- **时间**：{changed_at}",
    ]
    if error:
        lines.append(f"- **错误**：{error}")
    return {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{upstream_name} {label}",
            "text": "\n".join(lines),
        },
    }