"""Owner-facing cloud-efficiency digest (站内 + 飞书, per-owner aggregate).

The IT-only weekly digest (``cmdb.efficiency_digest``) tells IT admins
about the whole fleet. This module reaches the people who can actually
act: each server owner / subscription 业主 with PENDING actionable
recommendations gets ONE aggregated card per cycle — never one message
per machine — listing their top savings items with a deep link to
``/my/cloud/efficiency``.

Red-line compliance (2026-07-22 decision, DECISIONS.md): pushing to
employees/managers is normally forbidden, so this ships **default OFF**
behind the ``cloud_efficiency.owner_digest`` platform setting. The admin
console (效率页 → 触达设置) exposes the real switch, an optional canary
allowlist, and a recipient PREVIEW ("who would get what") so IT can
evaluate the blast radius before enabling. The weekly cron row stays
enabled; the platform setting is the single source of truth.

Aggregation contract:
* Only PENDING items count — verdict idle/oversized with no live triage
  (keep/acknowledge/snooze suppress; drifted/expired triage re-opens),
  same ``_effective_open_clause`` the 待处理 queue uses.
* Effective owner = stored ``owner_person_id``, else the owner inherited
  from the cloud subscription (``resolve_inherited_server_owners``).
* Owners must be ACTIVE employees (leavers / soft-deleted are dropped).
* Per-recipient weekly dedup via the notification outbox
  (``related_resource_type='cloud_efficiency_owner_digest'`` within the
  trailing 6 days), so a job re-run never double-sends.
"""
from __future__ import annotations

