from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import time
import uuid
from pathlib import Path
from typing import Any, Dict, Iterable, List, Sequence

from common.runtime_paths import app_data_root


MAX_CLOUD_CONCURRENT_TASKS = 8
WRITE_RESOURCE_RISK_GROUP = "write_resource"
READ_RESOURCE_RISK_GROUP = "read_resource"


def safe_name(value: str, *, fallback: str = "unknown") -> str:
    """把云名称、地区、任务名转换成安全文件夹名，避免特殊符号破坏路径。"""
    cleaned = re.sub(r"[^0-9A-Za-z\u4e00-\u9fff._-]+", "-", value.strip())
    cleaned = cleaned.strip("-._")
    return cleaned or fallback


def account_fingerprint(ak: str, sk: str = "") -> str:
    """只保存账号指纹，不把完整 AK/SK 写入日志目录名。"""
    ak = (ak or "").strip()
    sk = (sk or "").strip()
    suffix = ak[-6:] if len(ak) >= 6 else (ak or "unknown")
    digest = hashlib.sha256(f"{ak}:{sk}".encode("utf-8", errors="ignore")).hexdigest()[:10]
    return f"账号-{safe_name(suffix)}-{digest}"


def task_data_root() -> Path:
    """返回统一任务数据根目录。"""
    path = app_data_root() / "tasks"
    path.mkdir(parents=True, exist_ok=True)
    return path


def task_lock_root() -> Path:
    """返回任务锁目录，同账号同地区写资源任务用它避免互相抢资源。"""
    path = app_data_root() / "locks"
    path.mkdir(parents=True, exist_ok=True)
    return path


def legacy_task_lock_roots() -> List[Path]:
    """返回旧版本可能用过的锁目录，便携版启动时可一起清理残留。"""
    roots: List[Path] = []
    try:
        legacy = Path.home() / "Documents" / "多云管理" / "locks"
        if legacy != task_lock_root():
            roots.append(legacy)
    except Exception:
        pass
    return roots


def remove_empty_lock_dirs(root: Path) -> None:
    """清理空锁目录，避免旧目录一直误导排查。"""
    try:
        if not root.exists():
            return
        for child in sorted(root.rglob("*"), key=lambda item: len(item.parts), reverse=True):
            if child.is_dir():
                try:
                    child.rmdir()
                except OSError:
                    pass
        try:
            root.rmdir()
        except OSError:
            pass
    except OSError:
        pass


def cleanup_legacy_task_locks() -> None:
    """清理旧数据目录里的任务锁；当前便携包只使用 WaWaCloud/data/locks。"""
    for root in legacy_task_lock_roots():
        try:
            if not root.exists():
                continue
            shutil.rmtree(root, ignore_errors=True)
        except OSError:
            pass


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 task_lock_file_stale(path: Path, *, max_age_seconds: float = 60.0) -> bool:
    """判断 .lock/.run 是否已经失效，防止异常退出后永久挡住任务。"""
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except Exception:
        try:
            return (time.time() - path.stat().st_mtime) > max_age_seconds
        except OSError:
            return True
    pid = int(data.get("pid") or 0)
    if pid:
        return not pid_running(pid)
    status_path = str(data.get("status_path") or "")
    if status_path:
        try:
            status_data = json.loads(Path(status_path).read_text(encoding="utf-8-sig"))
            if str(status_data.get("status") or "").lower() in {"success", "failed", "stopped"}:
                return True
        except Exception:
            pass
    try:
        created_at = float(data.get("created_at") or 0)
    except (TypeError, ValueError):
        created_at = 0
    if created_at > 0:
        return (time.time() - created_at) > max_age_seconds
    try:
        return (time.time() - path.stat().st_mtime) > max_age_seconds
    except OSError:
        return True


def cleanup_stale_task_locks() -> None:
    """清理当前便携目录中已失效的 .lock/.run 文件。"""
    cleanup_legacy_task_locks()
    root = task_lock_root()
    for pattern in ("*.lock", "*.run"):
        for path in root.rglob(pattern):
            if task_lock_file_stale(path):
                try:
                    path.unlink(missing_ok=True)
                except OSError:
                    pass
    remove_empty_lock_dirs(root / "running")


def task_account_dir(cloud: str, account: str) -> Path:
    """返回某个云、某个账号的任务目录。"""
    path = task_data_root() / safe_name(cloud) / "accounts" / safe_name(account) / "tasks"
    path.mkdir(parents=True, exist_ok=True)
    return path


def make_task_id(action: str) -> str:
    """生成稳定可读的任务 ID，便于人工查日志。"""
    stamp = time.strftime("%Y%m%d-%H%M%S")
    return f"{stamp}-{safe_name(action)}-{uuid.uuid4().hex[:8]}"


def task_lock_keys(cloud: str, account: str, regions: Sequence[str], risk_group: str) -> List[str]:
    """生成资源写入锁；只有购买、搭建、删除、退订、重装这类任务才锁同地区。"""
    if risk_group != WRITE_RESOURCE_RISK_GROUP:
        return []
    keys: List[str] = []
    seen = set()
    for region in regions or ["global"]:
        safe_region = safe_name(region)
        if safe_region in seen:
            continue
        seen.add(safe_region)
        keys.append(f"{safe_name(cloud)}__{safe_name(account)}__{safe_region}__{risk_group}")
    return keys


def atomic_write_json(path: Path, data: Dict[str, Any]) -> None:
    """原子写 JSON，避免状态文件写到一半被界面读取。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    tmp.replace(path)


class TaskContext:
    """子进程任务上下文，云任务脚本通过它写日志、状态和结果。"""

    def __init__(self, spec: Dict[str, Any], task_dir: Path) -> None:
        self.spec = spec
        self.task_dir = task_dir
        # AK/SK 只通过子进程环境变量传入，不写进 task.json，避免导出日志时暴露密钥。
        self.ak = os.environ.get("WAWA_TASK_AK", "")
        self.sk = os.environ.get("WAWA_TASK_SK", "")
        self.payload = spec.get("payload") if isinstance(spec.get("payload"), dict) else {}
        self.log_path = task_dir / "run.log"
        self.status_path = task_dir / "status.json"
        self.result_path = task_dir / "result.json"
        self.stop_path = task_dir / "stop.flag"

    def log(self, message: str) -> None:
        text = str(message)
        if text and not text.endswith("\n"):
            text += "\n"
        with self.log_path.open("a", encoding="utf-8", errors="replace") as handle:
            handle.write(text)
            handle.flush()

    def update_status(self, status: str, message: str = "") -> None:
        data = {
            "task_id": self.spec.get("task_id", ""),
            "cloud": self.spec.get("cloud", ""),
            "account": self.spec.get("account", ""),
            "action": self.spec.get("action", ""),
            "title": self.spec.get("title", ""),
            "regions": self.spec.get("regions", []),
            "risk_group": self.spec.get("risk_group", ""),
            "status": status,
            "message": message,
            "task_dir": str(self.task_dir),
            "updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
        }
        atomic_write_json(self.status_path, data)

    def write_result(self, data: Dict[str, Any]) -> None:
        atomic_write_json(self.result_path, data)

    def should_stop(self) -> bool:
        return self.stop_path.exists()


def read_status_files(paths: Iterable[Path]) -> List[Dict[str, Any]]:
    """读取一组任务状态文件，坏文件跳过，不影响任务中心显示。"""
    rows: List[Dict[str, Any]] = []
    for path in paths:
        try:
            data = json.loads(path.read_text(encoding="utf-8-sig"))
        except Exception:
            continue
        if isinstance(data, dict):
            rows.append(data)
    return rows
