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 json
import logging
import os
import subprocess
import sys
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
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.feishu import RoiFeishuPublisher  # 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:
    mode = "内部15:30测试" if "--internal-test" in (extra_args or []) else "正式09:00任务"
    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 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_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 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 main() -> None:
    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: 内部")
    initialize_schema()
    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:
        scheduler.add_job(
            run_daily_roi,
            CronTrigger(
                hour=roi_config.report_hour,
                minute=roi_config.report_minute,
                timezone="Asia/Shanghai",
            ),
            id="daily_roi",
            name="日级ROI计算与逐行审批",
            max_instances=1,
            coalesce=True,
            misfire_grace_time=int(
                os.getenv("DAILY_ROI_MISFIRE_GRACE_SECONDS", "3600")
            ),
        )
    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(
            run_daily_roi,
            id="daily_roi_startup",
            name="启动时ROI计算",
        )
    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()