Agent/examples/auto_put_ad_mini/scan_creative_reviews.py

"""Scan Tencent official dynamic creative review results.

Run every 2 hours in production. This scanner only reads submitted creatives
from DB and writes official review results/rejection facts back to DB.
"""

import json
import logging
import os
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")

while str(_HERE) in sys.path:
    sys.path.remove(str(_HERE))
sys.path.insert(0, str(_HERE))

from tools.creative_review import scan_pending_reviews  # noqa: E402

logger = logging.getLogger("scan_creative_reviews")


def _setup_logging() -> None:
    """初始化日志配置：设置格式、降噪第三方 logger，并尝试挂载 SLS 上报。"""
    from logging_setup import setup_logging
    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))


def main() -> int:
    """扫描待审核创意的腾讯官方审核结果并回写 DB，返回退出码（0=成功，1=有错误）。"""
    _setup_logging()
    lookback_hours = int(os.getenv("CREATIVE_REVIEW_LOOKBACK_HOURS", "72"))
    limit = int(os.getenv("CREATIVE_REVIEW_SCAN_LIMIT", "1000"))
    batch_size = int(os.getenv("CREATIVE_REVIEW_BATCH_SIZE", "100"))

    logger.info(
        "event=creative_review stage=scan result=started lookback_hours=%d "
        "limit=%d batch_size=%d",
        lookback_hours, limit, batch_size,
    )
    started = time.monotonic()
    try:
        summary = scan_pending_reviews(
            lookback_hours=lookback_hours,
            limit=limit,
            batch_size=batch_size,
        )
    except Exception:
        logger.exception(
            "event=creative_review stage=scan result=failed duration_ms=%d",
            int((time.monotonic() - started) * 1000),
        )
        raise
    logger.info(
        "event=creative_review stage=scan result=%s duration_ms=%d summary=%s",
        "failed" if summary.get("errors") else "succeeded",
        int((time.monotonic() - started) * 1000),
        json.dumps(summary, ensure_ascii=False),
    )
    return 0 if not summary.get("errors") else 1


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