"""监控规则 API 路由 — HTTP 请求 → 调用 monitor_rules 模块 → 同步引擎内存态。

只做胶水: 校验 → 持久化 → 失效引擎内存态。不含评估逻辑。
"""
from __future__ import annotations

from pathlib import Path

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel

from app.strategy import monitor_rules

router = APIRouter(prefix="/api/monitor-rules", tags=["monitor-rules"])


def _data_dir(request: Request) -> Path:
    return request.app.state.repo.store.data_dir


def _sync_engine(request: Request) -> None:
    """保存/删除后,把最新规则集 reload 到引擎内存态。"""
    engine = getattr(request.app.state, "monitor_engine", None)
    if engine is not None:
        rules = monitor_rules.load_all(_data_dir(request))
        engine.set_rules(rules)


# ── Pydantic 模型 ───────────────────────────────────────
class ConditionModel(BaseModel):
    field: str
    op: str            # truth | > >= < <= == !=
    value: float | None = None   # op 非 truth 时必填


class RuleModel(BaseModel):
    id: str
    name: str
    enabled: bool = True
    type: str          # strategy | signal | price | market
    scope: str = "symbols"   # symbols | all | sector
    symbols: list[str] = []
    sector: str | None = None
    strategy_id: str | None = None
    direction: str = "entry"  # entry | exit | both
    conditions: list[ConditionModel] = []
    logic: str = "and"        # and | or
    cooldown_seconds: int = 3600
    severity: str = "info"    # info | warn | critical
    webhook_url: str = ""     # Webhook 推送地址 (推送到 QMT 等外部软件, 待定)
    webhook_enabled: bool = False
    message: str = ""
    # ladder 专属 (连板梯队封单监控)
    metric: str = "sealed_vol"   # sealed_vol=封单量(手) | sealed_amount=封单额(元)
    threshold: float = 0         # 封单 <= 此值时报警 (原始单位: 量=手, 额=元)


# ── 字段选项 ─────────────────────────────────────────────
@router.get("/options")
def get_options(request: Request):
    """返回可选字段、信号列、运算符、枚举,供前端表单使用。"""
    from app.indicators.pipeline import ENRICHED_COLUMNS
    from app.strategy.custom_signals import ALLOWED_FIELDS, load_all as load_csg

    # 阈值字段 (带中文标签)
    threshold_fields = [
        {"key": f, "label": ENRICHED_COLUMNS.get(f, f)}
        for f in sorted(ALLOWED_FIELDS)
    ]
    # 内置信号列 (布尔, 用于 op=truth)
    builtin_signals = [
        {"key": k, "label": v}
        for k, v in ENRICHED_COLUMNS.items()
        if k.startswith("signal_")
    ]
    # 自定义信号列 (csg_)
    custom_sigs = []
    try:
        for cs in load_csg(_data_dir(request)):
            if cs.get("enabled") is not False:
                custom_sigs.append({
                    "key": f"csg_{cs['id']}",
                    "label": cs.get("name", cs["id"]),
                })
    except Exception:
        pass

    return {
        "threshold_fields": threshold_fields,
        "builtin_signals": builtin_signals,
        "custom_signals": custom_sigs,
        "operators": [">", ">=", "<", "<=", "==", "!="],
        "types": [
            {"key": "signal", "label": "个股信号"},
            {"key": "price", "label": "价格/涨跌"},
            {"key": "market", "label": "市场异动"},
            {"key": "strategy", "label": "策略监控"},
        ],
        "scopes": [
            {"key": "symbols", "label": "指定股票"},
            {"key": "all", "label": "全市场"},
            {"key": "sector", "label": "板块"},
        ],
        "logics": [
            {"key": "and", "label": "全部满足 (AND)"},
            {"key": "or", "label": "任一满足 (OR)"},
        ],
        "severities": [
            {"key": "info", "label": "普通"},
            {"key": "warn", "label": "警告"},
            {"key": "critical", "label": "重要"},
        ],
        "directions": [
            {"key": "entry", "label": "买入"},
            {"key": "exit", "label": "卖出"},
            {"key": "both", "label": "买卖都报"},
        ],
    }


