|
|
@@ -589,13 +589,13 @@ async def execute_decisions(
|
|
|
exec_status_override = pause_result["exec_status"]
|
|
|
elif action in ("bid_up", "bid_down"):
|
|
|
final_bid = row.get("final_bid", row.get("recommended_bid"))
|
|
|
- if final_bid is None or final_bid == "":
|
|
|
+ if final_bid is None or final_bid == "" or pd.isna(final_bid):
|
|
|
audit.log({
|
|
|
"ad_id": ad_id,
|
|
|
"action": action,
|
|
|
"tier": 1,
|
|
|
"execution_status": "skipped",
|
|
|
- "reason": "无出价数据",
|
|
|
+ "reason": "无出价数据（NaN/空）",
|
|
|
})
|
|
|
continue
|
|
|
bid_fen = int(float(final_bid) * 100)
|
|
|
@@ -648,7 +648,17 @@ async def execute_decisions(
|
|
|
|
|
|
# ═══ Phase 2: Tier 2/3 — 审批 + 执行 ═══
|
|
|
if not df_tier2_3.empty:
|
|
|
- if IM_ENABLED:
|
|
|
+ # 决定审批状态
|
|
|
+ if filter_ad_ids is not None:
|
|
|
+ # 部分批准型：filter_ad_ids 已是上游审批结果，直接视为"已批准"，**严禁再发审批表**
|
|
|
+ logger.info(
|
|
|
+ "Tier 2/3 共 %d 个操作，filter_ad_ids 已上游批准，跳过 Phase 2 二次审批，直接执行",
|
|
|
+ len(df_tier2_3),
|
|
|
+ )
|
|
|
+ approval_status = "approved"
|
|
|
+ approved_ids = list({int(x) for x in filter_ad_ids})
|
|
|
+ rejected_ids = []
|
|
|
+ elif IM_ENABLED:
|
|
|
# 阻塞式审批：调用 send_approval_request(wait_for_reply=True)
|
|
|
logger.info("Tier 2/3 共 %d 个操作，发送 IM 审批并等待...", len(df_tier2_3))
|
|
|
|
|
|
@@ -674,139 +684,131 @@ async def execute_decisions(
|
|
|
if approval_result.metadata
|
|
|
else []
|
|
|
)
|
|
|
+ else:
|
|
|
+ # 兜底：IM 关 + 无 filter_ad_ids，全部标记 timeout
|
|
|
+ approval_status = "timeout"
|
|
|
+ approved_ids = []
|
|
|
+ rejected_ids = []
|
|
|
+
|
|
|
+ # 根据审批状态执行
|
|
|
+ if approval_status == "timeout":
|
|
|
+ # 超时：所有 Tier 2/3 标记为 timeout
|
|
|
+ timeout_count = len(df_tier2_3)
|
|
|
+ for _, row in df_tier2_3.iterrows():
|
|
|
+ audit.log({
|
|
|
+ "ad_id": int(row["ad_id"]),
|
|
|
+ "account_id": int(row.get("account_id", 0) or 0),
|
|
|
+ "action": row.get("final_action", row.get("action")),
|
|
|
+ "tier": int(row.get("tier", 2)),
|
|
|
+ "execution_status": "timeout",
|
|
|
+ "source": row.get("source", ""),
|
|
|
+ })
|
|
|
+ else:
|
|
|
+ # 执行已批准的广告
|
|
|
+ approved_set = set(int(x) for x in approved_ids)
|
|
|
+ rejected_set = set(int(x) for x in rejected_ids)
|
|
|
+
|
|
|
+ for _, row in df_tier2_3.iterrows():
|
|
|
+ ad_id = int(row["ad_id"])
|
|
|
+ account_id = int(row.get("account_id", 0) or 0)
|
|
|
+ action = row.get("final_action", row.get("action"))
|
|
|
+ tier = int(row.get("tier", 2))
|
|
|
|
|
|
- if approval_status == "timeout":
|
|
|
- # 超时：所有 Tier 2/3 标记为 timeout
|
|
|
- timeout_count = len(df_tier2_3)
|
|
|
- for _, row in df_tier2_3.iterrows():
|
|
|
+ if ad_id in rejected_set:
|
|
|
+ rejected_count += 1
|
|
|
audit.log({
|
|
|
- "ad_id": int(row["ad_id"]),
|
|
|
- "account_id": int(row.get("account_id", 0) or 0),
|
|
|
- "action": row.get("final_action", row.get("action")),
|
|
|
- "tier": int(row.get("tier", 2)),
|
|
|
- "execution_status": "timeout",
|
|
|
+ "ad_id": ad_id,
|
|
|
+ "account_id": account_id,
|
|
|
+ "action": action,
|
|
|
+ "tier": tier,
|
|
|
+ "execution_status": "rejected",
|
|
|
"source": row.get("source", ""),
|
|
|
})
|
|
|
- else:
|
|
|
- # 执行已批准的广告
|
|
|
- approved_set = set(int(x) for x in approved_ids)
|
|
|
- rejected_set = set(int(x) for x in rejected_ids)
|
|
|
-
|
|
|
- for _, row in df_tier2_3.iterrows():
|
|
|
- ad_id = int(row["ad_id"])
|
|
|
- account_id = int(row.get("account_id", 0) or 0)
|
|
|
- action = row.get("final_action", row.get("action"))
|
|
|
- tier = int(row.get("tier", 2))
|
|
|
-
|
|
|
- if ad_id in rejected_set:
|
|
|
- rejected_count += 1
|
|
|
- audit.log({
|
|
|
- "ad_id": ad_id,
|
|
|
- "account_id": account_id,
|
|
|
- "action": action,
|
|
|
- "tier": tier,
|
|
|
- "execution_status": "rejected",
|
|
|
- "source": row.get("source", ""),
|
|
|
- })
|
|
|
- continue
|
|
|
+ continue
|
|
|
|
|
|
- if ad_id not in approved_set:
|
|
|
- # 既不在 approved 也不在 rejected（部分审批场景遗漏）
|
|
|
- pending_approval += 1
|
|
|
- audit.log({
|
|
|
- "ad_id": ad_id,
|
|
|
- "account_id": account_id,
|
|
|
- "action": action,
|
|
|
- "tier": tier,
|
|
|
- "execution_status": "pending_approval",
|
|
|
- "source": row.get("source", ""),
|
|
|
- })
|
|
|
- continue
|
|
|
-
|
|
|
- # 已批准 → 执行
|
|
|
- pre_state = await executor.get_ad_state(ad_id, account_id)
|
|
|
- pause_extra: Dict[str, Any] = {}
|
|
|
-
|
|
|
- if action == "pause":
|
|
|
- pause_result = await _execute_pause(executor, row, audit, history)
|
|
|
- result = {"code": pause_result["code"], "message": pause_result["message"]}
|
|
|
- pause_extra = {
|
|
|
- "pause_scope": pause_result["scope"],
|
|
|
- "creative_results": pause_result["creative_results"],
|
|
|
- }
|
|
|
- exec_status_override = pause_result["exec_status"]
|
|
|
- elif action in ("bid_up", "bid_down"):
|
|
|
- final_bid = row.get("final_bid", row.get("recommended_bid"))
|
|
|
- if final_bid is None or final_bid == "":
|
|
|
- audit.log({
|
|
|
- "ad_id": ad_id,
|
|
|
- "action": action,
|
|
|
- "tier": tier,
|
|
|
- "execution_status": "skipped",
|
|
|
- "reason": "无出价数据",
|
|
|
- })
|
|
|
- continue
|
|
|
- bid_fen = int(float(final_bid) * 100)
|
|
|
- result = await executor.update_bid(ad_id, account_id, bid_fen)
|
|
|
- exec_status_override = None
|
|
|
- else:
|
|
|
- continue
|
|
|
-
|
|
|
- api_code = result.get("code", -1)
|
|
|
- if exec_status_override is not None:
|
|
|
- exec_status = exec_status_override
|
|
|
- else:
|
|
|
- exec_status = "success" if api_code == 0 else "failed"
|
|
|
-
|
|
|
- if exec_status in ("success", "partial"):
|
|
|
- approved_executed += 1
|
|
|
- change_pct = row.get("recommended_change_pct", 0)
|
|
|
- if isinstance(change_pct, str):
|
|
|
- try:
|
|
|
- change_pct = float(change_pct)
|
|
|
- except ValueError:
|
|
|
- change_pct = 0
|
|
|
- history.record_adjustment(str(ad_id), action, change_pct)
|
|
|
- else:
|
|
|
- failed += 1
|
|
|
-
|
|
|
- post_state = await executor.get_ad_state(ad_id, account_id) if exec_status in ("success", "partial") else None
|
|
|
-
|
|
|
- audit_entry = {
|
|
|
+ if ad_id not in approved_set:
|
|
|
+ # 既不在 approved 也不在 rejected（部分审批场景遗漏）
|
|
|
+ pending_approval += 1
|
|
|
+ audit.log({
|
|
|
"ad_id": ad_id,
|
|
|
"account_id": account_id,
|
|
|
"action": action,
|
|
|
"tier": tier,
|
|
|
- "pre_state": {
|
|
|
- "bid_amount": pre_state.get("bid_amount") if pre_state else None,
|
|
|
- "status": pre_state.get("configured_status") if pre_state else None,
|
|
|
- } if pre_state else None,
|
|
|
- "post_state": {
|
|
|
- "bid_amount": post_state.get("bid_amount") if post_state else None,
|
|
|
- "status": post_state.get("configured_status") if post_state else None,
|
|
|
- } if post_state else None,
|
|
|
- "api_code": api_code,
|
|
|
- "api_message": result.get("message", ""),
|
|
|
- "execution_status": f"approved_{exec_status}",
|
|
|
+ "execution_status": "pending_approval",
|
|
|
"source": row.get("source", ""),
|
|
|
+ })
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 已批准 → 执行
|
|
|
+ pre_state = await executor.get_ad_state(ad_id, account_id)
|
|
|
+ pause_extra: Dict[str, Any] = {}
|
|
|
+
|
|
|
+ if action == "pause":
|
|
|
+ pause_result = await _execute_pause(executor, row, audit, history)
|
|
|
+ result = {"code": pause_result["code"], "message": pause_result["message"]}
|
|
|
+ pause_extra = {
|
|
|
+ "pause_scope": pause_result["scope"],
|
|
|
+ "creative_results": pause_result["creative_results"],
|
|
|
}
|
|
|
- if pause_extra:
|
|
|
- audit_entry.update(pause_extra)
|
|
|
- audit.log(audit_entry)
|
|
|
- else:
|
|
|
- # IM 未启用：Tier 2/3 仅记录不执行
|
|
|
- logger.info("IM 未启用，Tier 2/3 共 %d 个操作仅记录不执行", len(df_tier2_3))
|
|
|
- pending_approval = len(df_tier2_3)
|
|
|
- for _, row in df_tier2_3.iterrows():
|
|
|
- audit.log({
|
|
|
- "ad_id": int(row["ad_id"]),
|
|
|
- "account_id": int(row.get("account_id", 0) or 0),
|
|
|
- "action": row.get("final_action", row.get("action")),
|
|
|
- "tier": int(row.get("tier", 2)),
|
|
|
- "execution_status": "pending_approval",
|
|
|
- "note": "IM未启用，操作仅记录",
|
|
|
+ exec_status_override = pause_result["exec_status"]
|
|
|
+ elif action in ("bid_up", "bid_down"):
|
|
|
+ final_bid = row.get("final_bid", row.get("recommended_bid"))
|
|
|
+ if final_bid is None or final_bid == "" or pd.isna(final_bid):
|
|
|
+ audit.log({
|
|
|
+ "ad_id": ad_id,
|
|
|
+ "action": action,
|
|
|
+ "tier": tier,
|
|
|
+ "execution_status": "skipped",
|
|
|
+ "reason": "无出价数据（NaN/空）",
|
|
|
+ })
|
|
|
+ continue
|
|
|
+ bid_fen = int(float(final_bid) * 100)
|
|
|
+ result = await executor.update_bid(ad_id, account_id, bid_fen)
|
|
|
+ exec_status_override = None
|
|
|
+ else:
|
|
|
+ continue
|
|
|
+
|
|
|
+ api_code = result.get("code", -1)
|
|
|
+ if exec_status_override is not None:
|
|
|
+ exec_status = exec_status_override
|
|
|
+ else:
|
|
|
+ exec_status = "success" if api_code == 0 else "failed"
|
|
|
+
|
|
|
+ if exec_status in ("success", "partial"):
|
|
|
+ approved_executed += 1
|
|
|
+ change_pct = row.get("recommended_change_pct", 0)
|
|
|
+ if isinstance(change_pct, str):
|
|
|
+ try:
|
|
|
+ change_pct = float(change_pct)
|
|
|
+ except ValueError:
|
|
|
+ change_pct = 0
|
|
|
+ history.record_adjustment(str(ad_id), action, change_pct)
|
|
|
+ else:
|
|
|
+ failed += 1
|
|
|
+
|
|
|
+ post_state = await executor.get_ad_state(ad_id, account_id) if exec_status in ("success", "partial") else None
|
|
|
+
|
|
|
+ audit_entry = {
|
|
|
+ "ad_id": ad_id,
|
|
|
+ "account_id": account_id,
|
|
|
+ "action": action,
|
|
|
+ "tier": tier,
|
|
|
+ "pre_state": {
|
|
|
+ "bid_amount": pre_state.get("bid_amount") if pre_state else None,
|
|
|
+ "status": pre_state.get("configured_status") if pre_state else None,
|
|
|
+ } if pre_state else None,
|
|
|
+ "post_state": {
|
|
|
+ "bid_amount": post_state.get("bid_amount") if post_state else None,
|
|
|
+ "status": post_state.get("configured_status") if post_state else None,
|
|
|
+ } if post_state else None,
|
|
|
+ "api_code": api_code,
|
|
|
+ "api_message": result.get("message", ""),
|
|
|
+ "execution_status": f"approved_{exec_status}",
|
|
|
"source": row.get("source", ""),
|
|
|
- })
|
|
|
+ }
|
|
|
+ if pause_extra:
|
|
|
+ audit_entry.update(pause_extra)
|
|
|
+ audit.log(audit_entry)
|
|
|
|
|
|
total_executed = executed + approved_executed
|
|
|
|
|
|
@@ -817,7 +819,7 @@ async def execute_decisions(
|
|
|
f" Tier 1 自动执行: {executed} 个成功 / {failed} 个失败",
|
|
|
]
|
|
|
|
|
|
- if IM_ENABLED and not df_tier2_3.empty:
|
|
|
+ if not df_tier2_3.empty and (filter_ad_ids is not None or IM_ENABLED):
|
|
|
output_lines.extend([
|
|
|
f" Tier 2/3 审批后执行: {approved_executed} 个成功",
|
|
|
f" 审批拒绝: {rejected_count} 个",
|
刘立冬