"""单 provider 全局互斥 — 智谱免费档 1 并发，跨线程/跨进程（API + worker）共享。"""
from __future__ import annotations

import fcntl
import os
import threading
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator

DEFAULT_EXCLUSIVE_PROVIDERS = frozenset({"zhipu"})

_THREAD_LOCKS: dict[str, threading.Lock] = {}
_THREAD_LOCKS_GUARD = threading.Lock()


class ProviderExclusiveBusy(Exception):
    """互斥锁等待超时 — 路由层应降级到下一 slot。"""

    def __init__(self, provider: str, *, timeout_sec: float) -> None:
        self.provider = provider
        self.timeout_sec = timeout_sec
        super().__init__(f"{provider} 正被占用（等待 {timeout_sec:.0f}s 超时）")


def _thread_lock(provider: str) -> threading.Lock:
    with _THREAD_LOCKS_GUARD:
        lock = _THREAD_LOCKS.get(provider)
        if lock is None:
            lock = threading.Lock()
            _THREAD_LOCKS[provider] = lock
        return lock


def is_exclusive_provider(name: str, *, exclusive_names: frozenset[str] | None = None) -> bool:
    return name in (exclusive_names or DEFAULT_EXCLUSIVE_PROVIDERS)


def resolve_exclusive_providers(api_cfg: dict) -> frozenset[str]:
    limits = api_cfg.get("provider_limits") or {}
    names: set[str] = set()
    if isinstance(limits, dict):
        for name, rule in limits.items():
            if isinstance(rule, dict) and rule.get("exclusive"):
                names.add(str(name))
    if not names:
        names.update(DEFAULT_EXCLUSIVE_PROVIDERS)
    return frozenset(names)


def resolve_lock_dir() -> Path:
    """互斥锁目录 — Docker 下用 persist 卷，使 api/worker 跨容器共享。"""
    import os

    persist = os.environ.get("NIANNIAN_PERSIST_DIR")
    if persist:
        return Path(persist).expanduser().resolve() / "locks"
    from niannian.core.config import resolve_data_home

    return resolve_data_home() / "locks"


def resolve_lock_timeout(api_cfg: dict, provider: str, *, default: float = 30.0) -> float:
    limits = api_cfg.get("provider_limits") or {}
    rule = limits.get(provider) if isinstance(limits, dict) else None
    if isinstance(rule, dict) and rule.get("lock_timeout_sec") is not None:
        return max(0.05, float(rule["lock_timeout_sec"]))
    return default


@contextmanager
def exclusive_provider_lock(
    provider: str,
    lock_dir: Path,
    *,
    timeout_sec: float = 30.0,
) -> Iterator[None]:
    """线程锁（同进程）+ 文件锁（API/worker 跨进程）。"""
    name = (provider or "unknown").strip().lower() or "unknown"
    deadline = time.monotonic() + max(0.05, float(timeout_sec))
    tlock = _thread_lock(name)

    while True:
        if tlock.acquire(blocking=False):
            break
        if time.monotonic() >= deadline:
            raise ProviderExclusiveBusy(provider, timeout_sec=timeout_sec) from None
        time.sleep(0.05)

    lock_dir.mkdir(parents=True, exist_ok=True)
    path = lock_dir / f"api_{name}.lock"
    fh = open(path, "a+", encoding="utf-8")
    try:
        while True:
            try:
                fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                break
            except BlockingIOError:
                if time.monotonic() >= deadline:
                    raise ProviderExclusiveBusy(provider, timeout_sec=timeout_sec) from None
                time.sleep(0.2)
        try:
            yield
        finally:
            fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
            fh.flush()
            os.fsync(fh.fileno())
    finally:
        fh.close()
        tlock.release()
