Agent/examples/tencent_realtime_control/operator_control.py

"""Preview and execute confirmed Feishu operator commands."""

from __future__ import annotations

import logging
import uuid
from datetime import datetime, timedelta
from typing import Any

from operator_commands import (
    ACTION_DAY_PAUSE,
    ACTION_RESUME,
    ACTION_STOP,
    ParsedCommand,
)
from storage import (
    advisory_lock,
    clear_operator_pause,
    create_operator_command_with_items,
    find_pending_command_conflict,
    insert_operator_command_item,
    load_ad_states,
    load_operator_command_items,
    load_operator_command,
    load_operator_pauses,
    load_realtime_accounts,
    set_operator_pause,
    transition_operator_command,
    update_operator_command,
    update_operator_command_item,
    upsert_ad_state,
)
from tencent_client import (
    ACTIVE_STATUS,
    SUSPEND_STATUS,
    PostWriteVerificationError,
    TencentWriteNotSentError,
    TencentWriteOutcomeUnknownError,
    TencentWriteRejectedError,
    TencentClient,
    current_bid_fen,
    resolve_bid_field,
)


COMMAND_PENDING = "PENDING_CONFIRMATION"
COMMAND_EXECUTING = "EXECUTING"
COMMAND_SUCCEEDED = "SUCCEEDED"
COMMAND_PARTIAL = "PARTIAL"
COMMAND_FAILED = "FAILED"
COMMAND_CANCELLED = "CANCELLED"
COMMAND_EXPIRED = "EXPIRED"
FINAL_COMMAND_STATUSES = {
    COMMAND_SUCCEEDED,
    COMMAND_PARTIAL,
    COMMAND_FAILED,
    COMMAND_CANCELLED,
    COMMAND_EXPIRED,
}

PAUSE_UNTIL_NEXT_DELIVERY = "UNTIL_NEXT_DELIVERY"
PAUSE_UNTIL_MANUAL = "UNTIL_MANUAL"
logger = logging.getLogger("tencent_realtime_control.operator_control")


def _managed_account_map() -> dict[int, dict[str, Any]]:
    return {
        int(row["account_id"]): row
        for row in load_realtime_accounts()
    }


def resolve_target_accounts(parsed: ParsedCommand) -> list[dict[str, Any]]:
    managed = _managed_account_map()
    if parsed.scope_type == "ALL":
        accounts = list(managed.values())
        if not accounts:
            raise ValueError("当前没有纳入实时控制的账户")
        return accounts
    missing = [account_id for account_id in parsed.account_ids if account_id not in managed]
    if missing:
        raise ValueError(f"账户不在实时控制范围: {missing}")
    return [managed[account_id] for account_id in parsed.account_ids]


def _next_delivery_start(now: datetime, start_hour: int) -> datetime:
    return datetime.combine(
        now.date() + timedelta(days=1),
        datetime.min.time().replace(hour=start_hour),
        now.tzinfo,
    )


def _delivery_end_allows_day(raw_end_date: Any, target_date: str) -> bool:
    end_date = str(raw_end_date or "").strip()
    return end_date in {"", "0"} or end_date >= target_date


def _load_today_metrics(
    client: TencentClient,
    account_id: int,
    adgroup_ids: list[int],
    now: datetime,
) -> dict[int, dict[str, int]]:
    metrics: dict[int, dict[str, int]] = {}
    for start in range(0, len(adgroup_ids), 100):
        metrics.update(
            client.get_today_ad_metrics(
                account_id,
                adgroup_ids[start:start + 100],
                now.date(),
            )
        )
    return metrics


def _operator_pause_is_active(state: dict[str, Any], now: datetime) -> bool:
    mode = str(state.get("operator_pause_mode") or "")
    if mode == PAUSE_UNTIL_MANUAL:
        return True
    if mode != PAUSE_UNTIL_NEXT_DELIVERY:
        return False
    resume_at = state.get("operator_resume_at")
    if not resume_at:
        return False
    if resume_at.tzinfo is None:
        resume_at = resume_at.replace(tzinfo=now.tzinfo)
    return resume_at > now


