Agent/examples/auto_put_ad_mini/merge_data.py

"""
数据合并脚本 — auto_put_ad_mini V3

独立工具：合并创意数据与广告状态数据

用法：
  # 合并最近 30 天
  .venv/bin/python3 examples/auto_put_ad_mini/merge_data.py --days 30

  # 合并单日
  .venv/bin/python3 examples/auto_put_ad_mini/merge_data.py --bizdate 20260411

  # 强制重新合并（覆盖已存在的文件）
  .venv/bin/python3 examples/auto_put_ad_mini/merge_data.py --days 30 --force
"""

import argparse
import logging
import sys
from datetime import datetime, timedelta
from pathlib import Path

# 把项目根目录加入 path
ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "examples" / "auto_put_ad_mini"))

from tools.data_query import _merge_single_day, _parse_bizdate

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)

_MINI_DIR = Path(__file__).resolve().parent
_MERGED_DIR = _MINI_DIR / "outputs" / "merged"


def merge_single_day(bizdate: str, force: bool = False) -> bool:
    """
    合并单日数据。

    Returns:
        True: 成功合并
        False: 合并失败
    """
    biz, _ = _parse_bizdate(bizdate)

    # 检查是否已存在
    merged_csv = _MERGED_DIR / f"merged_{biz}.csv"
    if merged_csv.exists() and not force:
        logger.info("✓ %s 合并文件已存在，跳过（使用 --force 强制重新合并）", biz)
        return True

    logger.info("→ 开始合并 %s", biz)

    df = _merge_single_day(biz)
    if df is not None:
        logger.info("✓ 合并成功: %d 行, %d 列", len(df), len(df.columns))
        return True
    else:
        logger.error("✗ 合并失败（源文件缺失）")
        return False


def merge_multiple_days(days: int, end_date: str = "yesterday", force: bool = False) -> None:
    """
    合并多日数据。

    Args:
        days: 合并天数
        end_date: 结束日期（yesterday 或 YYYYMMDD）
        force: 是否强制重新合并
    """
    if end_date == "yesterday":
        end_dt = datetime.now() - timedelta(days=1)
    else:
        end_dt = datetime.strptime(end_date, "%Y%m%d")

    logger.info("=" * 60)
    logger.info("开始合并 %d 天数据（结束日期: %s）", days, end_dt.strftime("%Y%m%d"))
    logger.info("=" * 60)

    success_count = 0
    fail_count = 0
    skip_count = 0

    for i in range(days):
        date_dt = end_dt - timedelta(days=i)
        bizdate = date_dt.strftime("%Y%m%d")

        merged_csv = _MERGED_DIR / f"merged_{bizdate}.csv"

        if merged_csv.exists() and not force:
            skip_count += 1
            continue

        if merge_single_day(bizdate, force):
            success_count += 1
        else:
            fail_count += 1

    logger.info("=" * 60)
    logger.info("合并完成: 成功 %d, 失败 %d, 跳过 %d", success_count, fail_count, skip_count)
    logger.info("=" * 60)

    # 列出已有文件
    merged_files = sorted(_MERGED_DIR.glob("merged_*.csv"))

    logger.info("合并文件 (%d 个):", len(merged_files))
    for f in merged_files[-5:]:  # 只显示最近 5 个
        size_kb = f.stat().st_size / 1024
        logger.info("  %s (%.1f KB)", f.name, size_kb)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="合并创意数据与广告状态（V3）")
    parser.add_argument("--bizdate", default="", help="单日合并: YYYYMMDD")
    parser.add_argument("--days", type=int, default=0, help="多日合并: 天数")
    parser.add_argument("--end_date", default="yesterday", help="结束日期: yesterday 或 YYYYMMDD")
    parser.add_argument("--force", action="store_true", help="强制重新合并（覆盖已存在的文件）")
    args = parser.parse_args()

    if args.bizdate:
        # 单日模式
        success = merge_single_day(args.bizdate, args.force)
        sys.exit(0 if success else 1)
    elif args.days > 0:
        # 多日模式
        merge_multiple_days(args.days, args.end_date, args.force)
    else:
        # 默认：合并 30 天
        merge_multiple_days(30, "yesterday", args.force)