Agent/examples/auto_put_ad_mini/roi_control/execution.py

"""Execute an approved ROI batch with Tencent read-back and state coordination."""

from __future__ import annotations

import json
import logging
import os
from datetime import date, datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
from typing import Any

from storage import (
    advisory_lock,
    load_ad_states,
    load_managed_accounts,
    sync_roi_base_bid,
)
from tencent_client import (
    ACTIVE_STATUS,
    SUSPEND_STATUS,
    PostWriteVerificationError,
    TencentClient,
    TencentWriteOutcomeUnknownError,
    current_bid_fen,
    resolve_bid_field,
)

from .config import RoiConfig
from .policy import ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID
from .repository import (
    FINAL_STATUSES,
    claim_run_for_execution,
    finalize_run,
    finalize_sheet_run_if_resolved,
    load_actions_by_ids,
    load_action_items,
    load_run,
    reject_run,
    update_action_item,
)


logger = logging.getLogger("auto_put_ad_mini.roi_execution")


def _round_fen(value: Decimal) -> int:
    return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))


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


def _failure_status(exc: Exception) -> str:
    if isinstance(exc, PostWriteVerificationError):
        return "VERIFY_FAILED"
    if isinstance(exc, TencentWriteOutcomeUnknownError):
        return "OUTCOME_UNKNOWN"
    return "FAILED"


def plan_scale_bid(
    *,
    current_bid_fen: int,
    base_bid_fen: int,
    initial_base_bid_fen: int,
    boosted_date: date | None,
    control_date: date,
    intraday_ratio: Decimal,
    scale_ratio: Decimal,
    max_base_ratio: Decimal,
) -> dict[str, Any]:
    """Plan a permanent base increase without reapplying a restored boost."""

    boosted_date_today = boosted_date == control_date
    expected_boosted = _round_fen(Decimal(base_bid_fen) * intraday_ratio)
    actively_boosted = boosted_date_today and current_bid_fen == expected_boosted
    if current_bid_fen != base_bid_fen and not actively_boosted:
        return {
            "allowed": False,
            "reason": (
                f"current bid={current_bid_fen} does not match managed "
                f"base={base_bid_fen} or intraday boost={expected_boosted}"
            ),
        }

    max_base = _round_fen(Decimal(initial_base_bid_fen) * max_base_ratio)
    new_base = min(
        _round_fen(Decimal(base_bid_fen) * scale_ratio),
        max_base,
    )
    if new_base <= base_bid_fen:
        return {
            "allowed": False,
            "reason": f"permanent base reached cap={max_base}",
            "cap_reached": True,
            "max_base_bid_fen": max_base,
        }
    target = (
        _round_fen(Decimal(new_base) * intraday_ratio)
        if actively_boosted
        else new_base
    )
    return {
        "allowed": True,
        "new_base_bid_fen": new_base,
        "target_bid_fen": target,
        "boosted_date_today": boosted_date_today,
        "preserve_intraday_boost": actively_boosted,
    }


