Agent/examples/tencent_realtime_control/feishu_command_service.py

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

from __future__ import annotations

import logging
import os
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from typing import Any
from zoneinfo import ZoneInfo

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

from command_intent_parser import CommandIntentParser
from operator_commands import (
    ACTION_CANCEL,
    ACTION_CONFIRM,
    ACTION_DAY_PAUSE,
    ACTION_REJECT,
    ACTION_RESUME,
    ACTION_STATUS,
    ACTION_STOP,
    ACTION_TODAY_SPEND,
    normalize_text,
    parse_command,
)
from operator_control import (
    cancel_command,
    execute_confirmed_command,
    pause_status_summary,
    preview_write_command,
)
from realtime_config import RealtimeControlConfig
from storage import (
    close_operator_draft,
    list_pending_operator_commands,
    load_active_operator_draft,
    load_operator_command_by_source,
    save_operator_draft,
)
from today_spend_query import query_today_spend


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.delivery_start_hour = self.config.next_delivery_hour
        self.confirmation_ttl_minutes = int(
            os.getenv("RTC_COMMAND_CONFIRM_TTL_MINUTES", "10")
        )
        self.draft_ttl_minutes = int(os.getenv("RTC_COMMAND_DRAFT_TTL_MINUTES", "5"))
        if self.confirmation_ttl_minutes < 1 or self.draft_ttl_minutes < 1:
            raise ValueError("command TTL values must be positive")
        self.intent_parser = CommandIntentParser()
        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",
        )
        self._conversation_locks: dict[tuple[str, str], threading.Lock] = {}
        self._conversation_locks_guard = threading.Lock()

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

    def _enqueue_message(self, event: FeishuMessageEvent) -> None:
        if not self._authorized(event):
            return
        self.executor.submit(self._handle_serialized, event)

    def _handle_serialized(self, event: FeishuMessageEvent) -> None:
        key = (event.chat_id, event.sender_open_id)
        with self._conversation_locks_guard:
            lock = self._conversation_locks.setdefault(key, threading.Lock())
        with lock:
            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:
        return (
            event.content_type in {"text", "post"}
            and event.chat_type == ChatType.GROUP
            and event.chat_id == self.allowed_chat_id
            and event.sender_open_id in self.allowed_open_ids
            and event.mentioned_bot
        )

    def _reply_pending_commands(self, event: FeishuMessageEvent, now: datetime) -> None:
        commands = list_pending_operator_commands(
            event.chat_id, event.sender_open_id, now
        )
        if not commands:
            self._reply(event, "当前没有待确认命令。")
            return
        lines = ["请指定要确认的命令ID："]
        for command in commands:
            lines.append(
                f"- {command['command_id']} / {_ACTION_LABELS.get(command['action'], command['action'])} "
                f"/ {command['preview_account_count']}个账户 "
                f"/ {command['preview_ad_count']}条广告 "
                f"/ 今日消耗{int(command.get('preview_cost_fen') or 0) / 100:.2f}元"
            )
        self._reply(event, "\n".join(lines))

    def _handle_command_id_action(
        self,
        event: FeishuMessageEvent,
        now: datetime,
    ) -> bool:
        text = normalize_text(event.content)
        if text == "确认":
            self._reply_pending_commands(event, now)
            return True
        if text in {"取消", "取消对话", "取消草稿"}:
            close_operator_draft(event.chat_id, event.sender_open_id, "CANCELLED")
            self._reply(event, "当前未完成的对话已取消。")
            return True
        if not text.startswith(("确认", "取消", "拒绝")):
            return False
        parsed = parse_command(text)
        if parsed is None:
            return False
        if parsed.action == ACTION_CANCEL:
            if (parsed.command_id or "").startswith("roi_"):
                raise ValueError("ROI批次请在审批表黄色列逐行处理")
            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']} 未取消，当前状态：{command['status']}"
            )
            self._reply(event, message)
            return True
        if parsed.action == ACTION_REJECT:
            self._reply(event, "ROI批次请在审批表黄色列逐行批准或拒绝。")
            return True
        if parsed.action != ACTION_CONFIRM:
            return False
        if (parsed.command_id or "").startswith("roi_"):
            self._reply(event, "ROI批次请在审批表黄色列逐行批准或拒绝。")
            return True
        if not self.apply:
            self._reply(event, "当前服务为 dry-run，禁止执行腾讯写操作。")
            return True
        command = execute_confirmed_command(
            parsed.command_id or "",
            sender_open_id=event.sender_open_id,
            now=now,
            start_hour=self.delivery_start_hour,
            lock_name=self.config.lock_name,
        )
        account_ids = ",".join(
            str(value) for value in command.get("target_account_ids") or []
        )
        self._reply(
            event,
            f"命令 {command['command_id']} 执行完成\n"
            f"- 状态：{command['status']}\n"
            f"- 账户：{account_ids or '无'}\n"
            f"- 预览时今日消耗："
            f"{int(command.get('preview_cost_fen') or 0) / 100:.2f}元\n"
            f"- 成功：{command.get('successes', 0)} 条\n"
            f"- 跳过：{command.get('skipped', 0)} 条\n"
            f"- 失败：{command.get('failures', 0)} 条",
        )
        return True

    def _save_incomplete_intent(
        self,
        event: FeishuMessageEvent,
        intent: Any,
        draft: dict[str, Any] | None,
        now: datetime,
    ) -> None:
        raw_text = "\n".join(
            value for value in ((draft or {}).get("raw_text"), event.content) if value
        )
        source_ids = list((draft or {}).get("source_message_ids") or [])
        source_ids.append(event.message_id)
        save_operator_draft({
            "draft_id": f"draft_{uuid.uuid4().hex}",
            "chat_id": event.chat_id,
            "sender_open_id": event.sender_open_id,
            "raw_text": raw_text,
            "action": intent.action,
            "scope_type": intent.scope_type,
            "account_ids": list(intent.account_ids),
            "missing_fields": list(intent.missing_fields or ("scope",)),
            "source_message_ids": source_ids,
            "expires_at": now + timedelta(minutes=self.draft_ttl_minutes),
        })
        if "clarification" in intent.missing_fields:
            question = "我没有准确理解该操作。请明确发送：暂停今天、持续停止、恢复投放或查询暂停状态。"
        else:
            if intent.action == ACTION_TODAY_SPEND:
                scope_hint = "自动化账户、全部账户，或一个/多个账户ID"
            else:
                scope_hint = "全部纳管账户，或一个/多个账户ID"
            question = (
                f"请指定操作范围：{scope_hint}。\n"
                f"该对话将在 {self.draft_ttl_minutes} 分钟后过期。"
            )
        self._reply(event, question)

    def _reply_preview(self, event: FeishuMessageEvent, command: dict[str, Any]) -> None:
        lines = [
            f"操作预览：{_ACTION_LABELS[command['action']]}",
            f"- 数据时间：{command['previewed_at'].strftime('%Y-%m-%d %H:%M:%S')}",
        ]
        if command["action"] == ACTION_DAY_PAUSE:
            lines.append(
                "- 次日自动投放："
                f"{command['resume_at'].strftime('%Y-%m-%d %H:%M')}（腾讯投放时段）"
            )
        elif command["action"] == ACTION_STOP:
            lines.append("- 恢复方式：后续发送恢复命令")
        lines.append("- 影响账户：")
        for row in command.get("account_summaries") or []:
            lines.append(
                f"  - {row['account_id']}：{row['ad_count']}条广告，"
                f"今日消耗{int(row['cost_fen']) / 100:.2f}元"
            )
        lines.extend([
            f"- 汇总：{command['preview_account_count']}个账户，"
            f"{command['preview_ad_count']}条广告，今日消耗"
            f"{int(command.get('preview_cost_fen') or 0) / 100:.2f}元",
            f"- 命令ID：{command['command_id']}",
            f"请在 {self.confirmation_ttl_minutes} 分钟内回复：确认 {command['command_id']}",
            f"取消命令请回复：取消 {command['command_id']}",
        ])
        self._reply(event, "\n".join(lines))

    def handle_message(self, event: FeishuMessageEvent) -> None:
        if not self._authorized(event):
            return
        now = datetime.now(SHANGHAI)
        try:
            existing = load_operator_command_by_source(event.message_id)
            if existing:
                self._reply(
                    event,
                    f"该消息已处理：{existing['command_id']}，"
                    f"当前状态 {existing['status']}。",
                )
                return
            if self._handle_command_id_action(event, now):
                return
            draft = load_active_operator_draft(
                event.chat_id, event.sender_open_id, now
            )
            try:
                intent = self.intent_parser.understand(event.content, draft=draft)
            except Exception as exc:
                logger.exception("Natural-language command parsing failed")
                self._reply(
                    event,
                    "自然语言理解暂时不可用，标准命令仍可使用：\n"
                    "- 暂停全部\n- 暂停 账户ID\n- 停止 账户ID\n- 恢复 账户ID",
                )
                return
            if intent is None:
                self._reply(
                    event,
                    "未识别为投放控制命令。可使用：暂停今天、持续停止、恢复投放、查询暂停状态。",
                )
                return
            if (
                draft
                and intent.action
                and intent.action != draft.get("action")
            ):
                close_operator_draft(
                    event.chat_id,
                    event.sender_open_id,
                    "SUPERSEDED",
                )
                draft = None
            if not intent.complete:
                self._save_incomplete_intent(event, intent, draft, now)
                return
            if draft:
                close_operator_draft(event.chat_id, event.sender_open_id, "COMPLETED")
            if intent.action == ACTION_STATUS:
                summary = pause_status_summary(now)
                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 intent.action == ACTION_TODAY_SPEND:
                summary = query_today_spend(
                    intent.to_parsed_command(),
                    now=now,
                )
                amount_label = (
                    "今日总消耗"
                    if summary["complete"]
                    else "成功账户消耗合计"
                )
                status = "完整" if summary["complete"] else "部分数据"
                self._reply(
                    event,
                    f"今日投放总览（{summary['scope_label']}）\n"
                    f"- 数据日期：{summary['data_date']}\n"
                    f"- 查询时间：{summary['queried_at'].strftime('%Y-%m-%d %H:%M:%S')}\n"
                    f"- 查询状态：{status}\n"
                    f"- 账户数量：{summary['account_count']} 个\n"
                    f"- 有消耗账户：{summary['spending_account_count']} 个\n"
                    f"- {amount_label}：{summary['cost_fen'] / 100:.2f} 元\n"
                    f"- 曝光：{summary['impressions']:,}\n"
                    f"- 点击：{summary['clicks']:,}\n"
                    f"- 转化：{summary['conversions']:,}\n"
                    f"- 查询失败账户：{summary['failed_account_count']} 个",
                )
                return
            combined_text = "\n".join(
                value for value in ((draft or {}).get("raw_text"), event.content) if value
            )
            command = preview_write_command(
                intent.to_parsed_command(),
                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.delivery_start_hour,
                raw_text=combined_text,
                parse_source=intent.parse_source,
                preview_lock_name=f"{self.config.lock_name}:operator-preview",
                intent={
                    "action": intent.action,
                    "scope_type": intent.scope_type,
                    "account_ids": list(intent.account_ids),
                    "confidence": intent.confidence,
                },
            )
            self._reply_preview(event, command)
        except Exception as exc:
            logger.exception("Feishu command failed")
            self._reply(event, f"命令处理失败：{exc}")