Agent/examples/auto_put_ad_mini/run_daily_service.py

#!/usr/bin/env python
"""Schedule daily creation and two-hour creative review scans."""

from __future__ import annotations

import argparse
import json
import logging
import os
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
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 config import TIME_SERIES_DEFAULT  # noqa: E402

# config import 会调整 sys.path；恢复本服务目录优先级，避免 im-client/tools.py
# 遮蔽 auto_put_ad_mini/tools 包。
while str(HERE) in sys.path:
    sys.path.remove(str(HERE))
sys.path.insert(0, str(HERE))

from db.connection import get_connection  # noqa: E402
from storage import advisory_lock, initialize_schema  # noqa: E402
from roi_control.config import AgencyWebhookConfig, RoiConfig  # noqa: E402
from roi_control.data_source import (  # noqa: E402
    SHANGHAI,
    SourceDataNotReadyError,
    validate_source_ready,
)
from roi_control.feishu import RoiFeishuPublisher  # noqa: E402
from roi_control.odps_client import ODPSClient  # noqa: E402
from roi_control.repository import load_formal_run_for_end_date  # noqa: E402
from tools.delivery_config import validate_region_dictionary  # noqa: E402


logger = logging.getLogger("auto_put_ad_mini.daily_service")


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


def _notify_roi_failure(
    *,
    extra_args: list[str] | None,
    error: str,
) -> None:
    if "--internal-test" in (extra_args or []):
        mode = "内部15:30测试"
    else:
        report_hour = int(os.getenv("DAILY_ROI_HOUR", "11"))
        report_minute = int(os.getenv("DAILY_ROI_MINUTE", "0"))
        mode = f"正式{report_hour:02d}:{report_minute:02d}起轮询任务"
    publisher: RoiFeishuPublisher | None = None
    try:
        publisher = RoiFeishuPublisher(require_chat_ids=False)
        publisher.send_service_alert(
            title="日级ROI任务失败",
            content=(
                f"任务：**{mode}**\n"
                f"错误：`{error[:500]}`\n"
                "本次已停止，不会生成或发送不完整报表。"
            ),
        )
    except Exception as exc:
        logger.error("Failed to send ROI failure alert: %s", exc)
    finally:
        if publisher is not None:
            publisher.close()


def _notify_roi_source_cutoff(
    *,
    target_date: str,
    cutoff_time: str,
    error: str,
) -> None:
    publisher: RoiFeishuPublisher | None = None
    try:
        publisher = RoiFeishuPublisher(require_chat_ids=False)
        publisher.send_service_alert(
            title="日级ROI数据截止仍未就绪",
            content=(
                f"目标数据日期：**{target_date}**\n"
                f"截止时间：**{cutoff_time}**\n"
                f"原因：`{error[:500]}`\n"
                "当天已停止自动轮询，不会使用旧分区，也不会生成或补发ROI报表。"
            ),
        )
    except Exception as exc:
        logger.error("Failed to send ROI source cutoff alert: %s", exc)
    finally:
        if publisher is not None:
            publisher.close()


def sync_enabled_delivery_templates() -> int:
    """Keep enabled DB templates aligned with the code's global delivery window."""
    serialized = json.dumps(TIME_SERIES_DEFAULT)
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            return cursor.execute(
                """
                UPDATE ad_delivery_template
                SET time_series_json=%s,
                    updated_by='ad_daily_service'
                WHERE enabled=TRUE
                  AND time_series_json<>%s
                """,
                (serialized, serialized),
            )
    finally:
        connection.close()


def _run_script(
    script_name: str,
    lock_name: str,
    extra_args: list[str] | None = None,
) -> None:
    with advisory_lock(lock_name) as acquired:
        if not acquired:
            logger.warning("Skip %s: another instance holds %s", script_name, lock_name)
            return
        logger.info("Starting %s", script_name)
        try:
            completed = subprocess.run(
                [sys.executable, str(HERE / script_name), *(extra_args or [])],
                cwd=HERE,
                check=False,
            )
        except Exception as exc:
            if script_name == "run_daily_roi.py":
                _notify_roi_failure(extra_args=extra_args, error=str(exc))
            raise
        if completed.returncode:
            if script_name == "run_daily_roi.py":
                _notify_roi_failure(
                    extra_args=extra_args,
                    error=f"process exited with code {completed.returncode}",
                )
            raise RuntimeError(
                f"{script_name} exited with code {completed.returncode}"
            )
        logger.info("Finished %s", script_name)


def run_creation() -> None:
    _run_script(
        "execute_creation_once.py",
        os.getenv("DAILY_CREATION_LOCK_NAME", "ad_daily_creation"),
    )