# ── 列表 ───────────────────────────────────────────────
@router.get("")
def list_rules(request: Request):
    rules = monitor_rules.load_all(_data_dir(request))
    # 按 created_at 倒序
    rules.sort(key=lambda r: r.get("created_at", ""), reverse=True)
    return {"rules": rules}


# ── 新建 / 更新 ────────────────────────────────────────
@router.post("")
def save_rule(req: RuleModel, request: Request):
    rule = monitor_rules.normalize(req.model_dump())
    # 连板梯队封单监控 (type=ladder) 依赖五档盘口数据, 需 Pro+ (DEPTH5_BATCH 能力)。
    # 无能力时拒绝创建, 避免规则存了却永远无法触发。
    if rule.get("type") == "ladder":
        from app.tickflow.capabilities import Cap
        capset = getattr(request.app.state, "capabilities", None)
        if capset is None or not capset.has(Cap.DEPTH5_BATCH):
            raise HTTPException(
                status_code=403,
                detail="封单监控需要 Pro+ 套餐 (批量五档能力),请升级后在「设置」页配置",
            )
    # 编辑现有规则时, 保留原 created_at (避免按时间排序时位置跳动)
    existing = monitor_rules.load_one(_data_dir(request), rule["id"])
    if existing and existing.get("created_at"):
        rule["created_at"] = existing["created_at"]
    try:
        monitor_rules.validate(rule)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    monitor_rules.save_one(_data_dir(request), rule)
    _sync_engine(request)
    return {"ok": True, "rule": rule}


# ── 删除 ───────────────────────────────────────────────
@router.delete("/{rule_id}")
def delete_rule(rule_id: str, request: Request):
    if not monitor_rules.ID_RE.match(rule_id):
        raise HTTPException(status_code=400, detail="规则 id 非法")
    deleted = monitor_rules.delete_one(_data_dir(request), rule_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="规则不存在")
    _sync_engine(request)
    return {"ok": True}


# ── 演示数据生成 (仅 Dev 页用) ─────────────────────────

import time as _time
from datetime import datetime, timezone


def _demo_rule(rule_id: str, name: str, rtype: str, scope: str, symbols: list[str],
               conditions: list[dict], logic: str = "or", cooldown: int = 3600,
               severity: str = "info", message: str = "",
               strategy_id: str | None = None, direction: str = "entry") -> dict:
    rule = monitor_rules.normalize({
        "id": rule_id,
        "name": name,
        "type": rtype,
        "scope": scope,
        "symbols": symbols,
        "conditions": conditions,
        "logic": logic,
        "cooldown_seconds": cooldown,
        "severity": severity,
        "message": message,
        "enabled": True,
    })
    if rtype == "strategy":
        rule["strategy_id"] = strategy_id
        rule["direction"] = direction
    return rule


_DEMO_RULES_TEMPLATE = [
    ("个股信号 · 茅台放量突破", "signal", "symbols", ["600519.SH"],
     [{"field": "signal_volume_surge", "op": "truth"},
      {"field": "signal_n_day_high", "op": "truth"}], "or", "info"),
    ("个股信号 · 宁德金叉", "signal", "symbols", ["300750.SZ"],
     [{"field": "signal_ma_golden_5_20", "op": "truth"}], "or", "info"),
    ("价格 · 平安跌幅监控", "price", "symbols", ["000001.SZ"],
     [{"field": "change_pct", "op": "<", "value": -0.03}], "or", "warn", "warn"),
    ("价格 · 比亚迪RSI超卖", "price", "symbols", ["002594.SZ"],
     [{"field": "rsi_14", "op": "<", "value": 30}], "and", "warn", "warn"),
    ("市场异动 · 全市场涨停", "market", "all", [],
     [{"field": "signal_limit_up", "op": "truth"}], "or", "critical", "critical"),
    ("市场异动 · 全市场炸板", "market", "all", [],
     [{"field": "signal_broken_limit_up", "op": "truth"}], "or", "warn", "warn"),
    ("市场异动 · 跌幅超5%", "market", "all", [],
     [{"field": "change_pct", "op": "<", "value": -0.05}], "or", "warn", "warn"),
    ("个股信号 · 茅台跌破MA20", "signal", "symbols", ["600519.SH"],
     [{"field": "signal_ma20_breakdown", "op": "truth"}], "or", "info"),
]

