"""同步任务：全市场日 K 线。

数据流：
  1. 选主源（按 priority 选第一个 is_available 的源）
  2. 增量判断：读 kline_stock.MAX(trade_date) 决定每只股票 start
  3. **Producer-Consumer 模式**：
     - N 个 fetcher 线程并发拉 K 线，把 (code, rows) 塞到 Queue
     - 1 个 writer 线程攒满 BATCH_SIZE 就 executemany + commit 一次
  4. 更新 stocks.kline_synced_at

设计要点：
  - 避免每只股票 commit（5000 只 = 5000 次 commit 太慢）
  - 减少并发数（避免新浪限流 / MySQL 连接数爆）
  - fetcher 间用 Queue 解耦，写库串行避免锁竞争
"""
from __future__ import annotations

import os
import queue
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError as FuturesTimeoutError
from datetime import datetime, timedelta
from typing import Any

import pandas as pd

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.kline_daily")

# 源优先级：QMT Bridge（本地） → 麦蕊 → 雪球 → 新浪
PRIMARY_PRIORITY = ["datasource_qmt_bridge", "datasource_mairui", "datasource_xueqiu", "datasource_xinlang"]
DEFAULT_START = "2018-01-01"
# 写库批量：攒够 BATCH_SIZE 行就 executemany 一次 + commit
BATCH_SIZE = 2000
# fetcher 并发：实测单线程最稳（5000 只 × 0.2s = 17 分钟，足够）
# 雪球 token 限速 10 RPS，单线程 0.1s/只 已经打满。多 worker 会触发风控 hang
DEFAULT_FETCH_WORKERS = 1
# 2026-07-08 教训: 数据源层虽然加了 fetch timeout (20s), task 层也加硬上限兜底,
# 防止数据源 SDK 升级后又被绕过。60s 对应 xueqiu/mairui 的 15-20s 上限 + DB upsert
# / DataFrame 处理余量 + 7月9日 mairui 间歇慢请求的容忍窗口。
FETCH_HARD_TIMEOUT = 180.0