def run_creation_once(
    *,
    account_ids: list[int],
    config_date: str | None = None,
) -> None:
    """Run one locked creation pass for explicitly selected eligible accounts."""
    extra_args: list[str] = []
    if config_date:
        extra_args.extend(["--config-date", config_date])
    for account_id in account_ids:
        extra_args.extend(["--account-id", str(account_id)])
    _run_script(
        "execute_creation_once.py",
        os.getenv("DAILY_CREATION_LOCK_NAME", "ad_daily_creation"),
        extra_args,
    )


def run_creative_review() -> None:
    _run_script(
        "scan_creative_reviews.py",
        os.getenv("DAILY_REVIEW_LOCK_NAME", "ad_creative_review_scan"),
    )


def run_daily_roi() -> None:
    _run_script(
        "run_daily_roi.py",
        os.getenv("DAILY_ROI_LOCK_NAME", "ad_daily_roi"),
        ["--send-feishu"],
    )


def _effective_shanghai_now(now: datetime | None = None) -> datetime:
    current = now or datetime.now(SHANGHAI)
    if current.tzinfo is None:
        current = current.replace(tzinfo=SHANGHAI)
    return current.astimezone(SHANGHAI)


def _daily_roi_times(
    config: RoiConfig,
    now: datetime,
) -> tuple[datetime, datetime]:
    start = now.replace(
        hour=config.report_hour,
        minute=config.report_minute,
        second=0,
        microsecond=0,
    )
    cutoff = now.replace(
        hour=config.daily_poll_cutoff_hour,
        minute=config.daily_poll_cutoff_minute,
        second=0,
        microsecond=0,
    )
    return start, cutoff


def _roi_poll_minutes(config: RoiConfig) -> str:
    values = {
        (config.report_minute + offset) % 60
        for offset in range(0, 60, config.daily_poll_interval_minutes)
    }
    return ",".join(str(value) for value in sorted(values))


def _attempt_daily_roi_when_ready(
    config: RoiConfig,
    *,
    now: datetime,
    at_cutoff: bool,
) -> str:
    target = (now - timedelta(days=1)).date()
    existing = load_formal_run_for_end_date(target)
    if existing:
        logger.info(
            "Skip daily ROI target=%s: formal run already exists run=%s status=%s",
            target.strftime("%Y%m%d"),
            existing.get("run_id"),
            existing.get("status"),
        )
        return "ALREADY_RAN"

    target_date = target.strftime("%Y%m%d")
    client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
    try:
        counts = validate_source_ready(client, target_date)
    except SourceDataNotReadyError as exc:
        if at_cutoff:
            logger.error(
                "Daily ROI source missed cutoff target=%s: %s",
                target_date,
                exc,
            )
            _notify_roi_source_cutoff(
                target_date=target_date,
                cutoff_time=(
                    f"{config.daily_poll_cutoff_hour:02d}:"
                    f"{config.daily_poll_cutoff_minute:02d}"
                ),
                error=str(exc),
            )
            return "CUTOFF_NOT_READY"
        logger.info("Daily ROI source not ready target=%s: %s", target_date, exc)
        return "WAITING_SOURCE"

    logger.info("Daily ROI source ready target=%s counts=%s", target_date, counts)
    run_daily_roi()
    return "TRIGGERED"


def poll_daily_roi(
    config: RoiConfig,
    *,
    now: datetime | None = None,
) -> str:
    current = _effective_shanghai_now(now)
    start, cutoff = _daily_roi_times(config, current)
    if current < start or current >= cutoff:
        return "OUTSIDE_WINDOW"
    return _attempt_daily_roi_when_ready(config, now=current, at_cutoff=False)


def finalize_daily_roi_at_cutoff(
    config: RoiConfig,
    *,
    now: datetime | None = None,
) -> str:
    current = _effective_shanghai_now(now)
    return _attempt_daily_roi_when_ready(config, now=current, at_cutoff=True)


def run_internal_roi_test() -> None:
    _run_script(
        "run_daily_roi.py",
        os.getenv("DAILY_ROI_INTERNAL_TEST_LOCK_NAME", "ad_daily_roi"),
        [
            "--internal-test",
            "--output-dir",
            str(HERE / "outputs" / "roi_control" / "internal_test"),
        ],
    )


def _positive_account_id(raw: str) -> int:
    try:
        account_id = int(raw)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(f"账户 ID 必须是正整数: {raw!r}") from exc
    if account_id <= 0:
        raise argparse.ArgumentTypeError(f"账户 ID 必须是正整数: {raw!r}")
    return account_id


