Agent/examples/tencent_realtime_control/run_control_service.py

#!/usr/bin/env python
"""Run Feishu controls and the ten-minute real-time control loop."""

from __future__ import annotations

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


ROOT = Path(__file__).resolve().parents[2]
AUTO_DIR = ROOT / "examples" / "auto_put_ad_mini"
for path in (ROOT, AUTO_DIR):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

from run_once import load_environment  # noqa: E402
from run_scheduler import run_forever  # noqa: E402
from storage import initialize_schema  # noqa: E402
from roi_control.config import RoiConfig  # noqa: E402


logger = logging.getLogger("tencent_realtime_control.service")


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


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--apply",
        action="store_true",
        help="Enable confirmed Feishu commands and real Tencent updates.",
    )
    return parser.parse_args()


def main() -> None:
    load_environment()
    args = parse_args()
    apply = args.apply or _env_flag("RTC_APPLY_ENABLED")
    roi_config = RoiConfig.from_env()
    initialize_schema()
    if os.getenv("RTC_COMMAND_ENABLED", "1").strip().lower() in {
        "1",
        "true",
        "yes",
    }:
        from feishu_command_service import FeishuCommandService

        FeishuCommandService(apply=apply).start()
        logger.info("Feishu command WebSocket started")
    else:
        logger.warning("Feishu command WebSocket is disabled")
    if roi_config.sheet_approval_enabled:
        from roi_control.sheet_approval import RoiSheetApprovalService

        RoiSheetApprovalService().start()
        logger.info("ROI sheet approval polling started")
    else:
        logger.warning("ROI sheet approval polling is disabled")
    run_forever(apply=apply)


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