Agent/examples/auto_put_ad_mini/execute_creation_once.py

"""模块 B 创意搭建子系统 — 主循环入口(P0-A / P0-C / P0-E,2026-06-09)。

数据流(三阶段,先审后挂):
  Phase 1 — 准备:
    扫账户 → 关联点过滤(读调控当日 latest_decisions 排除 pause)
    → find_ads_needing_creatives
    → 对每条广告 prepare_one_creative_for_ad(召回 + 上传图 + xcx/save + build body)
    → pending_records (含完整 Phase 3 用的 _request_body)
    → 写 outputs/data/creation_pending_{date}.json(追溯 + 独立 apply 用)

  Phase 2 — 审批(若 CREATION_APPROVAL_REQUIRED=True):
    run_approval_workflow 生成 20 列 xlsx → 上传飞书 sheet → 发链接 → 轮询读决策列
    actions {row_idx: approve/reject/hold}
    → 写回 records["action"]

  Phase 3 — 执行:
    对 action=approve 的 → POST /dynamic_creatives/add
    → 写 outputs/data/creation_run_{date}.json
    → 发飞书"执行汇报"消息

开关 CREATION_APPROVAL_REQUIRED=False 时跳过 Phase 2,全 records 直接 approve。
"""

import argparse
import json
import logging
import os
import sys
import time
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from datetime import date
from pathlib import Path

_HERE = Path(__file__).parent
sys.path.insert(0, str(_HERE.parent.parent))
sys.path.insert(0, str(_HERE))

from dotenv import load_dotenv  # noqa: E402
load_dotenv(_HERE / ".env")

from config import (  # noqa: E402
    ADS_PER_ACCOUNT,
    CREATION_APPROVAL_REQUIRED,
    CREATION_APPROVAL_TIMEOUT_MINUTES,
    CREATIVE_PREPARE_MAX_WORKERS,
    CREATIVE_PREPARE_TASK_BUFFER,
    MAX_SAME_LANDING_PER_AD_IN_RUN,
    TARGET_CREATIVES_PER_AD,
    WHITELIST_ACCOUNTS,
    get_creation_account_ids,
    now_in_timezone,
)

# config import 副作用会改 sys.path(把 im-client / agent/tools/builtin 顶到前面),
# 强占 sys.path[0],只影响本入口自己的 import resolution。
while str(_HERE) in sys.path:
    sys.path.remove(str(_HERE))
sys.path.insert(0, str(_HERE))

from tools.ad_creation import (  # noqa: E402
    build_ad_request_body,
    compute_fingerprint,
    enumerate_new_ad_candidates,
    post_ad_with_prepared_body,
)
from tools.audience_grant import ensure_account_audience_grant  # noqa: E402
from tools.creative_creation import (  # noqa: E402
    build_landing_candidate_pool,
    find_ads_needing_creatives,
    load_excluded_ad_ids_from_adjustment,
    prepare_one_creative_for_ad,
)
from tools.creative_material_usage import (  # noqa: E402
    load_recent_landing_usage_counts,
    load_recoverable_prepared_records,
    record_prepared_material_usage,
)
from tools.account_material_strategy import (  # noqa: E402
    MATERIAL_SOURCE_AI_GENERATED,
    MATERIAL_SOURCE_EXTERNAL_RECALL,
    load_account_material_strategy,
)
from tools.video_recall import get_account_crowd_package  # noqa: E402

from execute_creation_apply import (  # noqa: E402
    apply_pending_records,
    write_summary as write_apply_summary,
    _send_apply_summary_to_feishu,
)

logger = logging.getLogger("execute_creation_once")

MATERIAL_SOURCE_HISTORY = "history"


def _env_flag(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}


def _env_csv_set(name: str) -> set[str]:
    return {
        item.strip()
        for item in os.getenv(name, "").split(",")
        if item.strip()
    }


def _filter_creation_accounts(accounts: list[int], phase: str) -> list[int]:
    """Apply one-off run filters. Empty env means production default: no filter."""
    account_filter = {
        int(v)
        for v in _env_csv_set("CREATION_ONLY_ACCOUNT_IDS")
        if v.isdigit()
    }
    crowd_filter = _env_csv_set("CREATION_ONLY_CROWD_PACKAGES")
    if not account_filter and not crowd_filter:
        return accounts

    selected: list[int] = []
    for account_id in accounts:
        if account_filter and account_id not in account_filter:
            continue
        if crowd_filter:
            try:
                crowd_package = get_account_crowd_package(account_id)
            except Exception as e:
                logger.warning(
                    "[%s] account=%d 临时过滤读取 crowd_package 失败,跳过:%s",
                    phase, account_id, e,
                )
                continue
            if crowd_package not in crowd_filter:
                continue
        selected.append(account_id)

    logger.info(
        "[%s] 临时运行过滤 account_ids=%s crowd_packages=%s: %d -> %d",
        phase,
        sorted(account_filter) if account_filter else "ALL",
        sorted(crowd_filter) if crowd_filter else "ALL",
        len(accounts),
        len(selected),
    )
    return selected