def preview_write_command(
    parsed: ParsedCommand,
    *,
    now: datetime,
    source_message_id: str,
    chat_id: str,
    sender_open_id: str,
    sender_name: str | None,
    confirmation_ttl_minutes: int,
    start_hour: int,
    raw_text: str = "",
    parse_source: str = "deterministic",
    intent: dict[str, Any] | None = None,
    preview_lock_name: str = "tencent_operator_command_preview",
    tencent: TencentClient | None = None,
) -> dict[str, Any]:
    accounts = resolve_target_accounts(parsed)
    client = tencent or TencentClient()
    next_day = (now.date() + timedelta(days=1)).isoformat()
    preview_items: list[dict[str, Any]] = []
    account_summaries: list[dict[str, Any]] = []
    if parsed.action in {ACTION_DAY_PAUSE, ACTION_STOP}:
        for account in accounts:
            account_id = int(account["account_id"])
            operator_pauses = {
                int(row["adgroup_id"]): row
                for row in load_operator_pauses([account_id])
                if _operator_pause_is_active(row, now)
            }
            ads = [
                ad for ad in client.get_ads(account_id)
                if (
                    (
                        str(ad.get("configured_status") or "") == ACTIVE_STATUS
                        and (
                            parsed.action == ACTION_STOP
                            or not str(ad.get("begin_date") or "")
                            or str(ad.get("begin_date")) <= now.date().isoformat()
                        )
                        and (
                            parsed.action == ACTION_STOP
                            or _delivery_end_allows_day(
                                ad.get("end_date"), next_day
                            )
                        )
                        and (
                            parsed.action == ACTION_STOP
                            or int(ad.get("adgroup_id") or 0)
                            not in operator_pauses
                        )
                    )
                    or (
                        parsed.action == ACTION_STOP
                        and str(ad.get("configured_status") or "")
                        == SUSPEND_STATUS
                        and (
                            operator_pauses.get(
                                int(ad.get("adgroup_id") or 0),
                                {},
                            ).get("operator_pause_mode")
                            == PAUSE_UNTIL_NEXT_DELIVERY
                        )
                    )
                )
            ]
            metrics = _load_today_metrics(
                client, account_id, [int(ad["adgroup_id"]) for ad in ads], now
            )
            account_cost = 0
            for ad in ads:
                adgroup_id = int(ad["adgroup_id"])
                values = metrics.get(adgroup_id, {})
                account_cost += int(values.get("cost_fen") or 0)
                preview_items.append({
                    "account_id": account_id,
                    "audience_name": account.get("audience_name"),
                    "adgroup_id": adgroup_id,
                    "adgroup_name": ad.get("adgroup_name"),
                    "before_status": ad.get("configured_status"),
                    "target_status": (
                        SUSPEND_STATUS
                        if parsed.action == ACTION_STOP
                        else ad.get("configured_status")
                    ),
                    "preview_cost_fen": int(values.get("cost_fen") or 0),
                    "preview_impressions": int(values.get("impressions") or 0),
                    "preview_clicks": int(values.get("clicks") or 0),
                    "preview_conversions": int(values.get("conversions") or 0),
                    "previewed_at": now,
                    "execution_status": "PREVIEWED",
                })
            account_summaries.append({
                "account_id": account_id,
                "ad_count": len(ads),
                "cost_fen": account_cost,
            })
    elif parsed.action == ACTION_RESUME:
        pauses_by_account: dict[int, list[dict[str, Any]]] = {}
        for state in load_operator_pauses([int(row["account_id"]) for row in accounts]):
            if not _operator_pause_is_active(state, now):
                continue
            pauses_by_account.setdefault(int(state["account_id"]), []).append(state)
        for account in accounts:
            account_id = int(account["account_id"])
            states = pauses_by_account.get(account_id, [])
            ads = {
                int(ad.get("adgroup_id") or 0): ad
                for ad in client.get_ads(account_id)
            }
            ids = [int(state["adgroup_id"]) for state in states]
            metrics = _load_today_metrics(client, account_id, ids, now)
            account_cost = 0
            for state in states:
                adgroup_id = int(state["adgroup_id"])
                ad = ads.get(adgroup_id, {})
                values = metrics.get(adgroup_id, {})
                account_cost += int(values.get("cost_fen") or 0)
                preview_items.append({
                    "account_id": account_id,
                    "audience_name": account.get("audience_name"),
                    "adgroup_id": adgroup_id,
                    "adgroup_name": ad.get("adgroup_name") or state.get("adgroup_name"),
                    "before_status": ad.get("configured_status"),
                    "target_status": (
                        ACTIVE_STATUS
                        if state.get("operator_pause_mode") == PAUSE_UNTIL_MANUAL
                        else ad.get("configured_status")
                    ),
                    "preview_cost_fen": int(values.get("cost_fen") or 0),
                    "preview_impressions": int(values.get("impressions") or 0),
                    "preview_clicks": int(values.get("clicks") or 0),
                    "preview_conversions": int(values.get("conversions") or 0),
                    "previewed_at": now,
                    "execution_status": "PREVIEWED",
                })
            account_summaries.append({
                "account_id": account_id,
                "ad_count": len(states),
                "cost_fen": account_cost,
            })
    else:
        raise ValueError(f"Unsupported write action: {parsed.action}")

    if not preview_items:
        raise ValueError("当前范围内没有符合操作条件的广告")
    impacted_account_ids = sorted({int(row["account_id"]) for row in preview_items})
    impacted_accounts = [
        account for account in accounts
        if int(account["account_id"]) in impacted_account_ids
    ]
    account_summaries = [
        row for row in account_summaries if int(row["account_id"]) in impacted_account_ids
    ]
    command_id = f"cmd_{now.strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}"
    expires_at = now + timedelta(minutes=confirmation_ttl_minutes)
    resume_at = _next_delivery_start(now, start_hour) if parsed.action == ACTION_DAY_PAUSE else None
    totals = {
        "preview_cost_fen": sum(int(row["preview_cost_fen"]) for row in preview_items),
        "preview_impressions": sum(int(row["preview_impressions"]) for row in preview_items),
        "preview_clicks": sum(int(row["preview_clicks"]) for row in preview_items),
        "preview_conversions": sum(int(row["preview_conversions"]) for row in preview_items),
    }
    for row in preview_items:
        row["command_id"] = command_id
    command_record = {
        "command_id": command_id,
        "source_message_id": source_message_id,
        "chat_id": chat_id,
        "sender_open_id": sender_open_id,
        "sender_name": sender_name,
        "raw_text": raw_text,
        "parse_source": parse_source,
        "intent": intent or {},
        "action": parsed.action,
        "scope_type": parsed.scope_type,
        "target_account_ids": impacted_account_ids,
        "status": COMMAND_PENDING,
        "preview_account_count": len(impacted_accounts),
        "preview_ad_count": len(preview_items),
        **totals,
        "previewed_at": now,
        "resume_at": resume_at,
        "expires_at": expires_at,
    }
    with advisory_lock(preview_lock_name) as acquired:
        if not acquired:
            raise RuntimeError("其他运营命令正在生成预览，请稍后重试")
        conflict = find_pending_command_conflict(
            [(int(row["account_id"]), int(row["adgroup_id"])) for row in preview_items],
            now=now,
        )
        if conflict:
            raise ValueError(
                "操作范围与待确认命令冲突: "
                f"{conflict['command_id']}，账户 {conflict['account_id']}，"
                f"广告 {conflict['adgroup_id']}"
            )
        command = create_operator_command_with_items(command_record, preview_items)
    command["account_summaries"] = account_summaries
    return command