def main(argv: list[str] | None = None) -> None:
    parser = argparse.ArgumentParser(description="广告日级生产调度服务")
    parser.add_argument(
        "--run-creation-once",
        action="store_true",
        help="使用生产数据库锁执行一次广告/创意创建后退出",
    )
    parser.add_argument(
        "--account-id",
        action="append",
        type=_positive_account_id,
        dest="account_ids",
        help="一次性创建限定账户;必须与 --run-creation-once 一起使用,可重复",
    )
    parser.add_argument(
        "--config-date",
        help="历史复现时限定飞书配置日期；默认每账户使用日期最新的一行",
    )
    args = parser.parse_args(argv)
    if args.run_creation_once and not args.account_ids:
        parser.error("--run-creation-once 至少需要一个 --account-id")
    if not args.run_creation_once and (args.account_ids or args.config_date):
        parser.error("--account-id/--config-date 只能与 --run-creation-once 一起使用")

    roi_config = RoiConfig.from_env()
    if roi_config.internal_test_enabled:
        agency_config = AgencyWebhookConfig.from_env()
        if not (agency_config.webhooks or {}).get("内部"):
            raise ValueError("ROI internal test requires webhook route: 内部")
    region_path = validate_region_dictionary()
    logger.info("Validated Tencent region dictionary: %s", region_path)
    initialize_schema()
    if args.run_creation_once:
        run_creation_once(
            account_ids=args.account_ids,
            config_date=args.config_date,
        )
        return
    if _env_flag("DAILY_SYNC_DELIVERY_TEMPLATE", False):
        changed = sync_enabled_delivery_templates()
        logger.info("Synchronized %d enabled delivery template(s)", changed)

    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
    if _env_flag("DAILY_CREATION_ENABLED", False):
        creation_hour = int(os.getenv("DAILY_CREATION_HOUR", "10"))
        creation_minute = int(os.getenv("DAILY_CREATION_MINUTE", "30"))
        scheduler.add_job(
            run_creation,
            CronTrigger(
                hour=creation_hour,
                minute=creation_minute,
                timezone="Asia/Shanghai",
            ),
            id="daily_creation",
            name="广告与创意创建",
            max_instances=1,
            coalesce=True,
            misfire_grace_time=int(
                os.getenv("DAILY_CREATION_MISFIRE_GRACE_SECONDS", "3600")
            ),
        )
    if _env_flag("DAILY_REVIEW_ENABLED", False):
        scheduler.add_job(
            run_creative_review,
            IntervalTrigger(
                hours=int(os.getenv("DAILY_REVIEW_INTERVAL_HOURS", "2")),
                timezone="Asia/Shanghai",
            ),
            id="creative_review_scan",
            name="腾讯创意审核扫描",
            max_instances=1,
            coalesce=True,
            misfire_grace_time=1800,
        )
    if roi_config.daily_enabled:
        logger.info(
            "Daily ROI polling configured start=%02d:%02d interval=%dm cutoff=%02d:%02d",
            roi_config.report_hour,
            roi_config.report_minute,
            roi_config.daily_poll_interval_minutes,
            roi_config.daily_poll_cutoff_hour,
            roi_config.daily_poll_cutoff_minute,
        )
        scheduler.add_job(
            poll_daily_roi,
            CronTrigger(
                minute=_roi_poll_minutes(roi_config),
                timezone="Asia/Shanghai",
            ),
            id="daily_roi",
            name="日级ROI数据就绪轮询",
            args=[roi_config],
            max_instances=1,
            coalesce=True,
            misfire_grace_time=int(
                os.getenv("DAILY_ROI_MISFIRE_GRACE_SECONDS", "3600")
            ),
        )
        scheduler.add_job(
            finalize_daily_roi_at_cutoff,
            CronTrigger(
                hour=roi_config.daily_poll_cutoff_hour,
                minute=roi_config.daily_poll_cutoff_minute,
                timezone="Asia/Shanghai",
            ),
            id="daily_roi_cutoff",
            name="日级ROI数据就绪截止检查",
            args=[roi_config],
            max_instances=1,
            coalesce=True,
            misfire_grace_time=300,
        )
    if roi_config.internal_test_enabled:
        scheduler.add_job(
            run_internal_roi_test,
            CronTrigger(
                hour=roi_config.internal_test_hour,
                minute=roi_config.internal_test_minute,
                timezone="Asia/Shanghai",
            ),
            id="daily_roi_internal_test",
            name="日级ROI内部端到端测试",
            max_instances=1,
            coalesce=True,
            misfire_grace_time=int(
                os.getenv("ROI_INTERNAL_TEST_MISFIRE_GRACE_SECONDS", "3600")
            ),
        )
    if _env_flag("DAILY_RUN_ON_STARTUP", False):
        scheduler.add_job(
            run_creation,
            id="daily_creation_startup",
            name="启动时创建检查",
        )
    if _env_flag("DAILY_REVIEW_RUN_ON_STARTUP", False):
        scheduler.add_job(
            run_creative_review,
            id="creative_review_startup",
            name="启动时审核扫描",
        )
    if roi_config.daily_enabled and _env_flag("DAILY_ROI_RUN_ON_STARTUP", False):
        scheduler.add_job(
            poll_daily_roi,
            id="daily_roi_startup",
            name="启动时ROI数据就绪检查",
            args=[roi_config],
        )
    logger.info(
        "Daily service started jobs=%s",
        [job.id for job in scheduler.get_jobs()],
    )
    scheduler.start()


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