Agent/examples/tencent_realtime_control/run_once.py

#!/usr/bin/env python
"""Run one idempotent real-time CPM control cycle."""

from __future__ import annotations

import argparse
import json
import logging
import time
import uuid
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

from dotenv import load_dotenv


ROOT = Path(__file__).resolve().parent
SHANGHAI = ZoneInfo("Asia/Shanghai")
ACTIVE_STATUS = "AD_STATUS_NORMAL"
SUSPEND_STATUS = "AD_STATUS_SUSPEND"

DECISION_BOOST = "BOOST"
DECISION_RESTORE = "RESTORE"
DECISION_PAUSE = "PAUSE"
DECISION_CUTOFF = "CUTOFF"
DECISION_OFF_HOURS = "OFF_HOURS"
DECISION_WAIT_DATA = "WAIT_DATA"

logger = logging.getLogger("tencent_realtime_control")


def load_environment() -> None:
    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
    load_dotenv(Path.cwd() / ".env", override=False)


def decide(cpm: Decimal, low_cpm: Decimal, high_cpm: Decimal) -> str:
    if cpm > high_cpm:
        return DECISION_BOOST
    if cpm < low_cpm:
        return DECISION_PAUSE
    return DECISION_RESTORE


def boosted_bid(base_bid_fen: int, ratio: Decimal) -> int:
    return int(
        (Decimal(base_bid_fen) * ratio).quantize(
            Decimal("1"), rounding=ROUND_HALF_UP
        )
    )


def partition_lag_hours(partition: str, now: datetime) -> float:
    try:
        partition_time = datetime.strptime(partition[:10], "%Y%m%d%H").replace(
            tzinfo=SHANGHAI
        )
    except (TypeError, ValueError) as exc:
        raise ValueError(f"Invalid CPM partition: {partition!r}") from exc
    return (now - partition_time).total_seconds() / 3600


def should_refresh_inventory(
    daily_state: dict[str, Any],
    *,
    observed_partition: str,
    decision: str,
    now: datetime,
    refresh_minutes: int,
) -> bool:
    if (
        daily_state.get("last_observed_partition") != observed_partition
        or daily_state.get("last_decision") != decision
    ):
        return True
    refreshed_at = daily_state.get("last_inventory_refresh_at")
    if not refreshed_at:
        return True
    if refreshed_at.tzinfo is None:
        refreshed_at = refreshed_at.replace(tzinfo=SHANGHAI)
    return now - refreshed_at >= timedelta(minutes=refresh_minutes)


def target_for_ad(
    *,
    decision: str,
    control_date: Any,
    current_status: str,
    current_bid: int,
    base_bid: int,
    boosted_date: Any,
    paused_by_strategy: bool,
    operator_pause_expired: bool,
    bid_up_ratio: Decimal,
) -> dict[str, Any] | None:
    managed_suspend = paused_by_strategy or operator_pause_expired
    manual_suspend = current_status != ACTIVE_STATUS and not managed_suspend

    target_bid = base_bid
    target_status: str | None = None
    next_boosted_date = boosted_date
    next_paused = False
    pause_reason = None
    limit_reached = False
    defer_to_next_day = False

    if manual_suspend:
        if decision not in (DECISION_PAUSE, DECISION_CUTOFF):
            return None
        return {
            "target_bid_fen": base_bid,
            "target_status": None,
            "boosted_date": boosted_date,
            "paused_by_strategy": False,
            "pause_reason": None,
            "boost_limit_reached": False,
            "bid_change_needed": current_bid != base_bid,
            "status_change_needed": False,
            "defer_to_next_day": False,
        }

    if decision == DECISION_BOOST:
        if boosted_date == control_date:
            limit_reached = True
            raised_bid = boosted_bid(base_bid, bid_up_ratio)
            if current_bid == raised_bid:
                target_bid = raised_bid
        else:
            target_bid = boosted_bid(base_bid, bid_up_ratio)
            next_boosted_date = control_date
        if managed_suspend:
            target_status = ACTIVE_STATUS
    elif decision == DECISION_RESTORE:
        if managed_suspend:
            target_status = ACTIVE_STATUS
    elif decision == DECISION_PAUSE:
        if managed_suspend:
            target_status = ACTIVE_STATUS
        defer_to_next_day = True
    elif decision == DECISION_CUTOFF:
        target_status = SUSPEND_STATUS if current_status != SUSPEND_STATUS else None
        next_paused = True
        pause_reason = "cutoff"
    else:
        raise ValueError(f"Unsupported decision: {decision}")

    return {
        "target_bid_fen": target_bid,
        "target_status": target_status,
        "boosted_date": next_boosted_date,
        "paused_by_strategy": next_paused,
        "pause_reason": pause_reason,
        "boost_limit_reached": limit_reached,
        "bid_change_needed": current_bid != target_bid,
        "status_change_needed": target_status is not None
        and target_status != current_status,
        "defer_to_next_day": defer_to_next_day,
    }