def _setup_logging() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
        datefmt="%H:%M:%S",
    )
    for noisy in ("httpx", "httpcore", "config", "db.config"):
        logging.getLogger(noisy).setLevel(logging.WARNING)

    # SLS 上报(2026-06-11 接入,K8s pod 直发)— 配置缺失自动降级,不阻塞主链路
    try:
        from tools.sls_setup import attach_sls_handler
        attach_sls_handler()
    except Exception as e:
        logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)


def _fetch_existing_fingerprints_for_account(account_id: int) -> set[str]:
    """Task 27:拉账户下所有非 DELETED 广告 → 反推 fingerprint 集合(预校验唯一性)。

    腾讯文档:NORMAL + SUSPEND 状态广告均占用唯一性槽位,DENIED 也算占用。
    list 接口默认返回所有非 DELETED 状态 ad,故无需显式 filter status。

    若腾讯 list 接口某些字段不返回(field 实测差异),降级:跳过该条 + log 警告。
    返回空集 → 上层退化为"不预校验",enumerate 仍能跑(原有行为)。
    """
    from tools.ad_api import _get

    fingerprints: set[str] = set()
    skipped = 0
    page = 1
    try:
        while True:
            r = _get("/adgroups/get", {
                "account_id": account_id, "page": page, "page_size": 100,
                "fields": [
                    "adgroup_id", "configured_status", "site_set",
                    "targeting", "scene_spec",
                ],
            })
            ads = (r.get("data") or {}).get("list") or []
            if not ads:
                break
            for ad in ads:
                tgt = ad.get("targeting") or {}
                sc = ad.get("scene_spec") or {}
                wp = sc.get("wechat_position") if sc else None
                site_set = ad.get("site_set") or []
                age = tgt.get("age") or []
                geo_regions = (tgt.get("geo_location") or {}).get("regions") or []
                custom_audience = tgt.get("custom_audience")
                if not site_set or not age:
                    skipped += 1
                    continue
                try:
                    fp = compute_fingerprint(
                        account_id=account_id,
                        site_set=site_set,
                        custom_audience=custom_audience,
                        age=age,
                        geo_regions=geo_regions,
                        wechat_position=wp,
                    )
                    fingerprints.add(fp)
                except Exception as e:
                    skipped += 1
                    logger.debug(
                        "[phase0] fingerprint 算失败 ad=%s: %s",
                        ad.get("adgroup_id"), e,
                    )
            if len(ads) < 100:
                break
            page += 1
    except Exception as e:
        logger.warning(
            "[phase0] account=%d 拉 fingerprint 集失败(降级为不预校验): %s",
            account_id, e,
        )
        return set()

    logger.info(
        "[phase0]   已存在 fingerprint=%d 个(skip 缺字段广告 %d 条)",
        len(fingerprints), skipped,
    )
    return fingerprints