import logging
from collections.abc import Collection
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.efficiency_service import (
    _actionable_conds,
    _effective_open_clause,
)
from app.domains.cmdb.models import (
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerEfficiencyTriage,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

OWNER_DIGEST_SETTING_KEY = "cloud_efficiency.owner_digest"
DIGEST_RESOURCE_TYPE = "cloud_efficiency_owner_digest"

# The scope ``environment`` vocabulary (CmdbCloudScope.environment). Config
# values outside this set are dropped on load so a typo can never silently
# exclude nothing / everything.
KNOWN_ENVIRONMENTS: frozenset[str] = frozenset({"dev", "staging", "prod"})

# A job re-run inside this window must not double-send to the same person.
_DEDUP_WINDOW_DAYS = 6


@dataclass
class OwnerDigestConfig:
    """Runtime config, stored in ``platform_settings``. ``enabled`` is the
    single real switch (the cron row stays on and no-ops when False)."""

    enabled: bool = False
    # Non-empty ⇒ canary mode: only these person ids receive the digest.
    canary_person_ids: list[str] = field(default_factory=list)
    # Top-N items listed inside one card.
    max_items_per_card: int = 5
    # Minimum per-recipient estimated monthly savings (USD) required to send
    # a card. A recipient whose flagged machines carry NO attributable cost
    # (savings null/0 — e.g. AWS/GKE boxes with no per-instance cost fact,
    # see efficiency.estimate_savings) would otherwise get an embarrassing
    # "预计每月可省 $0" nudge. The hard floor is 0 → recipients with a $0
    # aggregate are ALWAYS dropped; a positive value raises the bar further.
    # Mirrors the fleet-wide guard the IT digest already applies
    # (handlers.run_efficiency_digest: skip when savings <= 0).
    min_savings_usd: float = 0.0
    # Machines whose cloud subscription (scope) carries one of these
    # environments are dropped from the digest entirely. A machine inherits
    # its subscription's environment (dev / staging / prod); machines that
    # can't be attributed to any subscription are NOT dropped — unknown is
    # not "excluded". Typical use: ["prod"] — production capacity is
    # managed outside the owner's self-service remit, so nudging owners
    # about prod boxes is noise.
    exclude_environments: list[str] = field(default_factory=list)

    @classmethod
    def from_setting(cls, value: Any) -> OwnerDigestConfig:
        if not isinstance(value, dict):
            return cls()
        raw_ids = value.get("canary_person_ids") or []
        raw_envs = value.get("exclude_environments") or []
        try:
            min_sav = float(value.get("min_savings_usd") or 0.0)
        except (TypeError, ValueError):
            min_sav = 0.0
        return cls(
            enabled=bool(value.get("enabled", False)),
            canary_person_ids=[str(x) for x in raw_ids if x],
            max_items_per_card=max(
                1, min(20, int(value.get("max_items_per_card") or 5)),
            ),
            exclude_environments=[
                str(x) for x in raw_envs if x in KNOWN_ENVIRONMENTS
            ],
            min_savings_usd=max(0.0, min_sav),
        )

    def to_setting(self) -> dict[str, Any]:
        return {
            "enabled": self.enabled,
            "canary_person_ids": self.canary_person_ids,
            "max_items_per_card": self.max_items_per_card,
            "exclude_environments": self.exclude_environments,
            "min_savings_usd": self.min_savings_usd,
        }


async def load_owner_digest_config(db: AsyncSession) -> OwnerDigestConfig:
    row = await db.get(PlatformSetting, OWNER_DIGEST_SETTING_KEY)
    return OwnerDigestConfig.from_setting(row.value if row else None)


async def save_owner_digest_config(
    db: AsyncSession, config: OwnerDigestConfig, *, updated_by: Any = None,
) -> None:
    row = await db.get(PlatformSetting, OWNER_DIGEST_SETTING_KEY)
    if row is None:
        row = PlatformSetting(
            key=OWNER_DIGEST_SETTING_KEY,
            value=config.to_setting(),
            description=(
                "云成本优化 owner digest(站内+飞书,按 owner 每周聚合一张卡)。"
                "enabled=false 时周任务静默跳过;canary_person_ids 非空=灰度。"
            ),
            updated_by=updated_by,
        )
        db.add(row)
    else:
        row.value = config.to_setting()
        row.updated_by = updated_by
    await db.flush()


@dataclass
class OwnerDigestEntry:
    person_id: str
    display_name: str | None
    work_email: str | None
    server_count: int = 0
    estimated_monthly_savings_usd: float = 0.0
    # Top items by savings desc: {name, instance_id, verdict,
    # recommendation, savings, scope_name}
    items: list[dict[str, Any]] = field(default_factory=list)


async def collect_owner_digest(
    db: AsyncSession, *, max_items: int = 5,
    exclude_environments: Collection[str] = (),
    min_savings_usd: float = 0.0,
) -> list[OwnerDigestEntry]:
    """Aggregate PENDING actionable recommendations per effective owner.
    Returns entries sorted by estimated savings desc; owners that are not
    active employees are dropped. ``exclude_environments`` drops machines
    whose cloud subscription carries one of those environments (a machine
    inherits its subscription's environment); machines with no resolvable
    subscription are kept.

    ``min_savings_usd`` drops any recipient whose aggregate estimated
    monthly savings is at or below the floor: a recipient whose flagged
    machines carry no attributable cost (savings null/0) is NEVER worth a
    "预计每月可省 $0" nudge. The floor is always at least 0, so $0/negative
    aggregates are dropped even when the caller passes 0 — mirrors the
    fleet-wide guard the IT digest applies (handlers.run_efficiency_digest,
    'skip when savings <= 0')."""
    t = CmdbServerEfficiencyTriage
    rows = (
        await db.execute(
            select(CmdbServerEfficiency, CmdbServer)
            .join(CmdbServer, CmdbServer.id == CmdbServerEfficiency.server_id)
            .join(t, t.server_id == CmdbServerEfficiency.server_id, isouter=True)
            .where(
                CmdbServer.deleted_at.is_(None),
                *_actionable_conds(),
                _effective_open_clause(),
            )
        )
    ).all()
    if not rows:
        return []
    pairs: list[tuple[CmdbServerEfficiency, CmdbServer]] = [
        (eff, srv) for eff, srv in rows
    ]

    # Effective owner: stored owner wins; otherwise the subscription owner
    # (batch-resolved once for all unowned servers).
    from app.domains.cmdb.service import (
        resolve_inherited_server_owners,
        resolve_server_scope_ids,
    )

    if exclude_environments:
        excluded = {str(e) for e in exclude_environments}
        # Resolve EVERY candidate server to its subscription (owned servers
        # included — an explicitly-owned prod box is still a prod box) and
        # look up the subscription environments in one query.
        scope_by_server = await resolve_server_scope_ids(
            db, [srv for _, srv in pairs],
        )
        scope_env: dict[str, str] = {}
        scope_ids = set(scope_by_server.values())
        if scope_ids:
            from app.domains.cmdb.models import CmdbCloudScope

            env_rows = await db.execute(
                select(CmdbCloudScope.id, CmdbCloudScope.environment).where(
                    CmdbCloudScope.id.in_(scope_ids),
                )
            )
            scope_env = {str(sid): env for sid, env in env_rows.all()}
        pairs = [
            (eff, srv) for eff, srv in pairs
            if scope_env.get(scope_by_server.get(str(srv.id), ""), None)
            not in excluded
        ]
        if not pairs:
            return []

    unowned = [srv for _, srv in pairs if not srv.owner_person_id]
    inherited = (
        await resolve_inherited_server_owners(db, unowned) if unowned else {}
    )

    by_owner: dict[str, list[tuple[CmdbServerEfficiency, CmdbServer, str | None]]] = {}
    for eff, srv in pairs:
        scope_name = None
        if srv.owner_person_id:
            owner = str(srv.owner_person_id)
        else:
            info = inherited.get(str(srv.id))
            if not info or not info.get("person_id"):
                continue  # nobody answers for this box — IT digest covers it
            owner = str(info["person_id"])
            scope_name = info.get("scope_name")
        by_owner.setdefault(owner, []).append((eff, srv, scope_name))

    if not by_owner:
        return []

    # Only active, non-deleted employees receive anything.
    people = {
        str(p.id): p
        for p in (
            await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_(list(by_owner.keys())),
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                )
            )
        ).scalars()
    }

    entries: list[OwnerDigestEntry] = []
    for owner_id, triples in by_owner.items():
        person = people.get(owner_id)
        if person is None:
            continue
        total = Decimal("0")
        items: list[dict[str, Any]] = []
        for eff, srv, scope_name in triples:
            sv = eff.estimated_monthly_savings_usd
            if sv is not None:
                total += Decimal(str(sv))
            items.append({
                "server_id": str(srv.id),
                "name": srv.name or srv.instance_id,
                "instance_id": srv.instance_id,
                "verdict": eff.verdict,
                "recommendation": eff.recommendation,
                "estimated_monthly_savings_usd": (
                    float(sv) if sv is not None else None
                ),
                "scope_name": scope_name,
            })
        # Drop recipients with no meaningful savings to pitch. The floor is
        # clamped to >= 0 so a $0/negative aggregate is always dropped even
        # when min_savings_usd is 0 — never send "预计每月可省 $0".
        if float(total) <= max(0.0, min_savings_usd):
            continue
        items.sort(
            key=lambda x: x["estimated_monthly_savings_usd"] or 0,
            reverse=True,
        )
        entries.append(OwnerDigestEntry(
            person_id=owner_id,
            display_name=person.display_name,
            work_email=person.work_email,
            server_count=len(triples),
            estimated_monthly_savings_usd=float(total),
            items=items[:max_items],
        ))

    entries.sort(key=lambda e: e.estimated_monthly_savings_usd, reverse=True)
    return entries