def write_run_report(payload: dict[str, Any]) -> Path:
    output_dir = ROOT / "outputs"
    output_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now(SHANGHAI).strftime("%Y%m%d_%H%M%S")
    mode = "apply" if payload["apply"] else "dry_run"
    path = output_dir / f"control_{mode}_{timestamp}.json"
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return path


def execute_inventory_action(
    *,
    run_id: str,
    now: datetime,
    decision: str,
    observed_partition: str | None,
    observed_cpm: Decimal | None,
    apply: bool,
    accounts: list[dict[str, Any]],
    config: Any,
    tencent: Any,
    test_notification: bool = False,
) -> dict[str, Any]:
    from storage import (
        clear_operator_pause,
        insert_action_log,
        load_ad_states,
        upsert_ad_state,
    )
    from tencent_client import current_bid_fen, resolve_bid_field

    results: list[dict[str, Any]] = []
    failures = 0
    api_updates = 0

    for account in accounts:
        account_id = int(account["account_id"])
        control_mode = str(account.get("control_mode") or "FULL").upper()
        if control_mode == "PAUSE_ONLY" and decision not in (
            DECISION_PAUSE,
            DECISION_CUTOFF,
        ):
            continue
        states = load_ad_states(account_id)
        try:
            ads = tencent.get_ads(account_id)
        except Exception as exc:
            failures += 1
            results.append(
                {
                    "account_id": account_id,
                    "status": "account_fetch_failed",
                    "error": str(exc),
                }
            )
            continue

        for ad in ads:
            adgroup_id = int(ad.get("adgroup_id") or 0)
            current_status = str(ad.get("configured_status") or "")
            state = states.get(adgroup_id)
            paused_by_strategy = bool(
                (state or {}).get("paused_by_strategy")
            )
            operator_pause_mode = str(
                (state or {}).get("operator_pause_mode") or ""
            )
            operator_resume_at = (state or {}).get("operator_resume_at")
            if operator_resume_at and operator_resume_at.tzinfo is None:
                operator_resume_at = operator_resume_at.replace(tzinfo=SHANGHAI)
            operator_manually_opened = bool(
                operator_pause_mode == "UNTIL_MANUAL"
                and current_status == ACTIVE_STATUS
            )
            operator_pause_expired = bool(
                operator_pause_mode == "UNTIL_NEXT_DELIVERY"
                and operator_resume_at
                and operator_resume_at <= now
            )
            begin_date = str(ad.get("begin_date") or "")
            operator_schedule_manually_resumed = bool(
                operator_pause_mode == "UNTIL_NEXT_DELIVERY"
                and operator_resume_at
                and operator_resume_at > now
                and begin_date
                and begin_date < operator_resume_at.date().isoformat()
            )
            operator_pause_active = bool(
                operator_pause_mode
                and not operator_manually_opened
                and not operator_schedule_manually_resumed
                and not operator_pause_expired
            )

            if (
                operator_manually_opened
                or operator_schedule_manually_resumed
            ) and apply:
                clear_operator_pause(
                    account_id,
                    adgroup_id,
                    action="OPERATOR_MANUAL_OVERRIDE",
                    action_at=now,
                )
                results.append(
                    {
                        "account_id": account_id,
                        "audience_name": account.get("audience_name"),
                        "adgroup_id": adgroup_id,
                        "adgroup_name": ad.get("adgroup_name"),
                        "decision": "OPERATOR_MANUAL_OVERRIDE",
                        "before_status": current_status,
                        "target_status": current_status,
                        "status": "success",
                        "bid_change_needed": False,
                        "status_change_needed": False,
                    }
                )
                continue

            if operator_pause_expired:
                if apply and current_status == ACTIVE_STATUS:
                    clear_operator_pause(
                        account_id,
                        adgroup_id,
                        action="OPERATOR_DAY_PAUSE_EXPIRED",
                        action_at=now,
                    )
                # Legacy DAY_PAUSE rows used SUSPEND. Keep the old one-time
                # recovery plan only for those rows; new rows stay NORMAL and
                # resume automatically through begin_date/time_series.
                operator_pause_expired = current_status != ACTIVE_STATUS

            if operator_pause_active:
                continue

            if (
                current_status != ACTIVE_STATUS
                and not paused_by_strategy
                and not operator_pause_expired
                and (
                    state is None
                    or decision not in (DECISION_PAUSE, DECISION_CUTOFF)
                )
            ):
                continue

            bid_field = (
                str(state["bid_field"])
                if state
                else resolve_bid_field(ad, account.get("bid_scene"))
            )
            current_bid = current_bid_fen(ad, bid_field)
            if current_bid is None or current_bid <= 0:
                failures += 1
                results.append(
                    {
                        "account_id": account_id,
                        "adgroup_id": adgroup_id,
                        "status": "invalid_current_bid",
                        "bid_field": bid_field,
                    }
                )
                continue

            base_bid = int(state["base_bid_fen"]) if state else current_bid
            plan = target_for_ad(
                decision=decision,
                control_date=now.date(),
                current_status=current_status,
                current_bid=current_bid,
                base_bid=base_bid,
                boosted_date=(state or {}).get("boosted_date"),
                paused_by_strategy=paused_by_strategy,
                operator_pause_expired=operator_pause_expired,
                bid_up_ratio=config.bid_up_ratio,
            )
            if plan is None:
                continue

            bid_hold = bool((state or {}).get("bid_hold"))
            if bid_hold:
                plan["target_bid_fen"] = current_bid
                plan["bid_change_needed"] = False

            before_begin_date = str(ad.get("begin_date") or "")
            target_begin_date = None
            begin_date_change_needed = False
            if plan["defer_to_next_day"]:
                next_day = (now.date() + timedelta(days=1)).isoformat()
                target_begin_date = (
                    before_begin_date
                    if before_begin_date > next_day
                    else next_day
                )
                begin_date_change_needed = before_begin_date != target_begin_date

            needs_api = (
                plan["bid_change_needed"]
                or plan["status_change_needed"]
                or begin_date_change_needed
            )
            execution_status = "planned" if needs_api else "noop"
            error = None
            if apply and state is None:
                upsert_ad_state(
                    account_id=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=plan["paused_by_strategy"],
                    pause_reason=plan["pause_reason"],
                    last_action="REGISTER_BASE",
                    action_at=now,
                )
            elif (
                apply
                and decision in (DECISION_PAUSE, DECISION_CUTOFF)
                and plan["paused_by_strategy"]
            ):
                upsert_ad_state(
                    account_id=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=(state or {}).get("boosted_date"),
                    paused_by_strategy=True,
                    pause_reason=plan["pause_reason"],
                    last_action=f"{decision}_PENDING",
                    action_at=now,
                )
            if apply and needs_api:
                try:
                    if begin_date_change_needed:
                        tencent.update_ad_begin_dates(
                            account_id,
                            [adgroup_id],
                            str(target_begin_date),
                        )
                    if plan["bid_change_needed"] or plan["status_change_needed"]:
                        tencent.update_ad(
                            account_id,
                            adgroup_id,
                            bid_field=bid_field
                            if plan["bid_change_needed"]
                            else None,
                            target_bid_fen=plan["target_bid_fen"]
                            if plan["bid_change_needed"]
                            else None,
                            target_status=plan["target_status"]
                            if plan["status_change_needed"]
                            else None,
                        )
                    execution_status = "success"
                    api_updates += 1
                    time.sleep(0.15)
                except Exception as exc:
                    execution_status = "failed"
                    error = str(exc)
                    failures += 1

            result = {
                "account_id": account_id,
                "audience_name": account.get("audience_name"),
                "adgroup_id": adgroup_id,
                "adgroup_name": ad.get("adgroup_name"),
                "decision": decision,
                "bid_field": bid_field,
                "base_bid_fen": base_bid,
                "before_bid_fen": current_bid,
                "target_bid_fen": plan["target_bid_fen"],
                "before_status": current_status,
                "target_status": plan["target_status"],
                "boost_limit_reached": plan["boost_limit_reached"],
                "bid_hold": bid_hold,
                "bid_change_needed": plan["bid_change_needed"],
                "status_change_needed": plan["status_change_needed"],
                "before_begin_date": before_begin_date,
                "target_begin_date": target_begin_date,
                "begin_date_change_needed": begin_date_change_needed,
                "status": execution_status,
                "error": error,
            }
            results.append(result)

            if apply and execution_status != "failed":
                upsert_ad_state(
                    account_id=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=plan["boosted_date"],
                    paused_by_strategy=plan["paused_by_strategy"],
                    pause_reason=plan["pause_reason"],
                    last_action=decision,
                    action_at=now,
                )
                if operator_pause_expired:
                    clear_operator_pause(
                        account_id,
                        adgroup_id,
                        action="OPERATOR_DAY_PAUSE_EXPIRED",
                        action_at=now,
                    )
            if apply and (needs_api or state is None or execution_status == "failed"):
                insert_action_log(
                    {
                        "run_id": run_id,
                        "control_date": now.date(),
                        "observed_partition": observed_partition,
                        "observed_cpm": observed_cpm,
                        "decision": decision,
                        "account_id": account_id,
                        "adgroup_id": adgroup_id,
                        "adgroup_name": ad.get("adgroup_name"),
                        "bid_field": bid_field,
                        "base_bid_fen": base_bid,
                        "before_bid_fen": current_bid,
                        "target_bid_fen": plan["target_bid_fen"],
                        "before_status": current_status,
                        "target_status": plan["target_status"],
                        "apply_mode": True,
                        "execution_status": execution_status,
                        "error_message": error,
                    }
                )

    notification = {"status": "dry_run" if not apply else "no_actions"}
    if apply or test_notification:
        from feishu_notifier import send_action_notification

        report_actions: dict[int, list[dict[str, Any]]] = {}
        expected_statuses = {"success", "failed"} if apply else {"planned"}
        for result in results:
            if result.get("status") not in expected_statuses:
                continue
            if not (
                result.get("bid_change_needed")
                or result.get("begin_date_change_needed")
                or (
                    result.get("decision") in (DECISION_PAUSE, DECISION_CUTOFF)
                    and result.get("status_change_needed")
                )
            ):
                continue
            report_actions.setdefault(int(result["account_id"]), []).append(
                result
            )

        for account_id, account_results in report_actions.items():
            try:
                metrics = tencent.get_today_ad_metrics(
                    account_id,
                    [int(row["adgroup_id"]) for row in account_results],
                    now.date(),
                )
                for result in account_results:
                    ad_metrics = metrics.get(int(result["adgroup_id"]), {})
                    result["today_cost_fen"] = int(
                        ad_metrics.get("cost_fen") or 0
                    )
                    result["today_impressions"] = int(
                        ad_metrics.get("impressions") or 0
                    )
                    result["today_clicks"] = int(
                        ad_metrics.get("clicks") or 0
                    )
            except Exception as exc:
                logger.warning(
                    "Failed to fetch today's metrics account=%s: %s",
                    account_id,
                    exc,
                )

        notification = send_action_notification(
            run_id=run_id,
            now=now,
            results=results,
            bid_up_ratio=config.bid_up_ratio,
            decision=decision,
            observed_partition=observed_partition,
            observed_cpm=observed_cpm,
            test_mode=test_notification,
        )
        if apply:
            try:
                from feishu_notifier import (
                    send_operator_reconciliation_notification,
                )

                send_operator_reconciliation_notification(results)
            except Exception:
                logger.exception(
                    "Failed to notify operator manual override reconciliation"
                )

    return {
        "failures": failures,
        "api_updates": api_updates,
        "results": results,
        "notification": notification,
    }