def phase0_create_ads(
    target_ads: int = ADS_PER_ACCOUNT,
    config_date: date | None = None,
) -> list[dict]:
    """Phase 0:对每账户检查广告数,不足则建到 target_ads 条(模块 A,2026-06-09 P1-G)。

    流程:
      account → 查当前广告数(NORMAL+SUSPEND 算占用)→ 不足 target_ads →
      → 反查 fingerprint 集 → enumerate candidates(排除已有 fp)→
      → 飞书审批 → approve 项 → 真 POST /adgroups/add → 返回新建的 adgroup_id 列表

    Returns: 本次新建成功的广告 list[dict],每项含 {account_id, adgroup_id, adgroup_name, wechat_position}
    """
    from tools.ad_api import _get
    from tools.im_approval_ad_creation import run_ad_approval_workflow

    creation_accounts = get_creation_account_ids(config_date=config_date)
    if not creation_accounts:
        logger.error("[phase0] 待投放账户配置为空,退出")
        return []
    creation_accounts = _filter_creation_accounts(creation_accounts, "phase0")
    if not creation_accounts:
        logger.info("[phase0] 临时过滤后无待处理账户")
        return []

    # Task 26:NORMAL + SUSPEND 都算占用唯一性槽位(腾讯文档:删除前历史广告占槽位)
    OCCUPIED_STATUSES = {"AD_STATUS_NORMAL", "AD_STATUS_SUSPEND"}

    pending_ad_records: list[dict] = []

    for account_id in creation_accounts:
        logger.info("=" * 60)
        logger.info("[phase0] 账户 %d 处理开始", account_id)

        try:
            grant_result = ensure_account_audience_grant(account_id)
            logger.info(
                "[phase0]   人群包校验完成: status=%s audience_id=%s audience=%r",
                grant_result.get("status"),
                grant_result.get("audience_id"),
                grant_result.get("audience_name"),
            )
        except Exception as e:
            logger.exception("[phase0] account=%d 人群包授权/验证失败,跳过:%s", account_id, e)
            continue

        # Task 26:查所有非 DELETED 广告,NORMAL+SUSPEND 都算占用槽位
        try:
            r = _get("/adgroups/get", {
                "account_id": account_id, "page": 1, "page_size": 100,
                "fields": ["adgroup_id", "configured_status"],
            })
            all_ads = (r.get("data") or {}).get("list") or []
            normal_n = sum(1 for a in all_ads if a.get("configured_status") == "AD_STATUS_NORMAL")
            suspend_n = sum(1 for a in all_ads if a.get("configured_status") == "AD_STATUS_SUSPEND")
            occupied = [a for a in all_ads if a.get("configured_status") in OCCUPIED_STATUSES]
            logger.info(
                "[phase0]   广告数: 总 %d,NORMAL %d + SUSPEND %d = 占用 %d / 目标 %d",
                len(all_ads), normal_n, suspend_n, len(occupied), target_ads,
            )
        except Exception as e:
            logger.exception("[phase0] account=%d 查广告数失败: %s", account_id, e)
            continue

        to_create = max(0, target_ads - len(occupied))
        if to_create == 0:
            logger.info("[phase0]   占用槽位已满,跳过")
            continue

        # Task 27:fingerprint 预校验 — 反查现存广告的 fp 集合,enumerate 时排除
        existing_fps = _fetch_existing_fingerprints_for_account(account_id)

        # enumerate 候选(模块 A,差异化 wechat_position 有/无)
        try:
            candidates = enumerate_new_ad_candidates(
                account_id, count=to_create,
                existing_fingerprints=existing_fps,
            )
        except Exception as e:
            logger.exception("[phase0] account=%d enumerate 失败: %s", account_id, e)
            continue

        if not candidates:
            logger.warning("[phase0]   enumerate 返回 0 条候选,跳过")
            continue

        today = now_in_timezone().strftime("%Y-%m-%d")
        for c in candidates:
            age_range = ",".join(
                f"{a.get('min')}-{a.get('max')}" for a in c.age
            )
            # 新建广告默认开启;实际投放仍受创意审核/账户/预算控制。
            # 这里显式传 NORMAL,避免未来默认值变化影响端到端创建语义。
            body = build_ad_request_body(c, configured_status="AD_STATUS_NORMAL")
            rec = {
                "approval_date": today,
                "account_id": c.account_id,
                "audience_tier_label": c.audience_tier_label,
                "adgroup_name": c.adgroup_name,
                "site_set": c.site_set,
                "delivery_version": c.delivery_version,
                "wechat_position": c.wechat_position,
                "bid_amount_fen": c.bid_amount_fen,
                "bid_scene": c.bid_scene,
                "custom_cost_cap_fen": c.custom_cost_cap_fen,
                "automatic_site_enabled": c.automatic_site_enabled,
                "age_range": age_range,
                "fingerprint": c.fingerprint,
                "_request_body": body,
            }
            pending_ad_records.append(rec)

    if not pending_ad_records:
        logger.info("[phase0] 无广告需新建,Phase 0 结束")
        return []

    # 飞书审批(如需要)
    data_dir = _HERE / "outputs" / "data"
    if CREATION_APPROVAL_REQUIRED:
        logger.info(
            "[phase0] CREATION_APPROVAL_REQUIRED=True → 飞书审批 %d 条广告候选",
            len(pending_ad_records),
        )
        sheet_meta, actions = run_ad_approval_workflow(
            records=pending_ad_records,
            xlsx_output_dir=data_dir,
            timeout_minutes=CREATION_APPROVAL_TIMEOUT_MINUTES,
        )
        for i, rec in enumerate(pending_ad_records, start=1):
            rec["action"] = actions.get(i, "skip")
    else:
        for rec in pending_ad_records:
            rec["action"] = "approve"

    # POST 真建 approve 项
    created: list[dict] = []
    for rec in pending_ad_records:
        if rec.get("action") != "approve":
            continue
        adgroup_id = post_ad_with_prepared_body(
            account_id=int(rec["account_id"]),
            body=rec["_request_body"],
        )
        if adgroup_id:
            created.append({
                "account_id": rec["account_id"],
                "adgroup_id": adgroup_id,
                "adgroup_name": rec["adgroup_name"],
                "wechat_position": rec.get("wechat_position"),
            })

    logger.info(
        "[phase0] 完成:候选 %d 条,approve %d 条,实际建 %d 条",
        len(pending_ad_records),
        sum(1 for r in pending_ad_records if r.get("action") == "approve"),
        len(created),
    )
    return created