def cancel_command(command_id: str, sender_open_id: str, now: datetime) -> dict[str, Any]:
    command = load_operator_command(command_id)
    if not command:
        raise ValueError(f"命令不存在: {command_id}")
    if command["sender_open_id"] != sender_open_id:
        raise PermissionError("只能取消自己发起的命令")
    if command["status"] != COMMAND_PENDING:
        return command
    transition_operator_command(
        command_id,
        expected_statuses={COMMAND_PENDING},
        target_status=COMMAND_CANCELLED,
        executed_at=now,
    )
    return load_operator_command(command_id) or command


def _ensure_state(
    *,
    account: dict[str, Any],
    ad: dict[str, Any],
    states: dict[int, dict[str, Any]],
    now: datetime,
) -> dict[str, Any]:
    adgroup_id = int(ad["adgroup_id"])
    state = states.get(adgroup_id)
    if state:
        return state
    bid_field = resolve_bid_field(ad, account.get("bid_scene"))
    base_bid = current_bid_fen(ad, bid_field)
    if base_bid is None or base_bid <= 0:
        raise ValueError(
            f"广告缺少有效基础出价: account={account['account_id']} ad={adgroup_id}"
        )
    upsert_ad_state(
        account_id=int(account["account_id"]),
        adgroup_id=adgroup_id,
        adgroup_name=str(ad.get("adgroup_name") or ""),
        bid_field=bid_field,
        base_bid_fen=base_bid,
        boosted_date=None,
        paused_by_strategy=False,
        pause_reason=None,
        last_action="REGISTER_BASE",
        action_at=now,
    )
    state = {
        "account_id": int(account["account_id"]),
        "adgroup_id": adgroup_id,
        "bid_field": bid_field,
        "base_bid_fen": base_bid,
        "paused_by_strategy": False,
    }
    states[adgroup_id] = state
    return state


