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,
    insert_operator_command_item,
    load_ad_states,
    load_managed_accounts,
    load_operator_command,
    load_operator_pauses,
    set_operator_pause,
    transition_operator_command,
    update_operator_command,
    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_managed_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 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,
    tencent: TencentClient | None = None,
) -> dict[str, Any]:
    accounts = resolve_target_accounts(parsed)
    client = tencent or TencentClient()
    preview_ad_count = 0
    if parsed.action in {ACTION_DAY_PAUSE, ACTION_STOP}:
        for account in accounts:
            ads = client.get_ads(int(account["account_id"]))
            preview_ad_count += sum(
                str(ad.get("configured_status") or "") == ACTIVE_STATUS
                for ad in ads
            )
    elif parsed.action == ACTION_RESUME:
        preview_ad_count = len(
            load_operator_pauses([int(row["account_id"]) for row in accounts])
        )
    else:
        raise ValueError(f"Unsupported write action: {parsed.action}")

    command_id = f"cmd_{now.strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}"
    expires_at = now + timedelta(minutes=confirmation_ttl_minutes)
    command = create_operator_command(
        {
            "command_id": command_id,
            "source_message_id": source_message_id,
            "chat_id": chat_id,
            "sender_open_id": sender_open_id,
            "sender_name": sender_name,
            "action": parsed.action,
            "scope_type": parsed.scope_type,
            "target_account_ids": [int(row["account_id"]) for row in accounts],
            "status": COMMAND_PENDING,
            "preview_account_count": len(accounts),
            "preview_ad_count": preview_ad_count,
            "expires_at": expires_at,
        }
    )
    command["resume_at"] = (
        _next_delivery_start(now, start_hour)
        if parsed.action == ACTION_DAY_PAUSE
        else None
    )
    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], **values: Any) -> None:
    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,
) -> tuple[int, int]:
    account_id = int(account["account_id"])
    ads = client.get_ads(account_id)
    states = load_ad_states(account_id)
    successes = failures = 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 ad in ads:
        adgroup_id = int(ad.get("adgroup_id") or 0)
        before_status = str(ad.get("configured_status") or "")
        if before_status != ACTIVE_STATUS:
            _record_item(
                command,
                account,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=SUSPEND_STATUS,
                readback_status=before_status,
                execution_status="SKIPPED_NOT_ACTIVE",
            )
            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,
            )
            readback = client.update_ad(
                account_id,
                adgroup_id,
                target_status=SUSPEND_STATUS,
            )
            tencent_updated = True
            _record_item(
                command,
                account,
                adgroup_id=adgroup_id,
                adgroup_name=ad.get("adgroup_name"),
                before_status=before_status,
                target_status=SUSPEND_STATUS,
                readback_status=readback.get("configured_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,
                    adgroup_id=adgroup_id,
                    adgroup_name=ad.get("adgroup_name"),
                    before_status=before_status,
                    target_status=SUSPEND_STATUS,
                    readback_status=(
                        exc.actual.get("configured_status")
                        if isinstance(exc, PostWriteVerificationError)
                        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


def _execute_resume(
    command: dict[str, Any],
    account: dict[str, Any],
    *,
    now: datetime,
    client: TencentClient,
) -> tuple[int, int]:
    account_id = int(account["account_id"])
    pauses = load_operator_pauses([account_id])
    if not pauses:
        return 0, 0
    ads = {
        int(ad.get("adgroup_id") or 0): ad
        for ad in client.get_ads(account_id)
    }
    successes = failures = 0
    for state in pauses:
        adgroup_id = int(state["adgroup_id"])
        ad = ads.get(adgroup_id)
        if not ad:
            _record_item(
                command,
                account,
                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 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,
                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,
                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


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()
            successes = failures = 0
            for account in accounts:
                if command["action"] in {ACTION_DAY_PAUSE, ACTION_STOP}:
                    ok, failed = _execute_pause(
                        command,
                        account,
                        now=now,
                        start_hour=start_hour,
                        client=client,
                    )
                elif command["action"] == ACTION_RESUME:
                    ok, failed = _execute_resume(
                        command,
                        account,
                        now=now,
                        client=client,
                    )
                else:
                    raise ValueError(f"Unsupported command action: {command['action']}")
                successes += ok
                failures += failed
        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
        return result


def pause_status_summary() -> dict[str, Any]:
    account_ids = [int(row["account_id"]) for row in load_managed_accounts()]
    pauses = load_operator_pauses(account_ids)
    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}),
    }