def _wait_created_ads_visible(created_ads: list[dict], timeout_seconds: int = 60) -> None:
    """等待刚创建的广告在 /adgroups/get 中可见。

    腾讯 adgroups/add 成功后,list 接口存在短暂一致性延迟。若立刻进入 Phase 1,
    可能只扫到部分新广告,导致同一轮只给一条广告补创意。
    """
    if not created_ads:
        return

    from tools.ad_api import _get

    by_account: dict[int, set[int]] = {}
    for ad in created_ads:
        by_account.setdefault(int(ad["account_id"]), set()).add(int(ad["adgroup_id"]))

    deadline = time.time() + timeout_seconds
    remaining = {acct: set(ids) for acct, ids in by_account.items()}
    while time.time() < deadline and any(remaining.values()):
        for account_id, ids in list(remaining.items()):
            if not ids:
                continue
            try:
                resp = _get("/adgroups/get", {
                    "account_id": account_id,
                    "page": 1,
                    "page_size": 100,
                    "fields": ["adgroup_id", "configured_status", "system_status"],
                })
                visible = {
                    int(item["adgroup_id"])
                    for item in ((resp.get("data") or {}).get("list") or [])
                    if item.get("adgroup_id") is not None
                }
                ids.difference_update(visible)
            except Exception as e:
                logger.warning(
                    "[phase0] 等待新广告可见失败 account=%d: %s",
                    account_id, e,
                )
        if any(remaining.values()):
            logger.info("[phase0] 等待新广告在列表接口可见: %s", remaining)
            time.sleep(5)

    not_visible = {acct: sorted(ids) for acct, ids in remaining.items() if ids}
    if not_visible:
        logger.warning("[phase0] 部分新广告仍未在 list 可见,继续 Phase 1: %s", not_visible)
    else:
        logger.info("[phase0] 新建广告均已在 list 接口可见")


