Kris
4347d3f937
本次提交包含多项改进： 1. 修复钉钉、WhatsApp、Telegram等适配器的线程动作映射名称 2. 为SynologyChat、iMessage、Urbit等多款适配器新增配置Schema 3. 优化日志输出格式，合并多行日志调用为单行 4. 修复指数退避计算中的空格问题 5. 为QQBot凭证备份模块添加弃用警告 6. 新增多款适配器的凭证持久化存储逻辑 7. 优化Matrix、Nostr、DingDing等适配器的状态存储实现 8. 完善Discord、Slack、Signal等适配器的动作注册逻辑 9. 优化WhatsApp桥接器的QR码获取逻辑 10. 修复IRC适配器的配置比对与重连逻辑
271 lines
9.1 KiB
Python
271 lines
9.1 KiB
Python
"""DingDing 渠道 CLI 配置向导。
|
||
|
||
交互式引导用户配置钉钉机器人，支持 Stream 和 Webhook 两种模式。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from typing import Any
|
||
|
||
|
||
def run_dingding_setup() -> dict[str, Any] | None:
|
||
print("=== 钉钉机器人配置向导 ===\n")
|
||
|
||
has_existing = _detect_existing_config()
|
||
if has_existing:
|
||
choice = (
|
||
_prompt(
|
||
"检测到已有钉钉配置，是否使用已有配置？[编辑(e)/新建(n)/取消(c)]",
|
||
default="e",
|
||
)
|
||
.strip()
|
||
.lower()
|
||
)
|
||
if choice == "c":
|
||
print("配置已取消")
|
||
return None
|
||
if choice == "n":
|
||
return _run_new_config_flow()
|
||
return _run_edit_flow(has_existing)
|
||
|
||
return _run_new_config_flow()
|
||
|
||
|
||
def _run_new_config_flow() -> dict[str, Any] | None:
|
||
print("\n--- 新建钉钉应用 ---")
|
||
|
||
config = _prompt_credentials()
|
||
if config is None:
|
||
return None
|
||
|
||
config = _prompt_mode_selection(config)
|
||
config = _prompt_policy_selection(config)
|
||
config = _prompt_streaming_config(config)
|
||
config = _apply_recommendations(config)
|
||
|
||
return _confirm_and_return_config(config)
|
||
|
||
|
||
def _run_edit_flow(existing: dict[str, Any]) -> dict[str, Any] | None:
|
||
print("\n--- 编辑已有应用 ---")
|
||
accounts = existing.get("accounts", {}).get("default", {})
|
||
print(f"当前 App Key: {accounts.get('app_key', '未配置')}")
|
||
print(f"当前 Robot Code: {accounts.get('robot_code', '未配置')}")
|
||
print(f"当前模式: {existing.get('mode', 'stream')}")
|
||
|
||
config = dict(existing)
|
||
change_cred = _prompt("是否更换应用凭据？[y/N]", default="n").strip().lower()
|
||
if change_cred == "y":
|
||
new_creds = _prompt_credentials()
|
||
if new_creds:
|
||
config.update(new_creds)
|
||
|
||
config = _prompt_mode_selection(config)
|
||
config = _prompt_policy_selection(config)
|
||
config = _prompt_streaming_config(config)
|
||
config = _apply_recommendations(config)
|
||
|
||
return _confirm_and_return_config(config)
|
||
|
||
|
||
def _prompt_credentials() -> dict[str, Any] | None:
|
||
print("\n--- 应用凭据 ---")
|
||
print("凭据可在钉钉开放平台 (https://open.dingtalk.com) 获取")
|
||
print("路径: 应用开发 > 企业内部应用 > 应用详情 > 凭证与基础信息\n")
|
||
|
||
app_key = _prompt("App Key (Client ID)").strip()
|
||
if not app_key:
|
||
print("App Key 不能为空，配置取消")
|
||
return None
|
||
|
||
app_secret = _prompt("App Secret (Client Secret)").strip()
|
||
if not app_secret:
|
||
print("App Secret 不能为空，配置取消")
|
||
return None
|
||
|
||
robot_code = _prompt("Robot Code (机器人应用详情页获取，可选)", default="").strip()
|
||
|
||
return {
|
||
"accounts": {
|
||
"default": {
|
||
"app_key": app_key,
|
||
"app_secret": app_secret,
|
||
"robot_code": robot_code,
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
def _prompt_mode_selection(config: dict[str, Any]) -> dict[str, Any]:
|
||
print("\n--- 连接模式 ---")
|
||
print("1) Stream 模式 - 通过 WebSocket 长连接接收消息（推荐，无需公网IP）")
|
||
print("2) Webhook 模式 - 通过 HTTP 回调接收消息（需要公网可达的服务器）")
|
||
|
||
choice = _prompt("请选择 [1/2]", default="1").strip()
|
||
if choice == "2":
|
||
config["mode"] = "webhook"
|
||
else:
|
||
config["mode"] = "stream"
|
||
|
||
return config
|
||
|
||
|
||
def _prompt_policy_selection(config: dict[str, Any]) -> dict[str, Any]:
|
||
print("\n--- 安全策略配置 ---")
|
||
print("请选择群聊策略:")
|
||
print(" 1) allowlist - 仅允许白名单中的群聊访问（推荐）")
|
||
print(" 2) open - 所有群聊均可访问")
|
||
print(" 3) disabled - 禁用群聊功能")
|
||
|
||
choice = _prompt("请选择 [1/2/3]", default="1").strip()
|
||
policy_map = {"1": "allowlist", "2": "open", "3": "disabled"}
|
||
group_policy = policy_map.get(choice, "allowlist")
|
||
config["groupPolicy"] = group_policy
|
||
config["group_policy"] = group_policy
|
||
|
||
if group_policy == "allowlist":
|
||
print("\n请输入允许访问的群聊 ID（每行一个，空行结束）：")
|
||
print("提示: 群聊 ID 可从消息中获取，格式如 cidXXXXXXXXX")
|
||
allowlist: list[str] = []
|
||
while True:
|
||
entry = _prompt(" chat_id (直接回车结束)", default="").strip()
|
||
if not entry:
|
||
break
|
||
allowlist.append(entry)
|
||
if allowlist:
|
||
config["allowlist"] = allowlist
|
||
config["allowFrom"] = allowlist
|
||
|
||
print("\n请选择私聊策略:")
|
||
print(" 1) pairing - 所有人可私聊机器人（推荐）")
|
||
print(" 2) open - 开放模式")
|
||
print(" 3) allowlist - 仅白名单用户可私聊")
|
||
|
||
dm_choice = _prompt("请选择 [1/2/3]", default="1").strip()
|
||
dm_policy_map = {"1": "pairing", "2": "open", "3": "allowlist"}
|
||
dm_policy = dm_policy_map.get(dm_choice, "pairing")
|
||
config["dmPolicy"] = dm_policy
|
||
config["dm_policy"] = dm_policy
|
||
|
||
print("\n群聊中需要 @机器人 才触发回复吗？")
|
||
require = _prompt("[y/N]", default="y").strip().lower()
|
||
config["requireMention"] = require != "n"
|
||
|
||
return config
|
||
|
||
|
||
def _prompt_streaming_config(config: dict[str, Any]) -> dict[str, Any]:
|
||
print("\n--- 流式输出配置 ---")
|
||
print("1) block - 累积后一次性发送（推荐，兼容性好）")
|
||
print("2) partial - 实时编辑消息（需要机器人支持编辑消息权限）")
|
||
print("3) off - 关闭流式输出")
|
||
|
||
choice = _prompt("请选择 [1/2/3]", default="1").strip()
|
||
mode_map = {"1": "block", "2": "partial", "3": "off"}
|
||
stream_mode = mode_map.get(choice, "block")
|
||
config.setdefault("streaming", {})
|
||
config["streaming"]["mode"] = stream_mode
|
||
|
||
return config
|
||
|
||
|
||
def _apply_recommendations(config: dict[str, Any]) -> dict[str, Any]:
|
||
group_policy = config.get("groupPolicy", "allowlist")
|
||
if group_policy == "open":
|
||
print("\n⚠ 警告: 群聊策略设置为 'open'，所有群聊均可访问机器人。")
|
||
print(" 建议在正式环境使用 'allowlist' 模式。")
|
||
|
||
if not config.get("robotCode", "") and not config.get("accounts", {}).get("default", {}).get("robot_code"):
|
||
print("\n⚠ 提示: 未配置 Robot Code，表情回复和卡片功能可能不可用。")
|
||
print(" Robot Code 可在机器人应用详情页获取。")
|
||
|
||
return config
|
||
|
||
|
||
def _confirm_and_return_config(config: dict[str, Any]) -> dict[str, Any] | None:
|
||
print("\n=== 配置摘要 ===")
|
||
|
||
accounts = config.get("accounts", {}).get("default", {})
|
||
print(f" App Key: {accounts.get('app_key', '')[:10]}****")
|
||
print(f" Robot Code: {accounts.get('robot_code', '未配置')}")
|
||
print(f" 连接模式: {config.get('mode', 'stream')}")
|
||
print(f" 群聊策略: {config.get('groupPolicy', 'allowlist')}")
|
||
print(f" 私聊策略: {config.get('dmPolicy', 'pairing')}")
|
||
print(f" 需要 @机器人: {config.get('requireMention', True)}")
|
||
print(f" 流式模式: {config.get('streaming', {}).get('mode', 'block')}")
|
||
|
||
allowlist = config.get("allowlist", [])
|
||
if allowlist:
|
||
for item in allowlist[:5]:
|
||
print(f" 白名单: {item}")
|
||
if len(allowlist) > 5:
|
||
print(f" ... 共 {len(allowlist)} 项")
|
||
|
||
confirm = _prompt("\n是否保存配置？[y/N]", default="n").strip().lower()
|
||
if confirm != "y":
|
||
print("配置已取消")
|
||
return None
|
||
|
||
print("配置已保存")
|
||
return config
|
||
|
||
|
||
def save_config_to_file(config: dict[str, Any], filepath: str) -> None:
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2)
|
||
print(f"配置已写入: {filepath}")
|
||
|
||
|
||
def _detect_existing_config() -> dict[str, Any] | None:
|
||
env_app_key = os.environ.get("DINGDING_APP_KEY", "")
|
||
env_app_secret = os.environ.get("DINGDING_APP_SECRET", "")
|
||
env_robot_code = os.environ.get("DINGDING_ROBOT_CODE", "")
|
||
|
||
if env_app_key and env_app_secret:
|
||
return {
|
||
"accounts": {
|
||
"default": {
|
||
"app_key": env_app_key,
|
||
"app_secret": env_app_secret,
|
||
"robot_code": env_robot_code,
|
||
}
|
||
},
|
||
"mode": "stream",
|
||
}
|
||
|
||
config_paths = [
|
||
"dingding_config.json",
|
||
"config/dingding.json",
|
||
os.path.expanduser("~/.forcepilot/dingding.json"),
|
||
]
|
||
for path in config_paths:
|
||
if os.path.exists(path):
|
||
try:
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, OSError):
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def _prompt(text: str, default: str = "") -> str:
|
||
if default:
|
||
prompt_text = f"{text} [{default}]: "
|
||
else:
|
||
prompt_text = f"{text}: "
|
||
try:
|
||
return input(prompt_text)
|
||
except (EOFError, KeyboardInterrupt):
|
||
return default
|
||
|
||
|
||
if __name__ == "__main__":
|
||
result = run_dingding_setup()
|
||
if result:
|
||
save_config_to_file(result, "dingding_config.json")
|
||
else:
|
||
print("未生成配置")
|