# 策略类型单独声明 (格式不同: 含 strategy_id + direction)
_DEMO_STRATEGY_RULES: list[dict] = [
    {"name": "策略监控 · 趋势突破", "strategy_id": "trend_breakout", "direction": "entry"},
    {"name": "策略监控 · MACD金叉", "strategy_id": "macd_golden", "direction": "both"},
]


@router.post("/seed")
def seed_demo_rules(request: Request):
    """生成演示监控规则 (Dev 页用)。覆盖 signal/price/market/strategy 四类。"""
    ts = int(_time.time() * 1000)
    created = []
    i = 0
    for (name, rtype, scope, symbols, conditions, logic, severity, sev) in _DEMO_RULES_TEMPLATE:
        rule_id = f"demo_{ts}_{i}"
        rule = _demo_rule(rule_id, name, rtype, scope, symbols, conditions, logic, 3600, sev)
        monitor_rules.save_one(_data_dir(request), rule)
        created.append(rule_id)
        i += 1
    # 策略类型规则
    for sr in _DEMO_STRATEGY_RULES:
        rule_id = f"demo_{ts}_{i}"
        rule = _demo_rule(
            rule_id, sr["name"], "strategy", "all", [], [], "and", 3600, "info",
            strategy_id=sr["strategy_id"], direction=sr.get("direction", "entry"),
        )
        monitor_rules.save_one(_data_dir(request), rule)
        created.append(rule_id)
        i += 1
    _sync_engine(request)
    return {"ok": True, "generated": len(created), "ids": created}


