from __future__ import annotations

import json
import os
import time
from pathlib import Path
from typing import List, Sequence

from common.shared import localize_region_tokens
from common.task_runtime import (
    MAX_CLOUD_CONCURRENT_TASKS,
    account_fingerprint,
    cleanup_stale_task_locks,
    make_task_id,
    safe_name,
    task_lock_keys,
    task_lock_root,
)


class TaskGuardError(RuntimeError):
    """任务守卫启动失败，通常是并发上限或同账号同地区写资源锁冲突。"""


def _pid_running(pid: int) -> bool:
    """判断进程是否存活；用于清理异常退出后留下的旧锁。"""
    if pid <= 0:
        return False
    if os.name == "nt":
        try:
            import ctypes

            kernel32 = ctypes.windll.kernel32
            handle = kernel32.OpenProcess(0x1000, False, int(pid))
            if not handle:
                return False
            exit_code = ctypes.c_ulong()
            ok = kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code))
            kernel32.CloseHandle(handle)
            return bool(ok and exit_code.value == 259)
        except Exception:
            return False
    try:
        os.kill(int(pid), 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except Exception:
        return False


def _read_json(path: Path) -> dict:
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def _write_json(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def _friendly_action(action: str) -> str:
    text = str(action or "").strip()
    return {
        "购买并搭建": "一键购买并搭建",
        "购买": "购买地区",
        "搭建": "搭建选择地区",
    }.get(text, text or "资源任务")


def _friendly_region(region: object) -> str:
    text = str(region or "").strip()
    if not text:
        return ""
    return localize_region_tokens(text)


def _lock_conflict_detail(path: Path, key: str, requested_action: str) -> str:
    # 中文注释：锁冲突通常是用户同时点了购买/搭建/退订/删除，弹窗必须说清楚被哪个任务挡住，避免误以为软件卡死。
    data = _read_json(path)
    running_action = _friendly_action(str(data.get("action") or "资源任务"))
    requested = _friendly_action(requested_action)
    regions = data.get("regions")
    if isinstance(regions, list):
        region_text = "、".join(label for label in (_friendly_region(item) for item in regions) if label)
    else:
        region_text = ""
    lines = [
        f"因“{running_action}”任务正在处理同账号同地区资源，无法执行“{requested}”，请等候完毕。",
    ]
    if region_text:
        lines.append(f"冲突地区: {region_text}")
    lines.extend(["", f"锁文件: {path}"])
    return "\n".join(lines)


class TaskGuard:
    """给仍在界面线程中运行的旧任务补上跨软件进程的并发和写资源保护。"""

    def __init__(
        self,
        *,
        cloud: str,
        ak: str,
        sk: str,
        action: str,
        regions: Sequence[str],
        risk_group: str,
    ) -> None:
        self.cloud = cloud
        self.account = account_fingerprint(ak, sk)
        self.action = action
        self.regions = list(regions or ["global"])
        self.risk_group = risk_group
        self.task_id = make_task_id(action)
        self.running_path = task_lock_root() / "running" / f"{safe_name(cloud)}__{safe_name(self.account)}__{safe_name(self.task_id)}.run"
        self.lock_paths: List[Path] = []
        self.acquired = False

    def _running_files_for_cloud(self) -> List[Path]:
        root = task_lock_root() / "running"
        root.mkdir(parents=True, exist_ok=True)
        prefix = f"{safe_name(self.cloud)}__"
        live: List[Path] = []
        for path in root.glob(f"{prefix}*.run"):
            data = _read_json(path)
            pid = int(data.get("pid") or 0)
            if _pid_running(pid):
                live.append(path)
                continue
            try:
                path.unlink(missing_ok=True)
            except OSError:
                pass
        return live

    def _lock_stale(self, path: Path) -> bool:
        data = _read_json(path)
        pid = int(data.get("pid") or 0)
        if pid:
            return not _pid_running(pid)
        try:
            return time.time() - path.stat().st_mtime > 60
        except OSError:
            return True

    def acquire(self) -> "TaskGuard":
        if self.acquired:
            return self
        cleanup_stale_task_locks()
        live_count = len(self._running_files_for_cloud())
        if live_count >= MAX_CLOUD_CONCURRENT_TASKS:
            raise TaskGuardError(f"{self.cloud} 当前运行任务已达到 {MAX_CLOUD_CONCURRENT_TASKS} 个，请等待前面的任务完成。")
        _write_json(
            self.running_path,
            {
                "task_id": self.task_id,
                "cloud": self.cloud,
                "account": self.account,
                "action": self.action,
                "regions": self.regions,
                "risk_group": self.risk_group,
                "pid": os.getpid(),
                "created_at": time.time(),
            },
        )
        try:
            for key in task_lock_keys(self.cloud, self.account, self.regions, self.risk_group):
                path = task_lock_root() / f"{key}.lock"
                if path in self.lock_paths:
                    continue
                while True:
                    try:
                        fd = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
                        break
                    except FileExistsError as exc:
                        if path in self.lock_paths:
                            break
                        if self._lock_stale(path):
                            try:
                                path.unlink(missing_ok=True)
                                continue
                            except OSError:
                                pass
                        detail = _lock_conflict_detail(path, key, self.action)
                        raise TaskGuardError(detail) from exc
                if path in self.lock_paths:
                    continue
                with os.fdopen(fd, "w", encoding="utf-8") as handle:
                    handle.write(
                        json.dumps(
                            {
                                "task_id": self.task_id,
                                "cloud": self.cloud,
                                "account": self.account,
                                "action": self.action,
                                "regions": self.regions,
                                "risk_group": self.risk_group,
                                "pid": os.getpid(),
                                "created_at": time.time(),
                            },
                            ensure_ascii=False,
                        )
                    )
                self.lock_paths.append(path)
        except Exception:
            self.release()
            raise
        self.acquired = True
        return self

    def release(self) -> None:
        for path in self.lock_paths:
            try:
                path.unlink(missing_ok=True)
            except OSError:
                pass
        self.lock_paths = []
        try:
            self.running_path.unlink(missing_ok=True)
        except OSError:
            pass
        self.acquired = False
