dashboard-nanobot/backend/services/bot_service.py

535 lines
21 KiB
Python

import os
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from zoneinfo import ZoneInfo
from sqlmodel import Session
from core.settings import BOTS_WORKSPACE_ROOT
from models.bot import BotInstance
from providers.bot_workspace_provider import BotWorkspaceProvider
from schemas.bot import ChannelConfigRequest
from services.bot_storage_service import (
_normalize_env_params,
_read_bot_config,
_read_bot_resources,
_read_env_store,
_safe_float,
_safe_int,
_workspace_root,
normalize_bot_resource_limits,
write_bot_resource_limits,
)
workspace_provider = BotWorkspaceProvider(host_data_root=BOTS_WORKSPACE_ROOT)
def normalize_bot_system_timezone(raw: Any) -> str:
value = str(raw or "").strip()
if not value:
raise ValueError("System timezone is required")
try:
ZoneInfo(value)
except Exception as exc:
raise ValueError("Invalid system timezone. Use an IANA timezone such as Asia/Shanghai.") from exc
return value
def resolve_bot_runtime_env_params(bot_id: str, raw: Optional[Dict[str, str]] = None) -> Dict[str, str]:
env_params = _normalize_env_params(raw if isinstance(raw, dict) else _read_env_store(bot_id))
if "TZ" not in env_params:
raise RuntimeError(f"Missing required TZ in bot env settings: {bot_id}")
env_params["TZ"] = normalize_bot_system_timezone(env_params.get("TZ"))
return env_params
def normalize_channel_extra(raw: Any) -> Dict[str, Any]:
if not isinstance(raw, dict):
return {}
return raw
def _normalize_allow_from(raw: Any) -> List[str]:
rows: List[str] = []
if isinstance(raw, list):
for item in raw:
text = str(item or "").strip()
if text and text not in rows:
rows.append(text)
return rows or ["*"]
def read_global_delivery_flags(channels_cfg: Any) -> tuple[bool, bool]:
if not isinstance(channels_cfg, dict):
return False, False
return bool(channels_cfg.get("sendProgress")), bool(channels_cfg.get("sendToolHints"))
def channel_config_to_api(bot_id: str, channel_type: str, cfg: Dict[str, Any]) -> Dict[str, Any]:
ctype = str(channel_type or "").strip().lower()
enabled = bool(cfg.get("enabled", True))
port = max(1, min(int(cfg.get("port", 8080) or 8080), 65535))
extra: Dict[str, Any] = {}
external_app_id = ""
app_secret = ""
if ctype == "feishu":
external_app_id = str(cfg.get("appId") or "")
app_secret = str(cfg.get("appSecret") or "")
extra = {
"encryptKey": cfg.get("encryptKey", ""),
"verificationToken": cfg.get("verificationToken", ""),
"allowFrom": _normalize_allow_from(cfg.get("allowFrom", [])),
}
elif ctype == "dingtalk":
external_app_id = str(cfg.get("clientId") or "")
app_secret = str(cfg.get("clientSecret") or "")
extra = {"allowFrom": _normalize_allow_from(cfg.get("allowFrom", []))}
elif ctype == "telegram":
app_secret = str(cfg.get("token") or "")
extra = {
"proxy": cfg.get("proxy", ""),
"replyToMessage": bool(cfg.get("replyToMessage", False)),
"allowFrom": _normalize_allow_from(cfg.get("allowFrom", [])),
}
elif ctype == "slack":
external_app_id = str(cfg.get("botToken") or "")
app_secret = str(cfg.get("appToken") or "")
extra = {
"mode": cfg.get("mode", "socket"),
"replyInThread": bool(cfg.get("replyInThread", True)),
"groupPolicy": cfg.get("groupPolicy", "mention"),
"groupAllowFrom": cfg.get("groupAllowFrom", []),
"reactEmoji": cfg.get("reactEmoji", "eyes"),
}
elif ctype == "qq":
external_app_id = str(cfg.get("appId") or "")
app_secret = str(cfg.get("secret") or "")
extra = {"allowFrom": _normalize_allow_from(cfg.get("allowFrom", []))}
elif ctype == "weixin":
app_secret = ""
extra = {
"hasSavedState": (Path(BOTS_WORKSPACE_ROOT) / bot_id / ".nanobot" / "weixin" / "account.json").is_file(),
}
elif ctype == "email":
extra = {
"consentGranted": bool(cfg.get("consentGranted", False)),
"imapHost": str(cfg.get("imapHost") or ""),
"imapPort": int(cfg.get("imapPort") or 993),
"imapUsername": str(cfg.get("imapUsername") or ""),
"imapPassword": str(cfg.get("imapPassword") or ""),
"imapMailbox": str(cfg.get("imapMailbox") or "INBOX"),
"imapUseSsl": bool(cfg.get("imapUseSsl", True)),
"smtpHost": str(cfg.get("smtpHost") or ""),
"smtpPort": int(cfg.get("smtpPort") or 587),
"smtpUsername": str(cfg.get("smtpUsername") or ""),
"smtpPassword": str(cfg.get("smtpPassword") or ""),
"smtpUseTls": bool(cfg.get("smtpUseTls", True)),
"smtpUseSsl": bool(cfg.get("smtpUseSsl", False)),
"fromAddress": str(cfg.get("fromAddress") or ""),
"autoReplyEnabled": bool(cfg.get("autoReplyEnabled", True)),
"pollIntervalSeconds": int(cfg.get("pollIntervalSeconds") or 30),
"markSeen": bool(cfg.get("markSeen", True)),
"maxBodyChars": int(cfg.get("maxBodyChars") or 12000),
"subjectPrefix": str(cfg.get("subjectPrefix") or "Re: "),
"allowFrom": _normalize_allow_from(cfg.get("allowFrom", [])),
}
else:
external_app_id = str(
cfg.get("appId") or cfg.get("clientId") or cfg.get("botToken") or cfg.get("externalAppId") or ""
)
app_secret = str(
cfg.get("appSecret")
or cfg.get("clientSecret")
or cfg.get("secret")
or cfg.get("token")
or cfg.get("appToken")
or ""
)
extra = {
key: value
for key, value in cfg.items()
if key
not in {
"enabled",
"port",
"appId",
"clientId",
"botToken",
"externalAppId",
"appSecret",
"clientSecret",
"secret",
"token",
"appToken",
}
}
return {
"id": ctype,
"bot_id": bot_id,
"channel_type": ctype,
"external_app_id": external_app_id,
"app_secret": app_secret,
"internal_port": port,
"is_active": enabled,
"extra_config": extra,
"locked": ctype == "dashboard",
}
def channel_api_to_config(row: Dict[str, Any]) -> Dict[str, Any]:
ctype = str(row.get("channel_type") or "").strip().lower()
enabled = bool(row.get("is_active", True))
extra = normalize_channel_extra(row.get("extra_config"))
external_app_id = str(row.get("external_app_id") or "")
app_secret = str(row.get("app_secret") or "")
port = max(1, min(int(row.get("internal_port") or 8080), 65535))
if ctype == "feishu":
return {
"enabled": enabled,
"appId": external_app_id,
"appSecret": app_secret,
"encryptKey": extra.get("encryptKey", ""),
"verificationToken": extra.get("verificationToken", ""),
"allowFrom": _normalize_allow_from(extra.get("allowFrom", [])),
}
if ctype == "dingtalk":
return {
"enabled": enabled,
"clientId": external_app_id,
"clientSecret": app_secret,
"allowFrom": _normalize_allow_from(extra.get("allowFrom", [])),
}
if ctype == "telegram":
return {
"enabled": enabled,
"token": app_secret,
"proxy": extra.get("proxy", ""),
"replyToMessage": bool(extra.get("replyToMessage", False)),
"allowFrom": _normalize_allow_from(extra.get("allowFrom", [])),
}
if ctype == "slack":
return {
"enabled": enabled,
"mode": extra.get("mode", "socket"),
"botToken": external_app_id,
"appToken": app_secret,
"replyInThread": bool(extra.get("replyInThread", True)),
"groupPolicy": extra.get("groupPolicy", "mention"),
"groupAllowFrom": extra.get("groupAllowFrom", []),
"reactEmoji": extra.get("reactEmoji", "eyes"),
}
if ctype == "qq":
return {
"enabled": enabled,
"appId": external_app_id,
"secret": app_secret,
"allowFrom": _normalize_allow_from(extra.get("allowFrom", [])),
}
if ctype == "weixin":
return {
"enabled": enabled,
"token": app_secret,
}
if ctype == "email":
return {
"enabled": enabled,
"consentGranted": bool(extra.get("consentGranted", False)),
"imapHost": str(extra.get("imapHost") or ""),
"imapPort": max(1, min(int(extra.get("imapPort") or 993), 65535)),
"imapUsername": str(extra.get("imapUsername") or ""),
"imapPassword": str(extra.get("imapPassword") or ""),
"imapMailbox": str(extra.get("imapMailbox") or "INBOX"),
"imapUseSsl": bool(extra.get("imapUseSsl", True)),
"smtpHost": str(extra.get("smtpHost") or ""),
"smtpPort": max(1, min(int(extra.get("smtpPort") or 587), 65535)),
"smtpUsername": str(extra.get("smtpUsername") or ""),
"smtpPassword": str(extra.get("smtpPassword") or ""),
"smtpUseTls": bool(extra.get("smtpUseTls", True)),
"smtpUseSsl": bool(extra.get("smtpUseSsl", False)),
"fromAddress": str(extra.get("fromAddress") or ""),
"autoReplyEnabled": bool(extra.get("autoReplyEnabled", True)),
"pollIntervalSeconds": max(5, int(extra.get("pollIntervalSeconds") or 30)),
"markSeen": bool(extra.get("markSeen", True)),
"maxBodyChars": max(1, int(extra.get("maxBodyChars") or 12000)),
"subjectPrefix": str(extra.get("subjectPrefix") or "Re: "),
"allowFrom": _normalize_allow_from(extra.get("allowFrom", [])),
}
merged = dict(extra)
merged.update(
{
"enabled": enabled,
"appId": external_app_id,
"appSecret": app_secret,
"port": port,
}
)
return merged
def list_bot_channels_from_config(bot: BotInstance) -> List[Dict[str, Any]]:
config_data = _read_bot_config(bot.id)
channels_cfg = config_data.get("channels")
if not isinstance(channels_cfg, dict):
channels_cfg = {}
send_progress, send_tool_hints = read_global_delivery_flags(channels_cfg)
rows: List[Dict[str, Any]] = [
{
"id": "dashboard",
"bot_id": bot.id,
"channel_type": "dashboard",
"external_app_id": f"dashboard-{bot.id}",
"app_secret": "",
"internal_port": 9000,
"is_active": True,
"extra_config": {
"sendProgress": send_progress,
"sendToolHints": send_tool_hints,
},
"locked": True,
}
]
for ctype, cfg in channels_cfg.items():
if ctype in {"sendProgress", "sendToolHints", "dashboard"} or not isinstance(cfg, dict):
continue
rows.append(channel_config_to_api(bot.id, ctype, cfg))
return rows
def normalize_initial_bot_channels(bot_id: str, channels: Optional[List[ChannelConfigRequest]]) -> List[Dict[str, Any]]:
rows: List[Dict[str, Any]] = []
seen_types: set[str] = set()
for channel in channels or []:
ctype = (channel.channel_type or "").strip().lower()
if not ctype or ctype == "dashboard" or ctype in seen_types:
continue
seen_types.add(ctype)
rows.append(
{
"id": ctype,
"bot_id": bot_id,
"channel_type": ctype,
"external_app_id": (channel.external_app_id or "").strip() or f"{ctype}-{bot_id}",
"app_secret": (channel.app_secret or "").strip(),
"internal_port": max(1, min(int(channel.internal_port or 8080), 65535)),
"is_active": bool(channel.is_active),
"extra_config": normalize_channel_extra(channel.extra_config),
"locked": False,
}
)
return rows
def _read_workspace_md(bot_id: str, filename: str) -> str:
path = os.path.join(_workspace_root(bot_id), filename)
if not os.path.isfile(path):
raise RuntimeError(f"Missing required workspace file: {path}")
try:
with open(path, "r", encoding="utf-8") as file:
return file.read().strip()
except Exception as exc:
raise RuntimeError(f"Failed to read workspace file: {path}") from exc
def read_bot_runtime_snapshot(bot: BotInstance) -> Dict[str, Any]:
config_data = _read_bot_config(bot.id)
env_params = resolve_bot_runtime_env_params(bot.id)
provider_name = ""
provider_cfg: Dict[str, Any] = {}
providers_cfg = config_data.get("providers")
if isinstance(providers_cfg, dict):
for p_name, p_cfg in providers_cfg.items():
provider_name = str(p_name or "").strip()
if isinstance(p_cfg, dict):
provider_cfg = p_cfg
break
if not provider_name or not provider_cfg:
raise RuntimeError(f"Missing provider configuration in bot config: {bot.id}")
agents_defaults: Dict[str, Any] = {}
agents_cfg = config_data.get("agents")
if isinstance(agents_cfg, dict):
defaults = agents_cfg.get("defaults")
if isinstance(defaults, dict):
agents_defaults = defaults
if not agents_defaults:
raise RuntimeError(f"Missing agents.defaults in bot config: {bot.id}")
channels_cfg = config_data.get("channels")
send_progress, send_tool_hints = read_global_delivery_flags(channels_cfg)
llm_provider = provider_name or ""
llm_model = str(agents_defaults.get("model") or "")
api_key = str(provider_cfg.get("apiKey") or "").strip()
api_base = str(provider_cfg.get("apiBase") or "").strip()
if not llm_model:
raise RuntimeError(f"Missing model in bot config: {bot.id}")
if not api_key:
raise RuntimeError(f"Missing apiKey in bot config: {bot.id}")
if not api_base:
raise RuntimeError(f"Missing apiBase in bot config: {bot.id}")
api_base_lower = api_base.lower()
provider_alias = str(provider_cfg.get("dashboardProviderAlias") or "").strip().lower()
if llm_provider == "openai" and provider_alias in {"xunfei", "iflytek", "xfyun", "vllm"}:
llm_provider = "xunfei" if provider_alias in {"iflytek", "xfyun"} else provider_alias
elif llm_provider == "openai" and ("spark-api-open.xf-yun.com" in api_base_lower or "xf-yun.com" in api_base_lower):
llm_provider = "xunfei"
tools_cfg = config_data.get("tools")
if tools_cfg is not None and not isinstance(tools_cfg, dict):
raise RuntimeError(f"Invalid tools configuration in bot config: {bot.id}")
mcp_servers = tools_cfg.get("mcpServers") if isinstance(tools_cfg, dict) else None
soul_md = _read_workspace_md(bot.id, "SOUL.md")
resources = _read_bot_resources(bot.id, config_data=config_data)
return {
"llm_provider": llm_provider,
"llm_model": llm_model,
"api_key": api_key,
"api_base": api_base,
"temperature": _safe_float(agents_defaults.get("temperature"), 0.2),
"top_p": _safe_float(agents_defaults.get("topP"), 1.0),
"max_tokens": _safe_int(agents_defaults.get("maxTokens"), 8192),
"cpu_cores": resources["cpu_cores"],
"memory_mb": resources["memory_mb"],
"storage_gb": resources["storage_gb"],
"system_timezone": env_params["TZ"],
"send_progress": send_progress,
"send_tool_hints": send_tool_hints,
"soul_md": soul_md,
"agents_md": _read_workspace_md(bot.id, "AGENTS.md"),
"user_md": _read_workspace_md(bot.id, "USER.md"),
"tools_md": _read_workspace_md(bot.id, "TOOLS.md"),
"identity_md": _read_workspace_md(bot.id, "IDENTITY.md"),
"mcp_servers": mcp_servers if isinstance(mcp_servers, dict) else None,
}
def serialize_bot_detail(bot: BotInstance) -> Dict[str, Any]:
runtime = read_bot_runtime_snapshot(bot)
created_at = bot.created_at.isoformat() + "Z" if bot.created_at else None
updated_at = bot.updated_at.isoformat() + "Z" if bot.updated_at else None
return {
"id": bot.id,
"name": bot.name,
"enabled": bool(getattr(bot, "enabled", True)),
"access_password": bot.access_password or "",
"has_access_password": bool(str(bot.access_password or "").strip()),
"avatar_model": "base",
"avatar_skin": "blue_suit",
"image_tag": bot.image_tag,
"llm_provider": runtime["llm_provider"],
"llm_model": runtime["llm_model"],
"api_key": runtime["api_key"],
"api_base": runtime["api_base"],
"temperature": runtime["temperature"],
"top_p": runtime["top_p"],
"max_tokens": runtime["max_tokens"],
"cpu_cores": runtime["cpu_cores"],
"memory_mb": runtime["memory_mb"],
"storage_gb": runtime["storage_gb"],
"system_timezone": runtime["system_timezone"],
"send_progress": runtime["send_progress"],
"send_tool_hints": runtime["send_tool_hints"],
"soul_md": runtime["soul_md"],
"agents_md": runtime["agents_md"],
"user_md": runtime["user_md"],
"tools_md": runtime["tools_md"],
"identity_md": runtime["identity_md"],
"workspace_dir": bot.workspace_dir,
"docker_status": bot.docker_status,
"current_state": bot.current_state,
"last_action": bot.last_action,
"created_at": created_at,
"updated_at": updated_at,
}
def serialize_bot_list_entry(bot: BotInstance) -> Dict[str, Any]:
created_at = bot.created_at.isoformat() + "Z" if bot.created_at else None
updated_at = bot.updated_at.isoformat() + "Z" if bot.updated_at else None
return {
"id": bot.id,
"name": bot.name,
"enabled": bool(getattr(bot, "enabled", True)),
"has_access_password": bool(str(bot.access_password or "").strip()),
"image_tag": bot.image_tag,
"docker_status": bot.docker_status,
"current_state": bot.current_state,
"last_action": bot.last_action,
"created_at": created_at,
"updated_at": updated_at,
}
def _has_bot_workspace_config(bot_id: str) -> bool:
return (Path(BOTS_WORKSPACE_ROOT) / bot_id / ".nanobot" / "config.json").is_file()
def sync_bot_workspace_channels(
session: Session,
bot_id: str,
channels_override: Optional[List[Dict[str, Any]]] = None,
global_delivery_override: Optional[Dict[str, Any]] = None,
runtime_overrides: Optional[Dict[str, Any]] = None,
) -> None:
bot = session.get(BotInstance, bot_id)
if not bot:
raise RuntimeError(f"Bot not found: {bot_id}")
has_existing_config = _has_bot_workspace_config(bot_id)
if has_existing_config:
snapshot = read_bot_runtime_snapshot(bot)
bot_data: Dict[str, Any] = dict(snapshot)
else:
if not isinstance(runtime_overrides, dict):
raise RuntimeError(f"Missing required bot config for workspace sync: {bot_id}")
bot_data = {}
if isinstance(runtime_overrides, dict):
bot_data.update(runtime_overrides)
resources = normalize_bot_resource_limits(
bot_data.get("cpu_cores"),
bot_data.get("memory_mb"),
bot_data.get("storage_gb"),
)
bot_data.update(resources)
send_progress = bool(bot_data.get("send_progress", False))
send_tool_hints = bool(bot_data.get("send_tool_hints", False))
if isinstance(global_delivery_override, dict):
if "sendProgress" in global_delivery_override:
send_progress = bool(global_delivery_override.get("sendProgress"))
if "sendToolHints" in global_delivery_override:
send_tool_hints = bool(global_delivery_override.get("sendToolHints"))
if channels_override is not None:
channels_data = channels_override
elif has_existing_config:
channels_data = list_bot_channels_from_config(bot)
else:
channels_data = []
bot_data["send_progress"] = send_progress
bot_data["send_tool_hints"] = send_tool_hints
normalized_channels: List[Dict[str, Any]] = []
for row in channels_data:
ctype = str(row.get("channel_type") or "").strip().lower()
if not ctype or ctype == "dashboard":
continue
normalized_channels.append(
{
"channel_type": ctype,
"external_app_id": str(row.get("external_app_id") or ""),
"app_secret": str(row.get("app_secret") or ""),
"internal_port": max(1, min(int(row.get("internal_port") or 8080), 65535)),
"is_active": bool(row.get("is_active", True)),
"extra_config": normalize_channel_extra(row.get("extra_config")),
}
)
workspace_provider.write_workspace(bot_id=bot_id, bot_data=bot_data, channels=normalized_channels)
write_bot_resource_limits(bot_id, bot_data.get("cpu_cores"), bot_data.get("memory_mb"), bot_data.get("storage_gb"))