#!/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()