def _execute_pause(
    item: dict[str, Any],
    *,
    client: TencentClient,
    now: datetime,
) -> None:
    account_id = int(item["account_id"])
    adgroup_id = int(item["adgroup_id"])
    creative_id = int(item["dynamic_creative_id"])
    creative = client.get_dynamic_creative(account_id, creative_id)
    actual_adgroup_id = int(creative.get("adgroup_id") or 0)
    before_status = str(creative.get("configured_status") or "")
    if actual_adgroup_id != adgroup_id:
        update_action_item(
            int(item["id"]),
            execution_status="SKIPPED_MAPPING_MISMATCH",
            skip_reason=(
                f"creative belongs to adgroup={actual_adgroup_id}, expected={adgroup_id}"
            ),
            pre_state_json=creative,
            executed_at=now,
        )
        return
    if item["execution_status"] == "PREPARED" and before_status == SUSPEND_STATUS:
        update_action_item(
            int(item["id"]),
            before_status=item.get("before_status"),
            target_status=SUSPEND_STATUS,
            readback_status=before_status,
            execution_status="SUCCESS",
            readback_json=creative,
            executed_at=now,
        )
        return
    if before_status != ACTIVE_STATUS:
        update_action_item(
            int(item["id"]),
            before_status=before_status,
            target_status=SUSPEND_STATUS,
            readback_status=before_status,
            execution_status="SKIPPED_STATE_MISMATCH",
            skip_reason="creative is no longer active",
            pre_state_json=creative,
            executed_at=now,
        )
        return

    update_action_item(
        int(item["id"]),
        before_status=before_status,
        target_status=SUSPEND_STATUS,
        execution_status="PREPARED",
        pre_state_json=creative,
    )
    try:
        readback = client.update_dynamic_creative_status(
            account_id,
            creative_id,
            SUSPEND_STATUS,
        )
        update_action_item(
            int(item["id"]),
            readback_status=str(readback.get("configured_status") or ""),
            execution_status="SUCCESS",
            readback_json=readback,
            executed_at=now,
        )
    except Exception as exc:
        update_action_item(
            int(item["id"]),
            execution_status=_failure_status(exc),
            error_message=str(exc),
            readback_json=(exc.actual if isinstance(exc, PostWriteVerificationError) else None),
            executed_at=now,
        )


def _resume_prepared_scale(
    item: dict[str, Any],
    ad: dict[str, Any],
    *,
    client: TencentClient,
    now: datetime,
) -> None:
    details = json.loads(item.get("pre_state_json") or "{}")
    bid_field = str(item.get("bid_field") or "")
    current = current_bid_fen(ad, bid_field)
    before = int(item.get("before_bid_fen") or 0)
    target = int(item.get("target_bid_fen") or 0)
    if current == target:
        sync_roi_base_bid(
            account_id=int(item["account_id"]),
            adgroup_id=int(item["adgroup_id"]),
            adgroup_name=str(ad.get("adgroup_name") or ""),
            bid_field=bid_field,
            initial_base_bid_fen=int(item["initial_base_bid_fen"]),
            new_base_bid_fen=int(details["new_base_bid_fen"]),
            boosted_date=(
                now.date() if details.get("boosted_date_today") else None
            ),
            action_at=now,
        )
        update_action_item(
            int(item["id"]),
            execution_status="SUCCESS",
            readback_json=ad,
            executed_at=now,
        )
        return
    if current != before:
        update_action_item(
            int(item["id"]),
            execution_status="SKIPPED_STATE_MISMATCH",
            skip_reason=f"bid changed after preparation: before={before} current={current}",
            readback_json=ad,
            executed_at=now,
        )
        return
    readback = client.update_ad(
        int(item["account_id"]),
        int(item["adgroup_id"]),
        bid_field=bid_field,
        target_bid_fen=target,
    )
    sync_roi_base_bid(
        account_id=int(item["account_id"]),
        adgroup_id=int(item["adgroup_id"]),
        adgroup_name=str(ad.get("adgroup_name") or ""),
        bid_field=bid_field,
        initial_base_bid_fen=int(item["initial_base_bid_fen"]),
        new_base_bid_fen=int(details["new_base_bid_fen"]),
        boosted_date=(now.date() if details.get("boosted_date_today") else None),
        action_at=now,
    )
    update_action_item(
        int(item["id"]),
        execution_status="SUCCESS",
        readback_json=readback,
        executed_at=now,
    )