async def _recently_sent_person_ids(
    db: AsyncSession, person_ids: list[str],
) -> set[str]:
    """Recipients who already got an owner digest inside the dedup window."""
    from app.notifications.models import NotificationOutbox

    if not person_ids:
        return set()
    cutoff = datetime.now(UTC) - timedelta(days=_DEDUP_WINDOW_DAYS)
    rows = await db.execute(
        select(NotificationOutbox.recipient_person_id).where(
            NotificationOutbox.related_resource_type == DIGEST_RESOURCE_TYPE,
            NotificationOutbox.recipient_person_id.in_(person_ids),
            NotificationOutbox.created_at >= cutoff,
        )
    )
    return {str(pid) for pid in rows.scalars() if pid}


def _clean(v: object, limit: int = 60) -> str:
    """Cloud resource names are attacker-influenceable — strip control +
    Feishu-markdown chars and bound the length before card interpolation."""
    import re

    s = str(v or "")
    s = re.sub(r"[\x00-\x1f\x7f`*_~\[\]<>|]", " ", s).strip()
    return s[:limit] or "(unnamed)"


_VERDICT_ZH = {"idle": "闲置", "oversized": "超配"}
_RECO_ZH = {
    "stop": "建议关停", "downsize": "建议降配", "investigate": "建议排查",
    "keep": "建议保留", "none": "",
}


