"""在日常命令里顺带更新 pouch 自身和已拉取的外部 source。"""

from __future__ import annotations

import os
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

from pouch import yaml_io
from pouch.catalog import catalog_checkout_path, load_catalog
from pouch.config import load_auto_update_settings
from pouch.paths import POUCH_HOME, STATE_FILE
from pouch.sources import load_sources, source_checkout

GIT_TIMEOUT_SECONDS = 20
_SKIP_COMMANDS = {"update", "publish"}
_TRUTHY = {"1", "true", "yes", "on"}
_refreshed_paths: set[str] = set()


@dataclass(frozen=True)
class UpdateResult:
    name: str
    status: str
    detail: str = ""

    @property
    def changed(self) -> bool:
        return self.status == "updated"


def _print_err(msg: str) -> None:
    print(msg, file=sys.stderr)


def _run_git(
    repo: Path,
    *args: str,
    check: bool = False,
    timeout: int = GIT_TIMEOUT_SECONDS,
) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", "-C", str(repo), *args],
        check=check,
        text=True,
        capture_output=True,
        timeout=timeout,
    )


def _is_git_repo(path: Path) -> bool:
    if not path.is_dir():
        return False
    try:
        result = _run_git(path, "rev-parse", "--is-inside-work-tree")
    except (OSError, subprocess.TimeoutExpired):
        return False
    return result.returncode == 0 and result.stdout.strip() == "true"


def _working_tree_dirty(path: Path) -> bool:
    result = _run_git(path, "status", "--porcelain")
    return result.returncode != 0 or bool(result.stdout.strip())


def _head_sha(path: Path) -> str:
    result = _run_git(path, "rev-parse", "HEAD")
    return result.stdout.strip() if result.returncode == 0 else ""


def refresh_git_checkout(
    path: Path,
    ref: str | None = None,
    *,
    name: str,
    strict: bool = False,
    allow_dirty: bool = False,
) -> UpdateResult:
    key = str(path.resolve()) if path.exists() else str(path)
    if key in _refreshed_paths:
        return UpdateResult(name, "current")
    if not _is_git_repo(path):
        _refreshed_paths.add(key)
        result = UpdateResult(name, "skipped", "不是 git 仓库")
        if strict:
            raise SystemExit(f"{name} 不是 git 仓库: {path}")
        return result
    if not allow_dirty and _working_tree_dirty(path):
        _refreshed_paths.add(key)
        result = UpdateResult(name, "skipped", "工作区不干净")
        if strict:
            raise SystemExit(f"{name} 工作区不干净，拒绝更新: {path}")
        return result

    before = _head_sha(path)
    commands: list[tuple[str, ...]] = [("fetch", "--all", "--tags")]
    if ref:
        commands.append(("checkout", ref))
    commands.append(("pull", "--ff-only"))
    try:
        for command in commands:
            ran = _run_git(path, *command, check=strict)
            if ran.returncode != 0:
                detail = (ran.stderr or ran.stdout or "git 失败").strip()
                return UpdateResult(name, "failed", detail)
    except subprocess.CalledProcessError as exc:
        detail = ((exc.stderr or exc.stdout or str(exc))).strip()
        if strict:
            raise SystemExit(f"更新 {name} 失败: {detail}") from exc
        return UpdateResult(name, "failed", detail)
    except subprocess.TimeoutExpired:
        result = UpdateResult(name, "failed", "git 超时")
        if strict:
            raise SystemExit(f"更新 {name} 超时: {path}") from None
        return result
    except OSError as exc:
        result = UpdateResult(name, "failed", str(exc))
        if strict:
            raise SystemExit(f"更新 {name} 失败: {exc}") from exc
        return result

    after = _head_sha(path)
    _refreshed_paths.add(key)
    if before and after and before != after:
        return UpdateResult(name, "updated", after[:8])
    return UpdateResult(name, "current")


def update_pouch_home(*, strict: bool = False) -> UpdateResult:
    return refresh_git_checkout(
        POUCH_HOME,
        name="pouch",
        strict=strict,
        allow_dirty=strict,
    )


