from __future__ import annotations

import logging
import os
import re
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from niannian.storage.cloud_auth.legacy_import import import_legacy_providers
from niannian.storage.cloud_auth.refresh_baidu import refresh_baidu
from niannian.storage.cloud_auth.refresh_feimao import refresh_feimao
from niannian.storage.cloud_auth.refresh_google_drive import refresh_google_drive
from niannian.storage.cloud_auth.refresh_onedrive import refresh_onedrive
from niannian.storage.cloud_auth.refresh_quark import refresh_quark
from niannian.storage.cloud_auth.vault import read_vault, resolve_credentials_path, update_vault, write_vault
from niannian.storage.cloud_base import resolve_remote_root
from niannian.storage.manifest import utc_now_iso
from niannian.storage.quark_live import looks_like_quark_cookie
from niannian.storage.tier_chain import is_feimao_disabled
from niannian.storage.tier_notifier import TierNotifier

logger = logging.getLogger(__name__)

PROVIDERS = ("quark", "baidu", "onedrive", "google_drive")
_LEGACY_PROVIDERS = ("feimao",)
_REFRESHERS = {
    "quark": refresh_quark,
    "baidu": refresh_baidu,
    "onedrive": refresh_onedrive,
    "google_drive": refresh_google_drive,
}


@dataclass
class RefreshResult:
    ok: bool
    provider: str
    status: str = "connected"
    message: str = ""


@dataclass
class CloudAuthStatus:
    auth: str
    expires_in_hours: float | None = None
    error: str | None = None
    role: str | None = None
    reachability: str | None = None
    reachability_hint: str | None = None
    failover_eligible: bool | None = None
    mirror_status: str | None = None

    def to_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"auth": self.auth}
        if self.expires_in_hours is not None:
            out["expires_in_hours"] = round(self.expires_in_hours, 1)
        if self.error:
            out["error"] = self.error
        if self.role:
            out["role"] = self.role
        if self.reachability:
            out["reachability"] = self.reachability
        if self.reachability_hint:
            out["reachability_hint"] = self.reachability_hint
        if self.failover_eligible is not None:
            out["failover_eligible"] = self.failover_eligible
        if self.mirror_status:
            out["mirror_status"] = self.mirror_status
        return out


def _parse_expires(expires_at: str | None) -> datetime | None:
    if not expires_at:
        return None
    try:
        return datetime.fromisoformat(str(expires_at).replace("Z", "+00:00"))
    except ValueError:
        return None