def _execute_scale(
    item: dict[str, Any],
    account: dict[str, Any],
    *,
    client: TencentClient,
    config: RoiConfig,
    now: datetime,
) -> None:
    account_id = int(item["account_id"])
    adgroup_id = int(item["adgroup_id"])
    ad = client.get_ad(account_id, adgroup_id)
    if str(ad.get("configured_status") or "") != ACTIVE_STATUS:
        update_action_item(
            int(item["id"]),
            execution_status="SKIPPED_STATE_MISMATCH",
            skip_reason="ad is no longer active",
            pre_state_json=ad,
            executed_at=now,
        )
        return
    if item["execution_status"] == "PREPARED":
        try:
            _resume_prepared_scale(item, ad, client=client, now=now)
        except Exception as exc:
            update_action_item(
                int(item["id"]),
                execution_status=_failure_status(exc),
                error_message=str(exc),
                executed_at=now,
            )
        return

    states = load_ad_states(account_id)
    state = states.get(adgroup_id) or {}
    last_scaled = state.get("last_roi_scaled_at")
    if last_scaled and now < last_scaled + timedelta(days=config.scale_cooldown_days):
        update_action_item(
            int(item["id"]),
            execution_status="SKIPPED_COOLDOWN",
            skip_reason=f"last ROI scale at {last_scaled}",
            pre_state_json=ad,
            executed_at=now,
        )
        return

    bid_field = resolve_bid_field(ad, account.get("bid_scene"))
    current = current_bid_fen(ad, bid_field)
    if current is None or current <= 0:
        raise ValueError(f"ad has no valid {bid_field}")
    base = int(state.get("base_bid_fen") or current)
    initial = int(state.get("initial_base_bid_fen") or base)
    intraday_ratio = Decimal(os.getenv("RTC_BID_UP_RATIO", "1.05"))
    plan = plan_scale_bid(
        current_bid_fen=current,
        base_bid_fen=base,
        initial_base_bid_fen=initial,
        boosted_date=state.get("boosted_date"),
        control_date=now.date(),
        intraday_ratio=intraday_ratio,
        scale_ratio=config.scale_ratio,
        max_base_ratio=config.max_base_ratio,
    )
    if not plan["allowed"]:
        update_action_item(
            int(item["id"]),
            bid_field=bid_field,
            initial_base_bid_fen=initial,
            base_bid_fen=base,
            before_bid_fen=current,
            target_bid_fen=current,
            execution_status=(
                "SKIPPED_CAP_REACHED"
                if plan.get("cap_reached")
                else "SKIPPED_STATE_MISMATCH"
            ),
            skip_reason=str(plan["reason"]),
            pre_state_json=ad,
            executed_at=now,
        )
        return
    new_base = int(plan["new_base_bid_fen"])
    target = int(plan["target_bid_fen"])
    boosted_date_today = bool(plan["boosted_date_today"])
    prepared_state = {
        **ad,
        "managed_base_bid_fen": base,
        "new_base_bid_fen": new_base,
        "boosted_date_today": boosted_date_today,
        "preserve_intraday_boost": bool(plan["preserve_intraday_boost"]),
    }
    update_action_item(
        int(item["id"]),
        bid_field=bid_field,
        initial_base_bid_fen=initial,
        base_bid_fen=base,
        before_bid_fen=current,
        target_bid_fen=target,
        execution_status="PREPARED",
        pre_state_json=prepared_state,
    )
    try:
        readback = client.update_ad(
            account_id,
            adgroup_id,
            bid_field=bid_field,
            target_bid_fen=target,
        )
        sync_roi_base_bid(
            account_id=account_id,
            adgroup_id=adgroup_id,
            adgroup_name=str(ad.get("adgroup_name") or ""),
            bid_field=bid_field,
            initial_base_bid_fen=initial,
            new_base_bid_fen=new_base,
            boosted_date=now.date() if boosted_date_today else None,
            action_at=now,
        )
        update_action_item(
            int(item["id"]),
            execution_status="SUCCESS",
            readback_json=readback,
            executed_at=now,
        )
    except Exception as exc:
        update_action_item(
            int(item["id"]),
            execution_status=_failure_status(exc),
            error_message=str(exc),
            readback_json=(exc.actual if isinstance(exc, PostWriteVerificationError) else None),
            executed_at=now,
        )


