gold_monitor/app/wecom.py

# -*- coding: utf-8 -*-
"""企业微信群机器人 Webhook 推送。"""
from __future__ import annotations

import json
import logging
import urllib.error
import urllib.request
from typing import Any, Optional

from app import db as database
from app.config import Config

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

SETTING_WEBHOOK = "wecom_webhook_url"
SETTING_ENABLED = "wecom_enabled"
SETTING_DAILY_LAST = "wecom_daily_last"


def is_enabled(conn: Optional[Any] = None) -> bool:
    def _read(c) -> bool:
        flag = database.get_setting(c, SETTING_ENABLED, "")
        if flag == "0":
            return False
        if flag == "1":
            return True
        # 未配置开关时：有 Webhook 即视为开启
        return bool(webhook_url(c))

    if conn is not None:
        return _read(conn)
    with database.get_db() as c:
        return _read(c)


def webhook_url(conn: Optional[Any] = None) -> str:
    def _read(c) -> str:
        url = (database.get_setting(c, SETTING_WEBHOOK, "") or "").strip()
        if url:
            return url
        return (Config.WECOM_WEBHOOK_URL or "").strip()

    if conn is not None:
        return _read(conn)
    with database.get_db() as c:
        return _read(c)


def send_text(content: str, mentioned: Optional[list[str]] = None) -> tuple[bool, str]:
    """发送文本消息。返回 (ok, detail)。"""
    text = (content or "").strip()
    if not text:
        return False, "空消息"
    with database.get_db() as conn:
        if not is_enabled(conn):
            return False, "未启用"
        url = webhook_url(conn)
    if not url:
        return False, "未配置 Webhook"
    body: dict[str, Any] = {
        "msgtype": "text",
        "text": {"content": text[:2048]},
    }
    if mentioned:
        body["text"]["mentioned_list"] = mentioned
    raw = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=raw,
        headers={"Content-Type": "application/json; charset=utf-8"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=8) as resp:
            payload = resp.read().decode("utf-8", errors="replace")
        try:
            data = json.loads(payload)
        except json.JSONDecodeError:
            return True, payload[:200]
        errcode = int(data.get("errcode", 0) or 0)
        if errcode != 0:
            return False, data.get("errmsg") or payload[:200]
        return True, "ok"
    except urllib.error.HTTPError as exc:
        return False, f"HTTP {exc.code}"
    except Exception as exc:  # noqa: BLE001
        return False, str(exc)


def _side_cn(side: str) -> str:
    s = (side or "").lower()
    if s == "buy":
        return "买入"
    if s == "sell":
        return "卖出"
    return side or "—"


def notify_trade_open(
    account_name: str,
    ticket: str,
    symbol: str,
    side: str,
    volume: float,
    open_price: float,
    sl: Any = None,
    tp: Any = None,
) -> None:
    sl_txt = f"{float(sl):.3f}" if sl not in (None, "") else "—"
    tp_txt = f"{float(tp):.3f}" if tp not in (None, "") else "—"
    msg = (
        f"【开仓】{account_name or '账户'}\n"
        f"#{ticket} {_side_cn(side)} {volume:.2f} 手 {symbol}\n"
        f"开仓价 {float(open_price):.3f}  SL {sl_txt}  TP {tp_txt}"
    )
    ok, detail = send_text(msg)
    if not ok:
        log.info("wecom open skip/fail: %s", detail)


def notify_trade_close(
    account_name: str,
    ticket: str,
    symbol: str,
    side: str,
    volume: float,
    open_price: float,
    close_price: Any,
    profit: Any,
) -> None:
    px = f"{float(close_price):.3f}" if close_price not in (None, "") else "—"
    try:
        pnl = float(profit)
        pnl_txt = f"{pnl:+.2f}"
    except (TypeError, ValueError):
        pnl_txt = "—"
    msg = (
        f"【平仓】{account_name or '账户'}\n"
        f"#{ticket} {_side_cn(side)} {volume:.2f} 手 {symbol}\n"
        f"开 {float(open_price):.3f} → 平 {px}  盈亏 {pnl_txt}"
    )
    ok, detail = send_text(msg)
    if not ok:
        log.info("wecom close skip/fail: %s", detail)


def format_daily_account(account_name: str, day: str, stats: dict[str, Any]) -> str:
    closed = int(stats.get("closed") or 0)
    open_pos = int(stats.get("open_pos") or 0)
    trade_n = closed + open_pos
    profit = float(stats.get("profit") or 0.0)
    wins = int(stats.get("wins") or 0)
    losses = int(stats.get("losses") or 0)
    be = int(stats.get("breakeven") or 0)
    wr = float(stats.get("win_rate") or 0.0)
    return (
        f"【黄金日报】{account_name or '账户'} · {day}\n"
        f"当日交易：{trade_n} 笔（已平 {closed} · 持仓 {open_pos}）\n"
        f"已平盈亏：{profit:+.2f}\n"
        f"胜率：{wr:.0f}%（{wins}胜{losses}负{be}平）"
    )


def send_daily_reports(day: Optional[str] = None) -> tuple[bool, str]:
    """按账户推送北京时间当日汇总。"""
    with database.get_db() as conn:
        if not is_enabled(conn):
            return False, "未启用"
        if not webhook_url(conn):
            return False, "未配置 Webhook"
        accounts = database.list_accounts(conn)
        if not accounts:
            return False, "无账户"
        start_ts, end_ts, day_label = database.day_bounds_bjt(day)
        _ = start_ts, end_ts
        lines: list[str] = []
        any_trade = False
        for acc in accounts:
            aid = int(acc["id"])
            stats = database.trade_stats(conn, account_id=aid, day=day_label, period="today")
            closed = int(stats.get("closed") or 0)
            open_pos = int(stats.get("open_pos") or 0)
            if closed == 0 and open_pos == 0:
                continue
            any_trade = True
            lines.append(format_daily_account(acc.get("name") or f"#{aid}", day_label, stats))
        if not any_trade:
            lines.append(f"【黄金日报】{day_label}\n今日各账户无成交记录。")
        # 多账户合并一条，避免刷屏；过长则拆分
        chunks: list[str] = []
        buf = ""
        for block in lines:
            if not buf:
                buf = block
            elif len(buf) + len(block) + 2 < 1800:
                buf = buf + "\n\n" + block
            else:
                chunks.append(buf)
                buf = block
        if buf:
            chunks.append(buf)

    ok_all = True
    details = []
    for chunk in chunks:
        ok, detail = send_text(chunk)
        ok_all = ok_all and ok
        details.append(detail)
    return ok_all, "; ".join(details) if details else "empty"