async def send_owner_digest(db: AsyncSession) -> dict[str, Any]:
    """Cron body: aggregate → filter (canary, dedup) → one card per owner.
    Best-effort per recipient; returns a summary for the job result."""
    from app.notifications.service import (
        CardStat,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    config = await load_owner_digest_config(db)
    if not config.enabled:
        return {"status": "skipped", "reason": "owner digest disabled"}

    entries = await collect_owner_digest(
        db, max_items=config.max_items_per_card,
        exclude_environments=config.exclude_environments,
        min_savings_usd=config.min_savings_usd,
    )
    if not entries:
        return {"status": "skipped", "reason": "no pending recommendations"}

    if config.canary_person_ids:
        canary = set(config.canary_person_ids)
        entries = [e for e in entries if e.person_id in canary]
        if not entries:
            return {
                "status": "skipped",
                "reason": "no pending recommendations for canary recipients",
            }

    already = await _recently_sent_person_ids(
        db, [e.person_id for e in entries],
    )

    persons = {
        str(p.id): p
        for p in (
            await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_(
                        [e.person_id for e in entries if e.person_id not in already],
                    )
                )
            )
        ).scalars()
    }

    trace_stamp = datetime.now(UTC).strftime("%G-W%V")  # ISO week
    sent = 0
    failed = 0
    for entry in entries:
        if entry.person_id in already:
            continue
        person = persons.get(entry.person_id)
        if person is None:
            continue
        lines = []
        for it in entry.items:
            sv = it["estimated_monthly_savings_usd"]
            verdict = _VERDICT_ZH.get(it["verdict"], it["verdict"])
            reco = _RECO_ZH.get(it["recommendation"], it["recommendation"])
            scope_hint = (
                f"〔{_clean(it['scope_name'], 30)}〕" if it["scope_name"] else ""
            )
            lines.append(
                f"· {_clean(it['name'])}{scope_hint} — {verdict}"
                f"{' → ' + reco if reco else ''}"
                f"{f',预估月省 ${sv:,.0f}' if sv else ''}"
            )
        more = entry.server_count - len(entry.items)
        if more > 0:
            lines.append(f"· …另有 {more} 台,详见效率页")
        body_text = (
            f"你名下(含你作为业主的云订阅)有 {entry.server_count} 台机器"
            f"存在待处理的成本优化建议,按建议执行预计每月可省 "
            f"${entry.estimated_monthly_savings_usd:,.0f}。\n\n"
            + "\n".join(lines)
            + "\n\n可在「我的云资产 → 效率与成本优化」逐台处理:去优化、"
            "标记保留(需说明原因)或暂缓提醒。此为只读洞察,Helix 不会"
            "自动改动你的云资源。"
        )
        envelope = NotificationEnvelope(
            recipient=person,
            title=(
                f"💡 你有 {entry.server_count} 台云机器可优化 — "
                f"预计每月可省 ${entry.estimated_monthly_savings_usd:,.0f}"
            ),
            body_text=body_text,
            channels=["feishu"],
            card_subtitle=LocalizedText(
                zh="云成本优化 · 按机器归属聚合的每周提醒(可在效率页处理后不再提示)",
                en=(
                    "Cloud cost optimization · weekly reminder grouped by "
                    "machine ownership (handle them on the efficiency page "
                    "to stop the reminders)"
                ),
                ja=(
                    "クラウドコスト最適化 · マシンの担当ごとにまとめた週次リマインド"
                    "(効率ページで対応すると通知は届かなくなります)"
                ),
            ),
            card_stats=[
                CardStat(
                    label=LocalizedText(
                        zh="待处理机器",
                        en="Machines pending action",
                        ja="未対応のマシン",
                    ),
                    value=str(entry.server_count),
                ),
                CardStat(
                    label=LocalizedText(
                        zh="预估可省 / 月",
                        en="Estimated savings / month",
                        ja="削減見込み / 月",
                    ),
                    value=f"${entry.estimated_monthly_savings_usd:,.0f}",
                ),
            ],
            url="/my/cloud/efficiency",
            related_resource_type=DIGEST_RESOURCE_TYPE,
        )
        try:
            svc = NotificationService(db)
            await svc.dispatch(
                envelope,
                trace_id=f"cmdb.efficiency_owner_digest:{trace_stamp}:{entry.person_id}",
                actor_user_id=None,
            )
            await db.commit()
            sent += 1
        except Exception as exc:  # noqa: BLE001 — best-effort per recipient
            failed += 1
            logger.warning(
                "owner digest dispatch to %s failed: %s",
                entry.work_email, str(exc)[:160],
            )
            await db.rollback()

    return {
        "status": "ok",
        "recipients": len(entries),
        "deduped": len(already),
        "dispatched": sent,
        "failed": failed,
        "canary": bool(config.canary_person_ids),
        "exclude_environments": config.exclude_environments,
    }