def execute_roi_batch(
    run_id: str,
    *,
    sender_open_id: str,
    now: datetime,
    lock_name: str,
    tencent: TencentClient | None = None,
) -> dict[str, Any]:
    config = RoiConfig.from_env()
    if not config.apply_enabled:
        raise RuntimeError("ROI_APPLY_ENABLED=0, ROI Tencent writes are disabled")
    with advisory_lock(lock_name) as acquired:
        if not acquired:
            raise RuntimeError("实时控制正在执行，请稍后再次确认 ROI 批次")
        run = claim_run_for_execution(
            run_id,
            sender_open_id=sender_open_id,
            now=now,
        )
        if run.get("status") in FINAL_STATUSES:
            return run
        accounts = _managed_accounts()
        client = tencent or TencentClient()
        for item in load_action_items(run_id):
            if item["execution_status"] not in {"PENDING", "PREPARED"}:
                continue
            account_id = int(item["account_id"])
            account = accounts.get(account_id)
            if account is None:
                update_action_item(
                    int(item["id"]),
                    execution_status="SKIPPED_NOT_MANAGED",
                    skip_reason="account left automated management scope",
                    executed_at=now,
                )
                continue
            try:
                if item["action_type"] == ACTION_PAUSE_CREATIVE:
                    _execute_pause(item, client=client, now=now)
                elif item["action_type"] == ACTION_SCALE_BID:
                    _execute_scale(
                        item,
                        account,
                        client=client,
                        config=config,
                        now=now,
                    )
                else:
                    update_action_item(
                        int(item["id"]),
                        execution_status="SKIPPED_UNSUPPORTED",
                        skip_reason=f"unsupported action={item['action_type']}",
                        executed_at=now,
                    )
            except Exception as exc:
                logger.exception(
                    "ROI action failed run=%s item=%s", run_id, item["id"]
                )
                update_action_item(
                    int(item["id"]),
                    execution_status=_failure_status(exc),
                    error_message=str(exc),
                    executed_at=now,
                )
        return {"run_id": run_id, **finalize_run(run_id, now=now)}


def execute_approved_roi_actions(
    item_ids: list[int],
    *,
    now: datetime,
    lock_name: str,
    tencent: TencentClient | None = None,
) -> dict[str, Any]:
    """Execute only explicitly approved sheet rows without claiming the batch."""

    config = RoiConfig.from_env()
    if not config.apply_enabled:
        raise RuntimeError("ROI_APPLY_ENABLED=0, ROI Tencent writes are disabled")
    with advisory_lock(lock_name) as acquired:
        if not acquired:
            raise RuntimeError("实时控制正在执行，ROI表格审批稍后自动重试")
        accounts = _managed_accounts()
        client = tencent or TencentClient()
        run_ids: set[str] = set()
        processed = 0
        for item in load_actions_by_ids(item_ids):
            run_ids.add(str(item["run_id"]))
            if item.get("approval_status") != "APPROVED":
                continue
            if item["execution_status"] not in {"PENDING", "PREPARED"}:
                continue
            processed += 1
            account_id = int(item["account_id"])
            account = accounts.get(account_id)
            if account is None:
                update_action_item(
                    int(item["id"]),
                    execution_status="SKIPPED_NOT_MANAGED",
                    skip_reason="account left automated management scope",
                    executed_at=now,
                )
                continue
            try:
                if item["action_type"] == ACTION_PAUSE_CREATIVE:
                    _execute_pause(item, client=client, now=now)
                elif item["action_type"] == ACTION_SCALE_BID:
                    _execute_scale(
                        item,
                        account,
                        client=client,
                        config=config,
                        now=now,
                    )
                else:
                    update_action_item(
                        int(item["id"]),
                        execution_status="SKIPPED_UNSUPPORTED",
                        skip_reason=f"unsupported action={item['action_type']}",
                        executed_at=now,
                    )
            except Exception as exc:
                logger.exception(
                    "Approved ROI action failed run=%s item=%s",
                    item["run_id"],
                    item["id"],
                )
                update_action_item(
                    int(item["id"]),
                    execution_status=_failure_status(exc),
                    error_message=str(exc),
                    executed_at=now,
                )
        final = {
            run_id: finalize_sheet_run_if_resolved(run_id, now=now)
            for run_id in run_ids
        }
        return {"processed": processed, "runs": final}


def reject_roi_batch(
    run_id: str,
    *,
    sender_open_id: str,
    now: datetime,
) -> dict[str, Any]:
    return reject_run(run_id, sender_open_id=sender_open_id, now=now)


def roi_batch_status(run_id: str) -> dict[str, Any]:
    run = load_run(run_id)
    if not run:
        raise ValueError(f"ROI批次不存在: {run_id}")
    return run