dashboard-nanobot/backend/services/chat_command_service.py

import logging
import os
from typing import Any, Dict, List

from fastapi import HTTPException
from sqlmodel import Session

from core.docker_instance import docker_manager
from core.utils import _is_video_attachment_path, _is_visual_attachment_path
from models.bot import BotInstance
from services.bot_service import read_bot_runtime_snapshot
from services.platform_activity_service import record_activity_event
from services.platform_usage_service import create_usage_request, fail_latest_usage
from services.runtime_service import broadcast_runtime_packet, persist_runtime_packet
from services.workspace_service import resolve_workspace_path

logger = logging.getLogger("dashboard.backend")


def _normalize_message_media_item(value: Any) -> str:
    return str(value or "").strip().replace("\\", "/").lstrip("/")


def _normalize_message_media_list(raw: Any) -> List[str]:
    if not isinstance(raw, list):
        return []
    rows: List[str] = []
    for value in raw:
        normalized = _normalize_message_media_item(value)
        if normalized:
            rows.append(normalized)
    return rows


def _build_delivery_command(command: str, checked_attachments: List[str]) -> str:
    if not checked_attachments:
        return command

    attachment_block = "\n".join(f"- {path}" for path in checked_attachments)
    if all(_is_visual_attachment_path(path) for path in checked_attachments):
        has_video = any(_is_video_attachment_path(path) for path in checked_attachments)
        media_label = "图片/视频" if has_video else "图片"
        capability_hint = (
            "1) 附件已随请求附带；图片在可用时可直接作为多模态输入理解，视频请按附件路径处理。\n"
            if has_video
            else "1) 附件中的图片已作为多模态输入提供，优先直接理解并回答。\n"
        )
        if command:
            return (
                f"{command}\n\n"
                "[Attached files]\n"
                f"{attachment_block}\n\n"
                "【附件处理要求】\n"
                f"{capability_hint}"
                "2) 若当前模型或接口不支持直接理解该附件，请明确说明后再调用工具解析。\n"
                "3) 除非用户明确要求，不要先调用工具读取附件文件。\n"
                "4) 回复语言必须遵循 USER.md；若未指定，则与用户当前输入语言保持一致。\n"
                "5) 仅基于可见内容回答；看不清或无法确认的部分请明确说明，不要猜测。"
            )
        return (
            "请先处理已附带的附件列表：\n"
            f"{attachment_block}\n\n"
            f"请直接分析已附带的{media_label}并总结关键信息。\n"
            f"{'图片在可用时可直接作为多模态输入理解，视频请按附件路径处理。' if has_video else ''}\n"
            "若当前模型或接口不支持直接理解该附件，请明确说明后再调用工具解析。\n"
            "回复语言必须遵循 USER.md；若未指定，则与用户当前输入语言保持一致。\n"
            "仅基于可见内容回答；看不清或无法确认的部分请明确说明，不要猜测。"
        )

    command_has_paths = all(path in command for path in checked_attachments) if command else False
    if command and not command_has_paths:
        return (
            f"{command}\n\n"
            "[Attached files]\n"
            f"{attachment_block}\n\n"
            "Please process the attached file(s) listed above when answering this request.\n"
            "Reply language must follow USER.md. If not specified, use the same language as the user input."
        )
    if not command:
        return (
            "Please process the uploaded file(s) listed below:\n"
            f"{attachment_block}\n\n"
            "Reply language must follow USER.md. If not specified, use the same language as the user input."
        )
    return command


def send_bot_command(session: Session, bot_id: str, command: str, attachments: Any) -> Dict[str, Any]:
    request_id = ""
    try:
        bot = session.get(BotInstance, bot_id)
        if not bot:
            raise HTTPException(status_code=404, detail="Bot not found")
        runtime_snapshot = read_bot_runtime_snapshot(bot)

        normalized_attachments = _normalize_message_media_list(attachments)
        text_command = str(command or "").strip()
        if not text_command and not normalized_attachments:
            raise HTTPException(status_code=400, detail="Command or attachments is required")

        checked_attachments: List[str] = []
        for rel_path in normalized_attachments:
            _, target = resolve_workspace_path(bot_id, rel_path)
            if not os.path.isfile(target):
                raise HTTPException(status_code=400, detail=f"attachment not found: {rel_path}")
            checked_attachments.append(rel_path)
        delivery_media = [f"/root/.nanobot/workspace/{path.lstrip('/')}" for path in checked_attachments]

        display_command = text_command if text_command else "[attachment message]"
        delivery_command = _build_delivery_command(text_command, checked_attachments)

        request_id = create_usage_request(
            session,
            bot_id,
            display_command,
            attachments=checked_attachments,
            channel="dashboard",
            metadata={"attachment_count": len(checked_attachments)},
            provider=str(runtime_snapshot.get("llm_provider") or "").strip() or None,
            model=str(runtime_snapshot.get("llm_model") or "").strip() or None,
        )
        record_activity_event(
            session,
            bot_id,
            "command_submitted",
            request_id=request_id,
            channel="dashboard",
            detail="command submitted",
            metadata={"attachment_count": len(checked_attachments), "has_text": bool(text_command)},
        )
        session.commit()

        outbound_user_packet: Dict[str, Any] | None = None
        if display_command or checked_attachments:
            outbound_user_packet = {
                "type": "USER_COMMAND",
                "channel": "dashboard",
                "text": display_command,
                "media": checked_attachments,
                "request_id": request_id,
            }
            persist_runtime_packet(bot_id, outbound_user_packet)

        if outbound_user_packet:
            broadcast_runtime_packet(bot_id, outbound_user_packet)

        success = docker_manager.send_command(bot_id, delivery_command, media=delivery_media)
        if success:
            return {"success": True}

        detail = docker_manager.get_last_delivery_error(bot_id)
        fail_latest_usage(session, bot_id, detail or "command delivery failed")
        record_activity_event(
            session,
            bot_id,
            "command_failed",
            request_id=request_id,
            channel="dashboard",
            detail=(detail or "command delivery failed")[:400],
        )
        session.commit()
        broadcast_runtime_packet(
            bot_id,
            {
                "type": "AGENT_STATE",
                "channel": "dashboard",
                "payload": {
                    "state": "ERROR",
                    "action_msg": detail or "command delivery failed",
                },
            },
        )
        raise HTTPException(
            status_code=502,
            detail=f"Failed to deliver command to bot dashboard channel{': ' + detail if detail else ''}",
        )
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("send_bot_command failed for bot_id=%s", bot_id)
        try:
            session.rollback()
        except Exception:
            pass
        if request_id:
            try:
                fail_latest_usage(session, bot_id, str(exc))
                record_activity_event(
                    session,
                    bot_id,
                    "command_failed",
                    request_id=request_id,
                    channel="dashboard",
                    detail=str(exc)[:400],
                )
                session.commit()
            except Exception:
                try:
                    session.rollback()
                except Exception:
                    pass
        raise HTTPException(status_code=500, detail=f"Failed to process bot command: {exc}") from exc