Agent/examples/tencent_realtime_control/feishu_command_service.py

"""Feishu WebSocket entry for deterministic advertising control commands."""

from __future__ import annotations

import logging
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

from agent.tools.builtin.feishu.feishu_client import (
    ChatType,
    FeishuClient,
    FeishuMessageEvent,
)

from operator_commands import (
    ACTION_CANCEL,
    ACTION_CONFIRM,
    ACTION_DAY_PAUSE,
    ACTION_RESUME,
    ACTION_STATUS,
    ACTION_STOP,
    parse_command,
)
from operator_control import (
    cancel_command,
    execute_confirmed_command,
    pause_status_summary,
    preview_write_command,
)
from realtime_config import RealtimeControlConfig


SHANGHAI = ZoneInfo("Asia/Shanghai")
logger = logging.getLogger("tencent_realtime_control.feishu_commands")

_ACTION_LABELS = {
    ACTION_DAY_PAUSE: "仅暂停今天",
    ACTION_STOP: "持续停止",
    ACTION_RESUME: "恢复投放",
}


def _split_ids(raw: str) -> set[str]:
    normalized = str(raw or "").replace("，", ",").replace(" ", ",")
    return {value.strip() for value in normalized.split(",") if value.strip()}


class FeishuCommandService:
    def __init__(self, *, apply: bool) -> None:
        app_id = os.getenv("FEISHU_APP_ID", "").strip()
        app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
        if not app_id or not app_secret:
            raise RuntimeError("FEISHU_APP_ID and FEISHU_APP_SECRET are required")
        self.allowed_chat_id = (
            os.getenv("RTC_COMMAND_CHAT_ID", "").strip()
            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
        )
        self.allowed_open_ids = _split_ids(
            os.getenv("RTC_COMMAND_ALLOWED_OPEN_IDS", "")
            or os.getenv("FEISHU_OPERATOR_OPEN_ID", "")
        )
        if not self.allowed_chat_id:
            raise RuntimeError(
                "RTC_COMMAND_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID is required"
            )
        if not self.allowed_open_ids:
            raise RuntimeError(
                "RTC_COMMAND_ALLOWED_OPEN_IDS/FEISHU_OPERATOR_OPEN_ID is required"
            )
        self.apply = apply
        self.config = RealtimeControlConfig.from_env()
        self.confirmation_ttl_minutes = int(
            os.getenv("RTC_COMMAND_CONFIRM_TTL_MINUTES", "10")
        )
        if self.confirmation_ttl_minutes < 1:
            raise ValueError("RTC_COMMAND_CONFIRM_TTL_MINUTES must be positive")
        self.client = FeishuClient(app_id=app_id, app_secret=app_secret)
        self.executor = ThreadPoolExecutor(
            max_workers=int(os.getenv("RTC_COMMAND_WORKERS", "2")),
            thread_name_prefix="feishu-ad-command",
        )

    def start(self) -> Any:
        return self.client.start_websocket(
            on_message=self._enqueue_message,
            blocking=False,
        )

    def _enqueue_message(self, event: FeishuMessageEvent) -> None:
        self.executor.submit(self.handle_message, event)

    def _reply(self, event: FeishuMessageEvent, text: str) -> None:
        self.client.send_message(
            to=event.chat_id,
            text=text,
            reply_to_message_id=event.message_id,
        )

    def _authorized(self, event: FeishuMessageEvent) -> bool:
        if event.sender_open_id not in self.allowed_open_ids:
            return False
        if event.chat_type == ChatType.GROUP:
            return (
                event.chat_id == self.allowed_chat_id
                and event.mentioned_bot
            )
        return event.chat_type == ChatType.P2P

    def handle_message(self, event: FeishuMessageEvent) -> None:
        if event.content_type not in {"text", "post"}:
            return
        if not self._authorized(event):
            return
        try:
            parsed = parse_command(event.content)
            if parsed is None:
                return
            now = datetime.now(SHANGHAI)
            if parsed.action == ACTION_STATUS:
                summary = pause_status_summary()
                accounts = ",".join(str(value) for value in summary["accounts"]) or "无"
                self._reply(
                    event,
                    "当前运营暂停状态\n"
                    f"- 暂停广告：{summary['total']} 条\n"
                    f"- 仅暂停今天：{summary['until_next_delivery']} 条\n"
                    f"- 持续停止：{summary['until_manual']} 条\n"
                    f"- 涉及账户：{accounts}",
                )
                return
            if parsed.action == ACTION_CANCEL:
                command = cancel_command(
                    parsed.command_id or "",
                    event.sender_open_id,
                    now,
                )
                message = (
                    f"命令 {command['command_id']} 已取消"
                    if command["status"] == "CANCELLED"
                    else (
                        f"命令 {command['command_id']} 未取消，"
                        f"当前状态：{command['status']}"
                    )
                )
                self._reply(
                    event,
                    message,
                )
                return
            if parsed.action == ACTION_CONFIRM:
                if not self.apply:
                    self._reply(event, "当前服务为 dry-run，禁止执行腾讯写操作。")
                    return
                command = execute_confirmed_command(
                    parsed.command_id or "",
                    sender_open_id=event.sender_open_id,
                    now=now,
                    start_hour=self.config.start_hour,
                    lock_name=self.config.lock_name,
                )
                self._reply(
                    event,
                    f"命令 {command['command_id']} 执行完成\n"
                    f"- 状态：{command['status']}\n"
                    f"- 成功：{command.get('successes', 0)} 条\n"
                    f"- 失败：{command.get('failures', 0)} 条",
                )
                return

            command = preview_write_command(
                parsed,
                now=now,
                source_message_id=event.message_id,
                chat_id=event.chat_id,
                sender_open_id=event.sender_open_id,
                sender_name=event.sender_name,
                confirmation_ttl_minutes=self.confirmation_ttl_minutes,
                start_hour=self.config.start_hour,
            )
            resume_text = (
                command["resume_at"].strftime("%Y-%m-%d %H:%M")
                if command.get("resume_at")
                else "仅人工明确恢复"
            )
            self._reply(
                event,
                f"操作预览：{_ACTION_LABELS[command['action']]}\n"
                f"- 账户：{command['preview_account_count']} 个\n"
                f"- 预计涉及广告：{command['preview_ad_count']} 条\n"
                f"- 恢复时间：{resume_text}\n"
                f"- 命令ID：{command['command_id']}\n"
                f"请在 {self.confirmation_ttl_minutes} 分钟内回复："
                f"确认 {command['command_id']}\n"
                f"取消命令请回复：取消 {command['command_id']}",
            )
        except Exception as exc:
            logger.exception("Feishu command failed")
            self._reply(event, f"命令处理失败：{exc}")