def _record_item(
    command: dict[str, Any],
    account: dict[str, Any],
    *,
    item: dict[str, Any] | None = None,
    **values: Any,
) -> None:
    if item and item.get("id"):
        update_operator_command_item(int(item["id"]), **values)
        return
    insert_operator_command_item(
        {
            "command_id": command["command_id"],
            "account_id": int(account["account_id"]),
            "audience_name": account.get("audience_name"),
            **values,
        }
    )


def _execute_pause(
    command: dict[str, Any],
    account: dict[str, Any],
    *,
    now: datetime,
    start_hour: int,
    client: TencentClient,
    items: list[dict[str, Any]],
) -> tuple[int, int, int]:
    account_id = int(account["account_id"])
    try:
        ads = {
            int(ad.get("adgroup_id") or 0): ad
            for ad in client.get_ads(account_id)
        }
    except Exception as exc:
        for item in items:
            _record_item(
                command,
                account,
                item=item,
                execution_status="FAILED",
                error_message=f"Tencent ad query failed: {exc}",
            )
        return 0, len(items), 0
    states = load_ad_states(account_id)
    successes = failures = skipped = 0
    mode = (
        PAUSE_UNTIL_NEXT_DELIVERY
        if command["action"] == ACTION_DAY_PAUSE
        else PAUSE_UNTIL_MANUAL
    )
    resume_at = _next_delivery_start(now, start_hour) if mode == PAUSE_UNTIL_NEXT_DELIVERY else None
    for item in items:
        adgroup_id = int(item["adgroup_id"])
        ad = ads.get(adgroup_id)
        if not ad:
            _record_item(
                command, account, item=item,
                execution_status="FAILED", error_message="Tencent ad not found",
            )
            failures += 1
            continue
        before_status = str(ad.get("configured_status") or "")
        if before_status != ACTIVE_STATUS:
            existing_state = states.get(adgroup_id)
            if (
                command["action"] == ACTION_STOP
                and before_status == SUSPEND_STATUS
                and existing_state
                and existing_state.get("operator_pause_mode")
            ):
                set_operator_pause(
                    account_id=account_id,
                    adgroup_id=adgroup_id,
                    mode=PAUSE_UNTIL_MANUAL,
                    resume_at=None,
                    command_id=command["command_id"],
                    paused_from_status=(
                        existing_state.get("operator_paused_from_status") or ACTIVE_STATUS
                    ),
                    paused_at=now,
                )
                _record_item(
                    command,
                    account,
                    item=item,
                    adgroup_id=adgroup_id,
                    adgroup_name=ad.get("adgroup_name"),
                    before_status=before_status,
                    target_status=SUSPEND_STATUS,
                    readback_status=before_status,
                    execution_status="SUCCESS",
                )
                successes += 1
                continue
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=(
                    before_status
                    if mode == PAUSE_UNTIL_NEXT_DELIVERY
                    else SUSPEND_STATUS
                ),
                readback_status=before_status,
                execution_status="SKIPPED_NOT_ACTIVE",
            )
            skipped += 1
            continue
        if (
            mode == PAUSE_UNTIL_NEXT_DELIVERY
            and str(ad.get("begin_date") or "") > now.date().isoformat()
        ):
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=before_status,
                readback_status=before_status,
                execution_status="SKIPPED_ALREADY_DEFERRED",
            )
            skipped += 1
            continue
        if (
            mode == PAUSE_UNTIL_NEXT_DELIVERY
            and str(ad.get("end_date") or "")
            and str(ad.get("end_date")) < resume_at.date().isoformat()
        ):
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=before_status,
                readback_status=before_status,
                execution_status="SKIPPED_END_DATE",
            )
            skipped += 1
            continue
        tencent_updated = False
        try:
            _ensure_state(account=account, ad=ad, states=states, now=now)
            set_operator_pause(
                account_id=account_id,
                adgroup_id=adgroup_id,
                mode=mode,
                resume_at=resume_at,
                command_id=command["command_id"],
                paused_from_status=before_status,
                paused_at=now,
            )
            if mode == PAUSE_UNTIL_NEXT_DELIVERY:
                before_begin_date = str(ad.get("begin_date") or "")
                next_day = resume_at.date().isoformat()
                target_begin_date = max(before_begin_date, next_day)
                if before_begin_date != target_begin_date:
                    client.update_ad_begin_dates(
                        account_id,
                        [adgroup_id],
                        target_begin_date,
                    )
                    tencent_updated = True
                readback_status = before_status
                target_status = before_status
            else:
                readback = client.update_ad(
                    account_id,
                    adgroup_id,
                    target_status=SUSPEND_STATUS,
                )
                tencent_updated = True
                readback_status = readback.get("configured_status")
                target_status = SUSPEND_STATUS
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=target_status,
                readback_status=readback_status,
                execution_status="SUCCESS",
            )
            successes += 1
        except Exception as exc:
            if isinstance(
                exc,
                (PostWriteVerificationError, TencentWriteOutcomeUnknownError),
            ):
                tencent_updated = True
            safe_to_clear = isinstance(
                exc,
                (TencentWriteNotSentError, TencentWriteRejectedError),
            )
            if not tencent_updated and safe_to_clear:
                clear_operator_pause(
                    account_id,
                    adgroup_id,
                    action="OPERATOR_PAUSE_FAILED",
                    action_at=now,
                )
            try:
                _record_item(
                    command,
                    account,
                    item=item,
                    adgroup_id=adgroup_id,
                    adgroup_name=ad.get("adgroup_name"),
                    before_status=before_status,
                    target_status=(
                        before_status
                        if mode == PAUSE_UNTIL_NEXT_DELIVERY
                        else SUSPEND_STATUS
                    ),
                    readback_status=(
                        exc.actual.get("configured_status")
                        if isinstance(exc, PostWriteVerificationError)
                        else (
                            before_status
                            if mode == PAUSE_UNTIL_NEXT_DELIVERY
                            else SUSPEND_STATUS
                        )
                        if tencent_updated
                        else None
                    ),
                    execution_status=(
                        "VERIFY_FAILED"
                        if isinstance(exc, PostWriteVerificationError)
                        else "OUTCOME_UNKNOWN"
                        if isinstance(exc, TencentWriteOutcomeUnknownError)
                        else "AUDIT_FAILED"
                        if tencent_updated
                        else "FAILED"
                    ),
                    error_message=str(exc),
                )
            except Exception:
                logger.exception(
                    "Failed to persist operator command failure item "
                    "command=%s account=%s ad=%s",
                    command["command_id"],
                    account_id,
                    adgroup_id,
                )
            failures += 1
    return successes, failures, skipped