def update_cloned_catalogs(*, strict: bool = False) -> list[UpdateResult]:
    results: list[UpdateResult] = []
    seen: set[Path] = set()
    for name, entry in load_catalog().items():
        repo = str(entry.get("repo", ""))
        if repo.startswith("~"):
            continue
        dest = catalog_checkout_path(name, entry)
        resolved = dest.resolve() if dest.exists() else dest
        if resolved in seen or not dest.is_dir():
            continue
        seen.add(resolved)
        results.append(
            refresh_git_checkout(
                dest,
                str(entry.get("ref", "main") or "main"),
                name=f"catalog:{name}",
                strict=strict,
            )
        )
    return results


def update_custom_sources(*, strict: bool = False) -> list[UpdateResult]:
    results: list[UpdateResult] = []
    for name, entry in load_sources().items():
        if entry.get("local_path"):
            continue
        checkout = source_checkout(name, entry)
        if not checkout.is_dir():
            continue
        results.append(
            refresh_git_checkout(
                checkout,
                str(entry.get("ref", "main") or "main"),
                name=f"custom:{name}",
                strict=strict,
            )
        )
    return results


def run_updates(*, strict_pouch: bool = False) -> list[UpdateResult]:
    results = [update_pouch_home(strict=strict_pouch)]
    results.extend(update_cloned_catalogs(strict=False))
    results.extend(update_custom_sources(strict=False))
    return results


def load_state(path: Path | None = None) -> dict[str, Any]:
    path = path or STATE_FILE
    if not path.is_file():
        return {}
    data = yaml_io.safe_load(path.read_text(encoding="utf-8")) or {}
    return data if isinstance(data, dict) else {}


def save_state(data: dict[str, Any], path: Path | None = None) -> None:
    path = path or STATE_FILE
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        yaml_io.safe_dump(data, allow_unicode=True, sort_keys=False),
        encoding="utf-8",
    )


def record_auto_update(path: Path | None = None) -> None:
    state = load_state(path)
    state["last_auto_update"] = datetime.now(timezone.utc).isoformat()
    save_state(state, path)


def _interval_elapsed(state: dict[str, Any], interval_seconds: int) -> bool:
    if interval_seconds <= 0:
        return True
    raw = state.get("last_auto_update")
    if not raw:
        return True
    try:
        last = datetime.fromisoformat(str(raw))
    except ValueError:
        return True
    if last.tzinfo is None:
        last = last.replace(tzinfo=timezone.utc)
    elapsed = (datetime.now(timezone.utc) - last).total_seconds()
    return elapsed >= interval_seconds


def _env_disabled() -> bool:
    return os.environ.get("POUCH_NO_AUTO_UPDATE", "").strip().lower() in _TRUTHY


def _acquire_lock(path: Path) -> Any | None:
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        lock_file = path.open("a+")
    except OSError:
        return None
    try:
        import fcntl

        fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        lock_file.close()
        return None
    return lock_file


def _release_lock(lock_file: Any | None) -> None:
    if lock_file is None:
        return
    try:
        import fcntl

        fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
    except OSError:
        pass
    try:
        lock_file.close()
    except OSError:
        pass


def reset_refresh_cache() -> None:
    _refreshed_paths.clear()


def report_results(results: Iterable[UpdateResult], *, prefix: str = "[pouch]") -> None:
    changed = [item.name for item in results if item.changed]
    failed = [item for item in results if item.status == "failed"]
    if changed:
        _print_err(f"{prefix} 已自动更新: {', '.join(changed)}")
    for item in failed:
        detail = f": {item.detail}" if item.detail else ""
        _print_err(f"{prefix} 更新 {item.name} 失败，已跳过{detail}")


def maybe_auto_update(*, command: str | None = None, disabled: bool = False) -> list[UpdateResult]:
    if disabled or command in _SKIP_COMMANDS or _env_disabled():
        return []
    if not POUCH_HOME.is_dir():
        return []

    settings = load_auto_update_settings()
    if not settings.enabled:
        return []
    state = load_state()
    if not _interval_elapsed(state, settings.interval_seconds):
        return []

    lock_file = _acquire_lock(STATE_FILE.with_name("auto-update.lock"))
    if lock_file is None:
        return []
    try:
        # 拿到锁后再读一次，避免并发重复拉取。
        state = load_state()
        if not _interval_elapsed(state, settings.interval_seconds):
            return []
        results = run_updates(strict_pouch=False)
        record_auto_update()
        report_results(results)
        return results
    except Exception as exc:  # noqa: BLE001 — 自动更新不得打断用户命令
        _print_err(f"[pouch] 自动更新失败，已跳过: {exc}")
        return []
    finally:
        _release_lock(lock_file)