from __future__ import annotations

from datetime import datetime
import argparse
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import sys
import time
from typing import Any


ROOT = Path(__file__).resolve().parent
VENDOR_DIR = ROOT / "vendor"
if VENDOR_DIR.exists() and str(VENDOR_DIR) not in sys.path:
    sys.path.insert(0, str(VENDOR_DIR))
DASHBOARD_DIR = ROOT / "reports" / "dashboard"
STATUS_PATH = DASHBOARD_DIR / "update_status.json"
LOCK_PATH = DASHBOARD_DIR / ".daily_update.lock"
LOG_DIR = DASHBOARD_DIR / "logs"
STAGES = [
    ("M1 数据刷新", ROOT / "run_m1.py", 3 * 60 * 60),
    ("M2 因子研究", ROOT / "run_m2_research.py", 3 * 60 * 60),
    ("看板生成", ROOT / "run_dashboard.py", 15 * 60),
]


def check_environment() -> int:
    missing_modules = [name for name in ("pandas", "numpy", "pyarrow", "duckdb") if importlib.util.find_spec(name) is None]
    required_paths = [ROOT / "config" / "local_paths.toml", *(script for _, script, _ in STAGES)]
    missing_paths = [str(path) for path in required_paths if not path.exists()]
    source_paths: dict[str, str] = {}
    missing_source_paths: list[str] = []
    warnings: list[str] = []
    try:
        sys.path.insert(0, str(ROOT / "src"))
        from fundamental_mvp.config import load_settings

        settings = load_settings(ROOT / "config" / "local_paths.toml")
        source_paths = {
            "tdxRoot": str(settings.tdx_root),
            "shDaily": str(settings.tdx_root / "vipdoc" / "sh" / "lday"),
            "szDaily": str(settings.tdx_root / "vipdoc" / "sz" / "lday"),
            "finance": str(settings.tdx_root / "vipdoc" / "cw"),
            "gbbq": str(settings.gbbq_path),
        }
        required_sources = [
            settings.tdx_root,
            settings.tdx_root / "vipdoc" / "sh" / "lday",
            settings.tdx_root / "vipdoc" / "sz" / "lday",
            settings.tdx_root / "vipdoc" / "cw",
            settings.gbbq_path,
        ]
        missing_source_paths = [str(path) for path in required_sources if not path.exists()]
        if not settings.legacy_data_dir.exists():
            warnings.append(f"历史名称辅助目录不存在（可选）：{settings.legacy_data_dir}")
        if not settings.legacy_finance_parquet.exists() and not settings.financial_pit_path.exists():
            warnings.append("既无历史财务备用文件，也无已生成的财务缓存")
    except Exception as error:
        missing_source_paths.append(f"配置无法读取：{error}")
    result = {
        "python": sys.executable,
        "pythonVersion": sys.version.split()[0],
        "projectRoot": str(ROOT),
        "missingModules": missing_modules,
        "missingPaths": missing_paths,
        "sourcePaths": source_paths,
        "missingSourcePaths": missing_source_paths,
        "warnings": warnings,
        "ready": not missing_modules and not missing_paths and not missing_source_paths,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["ready"] else 1


def _now() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def _write_json_atomic(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f"{path.name}.tmp")
    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary.replace(path)


def _lock_owner_pid() -> int | None:
    """PID recorded in the lock file, or None when it is absent/unreadable."""
    try:
        text = LOCK_PATH.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None
    if "pid=" not in text:
        return None
    try:
        return int(text.split("pid=", 1)[1].split()[0])
    except (IndexError, ValueError):
        return None


def _pid_alive(pid: int | None) -> bool:
    if not pid or pid <= 0:
        return False
    if os.name == "nt":
        import ctypes

        PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
        STILL_ACTIVE = 259
        kernel32 = ctypes.windll.kernel32
        handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
        if not handle:
            return False
        try:
            code = ctypes.c_ulong()
            if not kernel32.GetExitCodeProcess(handle, ctypes.byref(code)):
                return False
            return code.value == STILL_ACTIVE
        finally:
            kernel32.CloseHandle(handle)
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def _acquire_lock() -> int:
    DASHBOARD_DIR.mkdir(parents=True, exist_ok=True)
    try:
        return os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError:
        owner = _lock_owner_pid()
        age_seconds = time.time() - LOCK_PATH.stat().st_mtime
        # A lock left behind by a process that no longer exists is stale: take it over
        # right away. Without this, an interrupted run (closing the console window, a
        # crash, a reboot) would keep rejecting every later run for up to six hours,
        # which looks exactly like "the update button stopped working".
        if owner is not None and not _pid_alive(owner):
            LOCK_PATH.unlink()
            return os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        if age_seconds <= 6 * 60 * 60:
            held = f"（pid={owner}）" if owner else ""
            raise RuntimeError(f"已有日更任务在运行{held}")
        LOCK_PATH.unlink()
        return os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY)