def _execute_resume(
    command: dict[str, Any],
    account: dict[str, Any],
    *,
    now: datetime,
    client: TencentClient,
    items: list[dict[str, Any]],
) -> tuple[int, int, int]:
    account_id = int(account["account_id"])
    pauses = {
        int(row["adgroup_id"]): row
        for row in load_operator_pauses([account_id])
    }
    try:
        ads = {
            int(ad.get("adgroup_id") or 0): ad
            for ad in client.get_ads(account_id)
        }
    except Exception as exc:
        for item in items:
            _record_item(
                command,
                account,
                item=item,
                execution_status="FAILED",
                error_message=f"Tencent ad query failed: {exc}",
            )
        return 0, len(items), 0
    successes = failures = skipped = 0
    for item in items:
        adgroup_id = int(item["adgroup_id"])
        state = pauses.get(adgroup_id)
        if not state:
            _record_item(
                command, account, item=item,
                execution_status="SKIPPED_NOT_PAUSED",
            )
            skipped += 1
            continue
        ad = ads.get(adgroup_id)
        if not ad:
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=state.get("adgroup_name"),
                execution_status="FAILED",
                error_message="Tencent ad not found",
            )
            failures += 1
            continue
        before_status = str(ad.get("configured_status") or "")
        try:
            readback_status = before_status
            target_status = None
            if (
                state.get("operator_pause_mode") == PAUSE_UNTIL_NEXT_DELIVERY
                and str(ad.get("begin_date") or "") > now.date().isoformat()
                and not bool(state.get("paused_by_strategy"))
            ):
                client.update_ad_begin_dates(
                    account_id,
                    [adgroup_id],
                    now.date().isoformat(),
                )
            elif (
                before_status == SUSPEND_STATUS
                and not bool(state.get("paused_by_strategy"))
            ):
                target_status = ACTIVE_STATUS
                readback = client.update_ad(
                    account_id,
                    adgroup_id,
                    target_status=ACTIVE_STATUS,
                )
                readback_status = str(readback.get("configured_status") or "")
            clear_operator_pause(
                account_id,
                adgroup_id,
                action="OPERATOR_RESUME",
                action_at=now,
            )
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=target_status,
                readback_status=readback_status,
                execution_status="SUCCESS",
            )
            successes += 1
        except Exception as exc:
            _record_item(
                command,
                account,
                item=item,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=ACTIVE_STATUS,
                execution_status="FAILED",
                error_message=str(exc),
            )
            failures += 1
    return successes, failures, skipped


