"""同步任务：mairui 日 K 级别技术指标 MACD / KDJ / BOLL。

数据源：mairui /hsstock/history/{indicator}/{symbol}/d/n/{licence}
  - macd → market_data.kline_stock_macd_daily (diff, dea, macd, ema12, ema26)
  - kdj  → market_data.kline_stock_kdj_daily  (k, d, j)
  - boll → market_data.kline_stock_boll_daily (upper, mid, lower ← mairui u/m/d)

设计（对齐 task_kline_daily）：
  - 全市场 A 股循环，每只票分别拉 3 个指标
  - 增量：读各指标表 MAX(trade_date)，只补新增区间（回看 5 天防漏）
  - 并发 fetch + 攒 BATCH 批量 upsert，写库串行加锁
  - stock_code 落库用 hermes 格式（SH600519），与项目其他表一致

参考实现：pre_limit_seeker/production/sync_mairui_indicators.py（parquet 版）。
"""
from __future__ import annotations

import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from typing import Any

from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask, _effective_sync_end
from app.core.utils.logging import get_logger

logger = get_logger("sync.mairui_indicators")

DEFAULT_START = "2018-01-01"
BATCH_SIZE = 2000
# mairui 钻石档 100 RPS；5 workers × 10 RPS/worker = 50 RPS，留 buffer
DEFAULT_FETCH_WORKERS = 5

# 指标 → (mairui 源字段, 落库列名, upsert 函数)
INDICATORS: dict[str, dict[str, Any]] = {
    "macd": {
        "src_fields": ["diff", "dea", "macd", "ema12", "ema26"],
        "db_cols": ["diff", "dea", "macd", "ema12", "ema26"],
        "upsert": db_ops.upsert_kline_stock_macd_daily_rows,
    },
    "kdj": {
        "src_fields": ["k", "d", "j"],
        "db_cols": ["k", "d", "j"],
        "upsert": db_ops.upsert_kline_stock_kdj_daily_rows,
    },
    "boll": {
        # mairui: u=上轨, m=中轨, d=下轨 → 落库 upper/mid/lower
        "src_fields": ["u", "m", "d"],
        "db_cols": ["upper", "mid", "lower"],
        "upsert": db_ops.upsert_kline_stock_boll_daily_rows,
    },
}