def phase1_prepare(
    target_creatives: int = TARGET_CREATIVES_PER_AD,
    config_date: date | None = None,
) -> list[dict]:
    """Phase 1:扫账户 → 关联点过滤 → 准备 pending records(不 POST 腾讯)。

    Returns:
        pending_records — 每个元素是 prepare_one_creative_for_ad 返回的 dict
    """
    creation_accounts = get_creation_account_ids(config_date=config_date)
    if not creation_accounts:
        logger.error("[phase1] 待投放账户配置为空,Phase 1 退出")
        return []
    creation_accounts = _filter_creation_accounts(creation_accounts, "phase1")
    if not creation_accounts:
        logger.info("[phase1] 临时过滤后无待处理账户")
        return []

    excluded_ad_ids = load_excluded_ad_ids_from_adjustment()
    logger.info("[phase1] 关联点过滤集合 size=%d", len(excluded_ad_ids))

    pending_records: list[dict] = []
    # 素材排重只读取已有明确结果的 DB 记录。
    # pending/prepared 没有审批或投放结果,不进入排重。
    # 但同一轮审批候选需要做展示去重,避免表格里同 crowd_package 反复出现同一素材。
    # 这个 set 只存在内存里,不写历史排重库;进程结束即失效。
    display_used_material_ids_by_crowd: dict[str, set[str]] = {}
    # landing_video 按素材来源拆池。历史素材和 AI 生成素材互不占用,但池内都严格去重。
    landing_usage_counts_by_crowd: dict[str, dict[str, dict[int, int]]] = {}

    def accept_pending_record(
        rec: dict,
        *,
        crowd_package: str,
        display_excluded_material_ids: set[str],
        crowd_landing_counts: dict[str, dict[int, int]],
        landing_counts_for_ad: dict[int, int],
        landing_max_uses: int,
        recovered: bool = False,
    ) -> bool:
        material_id = rec.get("_material_id")
        if material_id and str(material_id) in display_excluded_material_ids:
            logger.info(
                "[phase1] pending 接收排重丢弃 crowd=%r material=%s%s",
                crowd_package, material_id,
                " recovered" if recovered else "",
            )
            return False
        landing_video_id = rec.get("landing_video_id")
        record_material_source = str(rec.get("material_source") or "")
        record_material_id = str(rec.get("_material_id") or "")
        if (
            record_material_source == MATERIAL_SOURCE_EXTERNAL_RECALL
            or record_material_id.startswith("external:")
        ):
            actual_material_source = MATERIAL_SOURCE_EXTERNAL_RECALL
        elif (
            record_material_source == MATERIAL_SOURCE_AI_GENERATED
            or record_material_id.startswith("ai:")
        ):
            actual_material_source = MATERIAL_SOURCE_AI_GENERATED
        else:
            actual_material_source = MATERIAL_SOURCE_HISTORY
        if landing_video_id is not None:
            landing_video_id = int(landing_video_id)
            if landing_counts_for_ad.get(landing_video_id, 0) >= MAX_SAME_LANDING_PER_AD_IN_RUN:
                logger.info(
                    "[phase1] pending 接收排重丢弃 adgroup=%s landing=%d 同广告已达上限%s",
                    rec.get("adgroup_id"), landing_video_id,
                    " recovered" if recovered else "",
                )
                return False
            source_counts = crowd_landing_counts.setdefault(actual_material_source, {})
            if source_counts.get(landing_video_id, 0) >= landing_max_uses:
                logger.info(
                    "[phase1] pending 接收排重丢弃 crowd=%r material_source=%s landing=%d 同人群包已达上限%s",
                    crowd_package, actual_material_source, landing_video_id,
                    " recovered" if recovered else "",
                )
                return False

        pending_records.append(rec)
        if material_id:
            display_excluded_material_ids.add(str(material_id))
            logger.info(
                "[phase1] 本轮展示去重登记 crowd=%r material=%s size=%d%s",
                crowd_package, material_id,
                len(display_excluded_material_ids),
                " recovered" if recovered else "",
            )
        if landing_video_id is not None:
            landing_counts_for_ad[landing_video_id] = (
                landing_counts_for_ad.get(landing_video_id, 0) + 1
            )
            source_counts = crowd_landing_counts.setdefault(actual_material_source, {})
            source_counts[landing_video_id] = source_counts.get(landing_video_id, 0) + 1
            logger.info(
                "[phase1] 同人群包 landing 使用登记 crowd=%r material_source=%s landing=%d count=%d max=%d%s",
                crowd_package, actual_material_source, landing_video_id,
                source_counts[landing_video_id],
                landing_max_uses,
                " recovered" if recovered else "",
            )
            logger.info(
                "[phase1] 本轮同广告 landing 计数 adgroup=%s landing=%d count=%d limit=%d%s",
                rec.get("adgroup_id"), landing_video_id,
                landing_counts_for_ad[landing_video_id],
                MAX_SAME_LANDING_PER_AD_IN_RUN,
                " recovered" if recovered else "",
            )
        if not recovered:
            try:
                record_prepared_material_usage(rec)
            except Exception as e:
                logger.warning(
                    "[phase1] material usage 记录失败 account=%s adgroup=%s material=%s:%s",
                    rec.get("account_id"), rec.get("adgroup_id"),
                    rec.get("_material_id"), e,
                )
        return True

    for account_id in creation_accounts:
        logger.info("=" * 60)
        logger.info("[phase1] 账户 %d 处理开始", account_id)

        try:
            ads = find_ads_needing_creatives(account_id, min_creatives=target_creatives)
        except Exception as e:
            logger.exception("[phase1] account=%d find_ads 失败,跳过:%s", account_id, e)
            continue

        ads_after_filter = [a for a in ads if a["adgroup_id"] not in excluded_ad_ids]
        excluded_count = len(ads) - len(ads_after_filter)
        if excluded_count:
            logger.info(
                "[phase1] account=%d 关联点过滤排除 %d 条",
                account_id, excluded_count,
            )

        if not ads_after_filter:
            logger.info("[phase1] account=%d 无广告需补创意", account_id)
            continue

        try:
            crowd_package = get_account_crowd_package(account_id)
        except Exception as e:
            logger.exception(
                "[phase1] account=%d 读取 crowd_package 失败,跳过:%s",
                account_id, e,
            )
            continue
        display_excluded_material_ids = display_used_material_ids_by_crowd.setdefault(
            crowd_package, set(),
        )
        if crowd_package not in landing_usage_counts_by_crowd:
            try:
                landing_usage_counts_by_crowd[crowd_package] = load_recent_landing_usage_counts(
                    crowd_package,
                )
            except Exception as e:
                logger.warning(
                    "[phase1] crowd=%r 读取 landing 使用历史失败,仅使用本轮排重:%s",
                    crowd_package, e,
                )
                landing_usage_counts_by_crowd[crowd_package] = {
                    MATERIAL_SOURCE_HISTORY: {},
                    MATERIAL_SOURCE_AI_GENERATED: {},
                    MATERIAL_SOURCE_EXTERNAL_RECALL: {},
                }
            logger.info(
                "[phase1] crowd=%r 近期 landing 使用 history=%d ai_generated=%d external_recall=%d",
                crowd_package,
                len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_HISTORY, {})),
                len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_AI_GENERATED, {})),
                len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_EXTERNAL_RECALL, {})),
            )
        crowd_landing_counts = landing_usage_counts_by_crowd[crowd_package]
        crowd_landing_counts.setdefault(MATERIAL_SOURCE_HISTORY, {})
        crowd_landing_counts.setdefault(MATERIAL_SOURCE_AI_GENERATED, {})
        crowd_landing_counts.setdefault(MATERIAL_SOURCE_EXTERNAL_RECALL, {})

        try:
            material_strategy = load_account_material_strategy(account_id)
        except Exception as e:
            logger.warning(
                "[phase1] account=%d 读取素材策略失败,按 history 排重:%s",
                account_id, e,
            )
            material_strategy = None
        requested_material_source = (
            material_strategy.material_source
            if material_strategy is not None
            else MATERIAL_SOURCE_HISTORY
        )
        landing_max_uses = MAX_SAME_LANDING_PER_AD_IN_RUN
        landing_candidate_pool = None
        logger.info(
            "[phase1] account=%d landing 排重池=%s max_uses=%d",
            account_id, requested_material_source, landing_max_uses,
        )

        for ad in ads_after_filter:
            adgroup_id = ad["adgroup_id"]
            already_have = ad["creative_count"]
            to_add = max(0, target_creatives - already_have)
            prepared_for_ad = 0
            failed_prepare_for_ad = 0
            failed_landing_ids_for_ad: set[int] = set()
            landing_counts_for_ad: dict[int, int] = {}
            logger.info(
                "[phase1]   adgroup=%d(have=%d need=%d)",
                adgroup_id, already_have, to_add,
            )
            recovered_records = load_recoverable_prepared_records(
                account_id=account_id,
                adgroup_id=adgroup_id,
                crowd_package=crowd_package,
                material_source=requested_material_source,
                limit=to_add,
            )
            recovered_accepted = 0
            for rec in recovered_records:
                if accept_pending_record(
                    rec,
                    crowd_package=crowd_package,
                    display_excluded_material_ids=display_excluded_material_ids,
                    crowd_landing_counts=crowd_landing_counts,
                    landing_counts_for_ad=landing_counts_for_ad,
                    landing_max_uses=landing_max_uses,
                    recovered=True,
                ):
                    recovered_accepted += 1
            if recovered_records:
                prepared_for_ad += recovered_accepted
                logger.info(
                    "[phase1]   adgroup=%d 恢复未提交 prepared records=%d accepted=%d remaining=%d",
                    adgroup_id, len(recovered_records), recovered_accepted,
                    max(0, to_add - prepared_for_ad),
                )

            def prepare_attempt(
                *,
                attempt_no: int,
                excluded_material_ids_snapshot: set[str],
                excluded_landing_ids_snapshot: set[int],
            ) -> dict | None:
                nonlocal landing_candidate_pool
                try:
                    if landing_candidate_pool is None:
                        landing_candidate_pool = build_landing_candidate_pool(account_id)
                    return prepare_one_creative_for_ad(
                        account_id, adgroup_id,
                        excluded_material_ids=excluded_material_ids_snapshot,
                        excluded_landing_ids=excluded_landing_ids_snapshot,
                        landing_candidates=landing_candidate_pool,
                        failed_landing_ids=failed_landing_ids_for_ad,
                    )
                except Exception as e:
                    logger.exception(
                        "[phase1] adgroup=%d attempt=%d 准备失败:%s",
                        adgroup_id, attempt_no, e,
                    )
                    return None

            def current_excluded_landing_ids() -> set[int]:
                landing_excluded_for_ad = {
                    vid
                    for vid, count in landing_counts_for_ad.items()
                    if count >= MAX_SAME_LANDING_PER_AD_IN_RUN
                }
                landing_excluded_for_source = {
                    vid
                    for vid, count in crowd_landing_counts
                    .get(requested_material_source, {})
                    .items()
                    if count >= landing_max_uses
                }
                return landing_excluded_for_source | landing_excluded_for_ad

            remaining_to_add = max(0, to_add - prepared_for_ad)
            max_workers = max(1, int(CREATIVE_PREPARE_MAX_WORKERS))
            if max_workers > 1:
                logger.warning(
                    "[phase1] CREATIVE_PREPARE_MAX_WORKERS=%d 已配置但当前 prepare_one_creative_for_ad "
                    "包含图片上传/xcx-save 等外部副作用,为避免重复创建落地计划,本轮强制串行执行",
                    max_workers,
                )
                max_workers = 1
            task_buffer = max(remaining_to_add, int(CREATIVE_PREPARE_TASK_BUFFER))
            task_limit = min(max(0, remaining_to_add * 2), task_buffer)
            if remaining_to_add > 0 and max_workers > 1:
                logger.info(
                    "[phase1]   adgroup=%d 并行准备 start remaining=%d workers=%d task_limit=%d",
                    adgroup_id, remaining_to_add, max_workers, task_limit,
                )
                submitted = 0
                accepted_parallel = 0
                dropped_parallel = 0
                failed_parallel = 0
                started_at = time.monotonic()
                with ThreadPoolExecutor(max_workers=max_workers) as executor:
                    futures = {}

                    def submit_next() -> None:
                        nonlocal submitted
                        # 已接收 + 运行中任务不超过缺口,避免并发准备产出无法追踪的多余副作用。
                        if (
                            submitted >= task_limit
                            or prepared_for_ad + len(futures) >= to_add
                        ):
                            return
                        submitted += 1
                        attempt_no = submitted
                        future = executor.submit(
                            prepare_attempt,
                            attempt_no=attempt_no,
                            excluded_material_ids_snapshot=set(display_excluded_material_ids),
                            excluded_landing_ids_snapshot=current_excluded_landing_ids(),
                        )
                        futures[future] = attempt_no

                    for _ in range(min(max_workers, task_limit)):
                        submit_next()

                    while futures and prepared_for_ad < to_add:
                        done, _ = wait(futures, return_when=FIRST_COMPLETED)
                        for future in done:
                            futures.pop(future, None)
                            rec = future.result()
                            if not rec:
                                failed_prepare_for_ad += 1
                                failed_parallel += 1
                            elif accept_pending_record(
                                rec,
                                crowd_package=crowd_package,
                                display_excluded_material_ids=display_excluded_material_ids,
                                crowd_landing_counts=crowd_landing_counts,
                                landing_counts_for_ad=landing_counts_for_ad,
                                landing_max_uses=landing_max_uses,
                            ):
                                prepared_for_ad += 1
                                accepted_parallel += 1
                            else:
                                dropped_parallel += 1
                            if prepared_for_ad < to_add:
                                submit_next()
                            else:
                                break
                    if prepared_for_ad >= to_add:
                        logger.info(
                            "[phase1]   adgroup=%d 并行准备 target reached accepted=%d",
                            adgroup_id, accepted_parallel,
                        )
                        cancelled = 0
                        for pending_future in futures:
                            if pending_future.cancel():
                                cancelled += 1
                        if cancelled:
                            logger.info(
                                "[phase1]   adgroup=%d 并行准备 cancelled_pending=%d",
                                adgroup_id, cancelled,
                            )
                    for future in futures:
                        if future.cancelled():
                            continue
                        rec = future.result()
                        if rec:
                            dropped_parallel += 1
                            logger.info(
                                "[phase1]   adgroup=%d 并行结果丢弃:target reached material=%s landing=%s",
                                adgroup_id, rec.get("_material_id"), rec.get("landing_video_id"),
                            )
                logger.info(
                    "[phase1]   adgroup=%d 并行准备 done submitted=%d accepted=%d dropped=%d failed=%d elapsed=%.1fs",
                    adgroup_id, submitted, accepted_parallel, dropped_parallel,
                    failed_parallel, time.monotonic() - started_at,
                )
            else:
                for attempt_no in range(1, remaining_to_add + 1):
                    rec = prepare_attempt(
                        attempt_no=attempt_no,
                        excluded_material_ids_snapshot=set(display_excluded_material_ids),
                        excluded_landing_ids_snapshot=current_excluded_landing_ids(),
                    )

                    if rec:
                        if accept_pending_record(
                            rec,
                            crowd_package=crowd_package,
                            display_excluded_material_ids=display_excluded_material_ids,
                            crowd_landing_counts=crowd_landing_counts,
                            landing_counts_for_ad=landing_counts_for_ad,
                            landing_max_uses=landing_max_uses,
                        ):
                            prepared_for_ad += 1
                    else:
                        failed_prepare_for_ad += 1
                        # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break
                        # 同广告剩余 to_add 创意还能继续试,不被一次失败拖累
                        logger.info(
                            "[phase1]   adgroup=%d 本条创意 prepare 失败,试下一条",
                            adgroup_id,
                        )
            logger.info(
                "[phase1]   adgroup=%d 补创意完成 target=%d have_before=%d "
                "planned=%d prepared=%d failed_prepare=%d",
                adgroup_id, target_creatives, already_have, to_add,
                prepared_for_ad, failed_prepare_for_ad,
            )
            if prepared_for_ad < to_add:
                logger.warning(
                    "[phase1]   adgroup=%d 未补满:缺口=%d。详细原因见上方 "
                    "prepare_one_creative source_summary/失败日志",
                    adgroup_id, to_add - prepared_for_ad,
                )

    logger.info("=" * 60)
    logger.info("[phase1] 准备完成,共 %d 条 pending records", len(pending_records))
    return pending_records


