#!/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
import time
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.feishu import RoiFeishuPublisher  # 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:
    mode = "内部15:30测试" if "--internal-test" in (extra_args or []) else "正式11: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:
    task_args = extra_args or []
    started = time.monotonic()
    logger.info(
        "event=scheduled_task task=%s stage=lock_wait result=started lock=%s args=%s",
        script_name, lock_name, task_args,
    )
    with advisory_lock(lock_name) as acquired:
        if not acquired:
            logger.warning(
                "event=scheduled_task task=%s stage=lock_wait result=skipped "
                "reason=lock_held lock=%s",
                script_name, lock_name,
            )
            return
        logger.info(
            "event=scheduled_task task=%s stage=subprocess result=started lock=%s",
            script_name, lock_name,
        )
        try:
            completed = subprocess.run(
                [sys.executable, str(HERE / script_name), *task_args],
                cwd=HERE,
                check=False,
            )
        except Exception as exc:
            logger.exception(
                "event=scheduled_task task=%s stage=subprocess result=failed "
                "duration_ms=%d",
                script_name, int((time.monotonic() - started) * 1000),
            )
            if script_name == "run_daily_roi.py":
                _notify_roi_failure(extra_args=extra_args, error=str(exc))
            raise
        if completed.returncode:
            logger.error(
                "event=scheduled_task task=%s stage=subprocess result=failed "
                "returncode=%d duration_ms=%d",
                script_name, completed.returncode,
                int((time.monotonic() - started) * 1000),
            )
            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(
            "event=scheduled_task task=%s stage=subprocess result=succeeded "
            "returncode=0 duration_ms=%d",
            script_name, int((time.monotonic() - started) * 1000),
        )


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 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:
    from logging_setup import setup_logging
    setup_logging(level=os.getenv("LOG_LEVEL", "INFO"))
    logger.info("event=daily_service stage=startup result=started")
    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:
        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(
        "event=daily_service stage=scheduler result=started jobs=%s",
        [job.id for job in scheduler.get_jobs()],
    )
    scheduler.start()


if __name__ == "__main__":
    main()
