from __future__ import annotations import asyncio import base64 import hashlib import hmac import json import re from contextlib import suppress from datetime import UTC, datetime from typing import Any from aiohttp import ClientError, ClientSession, ClientTimeout import wbb from wbb.utils.dbassistant import ( append_conversation_message, claim_feishu_message_notification, claim_update, classify_handoff, dead_letter_update, due_digest_connections, get_account_settings, get_business_connection, get_conversation, get_conversation_by_chat, get_knowledge_candidate, get_knowledge_source, get_or_create_conversation, get_source_event, get_source_event_by_key, invalidate_source_event_candidates, list_business_connections, list_business_sources, list_enabled_business_sources, load_update_offset, mark_digest_sent, mark_feishu_message_notification, mark_source_event_deleted, mark_update_done, mark_update_failed, match_knowledge, normalize_account_settings, pause_conversation_for_human, recent_conversation_messages, record_source_event, replace_event_candidates, reserve_ai_usage, resume_conversation, runtime_status, save_update_offset, set_conversation_handoff, touch_business_connection, update_conversation_summary, update_runtime_status, update_source_event_extraction, 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 FeishuWebhookError(RuntimeError): pass class FeishuWebhookClient: def __init__(self, session: ClientSession) -> None: self._session = session @staticmethod def _signature(secret: str, timestamp: int) -> str: sign_key = f"{timestamp}\n{secret}".encode() digest = hmac.new(sign_key, digestmod=hashlib.sha256).digest() return base64.b64encode(digest).decode() async def send(self, webhook_url: str, text: str, *, signing_secret: str = "") -> None: payload: dict[str, Any] = { "msg_type": "text", "content": {"text": str(text)[:4000]}, } await self._send_payload( webhook_url, payload, signing_secret=signing_secret ) async def send_post( self, webhook_url: str, *, title: str, content: list[list[dict[str, str]]], signing_secret: str = "", ) -> None: payload: dict[str, Any] = { "msg_type": "post", "content": { "post": { "zh_cn": { "title": str(title)[:100], "content": content, } } }, } await self._send_payload( webhook_url, payload, signing_secret=signing_secret ) async def _send_payload( self, webhook_url: str, payload: dict[str, Any], *, signing_secret: str = "", ) -> None: if signing_secret: timestamp = int(utc_now().timestamp()) payload.update( { "timestamp": str(timestamp), "sign": self._signature(signing_secret, timestamp), } ) last_error: FeishuWebhookError | None = None for attempt in range(3): try: async with self._session.post( webhook_url, json=payload, timeout=ClientTimeout(total=8), ) as response: data = await response.json(content_type=None) except (ClientError, TimeoutError, ValueError) as exc: last_error = FeishuWebhookError("无法连接飞书群机器人 Webhook。") if attempt < 2: await asyncio.sleep(2**attempt) continue raise last_error from exc raw_code = ( data.get("code", data.get("StatusCode", -1)) if isinstance(data, dict) else -1 ) try: code = int(raw_code) except (TypeError, ValueError): code = -1 if response.status == 200 and code == 0: return last_error = FeishuWebhookError( f"飞书群机器人 Webhook 请求失败（HTTP {response.status}，code={code}）。" ) if (response.status == 429 or response.status >= 500) and attempt < 2: retry_after = response.headers.get("Retry-After", "") try: wait_seconds = max(1, min(int(retry_after), 30)) except (TypeError, ValueError): wait_seconds = 2**attempt await asyncio.sleep(wait_seconds) continue raise last_error raise last_error or FeishuWebhookError("飞书群机器人 Webhook 请求失败。") 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], } def _short_string_list(value: Any, *, items: int, length: int) -> list[str]: raw_items = value if isinstance(value, list) else [] result: list[str] = [] for item in raw_items: text = " ".join(str(item or "").strip().split())[:length] if text and text.casefold() not in {existing.casefold() for existing in result}: result.append(text) return result[:items] def parse_knowledge_extraction(value: str) -> list[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) or not isinstance(payload.get("items"), list): raise AssistantProviderError("知识提取结果必须包含 items 数组。") result: list[dict[str, Any]] = [] for raw_item in payload["items"][:5]: if not isinstance(raw_item, dict): raise AssistantProviderError("知识候选必须是 JSON 对象。") question = " ".join(str(raw_item.get("question") or "").strip().split())[:300] answer = str(raw_item.get("answer") or "").strip()[:4000] if not question or not answer: raise AssistantProviderError("知识候选缺少标准问题或答案。") try: confidence = float(raw_item.get("confidence") or 0) except (TypeError, ValueError): confidence = 0 result.append( { "question": question, "aliases": _short_string_list( raw_item.get("aliases"), items=20, length=200 ), "keywords": _short_string_list( raw_item.get("keywords"), items=30, length=50 ), "answer": answer, "tags": _short_string_list(raw_item.get("tags"), items=20, length=30), "confidence": max(0.0, min(confidence, 1.0)), } ) return result def redact_business_knowledge_text(value: str) -> str: text = re.sub( r"(? 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 extract_knowledge( self, *, source_type: str, source_title: str, content: str, question_context: str = "", ) -> list[dict[str, Any]]: if not self.configured: raise AssistantProviderError("OpenAI 兼容模型尚未配置。") if source_type == "business": safe_content = redact_business_knowledge_text(content) safe_question = redact_business_knowledge_text(question_context) source_rule = ( "这是 Business 人工接待对话。customer_question 只能作为标准问题参考，" "答案事实只能来自 human_reply。不得把客户自述、联系方式、用户名、" "投诉、退款要求或未经确认的承诺写入知识。" ) source_payload = { "customer_question": safe_question, "human_reply": safe_content, } else: source_rule = ( "这是频道或群组内容。只提取明确、稳定、可复用的业务事实；" "忽略寒暄、广告口号、个人观点、临时活动、投诉、退款争议和未经确认的承诺。" "没有稳定事实时返回空 items。" ) source_payload = {"content": content} system_prompt = ( "你是知识库整理器，只提取输入中明确存在的事实，不得推测或补全。" f"{source_rule}" "只输出一个 JSON 对象，不使用 Markdown，格式为 " '{"items":[{"question":"","aliases":[],"keywords":[],"answer":"",' '"tags":[],"confidence":0.0}]}。items 最多 5 条。' ) request_body = { "model": self.model, "temperature": 0.1, "max_tokens": self.max_output_tokens, "messages": [ {"role": "system", "content": system_prompt}, { "role": "user", "content": json.dumps( { "source_type": source_type, "source_title": source_title, **source_payload, }, 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("知识提取模型暂时不可用。") from exc if response.status != 200: raise AssistantProviderError(f"知识提取模型返回 HTTP {response.status}。") try: result = payload["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError) as exc: raise AssistantProviderError("知识提取模型响应格式无效。") from exc return parse_knowledge_extraction(str(result)) 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 KnowledgeIngestionService: def __init__(self, provider: OpenAICompatibleAssistant) -> None: self.provider = provider async def ingest( self, source: dict[str, Any], *, event_key: str, content: str, metadata: dict[str, Any], question_context: str = "", auto_publish: bool = False, force: bool = False, ) -> list[dict[str, Any]]: event, changed = await record_source_event( source, event_key=event_key, content=content, metadata={ **metadata, "question_context": question_context, "trusted_author": bool(metadata.get("trusted_author")), }, ) if not changed and not force: return [] try: items = await self.provider.extract_knowledge( source_type=str(metadata.get("effective_source_type") or source["source_type"]), source_title=str(source.get("title") or ""), content=content, question_context=question_context, ) return await replace_event_candidates( source, event, items, auto_publish=auto_publish ) except Exception as exc: await invalidate_source_event_candidates( str(event["event_id"]), reason="extraction_failed" ) await update_source_event_extraction( str(event["event_id"]), status="failed", error=str(exc) ) raise async def regenerate_candidate(self, candidate_id: str) -> list[dict[str, Any]]: candidate = await get_knowledge_candidate(candidate_id) if not candidate: raise AssistantProviderError("未找到知识候选。") source = await get_knowledge_source(str(candidate["source_id"])) event = await get_source_event(str(candidate["event_id"])) if not source or not event or event.get("deleted"): raise AssistantProviderError("知识来源或原始事件已失效。") metadata = event.get("metadata") if isinstance(event.get("metadata"), dict) else {} auto_publish = bool( source.get("publication_mode") == "auto" and metadata.get("trusted_author") ) return await self.ingest( source, event_key=str(event["event_key"]), content=str(event.get("content") or ""), metadata=metadata, question_context=str(metadata.get("question_context") or ""), auto_publish=auto_publish, force=True, ) class BusinessAssistantRuntime: def __init__( self, *, token: str, session: ClientSession, provider: OpenAICompatibleAssistant | None = None, ) -> None: self.api = TelegramBusinessApi(token, session) self.feishu = FeishuWebhookClient(session) self.provider = provider or provider_from_wbb(session) self.ingestion = KnowledgeIngestionService(self.provider) 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) human_text = str(message.get("text") or message.get("caption") or "").strip() await append_conversation_message( conversation["conversation_id"], direction="human", telegram_message_id=message_id, text=human_text, sender_id=owner_id, ) await pause_conversation_for_human( conversation["conversation_id"], hours=int(settings.get("human_pause_hours") or 24), ) if human_text: await self._learn_from_human_reply( connection_id=connection_id, conversation_id=str(conversation["conversation_id"]), chat_id=chat_id, message_id=message_id, owner_id=owner_id, human_text=human_text, ) return await self._handle_customer_message( connection=connection, conversation=conversation, message=message, sender=sender, ) async def _learn_from_human_reply( self, *, connection_id: str, conversation_id: str, chat_id: int, message_id: int, owner_id: int, human_text: str, ) -> None: sources = await list_enabled_business_sources(connection_id) if not sources: return messages = await recent_conversation_messages(conversation_id, limit=20) customer_question = next( ( str(item.get("text") or "") for item in reversed(messages) if item.get("direction") == "incoming" and str(item.get("text") or "") != "[非文本消息]" ), "", ) if not customer_question: return for source in sources: event_key = f"business:{chat_id}:{message_id}" existing_event = await get_source_event_by_key( str(source["source_id"]), event_key ) existing_metadata = ( existing_event.get("metadata") if existing_event and isinstance(existing_event.get("metadata"), dict) else {} ) source_question = str( existing_metadata.get("question_context") or customer_question ) if any( term in f"{source_question}\n{human_text}".casefold() for term in ("承诺", "保证", "投诉", "退款", "退钱", "赔偿", "律师", "起诉") ): continue try: await self.ingestion.ingest( source, event_key=event_key, content=human_text, question_context=source_question, metadata={ "effective_source_type": "business", "chat_id": int(chat_id), "message_id": int(message_id), "author_id": int(owner_id), "trusted_author": True, }, auto_publish=source.get("publication_mode") == "auto", ) except Exception as exc: await update_runtime_status( { "knowledge_ingestion_last_error": str(exc)[:1000], "knowledge_ingestion_last_error_at": utc_now(), } ) 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"}, ) await self._notify_feishu_customer_message( connection=connection, conversation=conversation, message=message, sender=sender, settings=settings, text=text, ) 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 _notify_feishu_customer_message( self, *, connection: dict[str, Any], conversation: dict[str, Any], message: dict[str, Any], sender: dict[str, Any], settings: dict[str, Any], text: str, ) -> None: webhook_url = str(settings.get("feishu_webhook_url") or "") if not settings.get("feishu_webhook_enabled") or not webhook_url: return conversation_id = str(conversation["conversation_id"]) message_id = int(message.get("message_id") or 0) if not await claim_feishu_message_notification(conversation_id, message_id): return user_name = " ".join( str(item).strip() for item in (sender.get("first_name"), sender.get("last_name")) if str(item or "").strip() ) if not user_name: user_name = ( f"@{sender.get('username')}" if sender.get("username") else str((message.get("chat") or {}).get("id") or "未知") ) account = connection.get("user") or {} account_name = " ".join( str(item).strip() for item in (account.get("first_name"), account.get("last_name")) if str(item or "").strip() ) or (f"@{account.get('username')}" if account.get("username") else "Business 账号") message_type = "文本" if text else "非文本" preview_text = str( message.get("text") or message.get("caption") or "" ).strip() content: list[list[dict[str, str]]] = [ [{"tag": "text", "text": f"账号：{account_name}"}], [{"tag": "text", "text": "用户："}], [{"tag": "text", "text": f"类型：{message_type}"}], ] telegram_username = str(sender.get("username") or "").strip().lstrip("@") if re.fullmatch(r"[A-Za-z0-9_]{1,64}", telegram_username): content[1].append( { "tag": "a", "text": user_name, "href": f"https://t.me/{telegram_username}", } ) else: content[1].append({"tag": "text", "text": user_name}) if settings.get("feishu_message_preview_enabled"): content.append( [ { "tag": "text", "text": f"内容：{preview_text[:500] if preview_text else '[非文本消息]'}", } ] ) try: message_time = datetime.fromtimestamp( int(message.get("date") or 0), UTC ).strftime("%Y-%m-%d %H:%M:%S UTC") except (OSError, OverflowError, TypeError, ValueError): message_time = "未知" content.append([{"tag": "text", "text": f"时间：{message_time}"}]) try: await self.feishu.send_post( webhook_url, title="消息提醒", content=content, signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""), ) except FeishuWebhookError as exc: await mark_feishu_message_notification( conversation_id, message_id, status="failed", error=str(exc) ) await update_runtime_status( { "feishu_notification_last_error": str(exc)[:1000], "feishu_notification_last_error_at": utc_now(), } ) return await mark_feishu_message_notification( conversation_id, message_id, status="sent" ) await update_runtime_status( { "feishu_notification_last_error": "", "feishu_notification_last_sent_at": utc_now(), } ) async def test_feishu_webhook( self, connection_id: str, overrides: dict[str, Any] | None = None ) -> dict[str, Any]: connection = await self._resolve_connection(connection_id) current = await get_account_settings(connection_id) settings = normalize_account_settings(overrides or {}, previous=current) webhook_url = str(settings.get("feishu_webhook_url") or "") if not webhook_url: raise FeishuWebhookError("请先填写飞书群机器人 Webhook 地址。") account = connection.get("user") or {} account_name = " ".join( str(item).strip() for item in (account.get("first_name"), account.get("last_name")) if str(item or "").strip() ) or (f"@{account.get('username')}" if account.get("username") else "Business 账号") await self.feishu.send( webhook_url, "\n".join( [ "消息提醒", "飞书群机器人连接测试成功。", f"账号：{account_name}", ] ), signing_secret=str(settings.get("feishu_webhook_signing_secret") or ""), ) return {"ok": True, "account": account_name} 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 or message.get("sender_business_bot"): return await touch_business_connection(connection_id) conversation = await get_conversation_by_chat(connection_id, chat_id) if not conversation: return connection = await self._resolve_connection(connection_id) sender_id = int((message.get("from") or {}).get("id") or 0) owner_id = int((connection.get("user") or {}).get("id") or 0) message_id = int(message.get("message_id") or 0) text = str(message.get("text") or message.get("caption") or "").strip() if sender_id == owner_id: settings = await get_account_settings(connection_id) await append_conversation_message( conversation["conversation_id"], direction="human", telegram_message_id=-message_id, text=text or "[人工消息已编辑]", sender_id=owner_id, metadata={"edited_message_id": message_id}, ) await pause_conversation_for_human( conversation["conversation_id"], hours=int(settings.get("human_pause_hours") or 24), ) if text: await self._learn_from_human_reply( connection_id=connection_id, conversation_id=str(conversation["conversation_id"]), chat_id=chat_id, message_id=message_id, owner_id=owner_id, human_text=text, ) return await append_conversation_message( conversation["conversation_id"], direction="incoming", telegram_message_id=-message_id, text=text or "[消息已编辑]", sender_id=sender_id, metadata={"edited_message_id": message_id}, ) 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 for source in await list_business_sources(connection_id): for message_id in payload.get("message_ids") or []: await mark_source_event_deleted( str(source["source_id"]), f"business:{chat_id}:{int(message_id)}" ) 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