#!/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 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)