"""Cloud efficiency / cost-optimization REST endpoints (read-only insight).

Mounted under ``/api/v1/cmdb/efficiency``. Reads ``cmdb_server_efficiency``
(produced daily by ``cmdb.efficiency_compute``) + ``cmdb_server_metrics_daily``.

Surface:
  * ``GET  /summary``          — KPI: counts by verdict, total estimated
    monthly savings, per-cloud breakdown.
  * ``GET  /instances``        — paginated, filterable efficiency list
    (verdict / recommendation / cloud / min-savings), sorted by savings.
  * ``GET  /instances/{id}``   — one server's efficiency profile + the
    utilization daily series for the sparkline.
  * ``GET  /rules``            — current rule-engine thresholds.
  * ``PUT  /rules``            — tune thresholds (persisted to
    ``platform_settings['cloud_efficiency.rules']``).

RBAC: admin-tier reads gated by ``can_read_cloud`` (IT admin / cloud admin
/ finance / auditor / management); an ``employee`` sees only servers they
own (self + subordinates). Threshold edits gated by ``can_manage_cloud``.
This feature performs NO cloud writes — verdicts are advisory only.
"""
from __future__ import annotations

import logging
import uuid
from dataclasses import asdict
from typing import Any

from fastapi import APIRouter, Depends, Path, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import get_current_user
from app.core.dependencies import (
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import ForbiddenError, NotFoundError, ValidationError
from app.core.rbac import (
    can_manage_cloud,
    can_read_cloud,
    can_view_cloud_resources,
    get_subordinate_person_ids,
)
from app.core.response import paginated_response, success_response
from app.domains.cmdb import efficiency_service
from app.domains.cmdb.efficiency import (
    RULES_SETTING_KEY,
    EfficiencyRules,
    load_rules,
)
from app.domains.cmdb.visibility import (
    OwnerVisibility,
    is_server_visible,
    resolve_owner_visibility,
)
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="/api/v1/cmdb/efficiency",
    tags=["cmdb-efficiency"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


def _validate_optional_uuid(value: str | None, field: str) -> None:
    if value is None:
        return
    try:
        uuid.UUID(value)
    except (ValueError, AttributeError, TypeError):
        raise ValidationError(f"{field} is not a valid id") from None


async def _owner_scope(
    db: AsyncSession, user: AuthUser,
) -> OwnerVisibility | None:
    """``None`` → org-wide (admin-tier reader). Otherwise the resolved
    visibility of this user: servers they (or their subordinates) own at
    server level, plus every server attributed to a cloud subscription
    they (or their subordinates) own — the subscription owner is the 业主
    the recommendations are addressed to, so their machines must show up
    on the self-service page (server-level ``owner_person_id`` alone is
    too sparsely populated to carry the page)."""
    if can_read_cloud(user.role_names):
        return None
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    subs = await get_subordinate_person_ids(db, str(user.id))
    person_ids = [str(user.id), *subs] if subs else [str(user.id)]
    return await resolve_owner_visibility(db, person_ids)


async def _owner_scope_for_write(
    db: AsyncSession, user: AuthUser,
) -> OwnerVisibility | None:
    """WRITE-scope twin of :func:`_owner_scope`.

    The org-wide (``None``) branch requires cloud WRITE authority
    (``can_manage_cloud`` = global_admin / cloud_admin). The read-only
    admin tiers (auditor / finance / management) that ``_owner_scope``
    admits as org-wide *readers* must NOT be able to mutate state on
    servers they don't own — they fall through to the owner visibility
    (self + subordinates + owned subscriptions), so a triage write is
    confined to servers they answer for. Anyone below
    ``can_view_cloud_resources`` is refused.
    """
    if can_manage_cloud(user.role_names):
        return None
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    subs = await get_subordinate_person_ids(db, str(user.id))
    person_ids = [str(user.id), *subs] if subs else [str(user.id)]
    return await resolve_owner_visibility(db, person_ids)


@router.get("/summary")
async def efficiency_summary_endpoint(
    request: Request,
    integration_id: str | None = Query(default=None),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    _validate_optional_uuid(integration_id, "integration_id")
    visibility = await _owner_scope(db, user)
    data = await efficiency_service.efficiency_summary(
        db, visibility=visibility, integration_id=integration_id,
    )
    return success_response(data=data, request=request)


@router.get("/instances")
async def efficiency_instances_endpoint(
    request: Request,
    integration_id: str | None = Query(default=None),
    verdict: str | None = Query(default=None),
    recommendation: str | None = Query(default=None),
    min_savings: float | None = Query(default=None, ge=0),
    owner_person_id: str | None = Query(default=None),
    sort: str = Query(default="savings", pattern="^(savings|cost|cpu)$"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=100),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    _validate_optional_uuid(integration_id, "integration_id")
    _validate_optional_uuid(owner_person_id, "owner_person_id")
    visibility = await _owner_scope(db, user)
    items, total = await efficiency_service.list_instances(
        db,
        visibility=visibility,
        integration_id=integration_id,
        verdict=verdict,
        recommendation=recommendation,
        min_savings=min_savings,
        owner_person_id=owner_person_id,
        sort=sort,
        page=page,
        page_size=page_size,
    )
    return paginated_response(
        data=items, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/instances/{server_id}")
async def efficiency_instance_detail_endpoint(
    request: Request,
    server_id: str = Path(...),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    _validate_optional_uuid(server_id, "server_id")
    visibility = await _owner_scope(db, user)
    detail = await efficiency_service.get_instance_detail(db, server_id)
    # A non-owner gets the SAME 404 as a non-existent id (not 403) so the
    # endpoint isn't an existence oracle for servers they can't see.
    if detail is None or (
        visibility is not None
        and not await is_server_visible(db, server_id, visibility)
    ):
        raise NotFoundError("No efficiency profile for this server yet")
    return success_response(data=detail, request=request)


async def _scope_visibility(
    db: AsyncSession, user: AuthUser, scope_id: str,
) -> tuple[Any, OwnerVisibility | None]:
    """Resolve the subscription + the visibility filter to apply to its
    servers. Admin-tier readers and the subscription's own owner see every
    server attributed to the scope (``None``); anyone else is narrowed to
    their own visibility set (self + subordinates + owned subscriptions) —
    matching the self-service scoping of the top-level efficiency
    endpoints. A bad / absent scope id 404s."""
    scope = await efficiency_service.get_scope(db, scope_id)
    if scope is None:
        raise NotFoundError("Subscription not found")
    if can_read_cloud(user.role_names):
        return scope, None
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    if scope.owner_person_id and str(scope.owner_person_id) == str(user.id):
        return scope, None
    subs = await get_subordinate_person_ids(db, str(user.id))
    person_ids = [str(user.id), *subs] if subs else [str(user.id)]
    return scope, await resolve_owner_visibility(db, person_ids)


@router.get("/scopes/{scope_id}/summary")
async def efficiency_scope_summary_endpoint(
    request: Request,
    scope_id: str = Path(...),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Subscription-level rollup of the efficiency verdicts of the servers
    attributed to this cloud subscription (via the cost-facts attribution
    path). Lets an owner see their subscription's waste inside the 云订阅
    view instead of a disconnected global list."""
    _validate_optional_uuid(scope_id, "scope_id")
    scope, visibility = await _scope_visibility(db, user, scope_id)
    data = await efficiency_service.efficiency_summary(
        db, visibility=visibility, scope=scope,
    )
    return success_response(data=data, request=request)


@router.get("/scopes/{scope_id}/instances")
async def efficiency_scope_instances_endpoint(
    request: Request,
    scope_id: str = Path(...),
    actionable_only: bool = Query(default=False),
    pending_only: bool = Query(default=False),
    sort: str = Query(default="savings", pattern="^(savings|cost|cpu)$"),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=50, ge=1, le=100),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    _validate_optional_uuid(scope_id, "scope_id")
    scope, visibility = await _scope_visibility(db, user, scope_id)
    items, total = await efficiency_service.list_instances(
        db,
        visibility=visibility,
        scope=scope,
        actionable_only=actionable_only,
        pending_only=pending_only,
        sort=sort,
        page=page,
        page_size=page_size,
    )
    return paginated_response(
        data=items, total=total, page=page, page_size=page_size, request=request,
    )


class EfficiencyTriagePayload(BaseModel):
    """A non-destructive owner decision on a server's recommendation.
    NEVER executes a cloud action — records intent + clears the queue item.
    """
    action: str = Field(pattern="^(keep|acknowledge|snooze)$")
    reason: str | None = Field(default=None, max_length=1000)
    snooze_days: int | None = Field(default=None, ge=1, le=180)


@router.post("/instances/{server_id}/triage")
async def efficiency_triage_endpoint(
    request: Request,
    payload: EfficiencyTriagePayload,
    server_id: str = Path(...),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Owner (or admin-tier) records a lightweight triage on a server's
    efficiency recommendation: keep-as-is (with reason), acknowledge, or
    snooze. This is the "让大家都参与" participation loop — it does NOT stop
    / decommission / downsize anything; machine lifecycle stays in the
    existing management surface."""
    _validate_optional_uuid(server_id, "server_id")
    if not can_view_cloud_resources(user.role_names):
        raise ForbiddenError()
    if payload.action == "keep" and not (payload.reason and payload.reason.strip()):
        raise ValidationError("A reason is required to keep a flagged server")

    # Triage is a WRITE — use the write-scope so read-only admin tiers
    # (auditor / finance / management) can't mutate triage on servers they
    # don't own (only global_admin / cloud_admin get the org-wide path).
    visibility = await _owner_scope_for_write(db, user)
    # Non-admin may only triage a server inside their visibility (owned at
    # server level, or attributed to a subscription they own) — a foreign /
    # unknown server 404s (same non-oracle convention as the detail read).
    if visibility is not None and not await is_server_visible(db, server_id, visibility):
        raise NotFoundError("No efficiency profile for this server yet")

    row = await efficiency_service.submit_triage(
        db,
        server_id=server_id,
        action=payload.action,
        reason=payload.reason,
        snooze_days=payload.snooze_days,
        actor_person_id=str(user.id),
    )
    if row is None:
        raise NotFoundError("No efficiency profile for this server yet")

    await AuditService(db).log(
        trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.efficiency.triage",
        status="success",
        resource_type="cmdb_server",
        resource_id=server_id,
        detail={
            "action": payload.action,
            "reason": payload.reason,
            "snooze_days": payload.snooze_days,
            "verdict": row.get("verdict"),
        },
    )
    await db.commit()
    return success_response(data=row, request=request)


@router.get("/savings-outcomes")
async def efficiency_savings_outcomes_endpoint(
    request: Request,
    since_days: int | None = Query(default=None, ge=1, le=1830),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Leadership-facing OUTCOME rollup: money actually saved (realized),
    split by method + department, a monthly trend, and the still-open
    estimated opportunity. Read-only, org-wide — gated to the cloud-read
    tier (global_admin / cloud_admin / finance / auditor / management), the
    same audiences the leadership dashboards serve. NOT self-scoped: this is
    a whole-fleet conclusion, so an employee cannot call it."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb import savings_service

    data = await savings_service.savings_outcomes(db, since_days=since_days)
    return success_response(data=data, request=request)


@router.get("/by-department")
async def efficiency_by_department_endpoint(
    request: Request,
    integration_id: str | None = Query(default=None),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    _validate_optional_uuid(integration_id, "integration_id")
    visibility = await _owner_scope(db, user)
    data = await efficiency_service.savings_by_department(
        db, visibility=visibility, integration_id=integration_id,
    )
    return success_response(data=data, request=request)


@router.post("/instances/{server_id}/narrate")
async def efficiency_narrate_endpoint(
    request: Request,
    server_id: str = Path(...),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """On-demand (re)generate the AI narrative for one instance. Admin-tier
    (it triggers an LLM call)."""
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    _validate_optional_uuid(server_id, "server_id")
    from app.domains.cmdb.efficiency_ai import narrate_one
    result = await narrate_one(db, server_id)
    return success_response(data=result, request=request)


# ─── Owner digest (触达设置) ─────────────────────────────────────────

@router.get("/owner-digest/config")
async def owner_digest_config_get_endpoint(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Current owner-digest switch + canary list. Cloud-read tier."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.owner_digest import load_owner_digest_config

    config = await load_owner_digest_config(db)
    return success_response(data=config.to_setting(), request=request)


class OwnerDigestConfigPayload(BaseModel):
    """Partial update; unset fields keep their current value."""
    enabled: bool | None = None
    canary_person_ids: list[str] | None = Field(default=None, max_length=200)
    max_items_per_card: int | None = Field(default=None, ge=1, le=20)
    # Subscription environments whose machines are dropped from the digest
    # (a machine inherits its subscription's environment). Vocabulary is
    # validated against KNOWN_ENVIRONMENTS below.
    exclude_environments: list[str] | None = Field(default=None, max_length=8)
    # Minimum per-recipient estimated monthly savings (USD) to send a card.
    # 0 (the default floor) still drops $0/unattributed recipients; raise it
    # to suppress low-value nudges.
    min_savings_usd: float | None = Field(default=None, ge=0)


@router.put("/owner-digest/config")
async def owner_digest_config_put_endpoint(
    request: Request,
    payload: OwnerDigestConfigPayload,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Flip the owner-digest switch / canary list. Cloud WRITE tier — this
    is the single real gate the weekly cron honours (the cron row stays
    enabled), so the UI switch always reflects the true state."""
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.owner_digest import (
        KNOWN_ENVIRONMENTS,
        load_owner_digest_config,
        save_owner_digest_config,
    )

    for pid in payload.canary_person_ids or []:
        _validate_optional_uuid(pid, "canary_person_ids")
    for env in payload.exclude_environments or []:
        if env not in KNOWN_ENVIRONMENTS:
            raise ValidationError(
                f"exclude_environments: unknown environment {env!r}",
            )
    config = await load_owner_digest_config(db)
    if payload.enabled is not None:
        config.enabled = payload.enabled
    if payload.canary_person_ids is not None:
        config.canary_person_ids = [str(x) for x in payload.canary_person_ids]
    if payload.max_items_per_card is not None:
        config.max_items_per_card = payload.max_items_per_card
    if payload.min_savings_usd is not None:
        config.min_savings_usd = max(0.0, float(payload.min_savings_usd))
    if payload.exclude_environments is not None:
        config.exclude_environments = list(dict.fromkeys(
            payload.exclude_environments,
        ))
    await save_owner_digest_config(db, config, updated_by=user.id)

    await AuditService(db).log(
        trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.efficiency.owner_digest_config",
        status="success",
        resource_type="platform_setting",
        resource_id="cloud_efficiency.owner_digest",
        detail=config.to_setting(),
    )
    await db.commit()
    return success_response(data=config.to_setting(), request=request)


@router.get("/owner-digest/preview")
async def owner_digest_preview_endpoint(
    request: Request,
    exclude_environments: str | None = None,
    min_savings: float | None = Query(default=None, ge=0),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Dry-run of the next digest cycle: who would receive a card, with
    how many pending machines and how much estimated savings — the blast-
    radius check IT runs BEFORE flipping the switch. Read-only, no sends.
    Cloud-read tier.

    ``exclude_environments`` (comma-separated, e.g. ``prod`` or
    ``staging,prod``) overrides the SAVED env filter for this preview only,
    so the settings modal can show the blast radius of a not-yet-saved
    selection. Unknown values 422. Omitted ⇒ the saved config applies;
    an explicit empty string ⇒ no env filter.

    ``min_savings`` (USD, >= 0) likewise overrides the saved per-recipient
    savings floor for this preview only, so the modal reflects a not-yet-
    saved threshold. Omitted ⇒ the saved config floor applies."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from app.domains.cmdb.owner_digest import (
        KNOWN_ENVIRONMENTS,
        collect_owner_digest,
        load_owner_digest_config,
    )

    config = await load_owner_digest_config(db)
    effective_exclude = config.exclude_environments
    if exclude_environments is not None:
        effective_exclude = [
            e for e in (s.strip() for s in exclude_environments.split(",")) if e
        ]
        for env in effective_exclude:
            if env not in KNOWN_ENVIRONMENTS:
                raise ValidationError(
                    f"exclude_environments: unknown environment {env!r}",
                )
    effective_min_savings = config.min_savings_usd
    if min_savings is not None:
        effective_min_savings = max(0.0, float(min_savings))
    entries = await collect_owner_digest(
        db, max_items=config.max_items_per_card,
        exclude_environments=effective_exclude,
        min_savings_usd=effective_min_savings,
    )
    canary = set(config.canary_person_ids)
    data = {
        "config": config.to_setting(),
        "recipients": [
            {
                "person_id": e.person_id,
                "display_name": e.display_name,
                "work_email": e.work_email,
                "server_count": e.server_count,
                "estimated_monthly_savings_usd": e.estimated_monthly_savings_usd,
                "in_canary": (not canary) or (e.person_id in canary),
                "items": e.items,
            }
            for e in entries
        ],
        "totals": {
            "recipients": len(entries),
            "would_send": (
                sum(1 for e in entries if e.person_id in canary)
                if canary else len(entries)
            ),
            "servers": sum(e.server_count for e in entries),
            "estimated_monthly_savings_usd": sum(
                e.estimated_monthly_savings_usd for e in entries
            ),
        },
    }
    return success_response(data=data, request=request)


@router.get("/owner-digest/history")
async def owner_digest_history_endpoint(
    request: Request,
    limit: int = Query(default=8, ge=1, le=30),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """Recent executions of the weekly owner-digest cron — the "did it
    actually fire, and who did it reach?" proof the 触达设置 modal needs.

    The forward-looking preview shows who WOULD get a card next cycle; this
    is the backward-looking record so IT can trust the mechanism is live
    without hunting for the job's UUID in the scheduled-jobs admin page.
    Each run's summary (recipients / dispatched / deduped / failed) is
    lifted straight from the cron's stored ``result_summary``. Read-only,
    cloud-read tier."""
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    from sqlalchemy import select

    from app.domains.cmdb.owner_digest import load_owner_digest_config
    from app.scheduled_jobs import service as sched_service
    from app.scheduled_jobs.models import ScheduledJob

    config = await load_owner_digest_config(db)
    job = (
        await db.execute(
            select(ScheduledJob).where(
                ScheduledJob.handler_key == "cmdb.efficiency_owner_digest",
            )
        )
    ).scalar_one_or_none()

    def _iso(dt: Any) -> str | None:
        return dt.isoformat() if dt else None

    runs_out: list[dict[str, Any]] = []
    job_meta: dict[str, Any] = {}
    if job is not None:
        job_meta = {
            "last_run_at": _iso(job.last_run_at),
            "last_status": job.last_status,
            "run_count": job.run_count,
            "failure_count": job.failure_count,
            "cron_expr": job.cron_expr,
            "timezone": job.timezone,
        }
        runs = await sched_service.list_runs(db, job.id, limit=limit)
        for r in runs:
            summary = r.result_summary if isinstance(r.result_summary, dict) else {}
            runs_out.append({
                "started_at": _iso(r.started_at),
                "finished_at": _iso(r.finished_at),
                "status": r.status,
                "triggered_by": r.triggered_by,
                # Present on a successful send; None on skipped/failed.
                "recipients": summary.get("recipients"),
                "dispatched": summary.get("dispatched"),
                "deduped": summary.get("deduped"),
                "failed": summary.get("failed"),
                # A "skipped" run carries a human reason (disabled / no
                # pending recommendations) instead of counts.
                "skipped_reason": (
                    summary.get("reason") if summary.get("status") == "skipped"
                    else None
                ),
                "exclude_environments": summary.get("exclude_environments") or [],
                "error_message": r.error_message,
            })

    return success_response(
        data={"enabled": config.enabled, "job": job_meta, "runs": runs_out},
        request=request,
    )


@router.get("/rules")
async def efficiency_rules_get_endpoint(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    if not can_read_cloud(user.role_names):
        raise ForbiddenError()
    rules = await load_rules(db)
    return success_response(data=asdict(rules), request=request)


class EfficiencyRulesPayload(BaseModel):
    """Any subset of thresholds; unset fields keep their current value."""
    window_days: int | None = Field(default=None, ge=1, le=90)
    min_data_days: int | None = Field(default=None, ge=1, le=90)
    idle_cpu_p95_max: float | None = Field(default=None, ge=0, le=100)
    idle_net_bytes_max: float | None = Field(default=None, ge=0)
    oversized_cpu_p95_max: float | None = Field(default=None, ge=0, le=100)
    oversized_cpu_max_max: float | None = Field(default=None, ge=0, le=100)
    busy_cpu_p95_min: float | None = Field(default=None, ge=0, le=100)
    mem_high_pct: float | None = Field(default=None, ge=0, le=100)
    mem_idle_pct: float | None = Field(default=None, ge=0, le=100)
    downsize_savings_ratio: float | None = Field(default=None, ge=0, le=1)


@router.put("/rules")
async def efficiency_rules_put_endpoint(
    request: Request,
    payload: EfficiencyRulesPayload,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    overrides = {k: v for k, v in payload.model_dump().items() if v is not None}
    if not overrides:
        raise ValidationError("No threshold values supplied")

    # Merge over existing stored overrides so a partial PUT is additive.
    row = await db.get(PlatformSetting, RULES_SETTING_KEY)
    current = dict(row.value) if (row and isinstance(row.value, dict)) else {}
    current.update(overrides)
    if row is None:
        row = PlatformSetting(
            key=RULES_SETTING_KEY,
            value=current,
            description=(
                "云效率规则引擎阈值(闲置/超配判定 + 降配节省比例)。"
                "留空的键用代码默认。"
            ),
            updated_by=user.id,
        )
        db.add(row)
    else:
        row.value = current
        row.updated_by = user.id
    await db.flush()

    await AuditService(db).log(
        trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.efficiency.rules_update",
        status="success",
        resource_type="platform_setting",
        resource_id=RULES_SETTING_KEY,
        detail={"overrides": overrides},
    )
    await db.commit()

    merged = EfficiencyRules().merged(current)
    return success_response(data=asdict(merged), request=request)