market_intel/packages/notify/wecom.py

"""企业微信群机器人通知（独立实现，不依赖策略仓）。"""

from __future__ import annotations

import logging
import time
from typing import Any

import httpx

from packages.config import get_settings

log = logging.getLogger("notify.wecom")

_last_fault_key: str | None = None
_last_fault_ms: float = 0.0
_fault_active: bool = False


def _as_bool(raw: Any, 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()
    return _as_bool(getattr(s, "wecom_enabled", False))


def wecom_webhook_url() -> str:
    s = get_settings()
    return (getattr(s, "wecom_webhook_url", "") or "").strip()


def wecom_machine_name() -> str:
    s = get_settings()
    return (getattr(s, "wecom_machine_name", "") or "").strip()[:64]


def alert_fail_threshold() -> int:
    s = get_settings()
    try:
        return max(1, int(getattr(s, "alert_fail_threshold", 5) or 5))
    except (TypeError, ValueError):
        return 5


def build_markdown(*, tag: str, title: str, lines: list[str] | None = None) -> str:
    machine = wecom_machine_name()
    prefix = f"【{machine}】" if machine else ""
    parts = [
        f"## {prefix}{title}",
        f"> **标识**: `{tag}`",
        f"> **系统**: 比特骆驼行情采集分析",
        f"> **时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}",
    ]
    if machine:
        parts.append(f"> **机器**: {machine}")
    if lines:
        parts.append("")
        for ln in lines:
            parts.append(f"> {ln}" if not ln.startswith(">") else ln)
    return "\n".join(parts)


def post_markdown(content: str) -> tuple[bool, str]:
    if not wecom_enabled():
        return False, "未开启企业微信通知"
    url = wecom_webhook_url()
    if not url:
        return False, "未配置 Webhook"
    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=10.0) as client:
            r = client.post(url, json=payload)
            body = r.json() if r.content else {}
            if r.status_code != 200 or str(body.get("errcode", 0)) not in ("0", "0.0"):
                return False, f"webhook failed status={r.status_code} body={body}"
            return True, "ok"
    except Exception as e:  # noqa: BLE001
        return False, str(e)


def notify_test() -> tuple[bool, str]:
    md = build_markdown(
        tag="TEST",
        title="行情采集分析 · 测试推送",
        lines=["这是一条测试消息，说明企微 Webhook 可用。"],
    )
    return post_markdown(md)


def notify_collector_fault(
    *,
    error: str,
    consecutive_failures: int,
    dedup_sec: float = 300.0,
) -> tuple[bool, str]:
    """连续失败告警；同错误键在 dedup_sec 内不重复推。"""
    global _last_fault_key, _last_fault_ms, _fault_active
    threshold = alert_fail_threshold()
    if consecutive_failures < threshold:
        return False, f"below threshold ({consecutive_failures}<{threshold})"

    key = f"{consecutive_failures // threshold}:{(error or '')[:120]}"
    now = time.time()
    if (
        _last_fault_key == key
        and (now - _last_fault_ms) < dedup_sec
    ):
        return False, "dedup"
    _last_fault_key = key
    _last_fault_ms = now
    _fault_active = True

    # 脱敏：避免日志/推送里出现完整密钥形态串
    err_show = (error or "unknown").replace("\n", " ")[:300]
    md = build_markdown(
        tag="FAULT",
        title="行情采集异常",
        lines=[
            f"**连续失败**: {consecutive_failures}（阈值 {threshold}）",
            f"**错误**: {err_show}",
            "请检查 OKX 连通性 / 代理 / 合约是否可交易。",
        ],
    )
    ok, msg = post_markdown(md)
    if ok:
        log.info("wecom fault notified failures=%s", consecutive_failures)
    else:
        log.warning("wecom fault notify failed: %s", msg)
    return ok, msg


def notify_collector_recovered(*, consecutive_failures: int = 0) -> tuple[bool, str]:
    global _fault_active, _last_fault_key
    if not _fault_active:
        return False, "no active fault"
    _fault_active = False
    _last_fault_key = None
    md = build_markdown(
        tag="RECOVER",
        title="行情采集已恢复",
        lines=["采样已恢复正常。"],
    )
    ok, msg = post_markdown(md)
    if ok:
        log.info("wecom recovered notified")
    else:
        log.warning("wecom recover notify failed: %s", msg)
    return ok, msg


def reset_alert_state() -> None:
    """测试用。"""
    global _last_fault_key, _last_fault_ms, _fault_active
    _last_fault_key = None
    _last_fault_ms = 0.0
    _fault_active = False