def execute_confirmed_command(
    command_id: str,
    *,
    sender_open_id: str,
    now: datetime,
    start_hour: int,
    lock_name: str,
    tencent: TencentClient | None = None,
) -> dict[str, Any]:
    command = load_operator_command(command_id)
    if not command:
        raise ValueError(f"命令不存在: {command_id}")
    if command["sender_open_id"] != sender_open_id:
        raise PermissionError("只能确认自己发起的命令")
    if command["status"] in FINAL_COMMAND_STATUSES:
        return command
    if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}:
        raise RuntimeError(f"命令当前不可确认: {command['status']}")
    with advisory_lock(lock_name) as acquired:
        if not acquired:
            raise RuntimeError("实时控制正在执行，请稍后再次确认")
        command = load_operator_command(command_id) or command
        if command["status"] in FINAL_COMMAND_STATUSES:
            return command
        if command["status"] not in {COMMAND_PENDING, COMMAND_EXECUTING}:
            raise RuntimeError(f"命令当前不可确认: {command['status']}")
        if (
            command["status"] == COMMAND_PENDING
            and command.get("expires_at")
            and now.replace(tzinfo=None) > command["expires_at"]
        ):
            transition_operator_command(
                command_id,
                expected_statuses={COMMAND_PENDING},
                target_status=COMMAND_EXPIRED,
                executed_at=now,
            )
            return load_operator_command(command_id) or command
        if command["status"] == COMMAND_PENDING:
            transitioned = transition_operator_command(
                command_id,
                expected_statuses={COMMAND_PENDING},
                target_status=COMMAND_EXECUTING,
                confirmed_at=now,
            )
            if not transitioned:
                latest = load_operator_command(command_id) or command
                if latest["status"] in FINAL_COMMAND_STATUSES:
                    return latest
                raise RuntimeError(
                    f"命令状态并发变化: {latest['status']}"
                )
        try:
            managed = _managed_account_map()
            missing = [
                account_id
                for account_id in command["target_account_ids"]
                if account_id not in managed
            ]
            if missing:
                raise RuntimeError(
                    f"确认时账户已不在自动化管理范围: {missing}"
                )
            accounts = [
                managed[account_id]
                for account_id in command["target_account_ids"]
            ]
            client = tencent or TencentClient()
            frozen_items = load_operator_command_items(command_id)
            if not frozen_items:
                raise RuntimeError("命令缺少冻结广告明细，请重新发起")
            items_by_account: dict[int, list[dict[str, Any]]] = {}
            for item in frozen_items:
                items_by_account.setdefault(int(item["account_id"]), []).append(item)
            successes = failures = skipped = 0
            for account in accounts:
                if command["action"] in {ACTION_DAY_PAUSE, ACTION_STOP}:
                    ok, failed, ignored = _execute_pause(
                        command,
                        account,
                        now=now,
                        start_hour=start_hour,
                        client=client,
                        items=items_by_account.get(int(account["account_id"]), []),
                    )
                elif command["action"] == ACTION_RESUME:
                    ok, failed, ignored = _execute_resume(
                        command,
                        account,
                        now=now,
                        client=client,
                        items=items_by_account.get(int(account["account_id"]), []),
                    )
                else:
                    raise ValueError(f"Unsupported command action: {command['action']}")
                successes += ok
                failures += failed
                skipped += ignored
        except Exception as exc:
            update_operator_command(
                command_id,
                COMMAND_FAILED,
                executed_at=now,
                error_message=str(exc),
            )
            raise

        status = (
            COMMAND_FAILED
            if failures and not successes
            else COMMAND_PARTIAL
            if failures
            else COMMAND_SUCCEEDED
        )
        update_operator_command(command_id, status, executed_at=now)
        result = load_operator_command(command_id) or command
        result["successes"] = successes
        result["failures"] = failures
        result["skipped"] = skipped
        return result


def pause_status_summary(now: datetime) -> dict[str, Any]:
    account_ids = [int(row["account_id"]) for row in load_realtime_accounts()]
    pauses = [
        row for row in load_operator_pauses(account_ids)
        if _operator_pause_is_active(row, now)
    ]
    return {
        "total": len(pauses),
        "until_next_delivery": sum(
            row["operator_pause_mode"] == PAUSE_UNTIL_NEXT_DELIVERY
            for row in pauses
        ),
        "until_manual": sum(
            row["operator_pause_mode"] == PAUSE_UNTIL_MANUAL
            for row in pauses
        ),
        "accounts": sorted({int(row["account_id"]) for row in pauses}),
    }