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 json
import logging
import sys
import time
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,
    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
    find_ads_needing_creatives,
    load_excluded_ad_ids_from_adjustment,
    prepare_one_creative_for_ad,
)
from tools.creative_material_usage import record_prepared_material_usage  # noqa: E402
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")


def _setup_logging() -> None:
    """配置日志格式与级别，挂载 SLS 上报（配置缺失时自动降级）。"""
    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) -> 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()
    if not creation_accounts:
        logger.error("[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,
                "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) -> list[dict]:
    """Phase 1:扫账户 → 关联点过滤 → 准备 pending records(不 POST 腾讯)。

    Returns:
        pending_records — 每个元素是 prepare_one_creative_for_ad 返回的 dict
    """
    creation_accounts = get_creation_account_ids()
    if not creation_accounts:
        logger.error("[phase1] 待投放账户配置为空,Phase 1 退出")
        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 不进历史排重库,只在同一广告本轮候选内做限频。

    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(),
        )

        for ad in ads_after_filter:
            adgroup_id = ad["adgroup_id"]
            already_have = ad["creative_count"]
            to_add = max(0, target_creatives - already_have)
            landing_counts_for_ad: dict[int, int] = {}
            logger.info(
                "[phase1]   adgroup=%d(have=%d need=%d)",
                adgroup_id, already_have, to_add,
            )

            for _ in range(to_add):
                landing_excluded_for_ad = {
                    vid
                    for vid, count in landing_counts_for_ad.items()
                    if count >= MAX_SAME_LANDING_PER_AD_IN_RUN
                }
                try:
                    rec = prepare_one_creative_for_ad(
                        account_id, adgroup_id,
                        excluded_material_ids=display_excluded_material_ids,
                        excluded_landing_ids=landing_excluded_for_ad,
                    )
                except Exception as e:
                    logger.exception(
                        "[phase1] adgroup=%d 准备失败:%s", adgroup_id, e,
                    )
                    rec = None

                if rec:
                    pending_records.append(rec)
                    material_id = rec.get("_material_id")
                    if material_id:
                        display_excluded_material_ids.add(str(material_id))
                        logger.info(
                            "[phase1] 本轮展示去重登记 crowd=%r material=%s size=%d",
                            crowd_package, material_id,
                            len(display_excluded_material_ids),
                        )
                    landing_video_id = rec.get("landing_video_id")
                    if landing_video_id is not None:
                        landing_video_id = int(landing_video_id)
                        landing_counts_for_ad[landing_video_id] = (
                            landing_counts_for_ad.get(landing_video_id, 0) + 1
                        )
                        logger.info(
                            "[phase1] 本轮同广告 landing 计数 adgroup=%d landing=%d count=%d limit=%d",
                            adgroup_id, landing_video_id,
                            landing_counts_for_ad[landing_video_id],
                            MAX_SAME_LANDING_PER_AD_IN_RUN,
                        )
                    try:
                        record_prepared_material_usage(rec)
                    except Exception as e:
                        logger.warning(
                            "[phase1] material usage 记录失败 account=%d adgroup=%d material=%s:%s",
                            account_id, adgroup_id, rec.get("_material_id"), e,
                        )
                else:
                    # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break
                    # 同广告剩余 to_add 创意还能继续试,不被一次失败拖累
                    logger.info(
                        "[phase1]   adgroup=%d 本条创意 prepare 失败,试下一条",
                        adgroup_id,
                    )

    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() -> 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()
        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},
        }

    # Phase 0:模块 A 建广告(满足每账户 ADS_PER_ACCOUNT 条)
    logger.info("=" * 60)
    logger.info("[main] Phase 0 启动 — 模块 A 检查 + 建广告")
    created_ads = phase0_create_ads()
    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()
    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:
    """模块 B 创意搭建子系统主入口：初始化日志 → 运行主循环 → 输出汇总。"""
    _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()
    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())