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