108 lines
6.3 KiB
Python
108 lines
6.3 KiB
Python
"""Create the local demonstration through the application's real HTTP API."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
LOCAL = Path(__file__).resolve().parent
|
|
STATE = LOCAL / ".state"
|
|
API = "http://127.0.0.1:8018/api/v1/"
|
|
|
|
|
|
def checked(response: httpx.Response) -> dict | list:
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def main() -> None:
|
|
accounts = json.loads((STATE / "test-accounts.json").read_text())
|
|
with httpx.Client(base_url=API, timeout=600, trust_env=False) as client:
|
|
admin = accounts["admin"]
|
|
login = checked(client.post("users/login", data={"username": admin["email"], "password": admin["password"]}))
|
|
client.headers["Authorization"] = "Bearer " + login["access_token"]
|
|
org_id = login["user"]["organization_id"]
|
|
hours = {day: {"start": "00:00", "end": "23:59", "enabled": True} for day in (
|
|
"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"
|
|
)}
|
|
checked(client.patch(f"organizations/{org_id}", json={"business_hours": hours}))
|
|
roles = checked(client.get("roles"))
|
|
role_id = next(role["id"] for role in roles if role["name"] == "Agent")
|
|
users = checked(client.get("users"))
|
|
staff = []
|
|
for key in ("agent_a", "agent_b"):
|
|
account = accounts[key]
|
|
user = next((item for item in users if item["email"] == account["email"]), None)
|
|
if user is None:
|
|
user = checked(client.post("users", json={
|
|
"email": account["email"], "full_name": account["name"],
|
|
"password": account["password"], "role_id": role_id,
|
|
"see_all_ai_chats": False, "see_all_org_chats": False,
|
|
}))
|
|
accounts[key]["id"] = user["id"]
|
|
staff.append(user)
|
|
|
|
groups = checked(client.get("groups"))
|
|
group = next((item for item in groups if item["name"] == "Local Support"), None)
|
|
if group is None:
|
|
group = checked(client.post("groups", json={"name": "Local Support"}))
|
|
members = {user["id"] for user in group.get("users", [])}
|
|
for user in staff:
|
|
if user["id"] not in members:
|
|
checked(client.post(f"groups/{group['id']}/users/{user['id']}"))
|
|
|
|
agents = checked(client.get("agent/list"))
|
|
agent = next((item for item in agents if item["name"] == "网站客服演示"), None)
|
|
if agent is None:
|
|
agent = checked(client.post("agent", json={
|
|
"name": "网站客服演示", "agent_type": "customer_support",
|
|
"instructions": [
|
|
"你是本地客服测试助手，全程使用简体中文，回答简短清楚。",
|
|
"涉及青禾测试套餐、退款期限、校验码时，必须先查询知识库，不得凭常识编造。",
|
|
"用户明确要求转人工、真人或客服人员时，立即转人工，不要继续追问。",
|
|
"本环境的政策、套餐和校验码都是虚构测试数据，不代表真实公司的业务。",
|
|
],
|
|
"ai_replies_enabled": True, "transfer_to_human": True,
|
|
"ask_for_rating": True, "handoff_collect_email": False,
|
|
"handoff_collect_name": False, "ticketing_enabled": False,
|
|
"allow_attachments": True,
|
|
}))
|
|
checked(client.put(f"agent/{agent['id']}/groups", json=[group["id"]]))
|
|
checked(client.post(f"agent/{agent['id']}/customization", json={
|
|
"chat_style": "CHATBOT", "collect_email": False,
|
|
"accent_color": "#2563eb", "welcome_title": "网站智能客服",
|
|
"welcome_subtitle": "AI 解答 · 人工接待 · 服务评价",
|
|
"welcome_message": "你好！我是测试客服，可以查询演示知识库，也可以帮你转人工。",
|
|
"quick_actions": [], "chat_initiation_messages": [], "allow_new_chat": True,
|
|
}))
|
|
widgets = checked(client.get(f"widgets/agent/{agent['id']}"))
|
|
widget = widgets[0] if widgets else checked(client.post("widgets", json={"name": "Local Website", "agent_id": agent["id"]}))
|
|
environment = {"org_id": org_id, "agent_id": agent["id"], "widget_id": widget["id"],
|
|
"group_id": group["id"], "staff": [{"id": user["id"], "email": user["email"]} for user in staff],
|
|
"demo_url": "http://127.0.0.1:3018/local-demo.html"}
|
|
(STATE / "test-accounts.json").write_text(json.dumps(accounts, ensure_ascii=False, indent=2))
|
|
(STATE / "test-environment.json").write_text(json.dumps(environment, ensure_ascii=False, indent=2))
|
|
template = (LOCAL / "demo.template.html").read_text()
|
|
(STATE / "dashboard/local-demo.html").write_text(template.replace("__WIDGET_ID__", widget["id"]))
|
|
print("Created two scoped human agents, a support group, and the website widget.", flush=True)
|
|
|
|
existing = checked(client.get(f"knowledge/agent/{agent['id']}"))["knowledge"]
|
|
title = "青禾测试套餐演示规则"
|
|
if not any(item.get("name") == title or item.get("source") == title for item in existing):
|
|
print("Indexing the demonstration knowledge source; the first run downloads the embedding model.", flush=True)
|
|
result = checked(client.post("knowledge/add/text", json={
|
|
"org_id": org_id, "agent_id": agent["id"], "title": title,
|
|
"content": "这是一份技术验证专用的虚构知识库，不代表任何真实公司的政策。\n"
|
|
"青禾测试套餐的退款期限为购买后17天内；客服校验码为QH-7319。\n"
|
|
"申请退款需要提供演示订单编号，不需要提供银行卡密码。\n"
|
|
"人工客服可以处理演示订单核对、退款申请和复杂问题；用户要求人工时应立即转接。\n"
|
|
"服务完成后，访客可以提交1到5星评分和文字评价，管理员可以查看记录。",
|
|
}))
|
|
environment["knowledge_id"] = result["knowledge"]["id"]
|
|
(STATE / "test-environment.json").write_text(json.dumps(environment, ensure_ascii=False, indent=2))
|
|
print("Local demonstration is ready:", environment["demo_url"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|