"""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}")