class CloudCredentialManager:
    """Vault 鉴权 — get/refresh/login + legacy import。"""

    def __init__(self, home: Path, *, config: dict[str, Any] | None = None) -> None:
        self._home = home
        self._config = config or {}
        self._path = resolve_credentials_path(home, self._config)
        cred_cfg = (self._config.get("storage") or {}).get("credentials") or {}
        self._refresh_cfg = cred_cfg.get("refresh") or {}
        self._imported_legacy = False
        self._quark_persist_last_at = 0.0
        self._boot_phase = self._is_docker_mode()

    def _is_docker_mode(self) -> bool:
        mode = os.environ.get("NIANNIAN_MODE", (self._config.get("runtime") or {}).get("mode", "auto"))
        return mode == "docker"

    def _auth_cfg(self) -> dict[str, Any]:
        return dict(((self._config.get("storage") or {}).get("cloud") or {}).get("auth") or {})

    def boot_refresh_skip_providers(self) -> tuple[str, ...]:
        skip = self._auth_cfg().get("boot_refresh_skip") or ["onedrive", "google_drive"]
        return tuple(str(p) for p in skip)

    def _should_skip_boot_refresh(self, provider: str) -> bool:
        if not self._boot_phase or not self._is_docker_mode():
            return False
        return provider in self.boot_refresh_skip_providers()

    def end_boot_phase(self) -> None:
        self._boot_phase = False

    @property
    def boot_phase(self) -> bool:
        return self._boot_phase

    @property
    def path(self) -> Path:
        return self._path

    def ensure_legacy_import(self, providers: dict[str, Any] | None = None) -> int:
        if self._imported_legacy and self._path.is_file():
            return 0
        vault = read_vault(self._path)
        count = import_legacy_providers(vault, providers)
        if count:
            write_vault(self._path, vault)
            logger.info("legacy_import: imported %s provider token(s) into vault", count)
        self._imported_legacy = True
        return count

    def _entry(self, provider: str) -> dict[str, Any]:
        vault = read_vault(self._path)
        return dict(vault.get(provider) or {})

    def _tier_notify_cfg(self) -> dict[str, Any]:
        return dict((self._config.get("notifications") or {}).get("tier") or {})

    def _emit_auth_alert(self, provider: str, message: str) -> None:
        TierNotifier(self._home, self._tier_notify_cfg()).emit(
            "auth_refresh_failed",
            provider=provider,
            severity="warning",
            message=message or f"{provider} 鉴权失效，请重新登录",
        )

    def get_access_token(self, provider: str, *, allow_refresh: bool | None = None) -> str | None:
        entry = self._entry(provider)
        if provider == "feimao":
            raw = entry.get("session_token")
            return str(raw) if raw else None
        if provider == "quark" and entry.get("cookie"):
            return None

        token = entry.get("access_token")
        if token:
            expires = _parse_expires(entry.get("expires_at"))
            if expires and expires <= datetime.now(timezone.utc):
                may_refresh = allow_refresh
                if may_refresh is None:
                    may_refresh = not self._should_skip_boot_refresh(provider)
                if may_refresh:
                    result = self.refresh(provider)
                    entry = self._entry(provider)
                    if not result.ok and result.status != "degraded":
                        return str(token) if entry.get("access_token") else None
                    token = entry.get("access_token")
                    expires = _parse_expires(entry.get("expires_at"))
                else:
                    return str(token)
            if token and (not expires or expires > datetime.now(timezone.utc)):
                return str(token)
            if token and entry.get("status") in ("connected", "degraded"):
                return str(token)

        if entry.get("status") not in ("connected", "degraded"):
            return None
        if provider in ("baidu", "onedrive", "google_drive"):
            return None
        refresh = entry.get("refresh_token")
        return str(refresh) if refresh else None

    def get_refresh_token(self, provider: str) -> str | None:
        entry = self._entry(provider)
        raw = entry.get("refresh_token")
        return str(raw) if raw else None

    def get_session_token(self, provider: str) -> str | None:
        if provider != "feimao":
            return None
        entry = self._entry(provider)
        raw = entry.get("session_token")
        return str(raw) if raw else None

    def get_cookie(self, provider: str) -> str | None:
        if provider != "quark":
            return None
        entry = self._entry(provider)
        raw = entry.get("cookie")
        return str(raw) if raw else None

    def set_login_token(self, provider: str, token: str, *, kind: str | None = None) -> None:
        if provider not in PROVIDERS and provider not in _LEGACY_PROVIDERS:
            raise ValueError(f"unknown provider: {provider}")

        def mutator(vault: dict[str, Any]) -> None:
            entry = vault.setdefault(provider, {})
            if provider == "quark" and (kind == "cookie" or looks_like_quark_cookie(token)):
                entry["cookie"] = token
                entry["auth_kind"] = "cookie"
                entry.pop("refresh_token", None)
                entry.pop("access_token", None)
            elif provider == "feimao" or kind == "session_token":
                entry["session_token"] = token
                entry["auth_kind"] = "cookie" if ";" in token and "=" in token else "session"
                entry.pop("cookie", None)
            else:
                entry["refresh_token"] = token
                entry["access_token"] = f"{token[:12]}_access"
                entry.pop("cookie", None)
                entry["auth_kind"] = "bearer"
            entry["status"] = "connected"
            entry["updated_at"] = utc_now_iso()
            entry.pop("error", None)

        update_vault(self._path, mutator)

    @staticmethod
    def _merge_ctoken_into_cookie(cookie: str, session_token: str) -> str:
        token = session_token.strip()
        if not token:
            return cookie
        if re.search(r"\bctoken=", cookie):
            return re.sub(r"\bctoken=[^;]*", f"ctoken={token}", cookie)
        sep = "" if cookie.rstrip().endswith(";") or not cookie.strip() else "; "
        return f"{cookie.rstrip('; ')}{sep}ctoken={token}"

    def update_quark_cookie(
        self,
        cookie: str,
        *,
        session_token: str | None = None,
        min_interval_seconds: float = 60.0,
    ) -> bool:
        """Persist refreshed quark cookie to Vault when changed; throttle frequent writes."""
        new_cookie = str(cookie).strip()
        if session_token:
            new_cookie = self._merge_ctoken_into_cookie(new_cookie, session_token)
        if not new_cookie or not looks_like_quark_cookie(new_cookie):
            return False

        entry = self._entry("quark")
        old_cookie = str(entry.get("cookie") or "")
        if new_cookie == old_cookie:
            return False

        now = time.monotonic()
        if (
            old_cookie
            and self._quark_persist_last_at
            and now - self._quark_persist_last_at < min_interval_seconds
        ):
            return False

        def mutator(vault: dict[str, Any]) -> None:
            item = vault.setdefault("quark", {})
            item["cookie"] = new_cookie
            item["auth_kind"] = "cookie"
            item["status"] = "connected"
            item["updated_at"] = utc_now_iso()
            item.pop("error", None)
            item.pop("refresh_token", None)
            item.pop("access_token", None)

        update_vault(self._path, mutator)
        self._quark_persist_last_at = now
        logger.debug("quark cookie persisted to vault (len=%s)", len(new_cookie))
        return True

    def mark_needs_relogin(self, provider: str, *, error: str = "auth_failed") -> None:
        def mutator(vault: dict[str, Any]) -> None:
            entry = vault.setdefault(provider, {})
            entry["status"] = "needs_relogin"
            entry["error"] = error
            entry["updated_at"] = utc_now_iso()

        update_vault(self._path, mutator)
        self._emit_auth_alert(provider, error)

    def logout(self, provider: str) -> None:
        def mutator(vault: dict[str, Any]) -> None:
            vault.pop(provider, None)

        update_vault(self._path, mutator)

    def refresh(self, provider: str, *, feimao_reachability: str | None = None) -> RefreshResult:
        if provider == "feimao":
            tier_cfg = ((self._config.get("storage") or {}).get("tier") or {})
            if is_feimao_disabled(tier_cfg):
                return RefreshResult(ok=True, provider=provider, status="disabled", message="feimao disabled")
            def mutator(vault: dict[str, Any]) -> None:
                entry = vault.setdefault(provider, {})
                refresh_feimao(entry, reachability=feimao_reachability)
            update_vault(self._path, mutator)
            entry = self._entry(provider)
            status = str(entry.get("status", "needs_relogin"))
            ok = status == "connected"
            msg = str(entry.get("error") or entry.get("refresh_skipped") or ("ok" if ok else "refresh failed"))
            return RefreshResult(ok=ok, provider=provider, status=status, message=msg)

        if provider not in PROVIDERS:
            return RefreshResult(ok=False, provider=provider, status="unknown", message="unknown provider")

        refresher = _REFRESHERS[provider]

        def mutator(vault: dict[str, Any]) -> None:
            entry = vault.setdefault(provider, {})
            if provider in ("baidu", "onedrive", "google_drive"):
                refresher(entry, config=self._config)
            else:
                refresher(entry)

        try:
            update_vault(self._path, mutator)
        except Exception as exc:
            logger.warning("%s refresh vault update failed: %s", provider, exc)
            return RefreshResult(ok=False, provider=provider, status="degraded", message=str(exc)[:200])
        entry = self._entry(provider)
        status = str(entry.get("status", "needs_relogin"))
        ok = status in ("connected", "degraded")
        msg = str(entry.get("error") or entry.get("refresh_skipped") or ("ok" if ok else "refresh failed"))
        if not ok and status == "needs_relogin":
            self._emit_auth_alert(provider, msg)
        return RefreshResult(ok=ok, provider=provider, status=status, message=msg)

    def refresh_if_needed(self, provider: str, *, feimao_reachability: str | None = None) -> RefreshResult:
        if not self._refresh_cfg.get("enabled", True):
            return RefreshResult(ok=True, provider=provider, status="connected", message="refresh disabled")
        entry = self._entry(provider)
        if entry.get("status") not in ("connected", "degraded"):
            return RefreshResult(ok=False, provider=provider, status=str(entry.get("status", "unconfigured")))
        expires = _parse_expires(entry.get("expires_at"))
        before_h = float(self._refresh_cfg.get("before_expiry_hours", 24))
        if expires and expires > datetime.now(timezone.utc) + timedelta(hours=before_h):
            return RefreshResult(ok=True, provider=provider, status="connected", message="not due")
        return self.refresh(provider, feimao_reachability=feimao_reachability)

    def refresh_boot_providers(self, *, feimao_reachability: str | None = None) -> dict[str, RefreshResult]:
        skip = set(self.boot_refresh_skip_providers())
        results: dict[str, RefreshResult] = {}
        for provider in PROVIDERS:
            if provider in skip:
                entry = self._entry(provider)
                results[provider] = RefreshResult(
                    ok=True,
                    provider=provider,
                    status=str(entry.get("status") or "connected"),
                    message="boot_refresh_skipped",
                )
                continue
            results[provider] = self.refresh_if_needed(provider, feimao_reachability=feimao_reachability)
        return results

    @staticmethod
    def _refresh_result_ok(result: RefreshResult) -> bool:
        return result.ok or result.message in ("not due", "refresh disabled", "boot_refresh_skipped")

    def refresh_all_if_needed(self, *, feimao_reachability: str | None = None) -> dict[str, RefreshResult]:
        results = {
            provider: self.refresh_if_needed(provider, feimao_reachability=feimao_reachability)
            for provider in PROVIDERS
        }
        tier_cfg = ((self._config.get("storage") or {}).get("tier") or {})
        if not is_feimao_disabled(tier_cfg):
            results["feimao"] = self.refresh_if_needed("feimao", feimao_reachability=feimao_reachability)
        return results

    def import_legacy_providers(self, providers: dict[str, Any] | None) -> int:
        def mutator(vault: dict[str, Any]) -> None:
            nonlocal count
            count = import_legacy_providers(vault, providers)

        count = 0
        update_vault(self._path, mutator)
        return count

    def status(self) -> dict[str, dict[str, Any]]:
        vault = read_vault(self._path)
        out: dict[str, dict[str, Any]] = {}
        for provider in list(PROVIDERS) + list(_LEGACY_PROVIDERS):
            entry = vault.get(provider) or {}
            auth = str(entry.get("status") or ("connected" if entry else "unconfigured"))
            if not entry:
                auth = "unconfigured"
            if provider == "feimao" and is_feimao_disabled(
                ((self._config.get("storage") or {}).get("tier") or {})
            ):
                auth = "disabled"
            expires = _parse_expires(entry.get("expires_at"))
            expires_in = None
            if expires:
                expires_in = max(0.0, (expires - datetime.now(timezone.utc)).total_seconds() / 3600.0)
            out[provider] = {
                "auth": auth,
                "auth_kind": entry.get("auth_kind") or ("cookie" if entry.get("cookie") else "bearer"),
                "expires_in_hours": round(expires_in, 1) if expires_in is not None else None,
                "error": entry.get("error"),
                "updated_at": entry.get("updated_at"),
            }
        return out

    def credentials_health(
        self,
        *,
        feimao_reachability: dict[str, Any] | None = None,
        mirror_state: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        base = self.status()
        remote_root = resolve_remote_root(self._config)
        feimao_cfg = ((self._config.get("storage") or {}).get("tier") or {}).get("feimao") or {}
        reach = feimao_reachability or {}
        mirror = mirror_state or {}
        vault = read_vault(self._path)

        def _live_mode(provider: str) -> str | None:
            entry = vault.get(provider) or {}
            if not entry or entry.get("status") not in ("connected", None):
                if not entry:
                    return None
            if provider == "quark":
                if entry.get("cookie"):
                    return "cookie_api"
                if entry.get("refresh_token") or entry.get("access_token"):
                    return "bearer"
                return None
            if provider == "baidu":
                if entry.get("access_token") or entry.get("refresh_token"):
                    return "pcs"
                return None
            if provider == "onedrive":
                if entry.get("access_token") or entry.get("refresh_token"):
                    return "graph"
                return None
            if provider == "google_drive":
                if entry.get("access_token") or entry.get("refresh_token"):
                    return "drive"
                return None
            if provider == "feimao":
                if entry.get("session_token"):
                    return "session"
                return None
            return None

        for provider in list(PROVIDERS) + list(_LEGACY_PROVIDERS):
            info = dict(base.get(provider) or {})
            info["remote_root"] = remote_root
            live = _live_mode(provider)
            if live:
                info["live"] = live
            base[provider] = info

        if base.get("google_drive"):
            gd = dict(base["google_drive"])
            gd.setdefault("role", "monthly_only")
            base["google_drive"] = gd

        tier_cfg = (self._config.get("storage") or {}).get("tier") or {}
        feimao = dict(base.get("feimao") or {})
        if is_feimao_disabled(tier_cfg):
            feimao["role"] = "disabled"
            feimao["auth"] = "disabled"
            base["feimao"] = feimao
            return base
        feimao["role"] = feimao_cfg.get("role", "last_resort")
        reach_status = reach.get("status", "unknown")
        feimao["reachability"] = reach_status
        if reach.get("hint"):
            feimao["reachability_hint"] = reach["hint"]
        feimao["failover_eligible"] = reach_status == "ok" and feimao.get("auth") == "connected"
        if mirror.get("feimao_skipped_reason"):
            feimao["mirror_status"] = "skipped"
        elif mirror.get("feimao_status") == "failed":
            feimao["mirror_status"] = "partial"
        elif mirror.get("feimao_status") == "ok":
            feimao["mirror_status"] = "ok"
        base["feimao"] = feimao
        return base
