TelegramBotAdmin/wbb/utils/dbassistant.py

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()}},
    )