class SyncMairuiIndicators(SyncTask):
    dataset_id = "mairui_indicators"

    def _run(
        self,
        *,
        trigger_source: str = "manual",
        indicators: list[str] | None = None,
        codes: list[str] | None = None,
        start: str | None = None,
        end: str | None = None,
        incremental: bool = True,
        max_workers: int = DEFAULT_FETCH_WORKERS,
        **kwargs,
    ) -> dict[str, Any]:
        targets = [i.lower() for i in (indicators or ["macd", "kdj", "boll"])]
        for i in targets:
            if i not in INDICATORS:
                return {"status": "error", "message": f"未知指标 {i!r}，可选 {list(INDICATORS)}"}

        source = self._pick_source()
        if source is None:
            return {"status": "error", "message": "mairui 数据源不可用（licence 未配置？）"}

        # 股票列表（hermes 格式）
        if codes:
            hermes_codes = [to_hermes(c) for c in codes]
        else:
            hermes_codes = [
                to_hermes(c) for c in db_ops.iter_stock_codes(active_only=True)
                if is_a_share_code(c)
            ]
        limit_raw = os.environ.get("MARKET_DATA_STOCK_LIMIT", "").strip()
        if limit_raw.isdigit() and int(limit_raw) > 0:
            hermes_codes = hermes_codes[: int(limit_raw)]
        if not hermes_codes:
            return {"status": "error", "message": "无股票代码（stocks 表为空？）"}

        _end = end or _effective_sync_end()
        end_date_obj = datetime.strptime(_end, "%Y-%m-%d").date()

        t0 = time.time()
        agg_ok = 0
        agg_fail = 0
        agg_rows = 0
        per_indicator: dict[str, dict[str, int]] = {}

        for ind in targets:
            self._progress(message=f"[{ind}] 规划增量区间...", current_step=ind)
            jobs = self._plan_jobs(ind, hermes_codes, start, incremental, end_date_obj)
            total = len(jobs)
            skipped = len(hermes_codes) - total
            logger.info(f"[mairui_indicators] {ind}: to_pull={total}, skip={skipped}")
            self._progress(
                message=f"[{ind}] 开始同步 {total} 只 (跳过 {skipped} 只已最新)",
                current=0, total=total, current_step=ind,
            )

            ok, fail, rows = self._sync_indicator(
                source, ind, jobs, _end, max_workers,
            )
            per_indicator[ind] = {"ok": ok, "fail": fail, "skip": skipped, "rows": rows}
            agg_ok += ok
            agg_fail += fail
            agg_rows += rows

        elapsed = round(time.time() - t0, 1)
        parts = ", ".join(
            f"{k}({v['ok']}成/{v['fail']}败/{v['rows']}行)"
            for k, v in per_indicator.items()
        )
        msg = f"指标 {parts}；共 {agg_rows} 行, {elapsed}s"
        logger.info(f"[mairui_indicators] {msg}")
        return {
            "status": "ok" if agg_fail == 0 else "warning",
            "message": msg,
            "ok": agg_ok,
            "fail": agg_fail,
            "rows": agg_rows,
            "elapsed_sec": elapsed,
            "per_indicator": per_indicator,
        }

    def _plan_jobs(
        self, indicator: str, hermes_codes: list[str],
        start: str | None, incremental: bool, end_date_obj,
    ) -> list[tuple[str, str]]:
        """返回 [(hermes_code, fetch_start), ...]。"""
        jobs: list[tuple[str, str]] = []
        if incremental and not start:
            for hc in hermes_codes:
                last = db_ops.get_indicator_max_date(indicator, hc)
                if last is None:
                    fetch_start = DEFAULT_START
                elif datetime.strptime(last, "%Y-%m-%d").date() >= end_date_obj:
                    continue
                else:
                    fetch_start = (
                        datetime.strptime(last, "%Y-%m-%d").date() - timedelta(days=5)
                    ).strftime("%Y-%m-%d")
                jobs.append((hc, fetch_start))
        else:
            fetch_start = start or DEFAULT_START
            jobs = [(hc, fetch_start) for hc in hermes_codes]
        return jobs

    def _sync_indicator(
        self, source, indicator: str, jobs: list[tuple[str, str]],
        end: str, max_workers: int,
    ) -> tuple[int, int, int]:
        cfg = INDICATORS[indicator]
        upsert_fn = cfg["upsert"]
        src_fields = cfg["src_fields"]
        db_cols = cfg["db_cols"]

        batch_lock = threading.Lock()
        batch_rows: list[dict] = []
        ok_lock = threading.Lock()
        ok_cnt = [0]
        fail_cnt = [0]
        rows_cnt = [0]

        def _flush():
            with batch_lock:
                if not batch_rows:
                    return
                to_write = list(batch_rows)
                batch_rows.clear()
            try:
                upsert_fn(to_write)
            except Exception as e:
                logger.error(f"[{indicator}] batch upsert 失败 ({len(to_write)} 行): {e}")

        def _fetch_one(hermes_code: str, fetch_start: str):
            code6 = to_code6(hermes_code)
            try:
                df = source.fetch_indicator_daily(indicator, code6, fetch_start, end)
            except Exception as e:
                return (hermes_code, None, str(e)[:120])
            if df is None or df.empty:
                return (hermes_code, None, "no data")
            rows = []
            for _, r in df.iterrows():
                td = r["trade_date"]
                # 2026-07-11 修复: mairui 可能返回整行 NaN/NaT(如 688 早期无指标),
                # float(NaT) 会抛 TypeError。用 pandas.isna 统一跳过无效值。
                row = {
                    "stock_code": hermes_code,
                    "trade_date": td.strftime("%Y-%m-%d") if hasattr(td, "strftime") else str(td)[:10],
                    "source": "mairui",
                }
                for src_f, db_c in zip(src_fields, db_cols):
                    v = r.get(src_f)
                    if v is None or (hasattr(v, "isna") and v.isna()) or (isinstance(v, float) and v != v):
                        row[db_c] = None
                    else:
                        try:
                            row[db_c] = float(v)
                        except Exception:
                            row[db_c] = None
                rows.append(row)
            return (hermes_code, rows, None)

        total = len(jobs)
        t0 = time.time()
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures = {pool.submit(_fetch_one, hc, fs): hc for hc, fs in jobs}
            done = 0
            for fut in as_completed(futures):
                done += 1
                hermes_code, rows, err = fut.result()
                if rows is None:
                    with ok_lock:
                        fail_cnt[0] += 1
                    continue
                with batch_lock:
                    batch_rows.extend(rows)
                    cur = len(batch_rows)
                with ok_lock:
                    ok_cnt[0] += 1
                    rows_cnt[0] += len(rows)
                if cur >= BATCH_SIZE:
                    _flush()
                if done % 100 == 0 or done == total:
                    el = time.time() - t0
                    rate = done / el if el > 0 else 0
                    self._progress(
                        message=f"[{indicator}] {done}/{total} OK:{ok_cnt[0]} FAIL:{fail_cnt[0]} {rate:.1f}只/秒",
                        current=done, total=total, current_step=indicator,
                    )
        _flush()
        return ok_cnt[0], fail_cnt[0], rows_cnt[0]

    def _pick_source(self):
        from app.core.datasource.registry import (
            get_health_status, is_source_ready, run_health_check,
        )
        if not get_health_status():
            run_health_check()
        ok, _ = is_source_ready("datasource_mairui")
        if ok:
            return ds_registry.get("datasource_mairui")
        return None