def run_cycle(
    *,
    now: datetime,
    apply: bool,
    cpm_override: Decimal | None = None,
    force_refresh: bool = False,
    test_notification: bool = False,
) -> dict[str, Any]:
    from odps_source import HourlyCpm, build_odps_client, fetch_latest_cpm
    from realtime_config import RealtimeControlConfig
    from storage import (
        advisory_lock,
        load_daily_state,
        load_realtime_accounts,
        save_daily_state,
    )
    from tencent_client import TencentClient

    config = RealtimeControlConfig.from_env()
    run_id = str(uuid.uuid4())
    payload: dict[str, Any] = {
        "run_id": run_id,
        "now": now.isoformat(),
        "apply": apply,
        "decision": None,
        "observed_partition": None,
        "observed_cpm": None,
        "summary": {},
    }

    with advisory_lock(config.lock_name) as acquired:
        if not acquired:
            payload["decision"] = "LOCKED_BY_OTHER_WORKER"
            return payload

        daily_state = load_daily_state(now.date())
        accounts = load_realtime_accounts()
        if now.hour < config.start_hour:
            payload["decision"] = DECISION_OFF_HOURS
            return payload

        if now.hour >= config.stop_hour:
            payload["decision"] = DECISION_CUTOFF
            if apply and bool(daily_state.get("cutoff_done")) and not force_refresh:
                payload["summary"] = {"status": "cutoff_already_done"}
                return payload
            execution = execute_inventory_action(
                run_id=run_id,
                now=now,
                decision=DECISION_RESTORE,
                observed_partition=None,
                observed_cpm=None,
                apply=apply,
                accounts=accounts,
                config=config,
                tencent=TencentClient(),
                test_notification=test_notification,
            )
            payload["summary"] = execution
            if apply and not execution["failures"]:
                save_daily_state(
                    now.date(),
                    cutoff_done=True,
                    last_decision=DECISION_CUTOFF,
                    last_inventory_refresh_at=now,
                    last_evaluated_at=now,
                )
            return payload

        if apply and not bool(daily_state.get("morning_recovery_done")):
            execution = execute_inventory_action(
                run_id=run_id,
                now=now,
                decision=DECISION_RESTORE,
                observed_partition=None,
                observed_cpm=None,
                apply=True,
                accounts=accounts,
                config=config,
                tencent=TencentClient(),
                test_notification=test_notification,
            )
            payload["decision"] = "MORNING_RECOVERY"
            payload["summary"] = execution
            if not execution["failures"]:
                save_daily_state(
                    now.date(),
                    morning_recovery_done=True,
                    last_inventory_refresh_at=now,
                    last_evaluated_at=now,
                )
            else:
                return payload

        if cpm_override is not None:
            signal = HourlyCpm(
                partition=f"{now.strftime('%Y%m%d%H')}_override",
                hour_of_day=now.hour,
                impressions=0,
                cpm=float(cpm_override),
            )
        else:
            signal = fetch_latest_cpm(build_odps_client(), now.date())
        earliest_signal_hour = max(0, config.start_hour - 1)
        if signal is None or signal.hour_of_day < earliest_signal_hour:
            payload["decision"] = DECISION_WAIT_DATA
            payload["summary"] = {
                "reason": (
                    "No same-day CPM partition at or after "
                    f"{earliest_signal_hour:02d}:00"
                )
            }
            return payload

        observed_cpm = Decimal(str(signal.cpm))
        lag_hours = partition_lag_hours(signal.partition, now)
        payload.update(
            {
                "observed_partition": signal.partition,
                "observed_cpm": float(observed_cpm),
                "impressions": signal.impressions,
                "partition_lag_hours": round(lag_hours, 4),
            }
        )
        if lag_hours < 0 or lag_hours > config.max_partition_lag_hours:
            payload["decision"] = DECISION_WAIT_DATA
            payload["summary"] = {
                "reason": (
                    f"CPM partition lag {lag_hours:.2f}h exceeds allowed "
                    f"{config.max_partition_lag_hours}h"
                )
            }
            return payload

        decision = decide(observed_cpm, config.low_cpm, config.high_cpm)
        payload.update(
            {
                "decision": decision,
            }
        )

        refresh = force_refresh or should_refresh_inventory(
            daily_state,
            observed_partition=signal.partition,
            decision=decision,
            now=now,
            refresh_minutes=config.inventory_refresh_minutes,
        )
        if apply and not refresh:
            save_daily_state(now.date(), last_evaluated_at=now)
            payload["summary"] = {"status": "same_signal_noop"}
            return payload

        execution = execute_inventory_action(
            run_id=run_id,
            now=now,
            decision=decision,
            observed_partition=signal.partition,
            observed_cpm=observed_cpm,
            apply=apply,
            accounts=accounts,
            config=config,
            tencent=TencentClient(),
            test_notification=test_notification,
        )
        payload["summary"] = execution
        if apply and not execution["failures"]:
            save_daily_state(
                now.date(),
                last_observed_partition=signal.partition,
                last_observed_cpm=observed_cpm,
                last_decision=decision,
                last_inventory_refresh_at=now,
                last_evaluated_at=now,
            )
        return payload


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Execute Tencent updates. Default is dry-run.",
    )
    parser.add_argument(
        "--at",
        help="Dry-run only: simulate Shanghai local time, ISO format.",
    )
    parser.add_argument(
        "--cpm-override",
        type=Decimal,
        help="Dry-run only: override latest CPM to validate decision branches.",
    )
    parser.add_argument("--force-refresh", action="store_true")
    parser.add_argument(
        "--test-notification",
        action="store_true",
        help=(
            "Dry-run only: use real inventory and metrics, then send a clearly "
            "marked simulated Feishu action sheet."
        ),
    )
    return parser.parse_args()


