DinQuant/backend_api_python/app/services/signal_notifier.py

"""
Strategy signal notification service.

This module implements per-strategy notification channels based on the frontend schema:

notification_config = {
  "channels": ["browser", "email", "phone", "telegram", "discord", "webhook"],
  "targets": {
    "email": "foo@example.com",
    "phone": "+15551234567",
    "telegram": "12345678 or @username",
    "discord": "https://discord.com/api/webhooks/...",
    "webhook": "https://example.com/webhook"
  }
}
"""

from __future__ import annotations

import hashlib
import hmac
import html
import json
import os
import smtplib
import time
from datetime import datetime, timezone
from email.message import EmailMessage
from typing import Any, Dict, List, Optional, Tuple
from zoneinfo import ZoneInfo

import requests

from app.utils.db import get_db_connection
from app.utils.logger import get_logger

logger = get_logger(__name__)


def _as_list(value: Any) -> List[str]:
    if value is None:
        return []
    if isinstance(value, (list, tuple)):
        return [str(x).strip() for x in value if str(x).strip()]
    s = str(value).strip()
    if not s:
        return []
    # Allow comma-separated inputs.
    if "," in s:
        return [x.strip() for x in s.split(",") if x.strip()]
    return [s]


def _safe_json(value: Any) -> Dict[str, Any]:
    if isinstance(value, dict):
        return value
    if isinstance(value, str) and value.strip():
        try:
            obj = json.loads(value)
            return obj if isinstance(obj, dict) else {}
        except Exception:
            return {}
    return {}


def _signal_meta(signal_type: str) -> Dict[str, str]:
    st = (signal_type or "").strip().lower()
    action = "signal"
    if st.startswith("open_"):
        action = "open"
    elif st.startswith("add_"):
        action = "add"
    elif st.startswith("close_"):
        action = "close"
    elif st.startswith("reduce_"):
        action = "reduce"

    side = "long" if "long" in st else ("short" if "short" in st else "")
    return {"action": action, "side": side, "type": st}


def _load_user_timezone_for_strategy(strategy_id: int) -> str:
    try:
        sid = int(strategy_id)
    except Exception:
        return ""
    try:
        with get_db_connection() as db:
            cur = db.cursor()
            cur.execute(
                """
                SELECT COALESCE(u.timezone, '') AS tz
                FROM qd_strategies_trading s
                JOIN qd_users u ON u.id = s.user_id
                WHERE s.id = ?
                """,
                (sid,),
            )
            row = cur.fetchone() or {}
            cur.close()
        return str(row.get("tz") or "").strip()
    except Exception:
        return ""


def _utc_ts_to_user_display(now: int, user_timezone: str) -> Tuple[str, str, str]:
    """Return (utc_iso, display_local_str, label_for_plaintext)."""
    iso = datetime.fromtimestamp(int(now), tz=timezone.utc).isoformat()
    utz = (user_timezone or "").strip()
    if not utz:
        return iso, iso, "Time (UTC)"
    try:
        dt = datetime.fromtimestamp(int(now), tz=timezone.utc).astimezone(ZoneInfo(utz))
        return iso, dt.strftime("%Y-%m-%d %H:%M:%S"), f"Time ({utz})"
    except Exception:
        return iso, iso, "Time (UTC)"


def _fmt_float(value: Any, *, max_decimals: int = 10) -> str:
    try:
        v = float(value or 0.0)
    except Exception:
        v = 0.0
    s = f"{v:.{int(max_decimals)}f}"
    s = s.rstrip("0").rstrip(".")
    return s or "0"


