|
|
@@ -0,0 +1,1047 @@
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import hashlib
|
|
|
+import re
|
|
|
+from datetime import UTC, datetime, timedelta
|
|
|
+from typing import Any
|
|
|
+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
|
|
|
+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": "已收到你的消息，这类内容需要人工处理，我已经通知负责人。",
|
|
|
+}
|
|
|
+
|
|
|
+VALID_NOTIFICATION_DESTINATIONS = {"owner", "ops", "both"}
|
|
|
+VALID_TONES = {"professional", "friendly", "concise"}
|
|
|
+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_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
|
|
|
+
|
|
|
+
|
|
|
+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 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"] = 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 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)
|
|
|
+ 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)),
|
|
|
+ "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 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 _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 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()}},
|
|
|
+ )
|
chendeben