from __future__ import annotations

import json
import logging
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from niannian.config import resolve_env_value
from niannian.storage.manifest import utc_now_iso

logger = logging.getLogger(__name__)

ALERTS_PATH = "sync/tier_alerts.jsonl"
MACOS_NOTIFY_PATH = "sync/tier_macos_notify.jsonl"


class TierNotifier:
    """三云 tier 告警：append-only jsonl + health recent_alerts；同 (code, provider) dedupe。"""

    def __init__(self, home: Path, cfg: dict[str, Any] | None = None) -> None:
        self._home = home
        self._cfg = cfg or {}
        self._path = home / ALERTS_PATH
        self._macos_path = home / MACOS_NOTIFY_PATH
        self._channels = self._cfg.get("channels") or {}

    @property
    def enabled(self) -> bool:
        return bool(self._cfg.get("enabled", True))

    def emit(self, code: str, *, severity: str = "warning", message: str = "", **fields: Any) -> dict[str, Any] | None:
        if not self.enabled:
            return None
        provider = fields.get("provider") or fields.get("to") or fields.get("from")
        if self._is_duplicate(code, str(provider) if provider else None):
            return None
        row: dict[str, Any] = {
            "code": code,
            "severity": severity,
            "message": message,
            "at": utc_now_iso(),
            **fields,
        }
        self._path.parent.mkdir(parents=True, exist_ok=True)
        with self._path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(row, ensure_ascii=False) + "\n")
        self._dispatch_channels(row)
        return row

    def _dispatch_channels(self, row: dict[str, Any]) -> None:
        channels = self._channels
        if channels.get("console", True) is False:
            pass
        webhook_cfg = channels.get("webhook") or {}
        if isinstance(webhook_cfg, dict) and webhook_cfg.get("enabled"):
            self._send_webhook(row, webhook_cfg)
        macos_cfg = channels.get("macos") or {}
        if isinstance(macos_cfg, dict) and macos_cfg.get("enabled"):
            self._enqueue_macos(row)

    def _send_webhook(self, row: dict[str, Any], webhook_cfg: dict[str, Any]) -> None:
        url = resolve_env_value(webhook_cfg.get("url"))
        if not url:
            logger.debug("tier webhook enabled but url missing")
            return
        provider = row.get("provider") or row.get("to") or row.get("from")
        payload = {
            "event": f"tier.{row.get('code')}",
            "severity": row.get("severity", "warning"),
            "provider": provider,
            "active_role": row.get("active_role"),
            "message": row.get("message") or row.get("code"),
            "at": row.get("at"),
        }
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        headers = {"Content-Type": "application/json"}
        secret = resolve_env_value(webhook_cfg.get("secret_header"))
        if secret:
            headers["X-Tier-Secret"] = str(secret)
        req = urllib.request.Request(str(url), data=body, headers=headers, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=8) as resp:
                if resp.status >= 400:
                    logger.warning("tier webhook HTTP %s", resp.status)
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            logger.warning("tier webhook failed: %s", exc)

    def _enqueue_macos(self, row: dict[str, Any]) -> None:
        """写入 bridge 可消费的 jsonl；DeviceGateway / bridge-client 发本地通知。"""
        provider = row.get("provider") or row.get("to") or row.get("from") or "tier"
        title = "念念 · 云盘"
        message = str(row.get("message") or row.get("code") or "tier 事件")
        out = {
            "title": title,
            "body": f"[{provider}] {message}",
            "severity": row.get("severity", "warning"),
            "code": row.get("code"),
            "provider": provider,
            "at": row.get("at"),
        }
        self._macos_path.parent.mkdir(parents=True, exist_ok=True)
        with self._macos_path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(out, ensure_ascii=False) + "\n")

    def recent_alerts(self, limit: int = 20) -> list[dict[str, Any]]:
        if not self._path.is_file():
            return []
        try:
            lines = self._path.read_text(encoding="utf-8").strip().splitlines()
        except OSError:
            return []
        out: list[dict[str, Any]] = []
        for line in reversed(lines):
            if not line.strip():
                continue
            try:
                out.append(json.loads(line))
            except json.JSONDecodeError:
                continue
            if len(out) >= limit:
                break
        return out

    def _is_duplicate(self, code: str, provider: str | None) -> bool:
        hours = float(self._cfg.get("dedupe_hours", 6))
        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
        for row in self.recent_alerts(limit=200):
            if row.get("code") != code:
                continue
            row_provider = row.get("provider") or row.get("to") or row.get("from")
            if str(row_provider or "") != str(provider or ""):
                continue
            at_raw = row.get("at")
            if not at_raw:
                continue
            try:
                at = datetime.fromisoformat(str(at_raw).replace("Z", "+00:00"))
            except ValueError:
                continue
            if at >= cutoff:
                return True
        return False