class SignalNotifier:
    """
    Notify signal events across channels.

    Notification configuration instructions:
    - Users configure their own notification settings (telegram_bot_token, telegram_chat_id, email, etc.) in the personal center
    - When creating policies/monitors, the system automatically uses user-configured notification targets

    Public service configuration (configured by administrator in system settings):
    - SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD, SMTP_FROM, SMTP_USE_TLS
      (Mail service, shared by all users)
    - TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER
      (SMS service, shared by all users)

    Optional environment variables:
    - SIGNAL_NOTIFY_TIMEOUT_SEC: HTTP timeout (default: 6)
    """

    def __init__(self) -> None:
        try:
            self.timeout_sec = float(os.getenv("SIGNAL_NOTIFY_TIMEOUT_SEC") or "6")
        except Exception:
            self.timeout_sec = 6.0

        # Public SMTP configuration (configured by administrator in system settings)
        self.smtp_host = (os.getenv("SMTP_HOST") or "").strip()
        try:
            self.smtp_port = int(os.getenv("SMTP_PORT") or "587")
        except Exception:
            self.smtp_port = 587
        self.smtp_user = (os.getenv("SMTP_USER") or "").strip()
        self.smtp_password = (os.getenv("SMTP_PASSWORD") or "").strip()
        self.smtp_from = (os.getenv("SMTP_FROM") or self.smtp_user or "").strip()
        self.smtp_use_tls = (os.getenv("SMTP_USE_TLS") or "true").strip().lower() == "true"
        # Some providers require implicit SSL (port 465). Support it via SMTP_USE_SSL.
        self.smtp_use_ssl = (os.getenv("SMTP_USE_SSL") or "").strip().lower() == "true"

        self.twilio_sid = (os.getenv("TWILIO_ACCOUNT_SID") or "").strip()
        self.twilio_token = (os.getenv("TWILIO_AUTH_TOKEN") or "").strip()
        self.twilio_from = (os.getenv("TWILIO_FROM_NUMBER") or "").strip()

    def notify_signal(
        self,
        *,
        strategy_id: int,
        strategy_name: str,
        symbol: str,
        signal_type: str,
        price: float = 0.0,
        stake_amount: float = 0.0,
        direction: str = "long",
        notification_config: Optional[Dict[str, Any]] = None,
        extra: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Dict[str, Any]]:
        cfg = _safe_json(notification_config or {})
        channels = _as_list(cfg.get("channels"))
        if not channels:
            channels = ["browser"]

        targets = _safe_json(cfg.get("targets") or {})
        extra = extra if isinstance(extra, dict) else {}

        user_tz = _load_user_timezone_for_strategy(int(strategy_id))
        payload = self._build_payload(
            strategy_id=strategy_id,
            strategy_name=strategy_name,
            symbol=symbol,
            signal_type=signal_type,
            price=price,
            stake_amount=stake_amount,
            direction=direction,
            extra=extra,
            user_timezone=user_tz,
        )
        rendered = self._render_messages(payload)
        title = rendered.get("title") or ""
        message_plain = rendered.get("plain") or ""

        results: Dict[str, Dict[str, Any]] = {}
        for ch in channels:
            c = (ch or "").strip().lower()
            if not c:
                continue
            try:
                if c == "browser":
                    ok, err = self._notify_browser(
                        strategy_id=strategy_id,
                        symbol=symbol,
                        signal_type=signal_type,
                        channels=channels,
                        title=title,
                        message=message_plain,
                        payload=payload,
                    )
                elif c == "webhook":
                    url = (targets.get("webhook") or "").strip()
                    ok, err = self._notify_webhook(
                        url=url,
                        payload=payload,
                        headers_override=(targets.get("webhook_headers") or targets.get("webhookHeaders") or None),
                        token_override=(targets.get("webhook_token") or targets.get("webhookToken") or None),
                        signing_secret_override=(
                            targets.get("webhook_signing_secret") or targets.get("webhookSigningSecret") or None
                        ),
                    )
                elif c == "discord":
                    url = (targets.get("discord") or "").strip()
                    ok, err = self._notify_discord(url=url, payload=payload, fallback_text=message_plain)
                elif c == "telegram":
                    chat_id = (targets.get("telegram") or "").strip()
                    # User's token takes priority, then falls back to env TELEGRAM_BOT_TOKEN.
                    token_override = ""
                    try:
                        token_override = str(
                            targets.get("telegram_bot_token")
                            or targets.get("telegram_token")
                            or cfg.get("telegram_bot_token")
                            or cfg.get("telegram_token")
                            or ""
                        ).strip()
                    except Exception:
                        token_override = ""
                    ok, err = self._notify_telegram(
                        chat_id=chat_id,
                        text=rendered.get("telegram_html") or message_plain,
                        token_override=token_override,
                        parse_mode="HTML",
                    )
                elif c == "email":
                    to_email = (targets.get("email") or "").strip()
                    ok, err = self._notify_email(
                        to_email=to_email,
                        subject=title,
                        body_text=message_plain,
                        body_html=rendered.get("email_html") or "",
                    )
                elif c == "phone":
                    to_phone = (targets.get("phone") or "").strip()
                    ok, err = self._notify_phone(to_phone=to_phone, body=message_plain)
                else:
                    ok, err = False, f"unsupported_channel:{c}"
            except Exception as e:
                ok, err = False, str(e)

            results[c] = {"ok": bool(ok), "error": (err or "")}
            if not ok and c in ("webhook", "discord"):
                # Keep logs high-signal and avoid leaking full URLs (webhook URLs contain secrets).
                logger.info(
                    f"notify failed: channel={c} strategy_id={strategy_id} symbol={symbol} signal={signal_type} err={err}"
                )

        return results

    def _build_payload(
        self,
        *,
        strategy_id: int,
        strategy_name: str,
        symbol: str,
        signal_type: str,
        price: float,
        stake_amount: float,
        direction: str,
        extra: Dict[str, Any],
        user_timezone: str = "",
    ) -> Dict[str, Any]:
        now = int(time.time())
        iso, disp, tlab = _utc_ts_to_user_display(now, user_timezone)
        meta = _signal_meta(signal_type)

        pending_id = None
        try:
            pending_id = int((extra or {}).get("pending_order_id") or 0) or None
        except Exception:
            pending_id = None

        return {
            "event": "qd.signal",
            "version": 1,
            "timestamp": now,
            "timestamp_iso": iso,
            "timestamp_display": disp,
            "time_label": tlab,
            "strategy": {
                "id": int(strategy_id),
                "name": str(strategy_name or ""),
            },
            "instrument": {
                "symbol": str(symbol or ""),
            },
            "signal": {
                "type": meta.get("type") or str(signal_type or ""),
                "action": meta.get("action") or "signal",
                "side": meta.get("side") or "",
                "direction": str(direction or ""),
            },
            "order": {
                "ref_price": float(price or 0.0),
                "stake_amount": float(stake_amount or 0.0),
            },
            "trace": {
                "pending_order_id": pending_id,
                "mode": str((extra or {}).get("mode") or ""),
            },
            "extra": extra or {},
        }

    def _render_messages(self, payload: Dict[str, Any]) -> Dict[str, str]:
        strategy = (payload or {}).get("strategy") or {}
        instrument = (payload or {}).get("instrument") or {}
        sig = (payload or {}).get("signal") or {}
        order = (payload or {}).get("order") or {}
        trace = (payload or {}).get("trace") or {}

        symbol = str(instrument.get("symbol") or "")
        stype = str(sig.get("type") or "")
        action = str(sig.get("action") or "").upper()
        side = str(sig.get("side") or "").upper()
        title = f"QD Signal | {symbol} | {action} {side}".strip()

        price_s = _fmt_float(order.get("ref_price") or 0.0, max_decimals=10)
        stake_s = _fmt_float(order.get("stake_amount") or 0.0, max_decimals=12)
        pending_id = int(trace.get("pending_order_id") or 0) if trace.get("pending_order_id") else 0
        mode = str(trace.get("mode") or "")
        ts_iso = str(payload.get("timestamp_iso") or "")
        ts_disp = str(payload.get("timestamp_display") or "") or ts_iso
        ts_lbl = str(payload.get("time_label") or "Time")

        plain_lines = [
            "QuantDinger Signal",
            f"Strategy: {strategy.get('name') or ''} (#{int(strategy.get('id') or 0)})",
            f"Symbol: {symbol}",
            f"Signal: {stype}",
            f"Price: {price_s}",
            f"Stake: {stake_s}",
        ]
        if pending_id:
            plain_lines.append(f"PendingOrder: {pending_id}")
        if mode:
            plain_lines.append(f"Mode: {mode}")
        if ts_disp:
            plain_lines.append(f"{ts_lbl}: {ts_disp}")

        # Telegram (HTML) message. Escape all dynamic values.
        t_strategy = f"{strategy.get('name') or ''} (#{int(strategy.get('id') or 0)})"
        telegram_lines = [
            "<b>QuantDinger Signal</b>",
            "",
            f"<b>Strategy</b>: <code>{html.escape(str(t_strategy))}</code>",
            f"<b>Symbol</b>: <code>{html.escape(symbol)}</code>",
            f"<b>Signal</b>: <code>{html.escape(stype)}</code>",
            f"<b>Price</b>: <code>{html.escape(price_s)}</code>",
            f"<b>Stake</b>: <code>{html.escape(stake_s)}</code>",
        ]
        if pending_id:
            telegram_lines.append(f"<b>PendingOrder</b>: <code>{pending_id}</code>")
        if mode:
            telegram_lines.append(f"<b>Mode</b>: <code>{html.escape(mode)}</code>")
        if ts_disp:
            telegram_lines.append(f"<b>{html.escape(ts_lbl)}</b>: <code>{html.escape(ts_disp)}</code>")
        telegram_html = "\n".join([x for x in telegram_lines if x is not None])

        # Email (HTML) message. Keep inline CSS for maximum compatibility.
        email_html = self._build_email_html(
            title_text="QuantDinger Signal",
            strategy_text=t_strategy,
            symbol=symbol,
            signal_type=stype,
            price_text=price_s,
            stake_text=stake_s,
            pending_id=pending_id or None,
            mode_text=mode or "",
            timestamp_display=ts_disp or "",
            time_row_label=ts_lbl or "Time",
        )

        return {
            "title": title,
            "plain": "\n".join(plain_lines),
            "telegram_html": telegram_html,
            "email_html": email_html,
        }

    def _build_email_html(
        self,
        *,
        title_text: str,
        strategy_text: str,
        symbol: str,
        signal_type: str,
        price_text: str,
        stake_text: str,
        pending_id: Optional[int],
        mode_text: str,
        timestamp_display: str,
        time_row_label: str,
    ) -> str:
        def esc(s: Any) -> str:
            return html.escape(str(s or ""))

        rows: List[Tuple[str, str]] = [
            ("Strategy", strategy_text),
            ("Symbol", symbol),
            ("Signal", signal_type),
            ("Price", price_text),
            ("Stake", stake_text),
        ]
        if pending_id:
            rows.append(("PendingOrder", str(int(pending_id))))
        if mode_text:
            rows.append(("Mode", mode_text))
        if timestamp_display:
            rows.append((time_row_label or "Time", timestamp_display))

        tr_html = "\n".join(
            [
                (
                    "<tr>"
                    "<td style='padding:10px 12px;border-top:1px solid #eaecef;color:#57606a;width:160px;'>"
                    f"{esc(k)}"
                    "</td>"
                    '<td style=\'padding:10px 12px;border-top:1px solid #eaecef;color:#24292f;font-family:ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", "Courier New", monospace;\'>'
                    f"{esc(v)}"
                    "</td>"
                    "</tr>"
                )
                for (k, v) in rows
            ]
        )

        return f"""\
<!doctype html>
<html>
  <body style="margin:0;padding:0;background:#f6f8fa;">
    <div style="max-width:640px;margin:0 auto;padding:24px;">
      <div style="background:#111827;color:#ffffff;padding:16px 18px;border-radius:12px 12px 0 0;">
        <div style="font-size:16px;letter-spacing:0.2px;font-weight:600;">{esc(title_text)}</div>
        <div style="margin-top:6px;font-size:12px;color:#d1d5db;">{esc(timestamp_display) if timestamp_display else ""}</div>
      </div>
      <div style="background:#ffffff;border:1px solid #eaecef;border-top:0;border-radius:0 0 12px 12px;overflow:hidden;">
        <table cellpadding="0" cellspacing="0" style="width:100%;border-collapse:collapse;">
          {tr_html}
        </table>
        <div style="padding:14px 16px;color:#6e7781;font-size:12px;border-top:1px solid #eaecef;">
          Generated by QuantDinger
        </div>
      </div>
    </div>
  </body>
</html>
"""

    def _notify_browser(
        self,
        *,
        strategy_id: Optional[int] = None,
        symbol: str,
        signal_type: str,
        channels: List[str],
        title: str,
        message: str,
        payload: Dict[str, Any],
        user_id: int = None,
    ) -> Tuple[bool, str]:
        try:
            # Get user_id from strategy if not provided
            if user_id is None:
                if strategy_id is not None:
                    try:
                        with get_db_connection() as db:
                            cur = db.cursor()
                            cur.execute("SELECT user_id FROM qd_strategies_trading WHERE id = ?", (int(strategy_id),))
                            row = cur.fetchone()
                            cur.close()
                        user_id = int((row or {}).get("user_id") or 1)
                    except Exception:
                        user_id = 1
                else:
                    user_id = 1
            sid = None if strategy_id is None else int(strategy_id)
            with get_db_connection() as db:
                cur = db.cursor()
                cur.execute(
                    """
                    INSERT INTO qd_strategy_notifications
                    (user_id, strategy_id, symbol, signal_type, channels, title, message, payload_json, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())
                    """,
                    (
                        int(user_id),
                        sid,
                        str(symbol or ""),
                        str(signal_type or ""),
                        ",".join([str(c) for c in (channels or [])]),
                        str(title or ""),
                        str(message or ""),
                        json.dumps(payload or {}, ensure_ascii=False),
                    ),
                )
                db.commit()
                cur.close()
            return True, ""
        except Exception as e:
            logger.warning(f"browser notify persist failed: {e}")
            logger.exception("browser.error")
            return False, str(e)

    def _notify_webhook(
        self,
        *,
        url: str,
        payload: Dict[str, Any],
        headers_override: Any = None,
        token_override: Any = None,
        signing_secret_override: Any = None,
    ) -> Tuple[bool, str]:
        """
        Generic webhook delivery.

        User configuration in the profile center:
        - webhook_url: webhook URL
        - webhook_token: optional Bearer token

        Supported features:
        - Custom headers: notification_config.targets.webhook_headers
        - Bearer Token: notification_config.targets.webhook_token
        - Signature verification: notification_config.targets.webhook_signing_secret
        - Automatic retry: retry once on 429/5xx
        """
        if not url:
            return False, "missing_webhook_url"
        if not (str(url).startswith("http://") or str(url).startswith("https://")):
            return False, "invalid_webhook_url"

        headers: Dict[str, str] = {
            "Content-Type": "application/json",
            "User-Agent": "QuantDinger/1.0 (+https://www.quantdinger.com)",
        }

        # Per-strategy header overrides (optional)
        wh = headers_override
        if isinstance(wh, str) and wh.strip():
            try:
                obj = json.loads(wh)
                wh = obj if isinstance(obj, dict) else None
            except Exception:
                wh = None
        if isinstance(wh, dict):
            for k, v in wh.items():
                kk = str(k or "").strip()
                if not kk:
                    continue
                headers[kk] = str(v if v is not None else "")

        # Auth (user's token from notification_config.targets.webhook_token)
        tok = str(token_override or "").strip()
        if tok and "Authorization" not in headers:
            headers["Authorization"] = f"Bearer {tok}"

        # Optional signing secret (per-strategy override, else env)
        signing_secret = (
            str(signing_secret_override or "").strip() or (os.getenv("SIGNAL_WEBHOOK_SIGNING_SECRET") or "").strip()
        )
        if signing_secret:
            try:
                ts = str(int(time.time()))
                body = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
                sig_base = (ts + ".").encode("utf-8") + body
                sig = hmac.new(signing_secret.encode("utf-8"), sig_base, hashlib.sha256).hexdigest()
                headers["X-QD-Timestamp"] = ts
                headers["X-QD-Signature"] = sig

                # Send raw bytes so signature matches what we sign.
                def _post_once(timeout: float) -> requests.Response:
                    return requests.post(url, data=body, headers=headers, timeout=timeout)
            except Exception as e:
                return False, f"webhook_signing_failed:{e}"
        else:

            def _post_once(timeout: float) -> requests.Response:
                return requests.post(url, json=payload, headers=headers, timeout=timeout)

        # Post with minimal retry on 429/5xx
        try:
            resp = _post_once(self.timeout_sec)
            if 200 <= resp.status_code < 300:
                return True, ""
            if resp.status_code in (429, 500, 502, 503, 504):
                try:
                    time.sleep(1.0)
                except Exception:
                    pass
                resp2 = _post_once(self.timeout_sec)
                if 200 <= resp2.status_code < 300:
                    return True, ""
                return False, f"http_{resp2.status_code}:{(resp2.text or '')[:300]}"
            return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
        except Exception as e:
            logger.exception("webhook.error")
            return False, str(e)

    def _notify_discord(self, *, url: str, payload: Dict[str, Any], fallback_text: str) -> Tuple[bool, str]:
        if not url:
            return False, "missing_discord_webhook_url"
        if not (str(url).startswith("http://") or str(url).startswith("https://")):
            return False, "invalid_discord_webhook_url"

        strategy = (payload or {}).get("strategy") or {}
        instrument = (payload or {}).get("instrument") or {}
        sig = (payload or {}).get("signal") or {}
        order = (payload or {}).get("order") or {}
        trace = (payload or {}).get("trace") or {}

        action = str(sig.get("action") or "").lower()
        color = 0x3498DB
        if action in ("open", "add"):
            color = 0x2ECC71
        if action in ("close", "reduce"):
            color = 0xE74C3C

        embed: Dict[str, Any] = {
            "title": "QuantDinger Signal",
            "color": int(color),
            "fields": [
                {
                    "name": "Strategy",
                    "value": f"{strategy.get('name') or ''} (#{int(strategy.get('id') or 0)})",
                    "inline": True,
                },
                {"name": "Symbol", "value": str(instrument.get("symbol") or ""), "inline": True},
                {"name": "Signal", "value": str(sig.get("type") or ""), "inline": False},
                {"name": "Price", "value": str(float(order.get("ref_price") or 0.0)), "inline": True},
                {"name": "Stake", "value": str(float(order.get("stake_amount") or 0.0)), "inline": True},
            ],
        }
        if payload.get("timestamp_iso"):
            embed["timestamp"] = str(payload.get("timestamp_iso") or "")
        if trace.get("pending_order_id"):
            embed["footer"] = {"text": f"pending_order_id={int(trace.get('pending_order_id'))}"}
        headers = {
            "Content-Type": "application/json",
            "User-Agent": "QuantDinger/1.0 (+https://www.quantdinger.com)",
        }

        def _post(payload_json: Dict[str, Any]) -> requests.Response:
            return requests.post(url, json=payload_json, headers=headers, timeout=self.timeout_sec)

        try:
            resp = _post({"content": "", "embeds": [embed]})
            if 200 <= resp.status_code < 300:
                return True, ""

            # Rate limit: retry once if Discord asks us to.
            if resp.status_code == 429:
                try:
                    data = resp.json() if resp is not None else {}
                    retry_after = float((data or {}).get("retry_after") or 1.0)
                    time.sleep(min(max(retry_after, 0.5), 3.0))
                except Exception:
                    try:
                        time.sleep(1.0)
                    except Exception:
                        pass
                resp_retry = _post({"content": "", "embeds": [embed]})
                if 200 <= resp_retry.status_code < 300:
                    return True, ""
                resp = resp_retry

            # Fallback: plain text (some servers reject embeds)
            try:
                resp2 = _post({"content": str(fallback_text or "")[:1900]})
                if 200 <= resp2.status_code < 300:
                    return True, ""
                # If fallback also fails, return the original error (more useful than fallback sometimes).
            except Exception:
                pass
            return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
        except Exception as e:
            logger.exception("discord.error")
            return False, str(e)

    def _notify_telegram(
        self,
        *,
        chat_id: str,
        text: str,
        token_override: str = "",
        parse_mode: str = "",
    ) -> Tuple[bool, str]:
        # Users must configure their own telegram_bot_token in the personal center
        token = (token_override or "").strip()
        if not token:
            return False, "missing_telegram_bot_token (please configure Telegram Bot Token in the personal center)"
        if not chat_id:
            return False, "missing_telegram_chat_id"
        url = f"https://api.telegram.org/bot{token}/sendMessage"
        try:
            data: Dict[str, Any] = {
                "chat_id": chat_id,
                "text": str(text or "")[:3900],
                "disable_web_page_preview": True,
            }
            if (parse_mode or "").strip():
                data["parse_mode"] = str(parse_mode).strip()
            resp = requests.post(
                url,
                data=data,
                timeout=self.timeout_sec,
            )
            if 200 <= resp.status_code < 300:
                return True, ""
            return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
        except Exception as e:
            logger.exception("telegram.error")
            return False, str(e)

    def _notify_email(self, *, to_email: str, subject: str, body_text: str, body_html: str = "") -> Tuple[bool, str]:
        if not to_email:
            logger.warning("email.skip: missing recipient (to_email empty)")
            return False, "missing_email_target"
        if not self.smtp_host:
            logger.warning(
                "email.skip: SMTP_HOST not configured (set in system env / admin Email settings); "
                "test notification and all outbound mail require it"
            )
            return False, "missing_SMTP_HOST"
        if not self.smtp_from:
            logger.warning("email.skip: SMTP_FROM not configured (usually same as SMTP_USER or a verified sender)")
            return False, "missing_SMTP_FROM"

        msg = EmailMessage()
        msg["From"] = self.smtp_from
        msg["To"] = to_email
        msg["Subject"] = str(subject or "Signal")
        msg.set_content(str(body_text or ""))
        if (body_html or "").strip():
            msg.add_alternative(str(body_html or ""), subtype="html")

        try:
            # Heuristic: if port is 465 and SMTP_USE_SSL is not explicitly set, assume SSL.
            use_ssl = bool(self.smtp_use_ssl) or int(self.smtp_port or 0) == 465
            smtp_timeout = max(self.timeout_sec, 20)
            if use_ssl:
                with smtplib.SMTP_SSL(self.smtp_host, self.smtp_port, timeout=smtp_timeout) as server:
                    server.ehlo()
                    if self.smtp_user and self.smtp_password:
                        server.login(self.smtp_user, self.smtp_password)
                    server.send_message(msg)
            else:
                with smtplib.SMTP(self.smtp_host, self.smtp_port, timeout=smtp_timeout) as server:
                    server.ehlo()
                    if self.smtp_use_tls:
                        server.starttls()
                        server.ehlo()
                    if self.smtp_user and self.smtp_password:
                        server.login(self.smtp_user, self.smtp_password)
                    server.send_message(msg)
            return True, ""
        except Exception as e:
            logger.exception("email.error")
            return False, str(e)

    def _notify_phone(self, *, to_phone: str, body: str) -> Tuple[bool, str]:
        # Optional provider: Twilio via REST (no extra dependency).
        if not to_phone:
            return False, "missing_phone_target"
        if not (self.twilio_sid and self.twilio_token and self.twilio_from):
            return False, "missing_TWILIO_config"
        url = f"https://api.twilio.com/2010-04-01/Accounts/{self.twilio_sid}/Messages.json"
        data = {"To": to_phone, "From": self.twilio_from, "Body": str(body or "")[:1500]}
        try:
            resp = requests.post(url, data=data, auth=(self.twilio_sid, self.twilio_token), timeout=self.timeout_sec)
            if 200 <= resp.status_code < 300:
                return True, ""
            return False, f"http_{resp.status_code}:{(resp.text or '')[:300]}"
        except Exception as e:
            logger.exception("phone.error")
            return False, str(e)

    def send_profile_test_notifications(
        self,
        *,
        user_id: int,
        channels: List[str],
        targets: Dict[str, Any],
        language: str = "en-US",
    ) -> Dict[str, Dict[str, Any]]:
        """
        Send a short test message to each selected channel (profile / notification settings).
        Used by POST /api/users/notification-settings/test.
        """
        lang = (language or "en-US").strip().lower()
        zh = lang.startswith("zh")
        title = "QuantDinger 通知测试" if zh else "QuantDinger notification test"
        plain = (
            "这是一条来自 QuantDinger 个人中心「通知设置」的测试消息。若您收到本条消息，说明该渠道配置正确。"
            if zh
            else "This is a test message from QuantDinger profile notification settings. "
            "If you received this, the channel is configured correctly."
        )
        html_body = f"<p>{html.escape(plain)}</p>"
        telegram_html = f"<b>{html.escape(title)}</b>\n\n{html.escape(plain)}"

        now = int(time.time())
        iso = datetime.now(timezone.utc).isoformat()
        test_payload: Dict[str, Any] = {
            "event": "qd.profile_test",
            "version": 1,
            "timestamp": now,
            "timestamp_iso": iso,
            "strategy": {"id": 0, "name": "Profile Test"},
            "instrument": {"symbol": "TEST"},
            "signal": {"type": "profile_test", "action": "test", "side": ""},
            "order": {"ref_price": 0.0, "stake_amount": 0.0},
            "trace": {},
            "extra": {"kind": "profile_test"},
        }

        results: Dict[str, Dict[str, Any]] = {}
        ch_list = _as_list(channels)
        if not ch_list:
            ch_list = ["browser"]

        for ch in ch_list:
            c = (ch or "").strip().lower()
            if not c:
                continue
            ok, err = False, ""
            try:
                if c == "browser":
                    ok, err = self._notify_browser(
                        strategy_id=None,
                        symbol="TEST",
                        signal_type="profile_test",
                        channels=ch_list,
                        title=title,
                        message=html_body,
                        payload=test_payload,
                        user_id=int(user_id),
                    )
                elif c == "telegram":
                    chat_id = str((targets or {}).get("telegram") or "").strip()
                    token_override = str(
                        (targets or {}).get("telegram_bot_token") or (targets or {}).get("telegram_token") or ""
                    ).strip()
                    ok, err = self._notify_telegram(
                        chat_id=chat_id,
                        text=telegram_html,
                        token_override=token_override,
                        parse_mode="HTML",
                    )
                elif c == "email":
                    to_email = str((targets or {}).get("email") or "").strip()
                    ok, err = self._notify_email(
                        to_email=to_email,
                        subject=title,
                        body_text=plain,
                        body_html=html_body,
                    )
                elif c == "phone":
                    to_phone = str((targets or {}).get("phone") or "").strip()
                    ok, err = self._notify_phone(to_phone=to_phone, body=f"{title}\n\n{plain}")
                elif c == "discord":
                    url = str((targets or {}).get("discord") or "").strip()
                    ok, err = self._notify_discord(url=url, payload=test_payload, fallback_text=f"{title}\n\n{plain}")
                elif c == "webhook":
                    url = str((targets or {}).get("webhook") or "").strip()
                    tok = str((targets or {}).get("webhook_token") or "").strip()
                    wh_payload = {
                        "event": "qd.profile_test",
                        "title": title,
                        "message": plain,
                        "timestamp": now,
                        "timestamp_iso": iso,
                    }
                    ok, err = self._notify_webhook(url=url, payload=wh_payload, token_override=tok or None)
                else:
                    ok, err = False, f"unsupported_channel:{c}"
            except Exception as e:
                ok, err = False, str(e)
            results[c] = {"ok": bool(ok), "error": (err or "")}

        return results