from __future__ import annotations import asyncio import hashlib import json import re from datetime import UTC, datetime, timedelta from typing import Any from urllib.parse import urlsplit from uuid import uuid4 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from pymongo import ASCENDING, DESCENDING from pymongo.errors import DuplicateKeyError from wbb import BOT_PROFILE_ID, db connectionsdb = db.business_assistant_connections settingsdb = db.business_assistant_settings knowledgedb = db.business_assistant_knowledge knowledge_sourcesdb = db.business_assistant_knowledge_sources source_eventsdb = db.business_assistant_source_events knowledge_candidatesdb = db.business_assistant_knowledge_candidates conversationsdb = db.business_assistant_conversations messagesdb = db.business_assistant_messages usagedb = db.business_assistant_usage updatesdb = db.business_assistant_updates deadlettersdb = db.business_assistant_dead_letters runtimedb = db.business_assistant_runtime DEFAULT_ACCOUNT_SETTINGS: dict[str, Any] = { "assistant_enabled": False, "system_prompt": ( "你是该 Telegram 账号的智能接待秘书。回答要简洁、礼貌，只能依据提供的知识条目陈述业务事实。" ), "language": "zh-CN", "tone": "professional", "account_daily_limit": 200, "customer_daily_limit": 20, "human_pause_hours": 24, "notification_destination": "owner", "ops_group_id": 0, "timezone": "Asia/Shanghai", "digest_enabled": False, "digest_time": "09:00", "handoff_message": "这个问题需要人工确认，我已经通知负责人，请稍候。", "unsupported_message": "已收到你的消息，这类内容需要人工处理，我已经通知负责人。", "feishu_webhook_enabled": False, "feishu_webhook_url": "", "feishu_webhook_signing_secret": "", "feishu_message_preview_enabled": False, "feishu_outgoing_message_enabled": False, } SECRET_ACCOUNT_SETTINGS = { "feishu_webhook_url", "feishu_webhook_signing_secret", } VALID_NOTIFICATION_DESTINATIONS = {"owner", "ops", "both"} VALID_TONES = {"professional", "friendly", "concise"} VALID_SOURCE_TYPES = {"channel", "group", "business"} VALID_PUBLICATION_MODES = {"review", "auto"} VALID_AUTHOR_POLICIES = {"admins_or_allowlist", "admins", "allowlist", "all"} VALID_CANDIDATE_STATUSES = {"pending", "published", "rejected", "stale"} HANDOFF_TERMS = ( "人工", "真人", "客服", "负责人", "转人工", "human", "agent", ) SENSITIVE_TERMS = ( "承诺", "保证", "投诉", "退款", "退钱", "赔偿", "律师", "起诉", "支付失败", "账号被盗", "密码", "验证码", ) _index_lock = asyncio.Lock() _indexes_ready = False class AssistantDataError(ValueError): def __init__(self, code: str, message: str): super().__init__(message) self.code = code def utc_now() -> datetime: return datetime.now(UTC) def as_utc(value: datetime) -> datetime: return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) def _scope(filters: dict[str, Any] | None = None, /, **values: Any) -> dict[str, Any]: return {"bot_id": BOT_PROFILE_ID, **(filters or {}), **values} def clean_text( value: Any, *, max_length: int, required: bool = False, preserve_lines: bool = False, ) -> str: text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", str(value or "")) text = text.strip() if not preserve_lines: text = " ".join(text.split()) if required and not text: raise AssistantDataError("required_field", "必填内容不能为空。") if len(text) > max_length: raise AssistantDataError("text_too_long", f"内容不能超过 {max_length} 个字符。") return text def _bounded_int( value: Any, name: str, *, minimum: int, maximum: int, ) -> int: try: parsed = int(value) except (TypeError, ValueError) as exc: raise AssistantDataError("invalid_setting", f"{name} 必须是整数。") from exc if not minimum <= parsed <= maximum: raise AssistantDataError( "invalid_setting", f"{name} 必须在 {minimum} 到 {maximum} 之间。", ) return parsed def _normalize_feishu_webhook_url(value: Any) -> str: url = clean_text(value, max_length=1000) if not url: return "" try: parsed = urlsplit(url) port = parsed.port except ValueError as exc: raise AssistantDataError("invalid_setting", "飞书 Webhook 地址无效。") from exc allowed_hosts = {"open.feishu.cn", "open.larksuite.com"} if ( parsed.scheme != "https" or (parsed.hostname or "").casefold() not in allowed_hosts or port not in {None, 443} or parsed.username or parsed.password or parsed.query or parsed.fragment or not re.fullmatch(r"/open-apis/bot/v2/hook/[A-Za-z0-9-]{8,128}", parsed.path) ): raise AssistantDataError( "invalid_setting", "仅支持飞书或 Lark 官方 HTTPS 群机器人 Webhook 地址。", ) return url def _normalize_string_list(value: Any, *, max_items: int, max_length: int) -> list[str]: values = value if isinstance(value, (list, tuple, set)) else str(value or "").split(",") normalized: list[str] = [] seen: set[str] = set() for item in values: text = clean_text(item, max_length=max_length) key = text.casefold() if text and key not in seen: normalized.append(text) seen.add(key) if len(normalized) > max_items: raise AssistantDataError("too_many_items", f"最多允许 {max_items} 项。") return normalized def _normalize_int_list(value: Any, *, max_items: int = 100) -> list[int]: values = value if isinstance(value, (list, tuple, set)) else str(value or "").split(",") normalized: list[int] = [] for item in values: if item in {None, ""}: continue try: parsed = int(item) except (TypeError, ValueError) as exc: raise AssistantDataError("invalid_source", "白名单用户 ID 必须是整数。") from exc if parsed and parsed not in normalized: normalized.append(parsed) if len(normalized) > max_items: raise AssistantDataError("too_many_items", f"最多允许 {max_items} 项。") return normalized async def ensure_assistant_indexes() -> None: global _indexes_ready if _indexes_ready: return async with _index_lock: if _indexes_ready: return await connectionsdb.create_index( [("bot_id", ASCENDING), ("connection_id", ASCENDING)], unique=True ) await connectionsdb.create_index( [("bot_id", ASCENDING), ("updated_at", DESCENDING)] ) await settingsdb.create_index( [("bot_id", ASCENDING), ("connection_id", ASCENDING)], unique=True ) await knowledgedb.create_index( [("bot_id", ASCENDING), ("entry_id", ASCENDING)], unique=True ) await knowledge_sourcesdb.create_index( [("bot_id", ASCENDING), ("source_id", ASCENDING)], unique=True ) await knowledge_sourcesdb.create_index( [("bot_id", ASCENDING), ("source_key", ASCENDING)], unique=True ) await knowledge_sourcesdb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("enabled", ASCENDING), ("updated_at", DESCENDING), ] ) await source_eventsdb.create_index( [("bot_id", ASCENDING), ("event_id", ASCENDING)], unique=True ) await source_eventsdb.create_index( [ ("bot_id", ASCENDING), ("source_id", ASCENDING), ("event_key", ASCENDING), ], unique=True, ) await source_eventsdb.create_index("expires_at", expireAfterSeconds=0) await knowledge_candidatesdb.create_index( [("bot_id", ASCENDING), ("candidate_id", ASCENDING)], unique=True ) await knowledge_candidatesdb.create_index( [ ("bot_id", ASCENDING), ("event_id", ASCENDING), ("item_index", ASCENDING), ], unique=True, ) await knowledge_candidatesdb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("status", ASCENDING), ("updated_at", DESCENDING), ] ) await knowledgedb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("enabled", ASCENDING), ("priority", DESCENDING), ] ) await conversationsdb.create_index( [("bot_id", ASCENDING), ("conversation_id", ASCENDING)], unique=True ) await conversationsdb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("chat_id", ASCENDING), ], unique=True, ) await conversationsdb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("status", ASCENDING), ("updated_at", DESCENDING), ] ) await messagesdb.create_index( [("bot_id", ASCENDING), ("message_key", ASCENDING)], unique=True ) await messagesdb.create_index( [ ("bot_id", ASCENDING), ("conversation_id", ASCENDING), ("created_at", ASCENDING), ] ) await messagesdb.create_index("expires_at", expireAfterSeconds=0) await usagedb.create_index( [ ("bot_id", ASCENDING), ("connection_id", ASCENDING), ("chat_id", ASCENDING), ("day", ASCENDING), ], unique=True, ) await updatesdb.create_index( [("bot_id", ASCENDING), ("update_id", ASCENDING)], unique=True ) await updatesdb.create_index("expires_at", expireAfterSeconds=0) await deadlettersdb.create_index( [("bot_id", ASCENDING), ("update_id", ASCENDING)], unique=True ) await deadlettersdb.create_index( [("bot_id", ASCENDING), ("created_at", DESCENDING)] ) await deadlettersdb.create_index("expires_at", expireAfterSeconds=0) await runtimedb.create_index( [("bot_id", ASCENDING), ("runtime_id", ASCENDING)], unique=True ) _indexes_ready = True def _public_user(user: Any) -> dict[str, Any]: value = user if isinstance(user, dict) else {} return { "id": int(value.get("id") or 0), "username": clean_text(value.get("username"), max_length=64), "first_name": clean_text(value.get("first_name"), max_length=128), "last_name": clean_text(value.get("last_name"), max_length=128), } async def upsert_business_connection(payload: dict[str, Any]) -> dict[str, Any]: await ensure_assistant_indexes() connection_id = clean_text(payload.get("id"), max_length=128, required=True) now = utc_now() rights = payload.get("rights") if isinstance(payload.get("rights"), dict) else {} document = { "bot_id": BOT_PROFILE_ID, "connection_id": connection_id, "user": _public_user(payload.get("user")), "user_chat_id": int(payload.get("user_chat_id") or 0), "rights": {str(key): bool(value) for key, value in rights.items() if value is True}, "is_enabled": bool(payload.get("is_enabled")), "connected_at": datetime.fromtimestamp(int(payload.get("date") or 0), UTC) if payload.get("date") else now, "last_event_at": now, "updated_at": now, } await connectionsdb.update_one( _scope(connection_id=connection_id), {"$set": document, "$setOnInsert": {"created_at": now}}, upsert=True, ) await settingsdb.update_one( _scope(connection_id=connection_id), { "$setOnInsert": { "bot_id": BOT_PROFILE_ID, "connection_id": connection_id, **DEFAULT_ACCOUNT_SETTINGS, "created_at": now, "updated_at": now, } }, upsert=True, ) return await get_business_connection(connection_id) or document async def get_business_connection(connection_id: str) -> dict[str, Any] | None: await ensure_assistant_indexes() return await connectionsdb.find_one(_scope(connection_id=str(connection_id))) async def touch_business_connection( connection_id: str, *, error: str = "", is_enabled: bool | None = None ) -> dict[str, Any] | None: now = utc_now() values: dict[str, Any] = { "last_event_at": now, "updated_at": now, "last_error": clean_text(error, max_length=1000), } if is_enabled is not None: values["is_enabled"] = bool(is_enabled) await connectionsdb.update_one( _scope(connection_id=str(connection_id)), {"$set": values}, ) return await get_business_connection(connection_id) async def list_business_connections( *, page: int = 1, page_size: int = 20 ) -> tuple[list[dict[str, Any]], int]: await ensure_assistant_indexes() page = max(1, int(page)) page_size = max(1, min(int(page_size), 100)) filters = _scope() total = await connectionsdb.count_documents(filters) cursor = ( connectionsdb.find(filters) .sort("updated_at", DESCENDING) .skip((page - 1) * page_size) .limit(page_size) ) items = [] async for item in cursor: item["settings"] = public_account_settings( await get_account_settings(item["connection_id"]) ) items.append(item) return items, total async def get_account_settings(connection_id: str) -> dict[str, Any]: await ensure_assistant_indexes() stored = await settingsdb.find_one(_scope(connection_id=str(connection_id))) or {} return { **DEFAULT_ACCOUNT_SETTINGS, **{key: value for key, value in stored.items() if key != "_id"}, "connection_id": str(connection_id), } def public_account_settings(settings: dict[str, Any]) -> dict[str, Any]: public = { key: value for key, value in settings.items() if key not in SECRET_ACCOUNT_SETTINGS } public.update( { "feishu_webhook_url": "", "feishu_webhook_signing_secret": "", "feishu_webhook_configured": bool(settings.get("feishu_webhook_url")), "feishu_signing_secret_configured": bool( settings.get("feishu_webhook_signing_secret") ), } ) return public def normalize_account_settings( values: dict[str, Any], *, previous: dict[str, Any] | None = None ) -> dict[str, Any]: current = {**DEFAULT_ACCOUNT_SETTINGS, **(previous or {})} if "assistant_enabled" in values: current["assistant_enabled"] = bool(values.get("assistant_enabled")) if "system_prompt" in values: current["system_prompt"] = clean_text( values.get("system_prompt"), max_length=4000, required=True, preserve_lines=True ) if "language" in values: current["language"] = clean_text(values.get("language"), max_length=32, required=True) if "tone" in values: tone = str(values.get("tone") or "") if tone not in VALID_TONES: raise AssistantDataError("invalid_setting", "接待语气无效。") current["tone"] = tone if "account_daily_limit" in values: current["account_daily_limit"] = _bounded_int( values.get("account_daily_limit"), "账号每日额度", minimum=1, maximum=100000 ) if "customer_daily_limit" in values: current["customer_daily_limit"] = _bounded_int( values.get("customer_daily_limit"), "客户每日额度", minimum=1, maximum=1000 ) if "human_pause_hours" in values: current["human_pause_hours"] = _bounded_int( values.get("human_pause_hours"), "人工暂停小时数", minimum=1, maximum=168 ) if "notification_destination" in values: destination = str(values.get("notification_destination") or "") if destination not in VALID_NOTIFICATION_DESTINATIONS: raise AssistantDataError("invalid_setting", "通知目的地无效。") current["notification_destination"] = destination if "ops_group_id" in values: try: current["ops_group_id"] = int(values.get("ops_group_id") or 0) except (TypeError, ValueError) as exc: raise AssistantDataError("invalid_setting", "运营群 ID 必须是整数。") from exc if "timezone" in values: timezone_name = clean_text(values.get("timezone"), max_length=64, required=True) try: ZoneInfo(timezone_name) except ZoneInfoNotFoundError as exc: raise AssistantDataError("invalid_setting", "时区名称无效。") from exc current["timezone"] = timezone_name if "digest_enabled" in values: current["digest_enabled"] = bool(values.get("digest_enabled")) if "digest_time" in values: digest_time = str(values.get("digest_time") or "") if not re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", digest_time): raise AssistantDataError("invalid_setting", "简报时间必须是 HH:MM。") current["digest_time"] = digest_time for key, max_length in (("handoff_message", 1000), ("unsupported_message", 1000)): if key in values: current[key] = clean_text(values.get(key), max_length=max_length, required=True) if "feishu_webhook_url" in values: current["feishu_webhook_url"] = _normalize_feishu_webhook_url( values.get("feishu_webhook_url") ) if "feishu_webhook_signing_secret" in values: current["feishu_webhook_signing_secret"] = clean_text( values.get("feishu_webhook_signing_secret"), max_length=512 ) if "feishu_webhook_enabled" in values: current["feishu_webhook_enabled"] = bool(values.get("feishu_webhook_enabled")) if "feishu_message_preview_enabled" in values: current["feishu_message_preview_enabled"] = bool( values.get("feishu_message_preview_enabled") ) if "feishu_outgoing_message_enabled" in values: current["feishu_outgoing_message_enabled"] = bool( values.get("feishu_outgoing_message_enabled") ) if current["feishu_webhook_enabled"] and not current["feishu_webhook_url"]: raise AssistantDataError( "invalid_setting", "开启飞书消息提醒前，请先填写群机器人 Webhook 地址。" ) return {key: current[key] for key in DEFAULT_ACCOUNT_SETTINGS} async def update_account_settings( connection_id: str, values: dict[str, Any] ) -> dict[str, Any]: connection = await get_business_connection(connection_id) if not connection: raise AssistantDataError("connection_not_found", "未找到 Business 连接。") previous = await get_account_settings(connection_id) normalized = normalize_account_settings(values, previous=previous) now = utc_now() await settingsdb.update_one( _scope(connection_id=str(connection_id)), { "$set": {**normalized, "updated_at": now}, "$setOnInsert": { "bot_id": BOT_PROFILE_ID, "connection_id": str(connection_id), "created_at": now, }, }, upsert=True, ) return await get_account_settings(connection_id) async def create_knowledge_entry( connection_id: str, values: dict[str, Any] ) -> dict[str, Any]: if not await get_business_connection(connection_id): raise AssistantDataError("connection_not_found", "未找到 Business 连接。") now = utc_now() entry = { "bot_id": BOT_PROFILE_ID, "entry_id": uuid4().hex, "connection_id": str(connection_id), "question": clean_text(values.get("question"), max_length=300, required=True), "aliases": _normalize_string_list(values.get("aliases"), max_items=20, max_length=200), "keywords": _normalize_string_list(values.get("keywords"), max_items=30, max_length=50), "answer": clean_text( values.get("answer"), max_length=4000, required=True, preserve_lines=True ), "tags": _normalize_string_list(values.get("tags"), max_items=20, max_length=30), "priority": _bounded_int( values.get("priority", 0), "知识优先级", minimum=-1000, maximum=1000 ), "enabled": bool(values.get("enabled", True)), "source_candidate_id": clean_text( values.get("source_candidate_id"), max_length=64 ), "source_references": values.get("source_references") if isinstance(values.get("source_references"), list) else [], "created_at": now, "updated_at": now, } await knowledgedb.insert_one(entry) return entry async def update_knowledge_entry(entry_id: str, values: dict[str, Any]) -> dict[str, Any]: filters = _scope(entry_id=str(entry_id)) current = await knowledgedb.find_one(filters) if not current: raise AssistantDataError("knowledge_not_found", "未找到知识条目。") update: dict[str, Any] = {} if "question" in values: update["question"] = clean_text(values.get("question"), max_length=300, required=True) if "aliases" in values: update["aliases"] = _normalize_string_list( values.get("aliases"), max_items=20, max_length=200 ) if "keywords" in values: update["keywords"] = _normalize_string_list( values.get("keywords"), max_items=30, max_length=50 ) if "answer" in values: update["answer"] = clean_text( values.get("answer"), max_length=4000, required=True, preserve_lines=True ) if "tags" in values: update["tags"] = _normalize_string_list(values.get("tags"), max_items=20, max_length=30) if "priority" in values: update["priority"] = _bounded_int( values.get("priority"), "知识优先级", minimum=-1000, maximum=1000 ) if "enabled" in values: update["enabled"] = bool(values.get("enabled")) if "source_candidate_id" in values: update["source_candidate_id"] = clean_text( values.get("source_candidate_id"), max_length=64 ) if "source_references" in values: references = values.get("source_references") if not isinstance(references, list): raise AssistantDataError("invalid_knowledge", "知识来源引用必须是数组。") update["source_references"] = references[:20] if not update: raise AssistantDataError("unchanged", "没有可保存的知识条目字段。") update["updated_at"] = utc_now() await knowledgedb.update_one(filters, {"$set": update}) return await knowledgedb.find_one(filters) or current async def delete_knowledge_entry(entry_id: str) -> None: result = await knowledgedb.delete_one(_scope(entry_id=str(entry_id))) if not result.deleted_count: raise AssistantDataError("knowledge_not_found", "未找到知识条目。") async def list_knowledge_entries( connection_id: str, *, query: str = "", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: await ensure_assistant_indexes() filters: dict[str, Any] = _scope(connection_id=str(connection_id)) normalized_query = clean_text(query, max_length=100) if normalized_query: pattern = re.escape(normalized_query) filters["$or"] = [ {"question": {"$regex": pattern, "$options": "i"}}, {"aliases": {"$regex": pattern, "$options": "i"}}, {"keywords": {"$regex": pattern, "$options": "i"}}, {"tags": {"$regex": pattern, "$options": "i"}}, ] page = max(1, int(page)) page_size = max(1, min(int(page_size), 100)) total = await knowledgedb.count_documents(filters) cursor = ( knowledgedb.find(filters) .sort([("priority", DESCENDING), ("updated_at", DESCENDING)]) .skip((page - 1) * page_size) .limit(page_size) ) return [item async for item in cursor], total def _knowledge_source_key(connection_id: str, source_type: str, chat_id: int) -> str: return f"{connection_id}:{source_type}:{int(chat_id)}" def normalize_knowledge_source( values: dict[str, Any], *, previous: dict[str, Any] | None = None ) -> dict[str, Any]: current = {**(previous or {}), **values} source_type = str(current.get("source_type") or "") if source_type not in VALID_SOURCE_TYPES: raise AssistantDataError("invalid_source", "知识来源类型无效。") try: chat_id = int(current.get("chat_id") or 0) except (TypeError, ValueError) as exc: raise AssistantDataError("invalid_source", "来源聊天 ID 必须是整数。") from exc if source_type == "business": chat_id = 0 elif not chat_id: raise AssistantDataError("invalid_source", "频道或群组来源必须填写聊天 ID。") publication_mode = str(current.get("publication_mode") or "review") if publication_mode not in VALID_PUBLICATION_MODES: raise AssistantDataError("invalid_source", "知识发布方式无效。") author_policy = str(current.get("author_policy") or "admins_or_allowlist") if author_policy not in VALID_AUTHOR_POLICIES: raise AssistantDataError("invalid_source", "来源作者策略无效。") linked_chat_id = 0 try: linked_chat_id = int(current.get("linked_chat_id") or 0) except (TypeError, ValueError) as exc: raise AssistantDataError("invalid_source", "关联讨论群 ID 必须是整数。") from exc return { "source_type": source_type, "chat_id": chat_id, "title": clean_text( current.get("title") or ("Business 日常对话" if source_type == "business" else ""), max_length=200, required=True, ), "publication_mode": publication_mode, "author_policy": author_policy, "allowed_user_ids": _normalize_int_list(current.get("allowed_user_ids")), "include_linked_chat": bool(current.get("include_linked_chat", False)) if source_type == "channel" else False, "linked_chat_id": linked_chat_id if source_type == "channel" else 0, "linked_chat_title": clean_text( current.get("linked_chat_title"), max_length=200 ) if source_type == "channel" else "", "backfill_days": _bounded_int( current.get("backfill_days", 30), "历史回溯天数", minimum=0, maximum=365 ), "backfill_limit": _bounded_int( current.get("backfill_limit", 100), "历史回溯条数", minimum=1, maximum=500 ), "enabled": bool(current.get("enabled", True)), "access_status": clean_text(current.get("access_status"), max_length=64), } async def create_knowledge_source( connection_id: str, values: dict[str, Any] ) -> dict[str, Any]: await ensure_assistant_indexes() if not await get_business_connection(connection_id): raise AssistantDataError("connection_not_found", "未找到 Business 连接。") normalized = normalize_knowledge_source(values) now = utc_now() source = { "bot_id": BOT_PROFILE_ID, "source_id": uuid4().hex, "source_key": _knowledge_source_key( str(connection_id), normalized["source_type"], normalized["chat_id"] ), "connection_id": str(connection_id), **normalized, "sync_status": "idle", "last_error": "", "created_at": now, "updated_at": now, } try: await knowledge_sourcesdb.insert_one(source) except DuplicateKeyError as exc: raise AssistantDataError("source_exists", "该连接已配置相同的知识来源。") from exc return source async def get_knowledge_source(source_id: str) -> dict[str, Any] | None: await ensure_assistant_indexes() return await knowledge_sourcesdb.find_one(_scope(source_id=str(source_id))) async def update_knowledge_source( source_id: str, values: dict[str, Any] ) -> dict[str, Any]: current = await get_knowledge_source(source_id) if not current: raise AssistantDataError("source_not_found", "未找到知识来源。") for field in ("connection_id", "source_type", "chat_id"): if field in values and str(values[field]) != str(current[field]): raise AssistantDataError("immutable_source", "来源连接、类型和聊天 ID 不可修改。") normalized = normalize_knowledge_source(values, previous=current) await knowledge_sourcesdb.update_one( _scope(source_id=str(source_id)), {"$set": {**normalized, "updated_at": utc_now()}}, ) return await get_knowledge_source(source_id) or current async def list_knowledge_sources( connection_id: str, *, page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: await ensure_assistant_indexes() filters = _scope(connection_id=str(connection_id)) page = max(1, int(page)) page_size = max(1, min(int(page_size), 100)) total = await knowledge_sourcesdb.count_documents(filters) cursor = ( knowledge_sourcesdb.find(filters) .sort("updated_at", DESCENDING) .skip((page - 1) * page_size) .limit(page_size) ) return [item async for item in cursor], total async def list_sources_for_chat( chat_id: int, *, enabled_only: bool = True ) -> list[dict[str, Any]]: await ensure_assistant_indexes() values: dict[str, Any] = { "$or": [ {"chat_id": int(chat_id)}, {"include_linked_chat": True, "linked_chat_id": int(chat_id)}, ] } if enabled_only: values["enabled"] = True filters = _scope(values) return [item async for item in knowledge_sourcesdb.find(filters)] async def list_enabled_sources_for_chat(chat_id: int) -> list[dict[str, Any]]: return await list_sources_for_chat(chat_id, enabled_only=True) async def list_enabled_business_sources(connection_id: str) -> list[dict[str, Any]]: await ensure_assistant_indexes() return [ item async for item in knowledge_sourcesdb.find( _scope( connection_id=str(connection_id), source_type="business", enabled=True, ) ) ] async def list_business_sources(connection_id: str) -> list[dict[str, Any]]: await ensure_assistant_indexes() return [ item async for item in knowledge_sourcesdb.find( _scope(connection_id=str(connection_id), source_type="business") ) ] async def update_knowledge_source_sync( source_id: str, *, status: str, error: str = "", processed: int | None = None, ) -> dict[str, Any]: values: dict[str, Any] = { "sync_status": clean_text(status, max_length=32, required=True), "last_error": clean_text(error, max_length=1000), "updated_at": utc_now(), } if status == "running": values["last_sync_started_at"] = utc_now() if status in {"completed", "failed"}: values["last_sync_finished_at"] = utc_now() if processed is not None: values["last_sync_processed"] = max(0, int(processed)) result = await knowledge_sourcesdb.update_one( _scope(source_id=str(source_id)), {"$set": values} ) if not result.matched_count: raise AssistantDataError("source_not_found", "未找到知识来源。") return await get_knowledge_source(source_id) or {} async def touch_knowledge_source(source_id: str, *, error: str = "") -> None: await knowledge_sourcesdb.update_one( _scope(source_id=str(source_id)), { "$set": { "last_event_at": utc_now(), "last_error": clean_text(error, max_length=1000), "updated_at": utc_now(), } }, ) async def _disable_candidate_entries(filters: dict[str, Any]) -> None: entry_ids = [ str(item.get("knowledge_entry_id") or "") async for item in knowledge_candidatesdb.find(filters) if item.get("knowledge_entry_id") ] if entry_ids: await knowledgedb.update_many( _scope({"entry_id": {"$in": entry_ids}}), {"$set": {"enabled": False, "updated_at": utc_now()}}, ) async def delete_knowledge_source(source_id: str) -> None: source = await get_knowledge_source(source_id) if not source: raise AssistantDataError("source_not_found", "未找到知识来源。") candidate_filters = _scope(source_id=str(source_id)) await _disable_candidate_entries(candidate_filters) await knowledge_candidatesdb.update_many( candidate_filters, {"$set": {"status": "stale", "stale_reason": "source_deleted", "updated_at": utc_now()}}, ) await source_eventsdb.update_many( _scope(source_id=str(source_id)), {"$set": {"deleted": True, "delete_reason": "source_deleted", "updated_at": utc_now()}}, ) await knowledge_sourcesdb.delete_one(_scope(source_id=str(source_id))) def source_event_id_for(source_id: str, event_key: str) -> str: digest = hashlib.sha256( f"{BOT_PROFILE_ID}:{source_id}:{event_key}".encode() ).hexdigest() return digest[:32] async def record_source_event( source: dict[str, Any], *, event_key: str, content: str, metadata: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], bool]: await ensure_assistant_indexes() cleaned_content = clean_text( content, max_length=20000, required=True, preserve_lines=True ) cleaned_key = clean_text(event_key, max_length=300, required=True) event_metadata = metadata if isinstance(metadata, dict) else {} content_hash = hashlib.sha256( json.dumps( {"content": cleaned_content, "metadata": event_metadata}, ensure_ascii=False, sort_keys=True, default=str, ).encode() ).hexdigest() filters = _scope(source_id=str(source["source_id"]), event_key=cleaned_key) current = await source_eventsdb.find_one(filters) if current and current.get("content_hash") == content_hash and not current.get("deleted"): return current, False now = utc_now() event_id = str(current.get("event_id")) if current else source_event_id_for( str(source["source_id"]), cleaned_key ) version = int(current.get("version") or 0) + 1 if current else 1 document = { "bot_id": BOT_PROFILE_ID, "event_id": event_id, "source_id": str(source["source_id"]), "connection_id": str(source["connection_id"]), "event_key": cleaned_key, "content": cleaned_content, "content_hash": content_hash, "metadata": event_metadata, "version": version, "deleted": False, "extraction_status": "pending", "extraction_error": "", "updated_at": now, "expires_at": now + timedelta(days=30), } await source_eventsdb.update_one( filters, {"$set": document, "$setOnInsert": {"created_at": now}}, upsert=True, ) return await source_eventsdb.find_one(filters) or document, True async def get_source_event(event_id: str) -> dict[str, Any] | None: await ensure_assistant_indexes() return await source_eventsdb.find_one(_scope(event_id=str(event_id))) async def get_source_event_by_key( source_id: str, event_key: str ) -> dict[str, Any] | None: await ensure_assistant_indexes() return await source_eventsdb.find_one( _scope(source_id=str(source_id), event_key=str(event_key)) ) async def update_source_event_extraction( event_id: str, *, status: str, error: str = "", item_count: int = 0 ) -> None: await source_eventsdb.update_one( _scope(event_id=str(event_id)), { "$set": { "extraction_status": clean_text(status, max_length=32, required=True), "extraction_error": clean_text(error, max_length=1000), "extracted_item_count": max(0, int(item_count)), "extracted_at": utc_now(), "updated_at": utc_now(), } }, ) async def invalidate_source_event_candidates( event_id: str, *, reason: str ) -> None: filters = _scope(event_id=str(event_id)) await _disable_candidate_entries(filters) await knowledge_candidatesdb.update_many( filters, { "$set": { "status": "stale", "stale_reason": clean_text(reason, max_length=100), "updated_at": utc_now(), } }, ) async def mark_source_event_deleted(source_id: str, event_key: str) -> bool: event = await source_eventsdb.find_one( _scope(source_id=str(source_id), event_key=str(event_key)) ) if not event: return False await source_eventsdb.update_one( _scope(event_id=str(event["event_id"])), {"$set": {"deleted": True, "updated_at": utc_now()}}, ) await invalidate_source_event_candidates( str(event["event_id"]), reason="source_message_deleted" ) return True def _normalize_candidate_item(item: dict[str, Any]) -> dict[str, Any]: try: confidence = float(item.get("confidence") or 0) except (TypeError, ValueError): confidence = 0 return { "question": clean_text(item.get("question"), max_length=300, required=True), "aliases": _normalize_string_list(item.get("aliases"), max_items=20, max_length=200), "keywords": _normalize_string_list(item.get("keywords"), max_items=30, max_length=50), "answer": clean_text( item.get("answer"), max_length=4000, required=True, preserve_lines=True ), "tags": _normalize_string_list(item.get("tags"), max_items=20, max_length=30), "confidence": max(0.0, min(confidence, 1.0)), } async def replace_event_candidates( source: dict[str, Any], event: dict[str, Any], items: list[dict[str, Any]], *, auto_publish: bool, ) -> list[dict[str, Any]]: await invalidate_source_event_candidates( str(event["event_id"]), reason="source_message_changed" ) existing = { int(item.get("item_index") or 0): item async for item in knowledge_candidatesdb.find( _scope(event_id=str(event["event_id"])) ) } results: list[dict[str, Any]] = [] metadata = event.get("metadata") if isinstance(event.get("metadata"), dict) else {} for index, raw_item in enumerate(items[:5]): normalized = _normalize_candidate_item(raw_item) current = existing.get(index) or {} now = utc_now() candidate_id = str(current.get("candidate_id") or uuid4().hex) candidate = { "bot_id": BOT_PROFILE_ID, "candidate_id": candidate_id, "connection_id": str(source["connection_id"]), "source_id": str(source["source_id"]), "event_id": str(event["event_id"]), "item_index": index, **normalized, "status": "pending", "stale_reason": "", "knowledge_entry_id": str(current.get("knowledge_entry_id") or ""), "source_snapshot": { "title": str(source.get("title") or ""), "source_type": str(source.get("source_type") or ""), "chat_id": int(metadata.get("chat_id") or source.get("chat_id") or 0), "message_id": int(metadata.get("message_id") or 0), "author_id": int(metadata.get("author_id") or 0), }, "updated_at": now, } await knowledge_candidatesdb.update_one( _scope(event_id=str(event["event_id"]), item_index=index), {"$set": candidate, "$setOnInsert": {"created_at": now}}, upsert=True, ) if auto_publish: candidate = await publish_knowledge_candidate(candidate_id) else: candidate = await get_knowledge_candidate(candidate_id) or candidate results.append(candidate) await update_source_event_extraction( str(event["event_id"]), status="completed", item_count=len(results) ) return results async def get_knowledge_candidate(candidate_id: str) -> dict[str, Any] | None: await ensure_assistant_indexes() return await knowledge_candidatesdb.find_one( _scope(candidate_id=str(candidate_id)) ) async def publish_knowledge_candidate(candidate_id: str) -> dict[str, Any]: candidate = await get_knowledge_candidate(candidate_id) if not candidate: raise AssistantDataError("candidate_not_found", "未找到知识候选。") if candidate.get("status") == "stale": raise AssistantDataError("candidate_stale", "来源已变更或删除，不能发布该候选。") if not await get_knowledge_source(str(candidate.get("source_id") or "")): await knowledge_candidatesdb.update_one( _scope(candidate_id=str(candidate_id)), { "$set": { "status": "stale", "stale_reason": "source_deleted", "updated_at": utc_now(), } }, ) raise AssistantDataError("source_not_found", "知识来源已删除，不能发布该候选。") reference = { **(candidate.get("source_snapshot") or {}), "source_id": str(candidate.get("source_id") or ""), "event_id": str(candidate.get("event_id") or ""), } values = { key: candidate.get(key) for key in ("question", "aliases", "keywords", "answer", "tags") } values.update( { "priority": round(float(candidate.get("confidence") or 0) * 100), "enabled": True, "source_candidate_id": str(candidate_id), "source_references": [reference], } ) entry_id = str(candidate.get("knowledge_entry_id") or "") if entry_id: try: entry = await update_knowledge_entry(entry_id, values) except AssistantDataError as exc: if exc.code != "knowledge_not_found": raise entry = await create_knowledge_entry(str(candidate["connection_id"]), values) else: entry = await create_knowledge_entry(str(candidate["connection_id"]), values) await knowledge_candidatesdb.update_one( _scope(candidate_id=str(candidate_id)), { "$set": { "status": "published", "knowledge_entry_id": str(entry["entry_id"]), "published_at": utc_now(), "updated_at": utc_now(), } }, ) return await get_knowledge_candidate(candidate_id) or candidate async def reject_knowledge_candidate(candidate_id: str) -> dict[str, Any]: candidate = await get_knowledge_candidate(candidate_id) if not candidate: raise AssistantDataError("candidate_not_found", "未找到知识候选。") entry_id = str(candidate.get("knowledge_entry_id") or "") if entry_id: await knowledgedb.update_one( _scope(entry_id=entry_id), {"$set": {"enabled": False, "updated_at": utc_now()}}, ) await knowledge_candidatesdb.update_one( _scope(candidate_id=str(candidate_id)), { "$set": { "status": "rejected", "rejected_at": utc_now(), "updated_at": utc_now(), } }, ) return await get_knowledge_candidate(candidate_id) or candidate async def list_knowledge_candidates( connection_id: str, *, status: str = "", source_id: str = "", query: str = "", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: await ensure_assistant_indexes() filters: dict[str, Any] = _scope(connection_id=str(connection_id)) if status: if status not in VALID_CANDIDATE_STATUSES: raise AssistantDataError("invalid_status", "知识候选状态无效。") filters["status"] = status if source_id: filters["source_id"] = str(source_id) normalized_query = clean_text(query, max_length=100) if normalized_query: pattern = re.escape(normalized_query) filters["$or"] = [ {"question": {"$regex": pattern, "$options": "i"}}, {"answer": {"$regex": pattern, "$options": "i"}}, {"keywords": {"$regex": pattern, "$options": "i"}}, ] page = max(1, int(page)) page_size = max(1, min(int(page_size), 100)) total = await knowledge_candidatesdb.count_documents(filters) cursor = ( knowledge_candidatesdb.find(filters) .sort("updated_at", DESCENDING) .skip((page - 1) * page_size) .limit(page_size) ) return [item async for item in cursor], total def _search_text(value: str) -> str: return re.sub(r"[^\w\u3400-\u9fff]+", "", value.casefold()) async def match_knowledge( connection_id: str, query: str, *, limit: int = 8 ) -> list[dict[str, Any]]: await ensure_assistant_indexes() normalized_query = _search_text(query) if not normalized_query: return [] entries = [ item async for item in knowledgedb.find( _scope(connection_id=str(connection_id), enabled=True) ) ] scored: list[tuple[int, dict[str, Any]]] = [] for entry in entries: score = int(entry.get("priority") or 0) phrases = [entry.get("question", ""), *(entry.get("aliases") or [])] for phrase in phrases: normalized_phrase = _search_text(str(phrase)) if not normalized_phrase: continue if normalized_query == normalized_phrase: score += 1000 elif normalized_phrase in normalized_query: score += 300 + min(len(normalized_phrase), 100) elif normalized_query in normalized_phrase and len(normalized_query) >= 4: score += 120 for keyword in entry.get("keywords") or []: normalized_keyword = _search_text(str(keyword)) if normalized_keyword and normalized_keyword in normalized_query: score += 80 if score > int(entry.get("priority") or 0): scored.append((score, entry)) scored.sort(key=lambda item: (item[0], item[1].get("priority", 0)), reverse=True) return [{**entry, "match_score": score} for score, entry in scored[: max(1, limit)]] def classify_handoff(text: str) -> str | None: normalized = text.casefold() if any(term in normalized for term in HANDOFF_TERMS): return "customer_requested_human" if any(term in normalized for term in SENSITIVE_TERMS): return "sensitive_request" return None def conversation_id_for(connection_id: str, chat_id: int) -> str: digest = hashlib.sha256(f"{BOT_PROFILE_ID}:{connection_id}:{int(chat_id)}".encode()).hexdigest() return digest[:32] async def get_or_create_conversation( connection_id: str, chat_id: int, *, customer: dict[str, Any] | None = None, ) -> dict[str, Any]: await ensure_assistant_indexes() conversation_id = conversation_id_for(connection_id, chat_id) now = utc_now() update: dict[str, Any] = {"updated_at": now, "last_message_at": now} if customer: update["customer"] = _public_user(customer) await conversationsdb.update_one( _scope(conversation_id=conversation_id), { "$set": update, "$setOnInsert": { "bot_id": BOT_PROFILE_ID, "conversation_id": conversation_id, "connection_id": str(connection_id), "chat_id": int(chat_id), "status": "auto", "summary": "", "handoff_reason": "", "created_at": now, }, }, upsert=True, ) return await conversationsdb.find_one(_scope(conversation_id=conversation_id)) or {} async def get_conversation(conversation_id: str) -> dict[str, Any] | None: await ensure_assistant_indexes() return await conversationsdb.find_one( _scope(conversation_id=str(conversation_id)) ) async def get_conversation_by_chat( connection_id: str, chat_id: int ) -> dict[str, Any] | None: await ensure_assistant_indexes() return await conversationsdb.find_one( _scope(connection_id=str(connection_id), chat_id=int(chat_id)) ) async def append_conversation_message( conversation_id: str, *, direction: str, telegram_message_id: int, text: str, sender_id: int = 0, metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: await ensure_assistant_indexes() if direction not in {"incoming", "assistant", "human"}: raise AssistantDataError("invalid_direction", "会话消息方向无效。") now = utc_now() message_key = f"{conversation_id}:{direction}:{int(telegram_message_id)}" document = { "bot_id": BOT_PROFILE_ID, "message_key": message_key, "conversation_id": str(conversation_id), "direction": direction, "telegram_message_id": int(telegram_message_id), "sender_id": int(sender_id or 0), "text": clean_text(text, max_length=12000, preserve_lines=True), "metadata": metadata or {}, "created_at": now, "expires_at": now + timedelta(days=30), } try: await messagesdb.insert_one(document) except DuplicateKeyError: return await messagesdb.find_one(_scope(message_key=message_key)) or document await conversationsdb.update_one( _scope(conversation_id=str(conversation_id)), {"$set": {"last_message_at": now, "updated_at": now}}, ) return document async def claim_feishu_message_notification( conversation_id: str, telegram_message_id: int, *, direction: str = "incoming" ) -> bool: if direction not in {"incoming", "human", "assistant"}: raise AssistantDataError("invalid_direction", "会话消息方向无效。") message_key = f"{conversation_id}:{direction}:{int(telegram_message_id)}" stale_before = utc_now() - timedelta(minutes=2) result = await messagesdb.find_one_and_update( _scope( { "message_key": message_key, "$or": [ { "metadata.feishu_notification_status": { "$nin": ["sending", "sent"] } }, { "metadata.feishu_notification_status": "sending", "metadata.feishu_notification_started_at": { "$lte": stale_before }, }, ], } ), { "$set": { "metadata.feishu_notification_status": "sending", "metadata.feishu_notification_started_at": utc_now(), } }, return_document=True, ) return bool(result) async def mark_feishu_message_notification( conversation_id: str, telegram_message_id: int, *, status: str, error: str = "", direction: str = "incoming", ) -> None: if status not in {"sent", "failed"}: raise AssistantDataError("invalid_status", "飞书通知状态无效。") if direction not in {"incoming", "human", "assistant"}: raise AssistantDataError("invalid_direction", "会话消息方向无效。") message_key = f"{conversation_id}:{direction}:{int(telegram_message_id)}" await messagesdb.update_one( _scope(message_key=message_key), { "$set": { "metadata.feishu_notification_status": status, "metadata.feishu_notification_error": clean_text( error, max_length=1000 ), "metadata.feishu_notification_finished_at": utc_now(), } }, ) async def recent_conversation_messages( conversation_id: str, *, limit: int = 12 ) -> list[dict[str, Any]]: cursor = ( messagesdb.find(_scope(conversation_id=str(conversation_id))) .sort([("created_at", DESCENDING), ("_id", DESCENDING)]) .limit(max(1, min(int(limit), 50))) ) items = [item async for item in cursor] items.reverse() return items async def set_conversation_handoff( conversation_id: str, reason: str, *, summary: str = "", ) -> dict[str, Any]: now = utc_now() update: dict[str, Any] = { "status": "handoff", "handoff_reason": clean_text(reason, max_length=200, required=True), "paused_until": None, "handoff_at": now, "updated_at": now, } if summary: update["summary"] = clean_text(summary, max_length=4000, preserve_lines=True) result = await conversationsdb.find_one_and_update( _scope(conversation_id=str(conversation_id)), {"$set": update}, return_document=True, ) if not result: raise AssistantDataError("conversation_not_found", "未找到客户会话。") return result async def pause_conversation_for_human( conversation_id: str, *, hours: int ) -> dict[str, Any]: now = utc_now() result = await conversationsdb.find_one_and_update( _scope(conversation_id=str(conversation_id)), { "$set": { "status": "human_paused", "handoff_reason": "human_reply_detected", "paused_until": now + timedelta(hours=max(1, min(int(hours), 168))), "updated_at": now, } }, return_document=True, ) if not result: raise AssistantDataError("conversation_not_found", "未找到客户会话。") return result async def resume_conversation(conversation_id: str) -> dict[str, Any]: result = await conversationsdb.find_one_and_update( _scope(conversation_id=str(conversation_id)), { "$set": { "status": "auto", "handoff_reason": "", "paused_until": None, "updated_at": utc_now(), } }, return_document=True, ) if not result: raise AssistantDataError("conversation_not_found", "未找到客户会话。") return result async def pause_conversation(conversation_id: str) -> dict[str, Any]: return await set_conversation_handoff(conversation_id, "admin_paused") async def close_conversation(conversation_id: str) -> dict[str, Any]: result = await conversationsdb.find_one_and_update( _scope(conversation_id=str(conversation_id)), {"$set": {"status": "closed", "closed_at": utc_now(), "updated_at": utc_now()}}, return_document=True, ) if not result: raise AssistantDataError("conversation_not_found", "未找到客户会话。") return result async def clear_conversation(conversation_id: str) -> dict[str, Any]: current = await get_conversation(conversation_id) if not current: raise AssistantDataError("conversation_not_found", "未找到客户会话。") await messagesdb.delete_many(_scope(conversation_id=str(conversation_id))) await conversationsdb.update_one( _scope(conversation_id=str(conversation_id)), { "$set": { "summary": "", "handoff_reason": "", "status": "auto", "paused_until": None, "cleared_at": utc_now(), "updated_at": utc_now(), } }, ) return await get_conversation(conversation_id) or current async def update_conversation_summary( conversation_id: str, summary: str ) -> None: cleaned = clean_text(summary, max_length=4000, preserve_lines=True) if cleaned: await conversationsdb.update_one( _scope(conversation_id=str(conversation_id)), {"$set": {"summary": cleaned, "updated_at": utc_now()}}, ) async def list_conversations( *, connection_id: str = "", status: str = "", page: int = 1, page_size: int = 20, ) -> tuple[list[dict[str, Any]], int]: await ensure_assistant_indexes() filters: dict[str, Any] = _scope() if connection_id: filters["connection_id"] = str(connection_id) if status: if status not in {"auto", "handoff", "human_paused", "closed"}: raise AssistantDataError("invalid_status", "会话状态无效。") filters["status"] = status page = max(1, int(page)) page_size = max(1, min(int(page_size), 100)) total = await conversationsdb.count_documents(filters) cursor = ( conversationsdb.find(filters) .sort("updated_at", DESCENDING) .skip((page - 1) * page_size) .limit(page_size) ) return [item async for item in cursor], total async def conversation_detail(conversation_id: str) -> dict[str, Any]: conversation = await get_conversation(conversation_id) if not conversation: raise AssistantDataError("conversation_not_found", "未找到客户会话。") conversation["messages"] = await recent_conversation_messages( conversation_id, limit=50 ) return conversation def _usage_day(now: datetime, timezone_name: str) -> str: try: zone = ZoneInfo(timezone_name) except ZoneInfoNotFoundError: zone = ZoneInfo("Asia/Shanghai") return as_utc(now).astimezone(zone).date().isoformat() async def reserve_ai_usage( connection_id: str, chat_id: int, settings: dict[str, Any], *, now: datetime | None = None, ) -> tuple[bool, str]: await ensure_assistant_indexes() current_time = as_utc(now or utc_now()) day = _usage_day(current_time, str(settings.get("timezone") or "Asia/Shanghai")) account_limit = int(settings.get("account_daily_limit") or 200) customer_limit = int(settings.get("customer_daily_limit") or 20) account_filter = _scope( connection_id=str(connection_id), chat_id=0, day=day ) customer_filter = _scope( connection_id=str(connection_id), chat_id=int(chat_id), day=day ) for filters, usage_scope in ( (account_filter, "account"), (customer_filter, "customer"), ): await usagedb.update_one( filters, { "$setOnInsert": { **filters, "scope": usage_scope, "call_count": 0, "created_at": current_time, "updated_at": current_time, } }, upsert=True, ) account = await usagedb.find_one_and_update( {**account_filter, "call_count": {"$lt": account_limit}}, {"$inc": {"call_count": 1}, "$set": {"updated_at": current_time}}, return_document=True, ) if not account: return False, "account_daily_limit" customer = await usagedb.find_one_and_update( {**customer_filter, "call_count": {"$lt": customer_limit}}, {"$inc": {"call_count": 1}, "$set": {"updated_at": current_time}}, return_document=True, ) if not customer: await usagedb.update_one( {**account_filter, "call_count": {"$gt": 0}}, {"$inc": {"call_count": -1}, "$set": {"updated_at": current_time}}, ) return False, "customer_daily_limit" return True, "" async def usage_metrics(*, connection_id: str = "", day: str = "") -> dict[str, Any]: await ensure_assistant_indexes() match = _scope() if connection_id: match["connection_id"] = str(connection_id) if day: match["day"] = str(day) calls = 0 customers: set[tuple[str, int]] = set() customer_match = { **match, "scope": {"$ne": "account"}, "chat_id": {"$ne": 0}, "call_count": {"$gt": 0}, } async for item in usagedb.find(customer_match): calls += int(item.get("call_count") or 0) customers.add((str(item.get("connection_id")), int(item.get("chat_id") or 0))) conversation_filters = _scope() if connection_id: conversation_filters["connection_id"] = str(connection_id) return { "ai_calls": calls, "customers": len(customers), "conversations": await conversationsdb.count_documents(conversation_filters), "handoffs": await conversationsdb.count_documents( {**conversation_filters, "status": "handoff"} ), "human_paused": await conversationsdb.count_documents( {**conversation_filters, "status": "human_paused"} ), } async def claim_update(update_id: int) -> bool: await ensure_assistant_indexes() now = utc_now() try: await updatesdb.insert_one( { "bot_id": BOT_PROFILE_ID, "update_id": int(update_id), "status": "processing", "attempts": 1, "created_at": now, "updated_at": now, "expires_at": now + timedelta(days=7), } ) return True except DuplicateKeyError: filters = _scope(update_id=int(update_id)) current = await updatesdb.find_one(filters) or {} if current.get("status") == "done": return False await updatesdb.update_one( filters, {"$inc": {"attempts": 1}, "$set": {"updated_at": now}}, ) return True async def mark_update_done(update_id: int) -> None: await updatesdb.update_one( _scope(update_id=int(update_id)), {"$set": {"status": "done", "updated_at": utc_now(), "last_error": ""}}, ) async def mark_update_failed(update_id: int, error: str) -> int: await updatesdb.update_one( _scope(update_id=int(update_id)), {"$set": {"status": "failed", "updated_at": utc_now(), "last_error": error[:1000]}}, ) current = await updatesdb.find_one(_scope(update_id=int(update_id))) or {} return int(current.get("attempts") or 1) async def dead_letter_update(update: dict[str, Any], error: str) -> None: update_id = int(update.get("update_id") or 0) now = utc_now() await deadlettersdb.update_one( _scope(update_id=update_id), { "$set": { "bot_id": BOT_PROFILE_ID, "payload": update, "error": error[:2000], "updated_at": now, "expires_at": now + timedelta(days=30), }, "$setOnInsert": {"created_at": now}, }, upsert=True, ) await mark_update_done(update_id) async def runtime_status() -> dict[str, Any]: await ensure_assistant_indexes() stored = await runtimedb.find_one( _scope(runtime_id="business_assistant") ) or {} return { "runtime_id": "business_assistant", "polling_state": "stopped", "business_mode_supported": False, "webhook_conflict": False, "webhook_url": "", "model_configured": False, "last_error": "", **{key: value for key, value in stored.items() if key != "_id"}, } async def update_runtime_status(values: dict[str, Any]) -> dict[str, Any]: await ensure_assistant_indexes() insert_values = { "bot_id": BOT_PROFILE_ID, "runtime_id": "business_assistant", "created_at": utc_now(), } if "offset" not in values: insert_values["offset"] = 0 await runtimedb.update_one( _scope(runtime_id="business_assistant"), { "$set": {**values, "updated_at": utc_now()}, "$setOnInsert": insert_values, }, upsert=True, ) return await runtime_status() async def load_update_offset() -> int: return int((await runtime_status()).get("offset") or 0) async def save_update_offset(offset: int) -> None: await update_runtime_status({"offset": int(offset), "last_poll_at": utc_now()}) async def due_digest_connections(now: datetime | None = None) -> list[dict[str, Any]]: current = as_utc(now or utc_now()) results: list[dict[str, Any]] = [] async for settings in settingsdb.find( _scope(digest_enabled=True, assistant_enabled=True) ): try: local = current.astimezone(ZoneInfo(str(settings.get("timezone")))) except ZoneInfoNotFoundError: local = current.astimezone(ZoneInfo("Asia/Shanghai")) day = local.date().isoformat() if local.strftime("%H:%M") != str(settings.get("digest_time") or "09:00"): continue if settings.get("last_digest_day") == day: continue connection = await get_business_connection(settings["connection_id"]) if connection and connection.get("is_enabled"): results.append({"connection": connection, "settings": settings, "day": day}) return results async def mark_digest_sent(connection_id: str, day: str) -> None: await settingsdb.update_one( _scope(connection_id=str(connection_id)), {"$set": {"last_digest_day": str(day), "last_digest_at": utc_now()}}, )