def _write_pending_records(records: list[dict], output_dir: Path) -> Path:
    """落 Phase 1 产物 JSON(供 Phase 3 独立 apply 用 / 追溯)。"""
    output_dir.mkdir(parents=True, exist_ok=True)
    now = now_in_timezone()
    date_str = now.strftime("%Y%m%d")
    ts = now.strftime("%H%M%S")
    out_path = output_dir / f"creation_pending_{date_str}_{ts}.json"
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(records, f, ensure_ascii=False, indent=2)
    return out_path


def phase2_approval(records: list[dict], xlsx_output_dir: Path) -> dict[int, str]:
    """Phase 2:飞书审批(若需要)。

    Returns: actions {row_idx: action}, 1-based row_idx
    """
    from tools.im_approval_creation import run_approval_workflow

    sheet_meta, actions = run_approval_workflow(
        records=records,
        xlsx_output_dir=xlsx_output_dir,
        timeout_minutes=CREATION_APPROVAL_TIMEOUT_MINUTES,
    )
    logger.info(
        "[phase2] 审批完成 sheet_url=%s actions=%d/%d",
        sheet_meta.get("url"), len(actions), len(records),
    )
    return actions


def run_once(config_date: str | None = None) -> dict:
    """完整主循环:Phase 0 → Phase 1 → Phase 2(若开关 True)→ Phase 3。"""
    run_started = now_in_timezone().isoformat()
    sync_stats: dict = {}

    # Phase -1:每日主流程启动时先同步飞书「自动化账户」配置表。
    logger.info("=" * 60)
    logger.info("[main] Phase -1 启动 — 同步飞书自动化账户配置")
    try:
        from sync_feishu_account_config import sync_from_feishu

        sync_stats = sync_from_feishu(config_date=config_date)
        logger.info("[main] Phase -1 完成: %s", sync_stats)
    except Exception as e:
        logger.exception("[main] Phase -1 同步飞书配置失败,本轮停止:%s", e)
        return {
            "run_started": run_started,
            "run_finished": now_in_timezone().isoformat(),
            "approval_required": CREATION_APPROVAL_REQUIRED,
            "sync_stats": {"error": str(e)},
            "phase0_created_ads": 0,
            "total": {"phase1_prepared": 0},
        }

    effective_config_date = date.fromisoformat(sync_stats["config_date"])

    # Phase 0:模块 A 建广告(满足每账户 ADS_PER_ACCOUNT 条)
    logger.info("=" * 60)
    if _env_flag("CREATION_SKIP_PHASE0"):
        logger.info("[main] CREATION_SKIP_PHASE0=True → 跳过模块 A 建广告")
        created_ads = []
    else:
        logger.info("[main] Phase 0 启动 — 模块 A 检查 + 建广告")
        created_ads = phase0_create_ads(config_date=effective_config_date)
        logger.info("[main] Phase 0 完成:本轮新建广告 %d 条", len(created_ads))
        _wait_created_ads_visible(created_ads)

    # Phase 1:模块 B 给所有广告(新+旧)补创意
    logger.info("=" * 60)
    logger.info("[main] Phase 1 启动 — 模块 B 给广告补创意")
    pending_records = phase1_prepare(config_date=effective_config_date)
    if not pending_records:
        logger.info("[main] Phase 1 无 pending records,主循环退出")
        return {
            "run_started": run_started,
            "run_finished": now_in_timezone().isoformat(),
            "approval_required": CREATION_APPROVAL_REQUIRED,
            "sync_stats": sync_stats,
            "phase0_created_ads": len(created_ads),
            "total": {"phase1_prepared": 0},
        }

    data_dir = _HERE / "outputs" / "data"
    pending_path = _write_pending_records(pending_records, data_dir)
    logger.info("[main] pending records 已写入: %s", pending_path)

    # Phase 2:审批(或 skip)
    if CREATION_APPROVAL_REQUIRED:
        logger.info("[main] CREATION_APPROVAL_REQUIRED=True → 进 Phase 2 飞书审批")
        actions = phase2_approval(pending_records, data_dir)
        for i, rec in enumerate(pending_records, start=1):
            rec["action"] = actions.get(i, "skip")
    else:
        logger.info("[main] CREATION_APPROVAL_REQUIRED=False → 全 records approve")
        for rec in pending_records:
            rec["action"] = "approve"

    # Phase 3:执行
    logger.info("=" * 60)
    logger.info("[main] Phase 3 执行启动")
    summary = apply_pending_records(pending_records)
    summary["approval_required"] = CREATION_APPROVAL_REQUIRED
    summary["pending_records_path"] = str(pending_path)
    summary_path = write_apply_summary(summary, data_dir)
    logger.info("[main] summary 已写入: %s", summary_path)

    # 发执行汇报
    _send_apply_summary_to_feishu(summary)

    summary["run_started"] = run_started
    summary["run_finished"] = now_in_timezone().isoformat()
    summary["sync_stats"] = sync_stats
    summary["phase0_created_ads"] = len(created_ads)
    summary["phase0_ads"] = created_ads
    return summary


