Agent/examples/auto_put_ad_mini/run_daily_roi.py

#!/usr/bin/env python
"""Run one versioned daily ROI computation and optional Feishu approval batch."""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from pathlib import Path

from dotenv import load_dotenv


HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
RTC_DIR = ROOT / "examples" / "tencent_realtime_control"
for path in (ROOT, HERE, RTC_DIR):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

load_dotenv(HERE / ".env", override=False)
load_dotenv(Path.cwd() / ".env", override=False)

from roi_control.service import run_daily_roi  # noqa: E402


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--end-date", help="ODPS end partition in YYYYMMDD")
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=HERE / "outputs" / "roi_control",
    )
    parser.add_argument(
        "--source-revision",
        help="Stable revision label for corrected upstream data",
    )
    parser.add_argument("--send-feishu", action="store_true")
    parser.add_argument(
        "--internal-test",
        action="store_true",
        help="Run an isolated full calculation and notify only the internal webhook",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    result = run_daily_roi(
        requested_end_date=args.end_date,
        output_dir=args.output_dir,
        send_feishu=args.send_feishu,
        source_revision=args.source_revision,
        internal_test=args.internal_test,
    )
    print(json.dumps(result, ensure_ascii=False, default=str))


if __name__ == "__main__":
    logging.basicConfig(
        level=os.getenv("LOG_LEVEL", "INFO"),
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )
    main()