def _last_error_line(path: Path, limit: int = 80) -> str:
    """Best-effort: pull the most informative line from a stage's output.

    The stage output ends with a Python traceback, whose final line is normally the
    exception message. This lets the status file carry an actionable reason instead
    of only "stage failed with exit code 1".

    Two noise sources are handled deliberately:
      * our own bracketed markers, e.g. "[2026-10-13T17:09:17+08:00] END ..."
      * multi-line engine errors (DuckDB) that end with a bare caret pointing at the
        offending token -- taking the literal last line would yield just "^".
    """
    try:
        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        return ""
    candidates = []
    for line in reversed(lines[-limit:]):
        text = line.strip()
        if not text or text.startswith("["):
            continue
        candidates.append(text)
    # Prefer a line that names an exception/error and carries a message.
    for text in candidates:
        if ("Error" in text or "Exception" in text) and ":" in text:
            return text
    # Otherwise the first line that actually contains words (skips "^", "~", "|", ...).
    for text in candidates:
        if any(ch.isalnum() for ch in text):
            return text
    return ""


def main() -> int:
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    started_at = _now()
    log_path = LOG_DIR / f"update_{datetime.now():%Y%m%d_%H%M%S}.log"
    status: dict[str, Any] = {
        "state": "running",
        "startedAt": started_at,
        "completedAt": None,
        "stage": "准备更新",
        "message": "日度研究数据正在更新",
        "logPath": str(log_path),
    }

    try:
        lock_fd = _acquire_lock()
    except RuntimeError as error:
        print(str(error), file=sys.stderr)
        return 2

    os.write(lock_fd, f"pid={os.getpid()} started={started_at}".encode("utf-8"))
    os.close(lock_fd)
    _write_json_atomic(STATUS_PATH, status)

    try:
        with log_path.open("w", encoding="utf-8") as log:
            log.write(f"[{started_at}] 日度更新开始\n")
            for index, (stage_name, script_path, timeout_seconds) in enumerate(STAGES, start=1):
                status.update({"stage": stage_name, "message": f"正在执行 {stage_name}", "stageIndex": index, "stageCount": len(STAGES)})
                _write_json_atomic(STATUS_PATH, status)
                stage_started = time.monotonic()
                log.write(f"\n[{_now()}] START {stage_name}\n")
                log.flush()
                result = subprocess.run(
                    [sys.executable, str(script_path)],
                    cwd=ROOT,
                    stdout=log,
                    stderr=subprocess.STDOUT,
                    text=True,
                    encoding="utf-8",
                    errors="replace",
                    timeout=timeout_seconds,
                    env={
                        **os.environ,
                        "PYTHONIOENCODING": "utf-8",
                        "PYTHONPATH": os.pathsep.join(
                            [str(VENDOR_DIR), str(ROOT / "src"), os.environ.get("PYTHONPATH", "")]
                        ).rstrip(os.pathsep),
                    },
                    check=False,
                )
                duration = round(time.monotonic() - stage_started, 1)
                # Read the real failure reason from the stage output BEFORE appending our
                # own END marker, so the status file is actionable.
                failure_detail = _last_error_line(log_path) if result.returncode != 0 else ""
                log.write(f"\n[{_now()}] END {stage_name} exit={result.returncode} duration={duration}s\n")
                log.flush()
                if result.returncode != 0:
                    message = f"{stage_name} 失败，退出码 {result.returncode}"
                    if failure_detail:
                        message = f"{message}：{failure_detail}"
                    raise RuntimeError(message)

        status.update({
            "state": "success",
            "completedAt": _now(),
            "stage": "更新完成",
            "message": "M1、M2 与研究看板已全部更新",
            "stageIndex": len(STAGES),
            "stageCount": len(STAGES),
        })
        version_path = DASHBOARD_DIR / "version.json"
        if version_path.exists():
            version = json.loads(version_path.read_text(encoding="utf-8"))
            status["runId"] = version.get("runId")
            status["version"] = version.get("version")
        _write_json_atomic(STATUS_PATH, status)
        print(status["message"])
        return 0
    except subprocess.TimeoutExpired as error:
        status.update({"state": "failed", "completedAt": _now(), "stage": status["stage"], "message": f"{status['stage']} 超时，旧看板仍可用", "error": str(error)})
        _write_json_atomic(STATUS_PATH, status)
        print(status["message"], file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        status.update({"state": "cancelled", "completedAt": _now(), "message": "更新已取消，旧看板仍可用"})
        _write_json_atomic(STATUS_PATH, status)
        print(status["message"], file=sys.stderr)
        return 130
    except Exception as error:
        status.update({"state": "failed", "completedAt": _now(), "message": f"{error}；旧看板仍可用", "error": repr(error)})
        _write_json_atomic(STATUS_PATH, status)
        print(status["message"], file=sys.stderr)
        return 1
    finally:
        if LOCK_PATH.exists():
            LOCK_PATH.unlink()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="刷新 M1/M2 数据并重建研究看板")
    parser.add_argument("--check", action="store_true", help="仅检查 Python 环境和必需文件")
    arguments = parser.parse_args()
    raise SystemExit(check_environment() if arguments.check else main())