from __future__ import annotations import asyncio import json import re from contextlib import suppress from datetime import UTC, datetime from typing import Any from aiohttp import ClientError, ClientSession import wbb from wbb.utils.dbassistant import ( append_conversation_message, claim_update, classify_handoff, dead_letter_update, due_digest_connections, get_account_settings, get_business_connection, get_conversation, get_conversation_by_chat, get_or_create_conversation, list_business_connections, load_update_offset, mark_digest_sent, mark_update_done, mark_update_failed, match_knowledge, pause_conversation_for_human, recent_conversation_messages, reserve_ai_usage, resume_conversation, runtime_status, save_update_offset, set_conversation_handoff, touch_business_connection, update_conversation_summary, update_runtime_status, upsert_business_connection, usage_metrics, utc_now, ) BUSINESS_ALLOWED_UPDATES = [ "business_connection", "business_message", "edited_business_message", "deleted_business_messages", ] BUSINESS_REPLY_WINDOW_SECONDS = 24 * 60 * 60 class TelegramBotApiError(RuntimeError): def __init__( self, message: str, *, error_code: int = 0, retry_after: int = 0, ) -> None: super().__init__(message) self.error_code = int(error_code or 0) self.retry_after = int(retry_after or 0) class AssistantProviderError(RuntimeError): pass class TelegramBusinessApi: def __init__(self, token: str, session: ClientSession) -> None: self._base_url = f"https://api.telegram.org/bot{token}" self._session = session async def call( self, method: str, payload: dict[str, Any] | None = None, ) -> Any: last_error: TelegramBotApiError | None = None for attempt in range(3): try: async with self._session.post( f"{self._base_url}/{method}", json=payload or {} ) as response: data = await response.json(content_type=None) except (ClientError, TimeoutError, ValueError) as exc: last_error = TelegramBotApiError("无法连接 Telegram Bot API。") if attempt < 2: await asyncio.sleep(2**attempt) continue raise last_error from exc if response.status == 200 and isinstance(data, dict) and data.get("ok"): return data.get("result") parameters = data.get("parameters") if isinstance(data, dict) else {} last_error = TelegramBotApiError( str(data.get("description") or "Telegram Bot API 请求失败。") if isinstance(data, dict) else "Telegram Bot API 请求失败。", error_code=int(data.get("error_code") or response.status) if isinstance(data, dict) else response.status, retry_after=int((parameters or {}).get("retry_after") or 0), ) retryable = last_error.error_code == 429 or response.status >= 500 if retryable and attempt < 2: await asyncio.sleep(last_error.retry_after or 2**attempt) continue raise last_error raise last_error or TelegramBotApiError("Telegram Bot API 请求失败。") async def get_me(self) -> dict[str, Any]: result = await self.call("getMe") return result if isinstance(result, dict) else {} async def get_webhook_info(self) -> dict[str, Any]: result = await self.call("getWebhookInfo") return result if isinstance(result, dict) else {} async def get_business_connection(self, connection_id: str) -> dict[str, Any]: result = await self.call( "getBusinessConnection", {"business_connection_id": str(connection_id)} ) return result if isinstance(result, dict) else {} async def get_updates( self, *, offset: int, poll_timeout: int = 30 ) -> list[dict[str, Any]]: result = await self.call( "getUpdates", { "offset": int(offset), "timeout": int(poll_timeout), "limit": 100, "allowed_updates": BUSINESS_ALLOWED_UPDATES, }, ) return [item for item in (result or []) if isinstance(item, dict)] async def send_message( self, chat_id: int, text: str, *, business_connection_id: str = "", reply_markup: dict[str, Any] | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { "chat_id": int(chat_id), "text": str(text)[:4096], } if business_connection_id: payload["business_connection_id"] = str(business_connection_id) if reply_markup: payload["reply_markup"] = reply_markup result = await self.call("sendMessage", payload) return result if isinstance(result, dict) else {} async def read_business_message( self, connection_id: str, chat_id: int, message_id: int, ) -> None: await self.call( "readBusinessMessage", { "business_connection_id": str(connection_id), "chat_id": int(chat_id), "message_id": int(message_id), }, ) def _strip_json_fence(value: str) -> str: text = value.strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text, flags=re.IGNORECASE) text = re.sub(r"\s*```$", "", text) return text.strip() def business_reply_window_open( message: dict[str, Any], *, now: datetime | None = None ) -> bool: try: sent_at = datetime.fromtimestamp(int(message.get("date") or 0), UTC) except (OSError, OverflowError, TypeError, ValueError): return False if sent_at.timestamp() <= 0: return False age = (now or utc_now()) - sent_at return age.total_seconds() <= BUSINESS_REPLY_WINDOW_SECONDS def parse_ai_decision(value: str, *, allowed_entry_ids: set[str]) -> dict[str, Any]: try: payload = json.loads(_strip_json_fence(value)) except (json.JSONDecodeError, TypeError) as exc: raise AssistantProviderError("模型没有返回有效的 JSON 结果。") from exc if not isinstance(payload, dict): raise AssistantProviderError("模型结果必须是 JSON 对象。") action = str(payload.get("action") or "") if action not in {"answer", "clarify", "handoff"}: raise AssistantProviderError("模型返回了不支持的接待动作。") reply = str(payload.get("reply") or "").strip() if action in {"answer", "clarify"} and not reply: raise AssistantProviderError("模型没有提供回复内容。") if len(reply) > 4000: raise AssistantProviderError("模型回复内容过长。") matched_ids = [ str(item) for item in payload.get("matched_entry_ids") or [] if str(item) in allowed_entry_ids ] if action == "answer" and not matched_ids: raise AssistantProviderError("业务回答没有引用知识条目。") return { "action": action, "reply": reply, "handoff_reason": str(payload.get("handoff_reason") or "ai_handoff")[:200], "matched_entry_ids": matched_ids, "summary": str(payload.get("summary") or "")[:4000], } class OpenAICompatibleAssistant: def __init__( self, session: ClientSession, *, base_url: str, api_key: str, model: str, timeout_seconds: int = 30, max_output_tokens: int = 600, ) -> None: self._session = session self.base_url = str(base_url).rstrip("/") self.api_key = str(api_key) self.model = str(model) self.timeout_seconds = max(5, min(int(timeout_seconds), 120)) self.max_output_tokens = max(100, min(int(max_output_tokens), 2000)) @property def configured(self) -> bool: return bool(self.base_url and self.api_key and self.model) async def decide( self, *, settings: dict[str, Any], conversation: dict[str, Any], messages: list[dict[str, Any]], knowledge: list[dict[str, Any]], customer_text: str, ) -> dict[str, Any]: if not self.configured: raise AssistantProviderError("OpenAI 兼容模型尚未配置。") knowledge_payload = [ { "entry_id": item["entry_id"], "question": item["question"], "answer": item["answer"], "tags": item.get("tags") or [], } for item in knowledge ] recent_payload = [ {"direction": item.get("direction"), "text": item.get("text", "")} for item in messages[-12:] ] system_prompt = ( f"{settings.get('system_prompt')}\n" "必须遵守：业务事实只能来自 knowledge；不得自行补全价格、承诺、退款或政策。" "仅输出一个 JSON 对象，不使用 Markdown。字段为 action、reply、handoff_reason、" "matched_entry_ids、summary。action 只能是 answer、clarify、handoff。" "answer 必须填写实际引用的 matched_entry_ids；无法可靠回答时使用 handoff。" ) request_body = { "model": self.model, "temperature": 0.2, "max_tokens": self.max_output_tokens, "messages": [ {"role": "system", "content": system_prompt}, { "role": "user", "content": json.dumps( { "language": settings.get("language"), "tone": settings.get("tone"), "existing_summary": conversation.get("summary", ""), "recent_messages": recent_payload, "knowledge": knowledge_payload, "customer_message": customer_text, }, ensure_ascii=False, ), }, ], } try: async with asyncio.timeout(self.timeout_seconds): async with self._session.post( f"{self.base_url}/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json=request_body, ) as response: payload = await response.json(content_type=None) except (ClientError, TimeoutError, ValueError) as exc: raise AssistantProviderError("OpenAI 兼容服务暂时不可用。") from exc if response.status != 200: raise AssistantProviderError(f"OpenAI 兼容服务返回 HTTP {response.status}。") try: content = payload["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError) as exc: raise AssistantProviderError("OpenAI 兼容服务响应格式无效。") from exc return parse_ai_decision( str(content), allowed_entry_ids={str(item["entry_id"]) for item in knowledge}, ) async def test_connection(self) -> dict[str, Any]: if not self.configured: raise AssistantProviderError("OpenAI 兼容模型尚未配置。") request_body = { "model": self.model, "temperature": 0, "max_tokens": 12, "messages": [{"role": "user", "content": "只回复 OK"}], } try: async with asyncio.timeout(self.timeout_seconds): async with self._session.post( f"{self.base_url}/chat/completions", headers={ "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", }, json=request_body, ) as response: payload = await response.json(content_type=None) except (ClientError, TimeoutError, ValueError) as exc: raise AssistantProviderError("无法连接 OpenAI 兼容服务。") from exc if response.status != 200: raise AssistantProviderError(f"模型测试失败：HTTP {response.status}。") return {"ok": True, "model": self.model, "response_id": payload.get("id", "")} def provider_from_wbb(session: ClientSession) -> OpenAICompatibleAssistant: return OpenAICompatibleAssistant( session, base_url=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_BASE_URL", "")), api_key=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_API_KEY", "")), model=str(getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_MODEL", "")), timeout_seconds=int( getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_TIMEOUT_SECONDS", 30) ), max_output_tokens=int( getattr(wbb, "BUSINESS_ASSISTANT_OPENAI_MAX_OUTPUT_TOKENS", 600) ), ) class BusinessAssistantRuntime: def __init__( self, *, token: str, session: ClientSession, provider: OpenAICompatibleAssistant | None = None, ) -> None: self.api = TelegramBusinessApi(token, session) self.provider = provider or provider_from_wbb(session) self._poll_task: asyncio.Task[None] | None = None self._digest_task: asyncio.Task[None] | None = None self._stopping = asyncio.Event() async def start(self) -> dict[str, Any]: if self._poll_task and not self._poll_task.done(): return await runtime_status() self._stopping.clear() await update_runtime_status( { "polling_state": "starting", "model_configured": self.provider.configured, "last_error": "", } ) try: me = await self.api.get_me() webhook = await self.api.get_webhook_info() except TelegramBotApiError as exc: return await update_runtime_status( {"polling_state": "error", "last_error": str(exc)} ) webhook_url = str(webhook.get("url") or "") supported = bool(me.get("can_connect_to_business")) await update_runtime_status( { "business_mode_supported": supported, "webhook_conflict": bool(webhook_url), "webhook_url": webhook_url, "bot_username": str(me.get("username") or ""), } ) if webhook_url: return await update_runtime_status( { "polling_state": "blocked", "last_error": "检测到 Telegram webhook，Business 长轮询未启动。", } ) if not supported: return await update_runtime_status( { "polling_state": "blocked", "last_error": "请先在 BotFather 为机器人开启 Business Mode。", } ) await self._refresh_connections() self._poll_task = asyncio.create_task( self._poll_loop(), name="business-assistant-poller" ) self._digest_task = asyncio.create_task( self._digest_loop(), name="business-assistant-digest" ) return await update_runtime_status( {"polling_state": "running", "last_error": "", "started_at": utc_now()} ) async def _refresh_connections(self) -> None: page = 1 while True: connections, total = await list_business_connections( page=page, page_size=100 ) for connection in connections: connection_id = str(connection["connection_id"]) try: payload = await self.api.get_business_connection(connection_id) await upsert_business_connection(payload) except TelegramBotApiError as exc: await touch_business_connection( connection_id, error=str(exc), is_enabled=False if exc.error_code in {400, 403} else None, ) if not connections or page * 100 >= total: return page += 1 async def stop(self) -> None: self._stopping.set() tasks = [task for task in (self._poll_task, self._digest_task) if task] for task in tasks: task.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) self._poll_task = None self._digest_task = None await update_runtime_status( {"polling_state": "stopped", "stopped_at": utc_now()} ) async def _poll_loop(self) -> None: offset = await load_update_offset() backoff = 1 while not self._stopping.is_set(): try: updates = await self.api.get_updates(offset=offset, poll_timeout=30) backoff = 1 await update_runtime_status( {"polling_state": "running", "last_poll_at": utc_now(), "last_error": ""} ) for update in updates: update_id = int(update.get("update_id") or 0) if update_id <= 0: continue await self.process_update(update) offset = max(offset, update_id + 1) await save_update_offset(offset) except asyncio.CancelledError: raise except TelegramBotApiError as exc: wait_for = exc.retry_after or backoff await update_runtime_status( {"polling_state": "retrying", "last_error": str(exc)} ) await asyncio.sleep(min(max(wait_for, 1), 60)) backoff = min(backoff * 2, 60) except Exception as exc: await update_runtime_status( {"polling_state": "retrying", "last_error": str(exc)[:1000]} ) await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) async def process_update(self, update: dict[str, Any]) -> None: update_id = int(update.get("update_id") or 0) if not await claim_update(update_id): return last_error = "" for attempt in range(1, 4): try: if isinstance(update.get("business_connection"), dict): await upsert_business_connection(update["business_connection"]) elif isinstance(update.get("business_message"), dict): await self._handle_business_message(update["business_message"]) elif isinstance(update.get("edited_business_message"), dict): await self._handle_edited_message(update["edited_business_message"]) elif isinstance(update.get("deleted_business_messages"), dict): await self._handle_deleted_messages(update["deleted_business_messages"]) await mark_update_done(update_id) return except TelegramBotApiError as exc: last_error = str(exc) if exc.retry_after: await asyncio.sleep(min(exc.retry_after, 60)) except Exception as exc: last_error = str(exc) if attempt < 3: await asyncio.sleep(attempt) await mark_update_failed(update_id, last_error) await dead_letter_update(update, last_error or "unknown update error") async def _resolve_connection(self, connection_id: str) -> dict[str, Any]: connection = await get_business_connection(connection_id) if connection: return connection payload = await self.api.get_business_connection(connection_id) return await upsert_business_connection(payload) async def _handle_business_message(self, message: dict[str, Any]) -> None: connection_id = str(message.get("business_connection_id") or "") if not connection_id or message.get("is_from_offline"): return if message.get("sender_business_bot"): return connection = await self._resolve_connection(connection_id) await touch_business_connection(connection_id) chat = message.get("chat") if isinstance(message.get("chat"), dict) else {} sender = message.get("from") if isinstance(message.get("from"), dict) else {} chat_id = int(chat.get("id") or 0) message_id = int(message.get("message_id") or 0) if not chat_id or not message_id: return owner_id = int((connection.get("user") or {}).get("id") or 0) owner_reply = int(sender.get("id") or 0) == owner_id conversation = await get_or_create_conversation( connection_id, chat_id, customer=None if owner_reply else sender, ) if owner_reply: settings = await get_account_settings(connection_id) await append_conversation_message( conversation["conversation_id"], direction="human", telegram_message_id=message_id, text=str(message.get("text") or message.get("caption") or ""), sender_id=owner_id, ) await pause_conversation_for_human( conversation["conversation_id"], hours=int(settings.get("human_pause_hours") or 24), ) return await self._handle_customer_message( connection=connection, conversation=conversation, message=message, sender=sender, ) async def _handle_customer_message( self, *, connection: dict[str, Any], conversation: dict[str, Any], message: dict[str, Any], sender: dict[str, Any], ) -> None: connection_id = str(connection["connection_id"]) chat_id = int((message.get("chat") or {}).get("id") or 0) message_id = int(message.get("message_id") or 0) text = str(message.get("text") or "").strip() settings = await get_account_settings(connection_id) await append_conversation_message( conversation["conversation_id"], direction="incoming", telegram_message_id=message_id, text=text or "[非文本消息]", sender_id=int(sender.get("id") or 0), metadata={"content_type": "text" if text else "unsupported"}, ) if not connection.get("is_enabled") or not settings.get("assistant_enabled"): return current = await get_conversation(conversation["conversation_id"]) or conversation if current.get("status") == "human_paused": paused_until = current.get("paused_until") if isinstance(paused_until, datetime) and ( paused_until.replace(tzinfo=UTC) if paused_until.tzinfo is None else paused_until ) <= utc_now(): current = await resume_conversation(current["conversation_id"]) else: return if current.get("status") in {"handoff", "closed"}: return if not business_reply_window_open(message): await self._handoff( connection, current, settings, "reply_window_expired", send_customer_notice=False, ) return rights = connection.get("rights") or {} if not rights.get("can_reply"): await self._handoff( connection, current, settings, "can_reply_missing", send_customer_notice=False ) return if rights.get("can_read_messages"): with suppress(TelegramBotApiError): await self.api.read_business_message(connection_id, chat_id, message_id) if not text: await self._handoff( connection, current, settings, "unsupported_message", customer_notice=str(settings.get("unsupported_message")), ) return classification = classify_handoff(text) if classification: await self._handoff(connection, current, settings, classification) return knowledge = await match_knowledge(connection_id, text) if not knowledge: await self._handoff(connection, current, settings, "knowledge_not_found") return allowed, quota_reason = await reserve_ai_usage( connection_id, chat_id, settings ) if not allowed: await self._handoff(connection, current, settings, quota_reason) return messages = await recent_conversation_messages(current["conversation_id"], limit=12) try: decision = await self.provider.decide( settings=settings, conversation=current, messages=messages, knowledge=knowledge, customer_text=text, ) except AssistantProviderError as exc: await update_runtime_status( {"provider_last_error": str(exc)[:1000], "provider_last_error_at": utc_now()} ) await self._handoff( connection, current, settings, "provider_error", ) return if decision["summary"]: await update_conversation_summary( current["conversation_id"], decision["summary"] ) if decision["action"] == "handoff": await self._handoff( connection, current, settings, decision["handoff_reason"], summary=decision["summary"], ) return response = await self.api.send_message( chat_id, decision["reply"], business_connection_id=connection_id, ) await append_conversation_message( current["conversation_id"], direction="assistant", telegram_message_id=int(response.get("message_id") or 0), text=decision["reply"], sender_id=int((response.get("sender_business_bot") or {}).get("id") or 0), metadata={ "action": decision["action"], "matched_entry_ids": decision["matched_entry_ids"], }, ) async def _handoff( self, connection: dict[str, Any], conversation: dict[str, Any], settings: dict[str, Any], reason: str, *, summary: str = "", customer_notice: str = "", send_customer_notice: bool = True, ) -> None: updated = await set_conversation_handoff( conversation["conversation_id"], reason, summary=summary ) notice = customer_notice or str(settings.get("handoff_message") or "") if send_customer_notice and notice and connection.get("rights", {}).get("can_reply"): try: response = await self.api.send_message( int(updated["chat_id"]), notice, business_connection_id=str(connection["connection_id"]), ) await append_conversation_message( updated["conversation_id"], direction="assistant", telegram_message_id=int(response.get("message_id") or 0), text=notice, metadata={"action": "handoff", "reason": reason}, ) except TelegramBotApiError as exc: await update_runtime_status( { "customer_notice_last_error": str(exc)[:1000], "customer_notice_last_error_at": utc_now(), } ) await self.notify_handoff(connection, updated, settings, reason) async def notify_handoff( self, connection: dict[str, Any], conversation: dict[str, Any], settings: dict[str, Any], reason: str, ) -> None: customer = conversation.get("customer") or {} display = " ".join( item for item in (customer.get("first_name"), customer.get("last_name")) if item ) or (f"@{customer.get('username')}" if customer.get("username") else str(conversation["chat_id"])) latest = await recent_conversation_messages(conversation["conversation_id"], limit=1) latest_text = str(latest[-1].get("text") or "")[:1000] if latest else "" text = ( "需要人工接待\n" f"账号：{(connection.get('user') or {}).get('first_name') or connection['connection_id']}\n" f"客户：{display}\n" f"原因：{reason}\n" f"原消息：chat_id={conversation['chat_id']} / " f"message_id={latest[-1].get('telegram_message_id') if latest else '未知'}\n" f"摘要：{conversation.get('summary') or latest_text or '暂无'}" ) markup = { "inline_keyboard": [ [ { "text": "恢复自动回复", "callback_data": f"ba:resume:{conversation['conversation_id']}", } ] ] } destinations: list[int] = [] destination = str(settings.get("notification_destination") or "owner") if destination in {"owner", "both"} and int(connection.get("user_chat_id") or 0): destinations.append(int(connection["user_chat_id"])) if destination in {"ops", "both"} and int(settings.get("ops_group_id") or 0): destinations.append(int(settings["ops_group_id"])) for chat_id in dict.fromkeys(destinations): try: await self.api.send_message(chat_id, text, reply_markup=markup) except TelegramBotApiError as exc: await update_runtime_status( { "notification_last_error": str(exc)[:1000], "notification_last_error_at": utc_now(), } ) async def _handle_edited_message(self, message: dict[str, Any]) -> None: connection_id = str(message.get("business_connection_id") or "") chat_id = int((message.get("chat") or {}).get("id") or 0) if not connection_id or not chat_id: return await touch_business_connection(connection_id) conversation = await get_conversation_by_chat(connection_id, chat_id) if not conversation: return await append_conversation_message( conversation["conversation_id"], direction="incoming", telegram_message_id=-int(message.get("message_id") or 0), text=str(message.get("text") or "[消息已编辑]"), sender_id=int((message.get("from") or {}).get("id") or 0), metadata={"edited_message_id": int(message.get("message_id") or 0)}, ) async def _handle_deleted_messages(self, payload: dict[str, Any]) -> None: connection_id = str(payload.get("business_connection_id") or "") chat_id = int((payload.get("chat") or {}).get("id") or 0) if connection_id: await touch_business_connection(connection_id) conversation = await get_conversation_by_chat(connection_id, chat_id) if not conversation: return await append_conversation_message( conversation["conversation_id"], direction="incoming", telegram_message_id=-abs(int((payload.get("message_ids") or [0])[0] or 0)) - 1, text="[消息已删除]", metadata={"deleted_message_ids": payload.get("message_ids") or []}, ) async def _digest_loop(self) -> None: while not self._stopping.is_set(): try: for item in await due_digest_connections(): connection = item["connection"] settings = item["settings"] metrics = await usage_metrics( connection_id=str(connection["connection_id"]), day=str(item["day"]), ) text = ( f"智能接待运营简报 · {item['day']}\n" f"客户数：{metrics['customers']}\n" f"AI 调用：{metrics['ai_calls']}\n" f"待人工：{metrics['handoffs']}\n" f"人工暂停：{metrics['human_paused']}" ) destinations: list[int] = [] destination = str(settings.get("notification_destination") or "owner") if destination in {"owner", "both"} and connection.get("user_chat_id"): destinations.append(int(connection["user_chat_id"])) if destination in {"ops", "both"} and settings.get("ops_group_id"): destinations.append(int(settings["ops_group_id"])) sent = False for chat_id in dict.fromkeys(destinations): try: await self.api.send_message(chat_id, text) sent = True except TelegramBotApiError as exc: await update_runtime_status( { "digest_last_error": str(exc)[:1000], "digest_last_error_at": utc_now(), } ) if sent: await mark_digest_sent(connection["connection_id"], item["day"]) except asyncio.CancelledError: raise except Exception as exc: await update_runtime_status({"digest_last_error": str(exc)[:1000]}) await asyncio.sleep(60) async def preview_answer(self, connection_id: str, text: str) -> dict[str, Any]: settings = await get_account_settings(connection_id) knowledge = await match_knowledge(connection_id, text) classification = classify_handoff(text) if classification or not knowledge: return { "action": "handoff", "reason": classification or "knowledge_not_found", "matched_entries": knowledge, } conversation = {"summary": ""} decision = await self.provider.decide( settings=settings, conversation=conversation, messages=[], knowledge=knowledge, customer_text=text, ) return {**decision, "matched_entries": knowledge} async def runtime_overview() -> dict[str, Any]: status = await runtime_status() status["usage"] = await usage_metrics() return status