# ── 封单监控模拟触发 (Dev 调试用) ─────────────────────
@router.post("/test-ladder")
def test_ladder(request: Request):
    """模拟触发所有 ladder 规则, 返回命中结果 (不落盘、不推送飞书)。

    用当前 depth_service 的封单数据 + enriched 最新日 close 构造 mock DataFrame,
    跑 _evaluate_ladder 判断哪些规则会触发。供 Dev 页面调试验证。
    """
    import polars as pl

    repo = request.app.state.repo
    depth_svc = getattr(request.app.state, "depth_service", None)
    engine = getattr(request.app.state, "monitor_engine", None)

    if not depth_svc:
        raise HTTPException(status_code=503, detail="depth 服务未初始化")
    if not engine or not engine.has_rule_type("ladder"):
        raise HTTPException(status_code=400, detail="无 ladder 类型监控规则")

    # 最新交易日
    latest = repo.enriched_latest_date()
    if not latest:
        raise HTTPException(status_code=400, detail="无 enriched 数据")

    # 取涨停+跌停封单 {symbol: vol}
    sealed: dict[str, int] = {}
    for is_down in (False, True):
        m = depth_svc.get_sealed_map(latest, is_down=is_down)
        for sym, info in m.items():
            vol = (info or {}).get("vol")
            if vol and vol > 0:
                sealed[sym] = vol

    if not sealed:
        raise HTTPException(status_code=400, detail="无封单数据 (depth 未拉取或无涨停/跌停股)")

    # 取这些 symbol 的 close (算封单额用)
    enriched_today, _ = repo.get_enriched_latest()
    cols = ["symbol", "close", "change_pct"]
    avail = [c for c in cols if c in enriched_today.columns]
    mock = enriched_today.select(avail).filter(pl.col("symbol").is_in(list(sealed.keys())))

    # 注入 _sealed_vol
    sealed_df = pl.DataFrame({
        "symbol": list(sealed.keys()),
        "_sealed_vol": list(sealed.values()),
    })
    mock = mock.join(sealed_df, on="symbol", how="inner")

    # 取所有 ladder 规则, 逐条纯条件判断 (绕过引擎 cooldown, 不污染 _last_fire)
    ladder_rules = [r for r in engine.rules.values() if r.get("type") == "ladder" and r.get("enabled", True)]
    all_events = []
    not_triggered = []

    for rule in ladder_rules:
        syms = rule.get("symbols", [])
        sym = syms[0] if syms else None
        metric = rule.get("metric", "sealed_vol")
        thr = rule.get("threshold", 0)
        direction = rule.get("direction", "up")
        warn_label = "炸板预警" if direction == "up" else "翘板预警"

        # 取该 symbol 的封单数据
        cur_vol = sealed.get(sym) if sym else None
        row = mock.filter(pl.col("symbol") == sym) if sym else mock.clear()
        cur_close = row["close"][0] if len(row) and "close" in row.columns else None
        cur_amt = (cur_vol * 100 * cur_close) if (cur_vol and cur_close) else None
        cur_val = cur_amt if metric == "sealed_amount" else cur_vol

        # 条件判断: 封单 > 0 且 比较值 <= 阈值
        if cur_val is not None and cur_val > 0 and cur_val <= thr:
            if metric == "sealed_amount":
                sv_text = f"{cur_val / 1e4:.0f}万元"
                th_text = f"{thr / 1e4:.0f}万元"
            else:
                sv_text = f"{cur_val:,.0f} 手"
                th_text = f"{thr:,.0f} 手"
            all_events.append({
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "symbol": sym,
                "name": sym,
                "type": warn_label,
                "message": f"{warn_label} · 封单 {sv_text} ≤ {th_text}",
                "severity": rule.get("severity", "warn"),
                "sealed_value": cur_val,
                "sealed_metric": metric,
                "current_sealed_vol": cur_vol,
                "current_sealed_amount": cur_amt,
            })
        else:
            reason = "封单数据缺失" if cur_val is None else (
                f"封单 {cur_val:,.0f} > 阈值 {thr:,.0f}" if cur_val > thr else "封单为 0"
            )
            not_triggered.append({
                "rule_id": rule["id"],
                "rule_name": rule.get("name", ""),
                "symbol": sym,
                "metric": metric,
                "threshold": thr,
                "current_value": cur_val,
                "current_sealed_vol": cur_vol,
                "current_sealed_amount": cur_amt,
                "reason": reason,
            })

    return {
        "ok": True,
        "as_of": str(latest),
        "sealed_count": len(sealed),
        "triggered": all_events,
        "not_triggered": not_triggered,
    }


