姚章浩
122 lines
6.5 KiB
Python
122 lines
6.5 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import getpass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from urllib.parse import quote, unquote, urlsplit
|
|
import subprocess
|
|
import sys
|
|
|
|
FIELDS = (
|
|
("WEKNORA_BASE_URL", "WeKnora 地址", "应用容器通过宿主机网络访问既有WeKnora。", "http://127.0.0.1:8080", False),
|
|
("WEKNORA_API_KEY", "WeKnora API Key", "DobotAI专用WeKnora API Key。", "<你的WeKnora API Key>", True),
|
|
("DOBOTAI_QUICK_AGENT_ID", "快速问答Agent ID", "默认问答实际调用的WeKnora Agent。", "<快速问答Agent ID>", False),
|
|
("DOBOTAI_REASONING_AGENT_ID", "智能推理Agent ID", "/think和持续智能推理实际调用的WeKnora Agent。", "<智能推理Agent ID>", False),
|
|
("DINGTALK_CLIENT_ID", "钉钉 Client ID", "钉钉应用Client ID/AppKey。", "dingxxxxxxxx", False),
|
|
("DINGTALK_CLIENT_SECRET", "钉钉 Client Secret", "钉钉应用Client Secret/AppSecret。", "<你的钉钉 AppSecret>", True),
|
|
)
|
|
|
|
def prompt_rds(values):
|
|
"""分字段录入阿里云RDS并组装DATABASE_URL；已有值作为默认。"""
|
|
existing = values.get("DATABASE_URL", "").strip()
|
|
host = port = database = user = password = ""
|
|
if existing:
|
|
u = urlsplit(existing)
|
|
host = u.hostname or ""
|
|
port = str(u.port or "") if u.port else ""
|
|
database = u.path.lstrip("/")
|
|
user = unquote(u.username or "")
|
|
password = unquote(u.password or "")
|
|
|
|
def ask(key, label, default, secret=False):
|
|
suffix = "（已配置，直接回车保留）" if default else "（必填）"
|
|
reader = getpass.getpass if secret else input
|
|
entered = reader(f"{label}{suffix}: ").strip()
|
|
return entered or default
|
|
|
|
print("\n阿里云RDS PostgreSQL（分字段填写）\n用途：DobotAI专用数据库；空库自动初始化，已有表只校验不改结构。")
|
|
host = ask("host", "RDS主机地址", host)
|
|
port = ask("port", "RDS端口", port or "5432")
|
|
database = ask("database", "数据库名", database)
|
|
user = ask("user", "数据库账号", user)
|
|
password = ask("password", "数据库密码", password, secret=True)
|
|
if not all((host, port, database, user, password)):
|
|
raise SystemExit("RDS字段不完整，未保存。")
|
|
if not database.startswith(("dobotai", "dobotai_")):
|
|
print("警告：数据库名不是DobotAI专用库，继续前请确认。")
|
|
values["DATABASE_URL"] = f"postgres://{quote(user, safe='')}:{quote(password, safe='')}@{host}:{port}/{database}"
|
|
|
|
OSS_FIELDS = (
|
|
("DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT", "OSS Endpoint", "私有Bucket的接入地址。", "https://oss-cn-shenzhen.aliyuncs.com", False),
|
|
("DOBOTAI_OUTBOUND_IMAGE_OSS_REGION", "OSS Region", "Bucket所在地域。", "cn-shenzhen", False),
|
|
("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET", "OSS Bucket", "外发图片专用私有Bucket。", "dobot-xxx", False),
|
|
("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID", "OSS AccessKey ID", "仅授权该Bucket的AK。", "<AK ID>", False),
|
|
("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET", "OSS AccessKey Secret", "仅授权该Bucket的SK。", "<AK Secret>", True),
|
|
("DOBOTAI_OUTBOUND_IMAGE_PUBLIC_BASE_URL", "图片稳定外发基础地址", "钉钉可见的稳定HTTPS入口，不含凭据。", "https://host/dingtalk/i", False),
|
|
)
|
|
|
|
def prompt_oss(values):
|
|
"""OSS整组录入：全部留空即禁用；任一项填写则六项必填。"""
|
|
print("\n回答图片外发OSS（可选，OSS整组录入；直接回车表示整组禁用）")
|
|
entered = {}
|
|
for key, label, description, example, secret in OSS_FIELDS:
|
|
default = values.get(key, "").strip()
|
|
suffix = "（已配置，直接回车保留）" if default else ""
|
|
print(f"用途：{description} 格式示例：{example}")
|
|
reader = getpass.getpass if secret else input
|
|
value = reader(f"{label}{suffix}: ").strip()
|
|
entered[key] = value or default
|
|
if not any(entered.values()):
|
|
for key, *_ in OSS_FIELDS:
|
|
values.pop(key, None)
|
|
return
|
|
missing = [label for key, label, *_ in OSS_FIELDS if not entered[key]]
|
|
if missing:
|
|
raise SystemExit("OSS为整组可选能力，以下项未填写：" + "、".join(missing))
|
|
values.update(entered)
|
|
|
|
|
|
|
|
def parse_env(path):
|
|
lines = path.read_text(encoding="utf-8").splitlines(); values = {}
|
|
for line in lines:
|
|
if line and not line.lstrip().startswith("#") and "=" in line:
|
|
key, value = line.split("=", 1); values[key.strip()] = value
|
|
return lines, values
|
|
|
|
def write_env(path, lines, values):
|
|
rendered=[]; written=set()
|
|
for line in lines:
|
|
if line and not line.lstrip().startswith("#") and "=" in line:
|
|
key=line.split("=",1)[0].strip()
|
|
if key in values: rendered.append(f"{key}={values[key]}"); written.add(key); continue
|
|
rendered.append(line)
|
|
rendered.extend(f"{key}={value}" for key,value in values.items() if key not in written)
|
|
temp=path.with_name(path.name+".new"); temp.write_text("\n".join(rendered)+"\n",encoding="utf-8"); os.chmod(temp,0o600); os.replace(temp,path)
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description="配置DobotAI阿里云RDS和渠道凭证")
|
|
parser.add_argument("--env-file",required=True); parser.add_argument("--schema-script",required=True); parser.add_argument("--start-script",required=True)
|
|
args=parser.parse_args(); path=Path(args.env_file)
|
|
if not path.is_file(): raise SystemExit("DobotAI环境文件不存在，请先执行根目录install-docker.sh。")
|
|
lines,values=parse_env(path)
|
|
prompt_rds(values)
|
|
prompt_oss(values)
|
|
for key,label,description,example,secret in FIELDS:
|
|
suffix="（已配置，直接回车保留）" if values.get(key,"").strip() else "（必填）"
|
|
print(f"\n{label}\n用途：{description}\n格式示例：{example}")
|
|
entered=(getpass.getpass if secret else input)(f"{label}{suffix}: ").strip()
|
|
if entered: values[key]=entered
|
|
missing=[label for key,label,*_ in FIELDS if not values.get(key,"").strip()]
|
|
if missing: raise SystemExit("以下必填项未配置："+"、".join(missing))
|
|
if values["DOBOTAI_QUICK_AGENT_ID"] == values["DOBOTAI_REASONING_AGENT_ID"]:
|
|
raise SystemExit("快速问答Agent ID与智能推理Agent ID必须不同。")
|
|
values["DOBOTAI_AGENT_BINDINGS"] = json.dumps({"dingtalk:*": values["DOBOTAI_QUICK_AGENT_ID"]}, separators=(",", ":"))
|
|
write_env(path,lines,values)
|
|
subprocess.run([args.schema_script], check=True)
|
|
os.execv(args.start_script,[args.start_script])
|
|
if __name__=="__main__":
|
|
try: main()
|
|
except KeyboardInterrupt: sys.exit(130)
|