|
|
@@ -1478,6 +1478,8 @@ def run_rejected_creative_cleanup(
|
|
|
"RTC_DB_LOCK_NAME", "tencent_realtime_control"
|
|
|
)
|
|
|
retryable_items = load_retryable_cleanup_items()
|
|
|
+ # 阶段一（锁外）：纯前置判断，无腾讯写。通过者进入 deletable_items。
|
|
|
+ deletable_items: list[dict[str, Any]] = []
|
|
|
for item in retryable_items if apply_enabled else []:
|
|
|
item_id = int(item["id"])
|
|
|
account_id = int(item["account_id"])
|
|
|
@@ -1512,8 +1514,7 @@ def run_rejected_creative_cleanup(
|
|
|
)
|
|
|
deferred += 1
|
|
|
continue
|
|
|
- outcome_unknown = item.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN"
|
|
|
- if not outcome_unknown:
|
|
|
+ if item.get("cleanup_status") != "WRITE_OUTCOME_UNKNOWN":
|
|
|
precondition_failure = cleanup_precondition_failure(
|
|
|
item,
|
|
|
scanned_accounts,
|
|
|
@@ -1529,23 +1530,36 @@ def run_rejected_creative_cleanup(
|
|
|
if status == "DEFERRED":
|
|
|
deferred += 1
|
|
|
continue
|
|
|
- with advisory_lock(write_lock_name) as acquired:
|
|
|
- if not acquired:
|
|
|
- update_cleanup_item(
|
|
|
- item_id,
|
|
|
- cleanup_status="DEFERRED",
|
|
|
- error_message="腾讯写锁被实时调控占用",
|
|
|
- )
|
|
|
- deferred += 1
|
|
|
- continue
|
|
|
+ deletable_items.append(item)
|
|
|
+
|
|
|
+ # 阶段二（锁内并发）：整个删除批次持锁一次，锁内并发回读/复审/删除。
|
|
|
+ # 每 worker 用独立 TencentClient（requests.Session 非线程安全），
|
|
|
+ # 避免共享 session 并发导致不可预测行为（与扫描阶段一致）。
|
|
|
+ if deletable_items:
|
|
|
+ delete_workers = min(
|
|
|
+ int(os.getenv("TENCENT_AD_DELETE_WORKERS", "4")),
|
|
|
+ len(deletable_items),
|
|
|
+ 16,
|
|
|
+ )
|
|
|
+
|
|
|
+ def delete_one(item, delete_client):
|
|
|
+ """锁内单条创意：回读 → 复审 → 删除；返回 (deleted, deferred, error)。"""
|
|
|
+ item_id = int(item["id"])
|
|
|
+ account_id = int(item["account_id"])
|
|
|
+ creative_id = int(item["dynamic_creative_id"])
|
|
|
try:
|
|
|
try:
|
|
|
- before = client.get_dynamic_creative(account_id, creative_id)
|
|
|
+ before = delete_client.get_dynamic_creative(
|
|
|
+ account_id, creative_id
|
|
|
+ )
|
|
|
except Exception as read_exc:
|
|
|
if (
|
|
|
item.get("cleanup_action") == DELETE_CREATIVE
|
|
|
- and item.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN"
|
|
|
- and str(read_exc).startswith("Dynamic creative not found:")
|
|
|
+ and item.get("cleanup_status")
|
|
|
+ == "WRITE_OUTCOME_UNKNOWN"
|
|
|
+ and str(read_exc).startswith(
|
|
|
+ "Dynamic creative not found:"
|
|
|
+ )
|
|
|
):
|
|
|
update_cleanup_item(
|
|
|
item_id,
|
|
|
@@ -1554,10 +1568,9 @@ def run_rejected_creative_cleanup(
|
|
|
readback_json=_json({"deleted_from_listing": True}),
|
|
|
deleted_at=effective_now,
|
|
|
)
|
|
|
- deleted += 1
|
|
|
- continue
|
|
|
+ return 1, 0, None
|
|
|
raise
|
|
|
- if outcome_unknown:
|
|
|
+ if item.get("cleanup_status") == "WRITE_OUTCOME_UNKNOWN":
|
|
|
precondition_failure = cleanup_precondition_failure(
|
|
|
item,
|
|
|
scanned_accounts,
|
|
|
@@ -1571,8 +1584,8 @@ def run_rejected_creative_cleanup(
|
|
|
error_message=reason,
|
|
|
)
|
|
|
if status == "DEFERRED":
|
|
|
- deferred += 1
|
|
|
- continue
|
|
|
+ return 0, 1, None
|
|
|
+ return 0, 0, None
|
|
|
action = str(item.get("cleanup_action") or "")
|
|
|
approval_status = str(
|
|
|
before.get("creative_set_approval_status") or ""
|
|
|
@@ -1586,7 +1599,8 @@ def run_rejected_creative_cleanup(
|
|
|
(
|
|
|
result
|
|
|
for result in fresh_results
|
|
|
- if _as_int(result.get("dynamic_creative_id")) == creative_id
|
|
|
+ if _as_int(result.get("dynamic_creative_id"))
|
|
|
+ == creative_id
|
|
|
),
|
|
|
None,
|
|
|
)
|
|
|
@@ -1594,12 +1608,14 @@ def run_rejected_creative_cleanup(
|
|
|
fresh_spend_error = None
|
|
|
if approval_status == CREATIVE_PARTIAL_NORMAL_STATUS:
|
|
|
try:
|
|
|
- fresh_cost_fen = client.get_dynamic_creative_costs(
|
|
|
- account_id,
|
|
|
- [creative_id],
|
|
|
- spend_start_date,
|
|
|
- spend_end_date,
|
|
|
- ).get(creative_id, 0)
|
|
|
+ fresh_cost_fen = (
|
|
|
+ delete_client.get_dynamic_creative_costs(
|
|
|
+ account_id,
|
|
|
+ [creative_id],
|
|
|
+ spend_start_date,
|
|
|
+ spend_end_date,
|
|
|
+ ).get(creative_id, 0)
|
|
|
+ )
|
|
|
except Exception as spend_exc:
|
|
|
fresh_spend_error = str(spend_exc)
|
|
|
fresh_action = determine_cleanup_action(
|
|
|
@@ -1610,10 +1626,7 @@ def run_rejected_creative_cleanup(
|
|
|
wechat_cost_threshold_fen=wechat_cost_threshold_fen,
|
|
|
spend_error=fresh_spend_error,
|
|
|
)
|
|
|
- if not _same_cleanup_action(
|
|
|
- action,
|
|
|
- fresh_action,
|
|
|
- ):
|
|
|
+ if not _same_cleanup_action(action, fresh_action):
|
|
|
if (
|
|
|
fresh_action
|
|
|
and fresh_action.get("cleanup_action") == ALERT_ONLY
|
|
|
@@ -1639,17 +1652,16 @@ def run_rejected_creative_cleanup(
|
|
|
deleted_at=None,
|
|
|
notified_at=None,
|
|
|
)
|
|
|
- continue
|
|
|
+ return 0, 0, None
|
|
|
update_cleanup_item(
|
|
|
item_id,
|
|
|
cleanup_status="SKIPPED_REVIEW_NOT_RECONFIRMED",
|
|
|
error_message="写锁内回读发现整创意删除条件已变化",
|
|
|
pre_state_json=_json(before),
|
|
|
)
|
|
|
- continue
|
|
|
-
|
|
|
+ return 0, 0, None
|
|
|
if action == DELETE_CREATIVE:
|
|
|
- readback = client.delete_dynamic_creative(
|
|
|
+ readback = delete_client.delete_dynamic_creative(
|
|
|
account_id, creative_id
|
|
|
)
|
|
|
update_cleanup_item(
|
|
|
@@ -1660,8 +1672,8 @@ def run_rejected_creative_cleanup(
|
|
|
readback_json=_json(readback),
|
|
|
deleted_at=effective_now,
|
|
|
)
|
|
|
- deleted += 1
|
|
|
- continue
|
|
|
+ return 1, 0, None
|
|
|
+ return 0, 0, None
|
|
|
except Exception as exc:
|
|
|
from tencent_client import (
|
|
|
PostWriteVerificationError,
|
|
|
@@ -1682,9 +1694,48 @@ def run_rejected_creative_cleanup(
|
|
|
),
|
|
|
error_message=str(exc)[:4000],
|
|
|
)
|
|
|
- delete_errors.append(
|
|
|
- f"account={account_id} creative={creative_id}: {exc}"
|
|
|
+ return 0, 0, f"account={account_id} creative={creative_id}: {exc}"
|
|
|
+
|
|
|
+ def run_delete(item):
|
|
|
+ if owned_tencent:
|
|
|
+ from tencent_client import TencentClient
|
|
|
+
|
|
|
+ delete_client = TencentClient()
|
|
|
+ delete_client.seed_access_tokens(prefetched_tokens)
|
|
|
+ else:
|
|
|
+ delete_client = client
|
|
|
+ try:
|
|
|
+ return delete_one(item, delete_client)
|
|
|
+ finally:
|
|
|
+ if delete_client is not client:
|
|
|
+ delete_client.session.close()
|
|
|
+
|
|
|
+ with advisory_lock(write_lock_name) as acquired:
|
|
|
+ if not acquired:
|
|
|
+ for item in deletable_items:
|
|
|
+ update_cleanup_item(
|
|
|
+ int(item["id"]),
|
|
|
+ cleanup_status="DEFERRED",
|
|
|
+ error_message="腾讯写锁被实时调控占用",
|
|
|
+ )
|
|
|
+ deferred += 1
|
|
|
+ else:
|
|
|
+ logger.info(
|
|
|
+ "creative delete started items=%d workers=%d",
|
|
|
+ len(deletable_items),
|
|
|
+ delete_workers,
|
|
|
)
|
|
|
+ with ThreadPoolExecutor(
|
|
|
+ max_workers=delete_workers,
|
|
|
+ thread_name_prefix="creative-delete",
|
|
|
+ ) as executor:
|
|
|
+ for d_deleted, d_deferred, d_error in executor.map(
|
|
|
+ run_delete, deletable_items
|
|
|
+ ):
|
|
|
+ deleted += d_deleted
|
|
|
+ deferred += d_deferred
|
|
|
+ if d_error:
|
|
|
+ delete_errors.append(d_error)
|
|
|
|
|
|
pending_notifications = load_unnotified_deleted_items(
|
|
|
include_discovered=not apply_enabled,
|
wangyunpeng