@router.post("/trigger-ladder")
def trigger_ladder(request: Request):
    """真实触发一次 ladder 预警 (落盘 + 飞书推送 + SSE), 供 Dev 调试验证完整效果。

    与 test-ladder 区别: 本端点会真的把预警写入 alerts.jsonl、推送飞书、触发 SSE,
    让用户看到真实的预警通知。绕过 cooldown 强制触发。
    """
    import time
    from app.services import alert_store

    repo = request.app.state.repo
    depth_svc = getattr(request.app.state, "depth_service", None)
    engine = getattr(request.app.state, "monitor_engine", None)
    quote_svc = getattr(request.app.state, "quote_service", None)

    if not depth_svc:
        raise HTTPException(status_code=503, detail="depth 服务未初始化")
    if not engine or not engine.has_rule_type("ladder"):
        raise HTTPException(status_code=400, detail="无 ladder 类型监控规则")

    latest = repo.enriched_latest_date()
    if not latest:
        raise HTTPException(status_code=400, detail="无 enriched 数据")

    # 取封单
    sealed: dict[str, int] = {}
    for is_down in (False, True):
        m = depth_svc.get_sealed_map(latest, is_down=is_down)
        for sym, info in m.items():
            vol = (info or {}).get("vol")
            if vol and vol > 0:
                sealed[sym] = vol
    if not sealed:
        raise HTTPException(status_code=400, detail="无封单数据")

    # 构造真实 rule_events (与 _evaluate_ladder 产出格式一致)
    import polars as pl
    enriched_today, _ = repo.get_enriched_latest()
    cols = [c for c in ["symbol", "close", "change_pct"] if c in enriched_today.columns]
    mock = enriched_today.select(cols).filter(pl.col("symbol").is_in(list(sealed.keys())))
    sealed_df = pl.DataFrame({"symbol": list(sealed.keys()), "_sealed_vol": list(sealed.values())})
    mock = mock.join(sealed_df, on="symbol", how="inner")

    now = time.time()
    rule_events: list[dict] = []
    name_map = {}
    try:
        inst = repo.get_instruments()
        if not inst.is_empty() and "name" in inst.columns:
            name_map = {r["symbol"]: r["name"] for r in inst.select(["symbol", "name"]).iter_rows(named=True) if r.get("name")}
    except Exception:  # noqa: BLE001
        pass

    for rule in engine.rules.values():
        if rule.get("type") != "ladder" or not rule.get("enabled", True):
            continue
        sym = rule.get("symbols", [""])[0] if rule.get("symbols") else ""
        metric = rule.get("metric", "sealed_vol")
        thr = rule.get("threshold", 0)
        direction = rule.get("direction", "up")
        warn_label = "炸板预警" if direction == "up" else "翘板预警"

        row = mock.filter(pl.col("symbol") == sym)
        if row.is_empty():
            continue
        cur_vol = row["_sealed_vol"][0]
        close_v = row["close"][0] if "close" in row.columns else None
        cur_val = cur_vol * 100 * close_v if metric == "sealed_amount" else cur_vol
        if not cur_val or cur_val <= 0 or cur_val > thr:
            continue  # 不满足条件, 跳过

        if metric == "sealed_amount":
            sv_text = f"{cur_val / 1e4:.0f}万元"
            th_text = f"{thr / 1e4:.0f}万元"
        else:
            sv_text = f"{cur_val:,.0f} 手"
            th_text = f"{thr:,.0f} 手"

        rule_events.append({
            "ts": int(now * 1000),
            "rule_id": rule["id"],
            "rule_name": rule.get("name", ""),
            "source": "ladder",
            "type": warn_label,
            "symbol": sym,
            "name": name_map.get(sym, sym),
            "message": f"{warn_label} · 封单 {sv_text} ≤ {th_text}",
            "price": close_v,
            "change_pct": row["change_pct"][0] if "change_pct" in row.columns else None,
            "signals": [],
            "severity": rule.get("severity", "warn"),
            "conditions": [],
            "logic": "and",
            "sealed_value": cur_val,
            "sealed_metric": metric,
        })

    if not rule_events:
        raise HTTPException(status_code=400, detail="当前无 ladder 规则满足触发条件 (封单均 > 阈值)")

    # 1. 落盘到 alerts.jsonl
    try:
        alert_store.append_many(repo.store.data_dir, rule_events)
    except Exception as e:  # noqa: BLE001
        pass  # 落盘失败不阻断推送

    # 2. SSE 推送 (入 pending_alerts 队列)
    if quote_svc:
        sse_alerts = [{
            "source": ev["source"], "type": ev["type"], "rule_id": ev["rule_id"],
            "strategy_id": None, "symbol": ev["symbol"], "name": ev["name"],
            "message": ev["message"], "price": ev["price"], "change_pct": ev["change_pct"],
            "signals": ev["signals"], "severity": ev["severity"],
            "conditions": ev["conditions"], "logic": ev["logic"],
        } for ev in rule_events]
        try:
            with quote_svc._lock:
                quote_svc._pending_alerts.extend(sse_alerts)
            quote_svc._alert_event.set()
        except Exception:  # noqa: BLE001
            pass

    # 3. 飞书推送
    if quote_svc:
        try:
            quote_svc._maybe_send_webhook(rule_events, engine)
        except Exception:  # noqa: BLE001
            pass

    return {
        "ok": True,
        "triggered": len(rule_events),
        "events": [{"symbol": ev["symbol"], "name": ev["name"], "message": ev["message"]} for ev in rule_events],
    }