def main() -> int:
    parser = argparse.ArgumentParser(description="执行广告与创意创建主流程")
    parser.add_argument(
        "--config-date",
        help="执行指定飞书配置日期,格式 YYYYMMDD 或 YYYY-MM-DD;默认当天",
    )
    args = parser.parse_args()

    _setup_logging()
    logger.info("=" * 60)
    logger.info("模块 B 创意搭建子系统 — 主循环启动")
    logger.info("TARGET_CREATIVES_PER_AD       = %d", TARGET_CREATIVES_PER_AD)
    logger.info("CREATION_APPROVAL_REQUIRED    = %s", CREATION_APPROVAL_REQUIRED)
    logger.info("WHITELIST_ACCOUNTS            = %s", WHITELIST_ACCOUNTS)
    logger.info("=" * 60)

    summary = run_once(config_date=args.config_date)
    t = summary.get("total") or {}
    logger.info("=" * 60)
    logger.info("[main] 主循环结束")
    if "phase1_prepared" in t:
        logger.info("  phase1_prepared = %d", t["phase1_prepared"])
    else:
        logger.info("  records         = %d", t.get("records", 0))
        logger.info("  approved        = %d", t.get("approved", 0))
        logger.info("  posted_ok       = %d", t.get("posted_ok", 0))
        logger.info("  posted_failed   = %d", t.get("posted_failed", 0))
    logger.info("=" * 60)
    return 0 if t.get("posted_failed", 0) == 0 else 1


if __name__ == "__main__":
    sys.exit(main())