class SyncKlineDaily(SyncTask):
    dataset_id = "kline_daily"

    def _run(
        self,
        *,
        trigger_source: str = "manual",
        start: str | None = None,
        end: str | None = None,
        codes: list[str] | None = None,
        incremental: bool = True,
        max_workers: int = DEFAULT_FETCH_WORKERS,
        **kwargs,
    ) -> dict[str, Any]:
        # 1. 获取所有可用源（按优先级），支持 per-stock fallback
        sources = self._get_available_sources()
        if not sources:
            return {"status": "error", "message": "所有 K 线数据源均不可用"}
        source_names = ", ".join(s.name for s in sources)

        self._progress(
            message=f"使用 [{source_names}] 同步日K线 (fetcher={max_workers}, batch={BATCH_SIZE})...",
            current_step=sources[0].key,
        )

        # 2. 拉股票列表
        if codes:
            stock_codes = [to_code6(c) for c in codes]
        else:
            stock_codes = [
                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:
            stock_codes = stock_codes[: int(limit_raw)]

        if not stock_codes:
            return {"status": "error", "message": "无股票代码（stocks 表为空？）"}

        # 3. 增量判断
        _end = end or _effective_sync_end()
        end_date_obj = datetime.strptime(_end, "%Y-%m-%d").date()
        jobs: list[tuple[str, str]] = []
        if incremental and not start:
            for c6 in stock_codes:
                last = db_ops.get_stock_kline_max_date(c6)
                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((c6, fetch_start))
        else:
            fetch_start = start or DEFAULT_START
            jobs = [(c, fetch_start) for c in stock_codes]

        total = len(jobs)
        skipped = len(stock_codes) - total
        self._progress(
            message=f"开始同步 {total} 只股票 (跳过 {skipped} 只已最新)",
            current=0, total=total,
        )

        # 4. 并发 fetch + 攒 batch 写库（线程安全）
        from concurrent.futures import ThreadPoolExecutor, as_completed
        t0 = time.time()
        batch_lock = threading.Lock()
        batch_rows: list[dict] = []
        pending_codes: dict[str, str] = {}
        ok_lock = threading.Lock()
        ok_cnt = [0]
        fallback_ok_cnt = [0]
        fail_cnt = [0]
        rows_cnt = [0]

        def _flush():
            with batch_lock:
                if not batch_rows:
                    return
                rows_to_write = list(batch_rows)
                codes_to_update = dict(pending_codes)
                batch_rows.clear()
                pending_codes.clear()
            try:
                db_ops.upsert_kline_stock(rows_to_write)
            except Exception as e:
                logger.error(f"batch upsert 失败 ({len(rows_to_write)} 行): {e}")
                return
            for c6, latest in codes_to_update.items():
                try:
                    db_ops.update_stock_kline_synced_at(c6, latest)
                except Exception as e:
                    logger.warning(f"update synced_at 失败 {c6}: {e}")

        def _rows_from_df(code6: str, df: pd.DataFrame) -> tuple[list[dict], str]:
            hermes = to_hermes(code6)
            rows = []
            for _, r in df.iterrows():
                td = r["trade_date"]
                rows.append({
                    "stock_code": hermes,
                    "trade_date": td.strftime("%Y-%m-%d") if hasattr(td, "strftime") else str(td)[:10],
                    "open": float(r["open"]),
                    "high": float(r["high"]),
                    "low": float(r["low"]),
                    "close": float(r["close"]),
                    "volume": float(r["volume"]),
                })
            latest = df["trade_date"].max()
            latest = latest.strftime("%Y-%m-%d") if hasattr(latest, "strftime") else str(latest)[:10]
            return (rows, latest)

        def _fetch_one(code6: str, fetch_start: str) -> tuple[str, list[dict] | None, str | None, bool]:
            errors = []
            for src in sources:
                try:
                    df = src.fetch_kline_daily(code6, fetch_start, _end)
                except Exception as e:
                    errors.append(f"{src.key}: {e}")
                    continue
                if df is None or df.empty:
                    errors.append(f"{src.key}: no data")
                    continue
                rows, latest = _rows_from_df(code6, df)
                is_fallback = src is not sources[0]
                return (code6, rows, latest, is_fallback)
            return (code6, None, "; ".join(errors)[:200], False)

        total = len(jobs)
        # 不用 `with ThreadPoolExecutor(...) as pool:` — 它的 __exit__ 默认 wait=True,
        # 一旦某个 worker 卡在 xueqiu IO,主线程会在 as_completed 触发 timeout 后
        # 仍被 __exit__ 阻塞等 worker 退出 → 进程挂死(2026-07-09 13:21 教训)。
        # 改成手动管理 + shutdown(wait=False),主线程能立刻退出。
        pool = ThreadPoolExecutor(max_workers=max_workers)
        futures = {pool.submit(_fetch_one, c6, fs): c6 for c6, fs in jobs}
        done_cnt = 0
        try:
            for future in as_completed(futures, timeout=FETCH_HARD_TIMEOUT):
                done_cnt += 1
                code6 = futures[future]
                try:
                    code6_r, rows, latest, is_fallback = future.result(timeout=0.1)
                except FuturesTimeoutError:
                    with ok_lock:
                        fail_cnt[0] += 1
                    logger.warning(f"[kline {code6}] 内部 race timeout, 记 fail")
                    continue
                except Exception as e:
                    with ok_lock:
                        fail_cnt[0] += 1
                    logger.warning(f"[kline {code6}] 异常: {e}")
                    continue
                if rows is None:
                    with ok_lock:
                        fail_cnt[0] += 1
                    logger.warning(f"[kline {code6}] {latest}")
                    continue
                if is_fallback:
                    with ok_lock:
                        fallback_ok_cnt[0] += 1
                with batch_lock:
                    batch_rows.extend(rows)
                    pending_codes[code6] = latest
                    cur_size = len(batch_rows)
                with ok_lock:
                    ok_cnt[0] += 1
                    rows_cnt[0] += len(rows)

                if cur_size >= BATCH_SIZE:
                    _flush()

                if done_cnt % 50 == 0 or done_cnt == total:
                    elapsed = time.time() - t0
                    rate = done_cnt / elapsed if elapsed > 0 else 0
                    self._progress(
                        message=f"进度 {done_cnt}/{total} OK:{ok_cnt[0]} FAIL:{fail_cnt[0]} {rate:.1f}只/秒",
                        current=done_cnt, total=total, current_step=code6,
                    )
        except FuturesTimeoutError:
            # 30s 内没 future 完成 → 所有 worker 都被卡死(数据源层 timeout 兜底失败)
            # 取消 pending futures,标 fail。shutdown(wait=False) 让主线程立刻返回,
            # 不被卡住的 worker 拖死。
            # 2026-07-11 修复: 防御性 cancel,避免异常处理链里变量被遮蔽导致
            # AttributeError: 'str' object has no attribute 'cancel'。
            stuck = [(f, c6) for f, c6 in futures.items() if not f.done()]
            logger.error(
                "所有 fetcher 卡死 (>%ss), %d 只股票未完成, 标 fail 后退出主循环",
                FETCH_HARD_TIMEOUT, len(stuck),
            )
            with ok_lock:
                fail_cnt[0] += len(stuck)
            for f, c6 in stuck:
                try:
                    if hasattr(f, "cancel"):
                        f.cancel()
                except Exception as cancel_err:
                    logger.warning(f"[kline] cancel future for {c6} 失败: {cancel_err}")
            try:
                pool.shutdown(wait=False)
            except Exception as shutdown_err:
                logger.warning(f"[kline] pool.shutdown(wait=False) 失败: {shutdown_err}")

            # ── 补偿：冷却后重入队列 ──
            # 雪球/mairui 偶发全 hang（服务端限流/网络抖动），等 30s 再试一次。
            # 单线程拉取场景下，卡住通常意味着当前这只股票请求 hang 死；
            # 冷却后把未完成的股票重新排队重试，救回临时故障。
            if stuck:
                cooldown = 30
                logger.warning(f"[kline] 冷却 {cooldown}s 后重试 {len(stuck)} 只…")
                time.sleep(cooldown)
                retry_ok = 0
                retry_fb = 0
                retry_rows = 0
                retry_pool = ThreadPoolExecutor(max_workers=max_workers)
                jobs_dict = {c6: fs for c6, fs in jobs}
                retry_futs = {}
                for _, c6 in stuck:
                    if c6 in jobs_dict:
                        retry_futs[retry_pool.submit(_fetch_one, c6, jobs_dict[c6])] = c6
                if retry_futs:
                    try:
                        for fut in as_completed(retry_futs, timeout=FETCH_HARD_TIMEOUT):
                            c6 = retry_futs[fut]
                            try:
                                code6_r, rows, latest, is_fallback = fut.result(timeout=0.1)
                            except Exception:
                                logger.warning(f"[kline {c6}] 重试异常")
                                continue
                            if rows is None:
                                logger.warning(f"[kline {c6}] 重试仍失败: {latest}")
                                continue
                            with batch_lock:
                                batch_rows.extend(rows)
                                pending_codes[c6] = latest
                                cur_size = len(batch_rows)
                            with ok_lock:
                                retry_ok += 1
                                retry_rows += len(rows)
                                if is_fallback:
                                    retry_fb += 1
                            if cur_size >= BATCH_SIZE:
                                _flush()
                    except FuturesTimeoutError:
                        logger.error(
                            "[kline] 重试仍卡死 (>%ss), %d 只股票放弃",
                            FETCH_HARD_TIMEOUT, sum(1 for f in retry_futs if not f.done()),
                        )
                    try:
                        retry_pool.shutdown(wait=False)
                    except Exception:
                        pass
                    with ok_lock:
                        ok_cnt[0] += retry_ok
                        fail_cnt[0] -= retry_ok
                        fallback_ok_cnt[0] += retry_fb
                        rows_cnt[0] += retry_rows
                    logger.warning(
                        f"[kline] 重试结果: {retry_ok}成 "
                        f"({retry_fb}只备胎) 共{retry_rows}行, 救回 {retry_ok} 只"
                    )
        else:
            pool.shutdown(wait=True)

        # 收尾 flush
        _flush()

        elapsed = round(time.time() - t0, 1)
        fb = fallback_ok_cnt[0]
        fb_info = f" ({fb}只备胎)" if fb else ""
        msg = f"日K线 {ok_cnt[0]}成 {fail_cnt[0]}败 {skipped}跳 共{rows_cnt[0]}行{fb_info}, {elapsed}s"
        return {
            "status": "ok" if fail_cnt[0] == 0 else "warning",
            "message": msg,
            "ok": ok_cnt[0],
            "fail": fail_cnt[0],
            "fallback_ok": fb,
            "skip": skipped,
            "rows": rows_cnt[0],
            "elapsed_sec": elapsed,
            "primary_source": sources[0].key if sources else "",
            "sources": [s.key for s in sources],
        }

    def _get_available_sources(self):
        from app.core.datasource.registry import is_source_ready, run_health_check, get_health_status
        if not get_health_status():
            run_health_check()
        available = []
        for key in PRIMARY_PRIORITY:
            ok, _ = is_source_ready(key)
            if ok:
                available.append(ds_registry.get(key))
        return available

    def _sync_one(self, primary, code6: str, start: str, end: str) -> dict:
        """保留这个方法以兼容外部调用（已不用，但单只测试可能用到）"""
        try:
            df = primary.fetch_kline_daily(code6, start, end)
        except Exception as e:
            return {"status": "fail", "error": str(e)}
        if df is None or df.empty:
            return {"status": "fail", "error": "no data"}
        hermes = to_hermes(code6)
        rows = []
        for _, r in df.iterrows():
            td = r["trade_date"]
            rows.append({
                "stock_code": hermes,
                "trade_date": td.strftime("%Y-%m-%d") if hasattr(td, "strftime") else str(td)[:10],
                "open": float(r["open"]),
                "high": float(r["high"]),
                "low": float(r["low"]),
                "close": float(r["close"]),
                "volume": float(r["volume"]),
            })
        try:
            db_ops.upsert_kline_stock(rows)
        except Exception as e:
            return {"status": "fail", "error": f"db write: {e}"}
        if rows:
            latest = max(r["trade_date"] for r in rows)
            try:
                db_ops.update_stock_kline_synced_at(code6, latest)
            except Exception:
                pass
        return {"status": "ok", "rows": len(rows)}