def main() -> int:
    load_environment()
    args = parse_args()
    if args.apply and (args.at or args.cpm_override is not None):
        raise ValueError("--at and --cpm-override are forbidden with --apply")
    if args.apply and args.test_notification:
        raise ValueError("--test-notification is forbidden with --apply")
    now = (
        datetime.fromisoformat(args.at).replace(tzinfo=SHANGHAI)
        if args.at
        else datetime.now(SHANGHAI)
    )
    payload = run_cycle(
        now=now,
        apply=args.apply,
        cpm_override=args.cpm_override,
        force_refresh=args.force_refresh,
        test_notification=args.test_notification,
    )
    report = write_run_report(payload)
    compact = {
        "decision": payload["decision"],
        "observed_partition": payload.get("observed_partition"),
        "observed_cpm": payload.get("observed_cpm"),
        "apply": payload["apply"],
        "api_updates": payload.get("summary", {}).get("api_updates", 0),
        "failures": payload.get("summary", {}).get("failures", 0),
        "notification": payload.get("summary", {}).get("notification", {}).get(
            "status"
        ),
        "report": str(report),
    }
    print(json.dumps(compact, ensure_ascii=False))
    return 1 if compact["failures"] else 0


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )
    try:
        raise SystemExit(main())
    except Exception as exc:
        logger.exception("Real-time control failed: %s", exc)
        raise SystemExit(1)