eth_hedge_sim/backend/app/notify/wecom.py

"""企业微信群机器人通知（独立模块）。"""

from __future__ import annotations

import asyncio
import logging
import time
from typing import Any

import httpx

from ..config import get_settings
from ..exchange.runtime import load_runtime_settings
from ..models.db import get_db

logger = logging.getLogger(__name__)

# 消息标识（Markdown 内展示，便于检索）
TAG_OPEN = "OPEN"
TAG_CLOSE = "CLOSE"
TAG_START = "START"
TAG_PAUSE = "PAUSE"
TAG_FAULT = "FAULT"
TAG_TEST = "TEST"

_last_fault_key: str | None = None
_last_fault_ms: float = 0.0
_FAULT_DEDUP_SEC = 300.0


def _as_bool(raw: str | None, default: bool = False) -> bool:
    if raw is None or raw == "":
        return default
    return str(raw).strip().lower() in ("1", "true", "yes", "on")


def wecom_enabled() -> bool:
    s = get_settings()
    if getattr(s, "wecom_enabled", False):
        return True
    try:
        return _as_bool(get_db().get_setting("wecom_enabled", "0"), False)
    except Exception:
        return False


def wecom_webhook_url() -> str:
    s = get_settings()
    url = (getattr(s, "wecom_webhook_url", None) or "").strip()
    if url:
        return url
    try:
        return (get_db().get_setting("wecom_webhook_url", "") or "").strip()
    except Exception:
        return ""


def wecom_machine_name() -> str:
    """多机推送区分用的机器名（设置页 / WECOM_MACHINE_NAME）。"""
    s = get_settings()
    name = (getattr(s, "wecom_machine_name", None) or "").strip()
    if name:
        return name[:64]
    try:
        return (get_db().get_setting("wecom_machine_name", "") or "").strip()[:64]
    except Exception:
        return ""


def venue_label() -> str | None:
    """实盘时返回「实盘·交易所」；SIM 不展示盘口标签。"""
    s = get_settings()
    if s.is_sim:
        return None
    try:
        ex = load_runtime_settings().exchange
    except Exception:
        ex = s.exchange or "okx"
    ex_u = str(ex).strip().lower()
    if ex_u in ("binance", "bn"):
        return "实盘·币安"
    if ex_u in ("gate", "gateio"):
        return "实盘·Gate"
    return "实盘·OKX"


def build_markdown(*, tag: str, title: str, lines: list[str] | None = None) -> str:
    body = "\n".join(f"> {ln}" if not ln.startswith(">") else ln for ln in (lines or []))
    machine = wecom_machine_name()
    venue = venue_label()
    prefix_parts: list[str] = []
    if machine:
        prefix_parts.append(f"【{machine}】")
    if venue:
        prefix_parts.append(f"【{venue}】")
    head = f"## {''.join(prefix_parts)}{title}"
    parts = [
        head,
        f"> **标识**: `{tag}`",
        f"> **时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}",
    ]
    if machine:
        parts.append(f"> **机器**: {machine}")
    if body:
        parts.append("")
        parts.append(body)
    return "\n".join(parts)


def _post_markdown_sync(content: str) -> tuple[bool, str]:
    if not wecom_enabled():
        return False, "未开启企业微信通知"
    url = wecom_webhook_url()
    if not url:
        return False, "未配置 Webhook"
    # 企业微信 markdown 上限约 4096 字节
    raw = content.encode("utf-8")
    if len(raw) > 4000:
        content = raw[:3900].decode("utf-8", errors="ignore") + "\n…"
    payload = {"msgtype": "markdown", "markdown": {"content": content}}
    try:
        with httpx.Client(timeout=8.0) as client:
            r = client.post(url, json=payload)
            data = r.json() if r.content else {}
            if r.status_code != 200 or int(data.get("errcode") or 0) != 0:
                return False, str(data.get("errmsg") or r.text or r.status_code)
            return True, "ok"
    except Exception as e:
        return False, str(e)


async def send_markdown(content: str) -> tuple[bool, str]:
    return await asyncio.to_thread(_post_markdown_sync, content)


def notify_async(content: str) -> None:
    """火忘：不阻塞策略循环。"""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        ok, msg = _post_markdown_sync(content)
        if not ok:
            logger.warning("wecom sync send failed: %s", msg)
        return

    async def _run() -> None:
        ok, msg = await send_markdown(content)
        if not ok:
            logger.warning("wecom send failed: %s", msg)

    loop.create_task(_run())


def notify_start() -> None:
    notify_async(
        build_markdown(
            tag=TAG_START,
            title="策略启动",
            lines=["策略已启动，可新开仓并盯目标平仓。"],
        )
    )


def notify_pause() -> None:
    notify_async(
        build_markdown(
            tag=TAG_PAUSE,
            title="策略暂停",
            lines=[
                "策略已暂停：**不再新开仓**。",
                "仍会盯盘：**目标平仓 + 到期平仓**。",
            ],
        )
    )


def notify_open(*, group_id: str, detail: str = "", extra: dict[str, Any] | None = None) -> None:
    extra = extra or {}
    lines = [
        f"**组**: `{group_id}`",
        f"**方向**: {extra.get('bias') or extra.get('option_side') or '—'}",
        f"**期权**: `{extra.get('option_inst_id') or '—'}`",
        f"**行权/到期**: {extra.get('strike') or '—'} / {extra.get('expiry_ymd') or '—'}",
    ]
    if detail:
        lines.append(f"**说明**: {detail}")
    notify_async(build_markdown(tag=TAG_OPEN, title="开仓成功", lines=lines))


def notify_close(
    *,
    reason: str,
    detail: str = "",
    group_id: str | None = None,
    data: dict[str, Any] | None = None,
) -> None:
    data = data or {}
    reason_zh = {
        "expiry": "到期平仓",
        "target_perp_only": "目标平仓·只平永续",
        "fixed_usdt": "目标平仓·双腿",
        "premium_multiple": "目标平仓·双腿",
        "emergency": "紧急全平",
        "emergency_perp": "紧急·只平永续",
        "manual": "手动全平",
        "perp_pending_retry": "续平永续",
    }.get(reason, reason)
    lines = [
        f"**原因**: {reason_zh} (`{reason}`)",
        f"**组**: `{group_id or data.get('group_id') or '—'}`",
    ]
    if detail:
        lines.append(f"**说明**: {detail}")
    net = data.get("net_pnl")
    if net is not None:
        lines.append(f"**净盈亏**: {net}")
    notify_async(build_markdown(tag=TAG_CLOSE, title=f"平仓 · {reason_zh}", lines=lines))


def notify_fault(*, title: str, detail: str, dedupe_key: str | None = None) -> None:
    global _last_fault_key, _last_fault_ms
    key = dedupe_key or f"{title}:{detail[:120]}"
    now = time.time()
    if key == _last_fault_key and now - _last_fault_ms < _FAULT_DEDUP_SEC:
        return
    _last_fault_key = key
    _last_fault_ms = now
    notify_async(
        build_markdown(
            tag=TAG_FAULT,
            title=f"故障 · {title}",
            lines=[f"**详情**: {detail[:500]}"],
        )
    )


def notify_test() -> tuple[bool, str]:
    venue = venue_label()
    machine = wecom_machine_name()
    lines = ["企业微信通知已连通。"]
    if machine:
        lines.append(f"机器名称: **{machine}**")
    if venue:
        lines.append(f"当前盘口标签: **{venue}**")
    content = build_markdown(
        tag=TAG_TEST,
        title="测试推送",
        lines=lines,
    )
    return _post_markdown_sync(content)