353 lines
11 KiB
Python
353 lines
11 KiB
Python
"""Chatwoot Agent Bot webhook → MaxKB（PAI）→ 回寫對話。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.request import Request, urlopen
|
||
|
||
from config import CONFIG
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_SESSION_LOCK = threading.Lock()
|
||
_SESSIONS: dict[str, dict[str, Any]] = {}
|
||
_SESSION_FILE = Path(__file__).resolve().parent / ".maxkb_cw_sessions.json"
|
||
|
||
|
||
def _load_sessions() -> None:
|
||
if not _SESSION_FILE.exists():
|
||
return
|
||
try:
|
||
data = json.loads(_SESSION_FILE.read_text(encoding="utf-8"))
|
||
if isinstance(data, dict):
|
||
_SESSIONS.update(data)
|
||
except Exception:
|
||
logger.exception("讀取 MaxKB session 快取失敗")
|
||
|
||
|
||
def _save_sessions() -> None:
|
||
try:
|
||
_SESSION_FILE.write_text(
|
||
json.dumps(_SESSIONS, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
except Exception:
|
||
logger.exception("寫入 MaxKB session 快取失敗")
|
||
|
||
|
||
_load_sessions()
|
||
|
||
|
||
def _http_json(
|
||
method: str,
|
||
url: str,
|
||
*,
|
||
headers: dict[str, str] | None = None,
|
||
body: Any = None,
|
||
timeout: float = 120,
|
||
) -> tuple[int, Any]:
|
||
data = None
|
||
if body is not None:
|
||
data = body if isinstance(body, (bytes, bytearray)) else json.dumps(
|
||
body, ensure_ascii=False
|
||
).encode("utf-8")
|
||
req_headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||
if headers:
|
||
req_headers.update(headers)
|
||
req = Request(url, data=data, headers=req_headers, method=method)
|
||
try:
|
||
with urlopen(req, timeout=timeout) as resp:
|
||
raw = resp.read()
|
||
try:
|
||
return resp.status, json.loads(raw.decode("utf-8") or "null")
|
||
except Exception:
|
||
return resp.status, raw.decode("utf-8", errors="replace")
|
||
except HTTPError as exc:
|
||
raw = exc.read()
|
||
try:
|
||
return exc.code, json.loads(raw.decode("utf-8") or "null")
|
||
except Exception:
|
||
return exc.code, raw.decode("utf-8", errors="replace")
|
||
except URLError as exc:
|
||
raise RuntimeError(f"HTTP 連線失敗: {url}: {exc}") from exc
|
||
|
||
|
||
def _maxkb_auth() -> str:
|
||
status, body = _http_json(
|
||
"POST",
|
||
f"{CONFIG.maxkb_base_url}/chat/api/auth/anonymous",
|
||
body={"access_token": CONFIG.maxkb_access_token},
|
||
timeout=30,
|
||
)
|
||
if status != 200 or not isinstance(body, dict) or body.get("code") != 200:
|
||
raise RuntimeError(f"MaxKB anonymous auth 失敗: {status} {body}")
|
||
token = body.get("data")
|
||
if not token:
|
||
raise RuntimeError("MaxKB anonymous auth 未回傳 token")
|
||
return str(token)
|
||
|
||
|
||
def _maxkb_open(jwt: str) -> str:
|
||
status, body = _http_json(
|
||
"GET",
|
||
f"{CONFIG.maxkb_base_url}/chat/api/open",
|
||
headers={"Authorization": f"Bearer {jwt}"},
|
||
timeout=30,
|
||
)
|
||
if status != 200 or not isinstance(body, dict) or body.get("code") != 200:
|
||
raise RuntimeError(f"MaxKB open 失敗: {status} {body}")
|
||
chat_id = body.get("data")
|
||
if not chat_id:
|
||
raise RuntimeError("MaxKB open 未回傳 chat_id")
|
||
return str(chat_id)
|
||
|
||
|
||
def _maxkb_ask(
|
||
jwt: str,
|
||
chat_id: str,
|
||
message: str,
|
||
form_data: dict[str, Any],
|
||
) -> str:
|
||
status, body = _http_json(
|
||
"POST",
|
||
f"{CONFIG.maxkb_base_url}/chat/api/chat_message/{chat_id}",
|
||
headers={"Authorization": f"Bearer {jwt}"},
|
||
body={
|
||
"message": message,
|
||
"stream": False,
|
||
"re_chat": False,
|
||
"form_data": form_data,
|
||
},
|
||
timeout=CONFIG.maxkb_timeout_sec,
|
||
)
|
||
if status != 200 or not isinstance(body, dict) or body.get("code") != 200:
|
||
raise RuntimeError(f"MaxKB chat 失敗: {status} {body}")
|
||
data = body.get("data") or {}
|
||
content = ""
|
||
if isinstance(data, dict):
|
||
content = str(data.get("content") or "").strip()
|
||
if not content:
|
||
answers = data.get("answer_list") or []
|
||
if answers and isinstance(answers[0], dict):
|
||
content = str(answers[0].get("content") or "").strip()
|
||
if not content:
|
||
raise RuntimeError(f"MaxKB 回覆為空: {body}")
|
||
return content
|
||
|
||
|
||
def _get_or_create_session(conversation_key: str) -> dict[str, Any]:
|
||
with _SESSION_LOCK:
|
||
sess = _SESSIONS.get(conversation_key)
|
||
if sess and sess.get("jwt") and sess.get("chat_id"):
|
||
return dict(sess)
|
||
|
||
jwt = _maxkb_auth()
|
||
chat_id = _maxkb_open(jwt)
|
||
sess = {
|
||
"jwt": jwt,
|
||
"chat_id": chat_id,
|
||
"updated_at": int(time.time()),
|
||
}
|
||
with _SESSION_LOCK:
|
||
_SESSIONS[conversation_key] = sess
|
||
_save_sessions()
|
||
return dict(sess)
|
||
|
||
|
||
def _refresh_session(conversation_key: str) -> dict[str, Any]:
|
||
with _SESSION_LOCK:
|
||
_SESSIONS.pop(conversation_key, None)
|
||
_save_sessions()
|
||
return _get_or_create_session(conversation_key)
|
||
|
||
|
||
def _chatwoot_reply(
|
||
account_id: int,
|
||
conversation_id: int,
|
||
content: str,
|
||
) -> None:
|
||
base = (
|
||
CONFIG.chatwoot_guest_api_base
|
||
or CONFIG.chatwoot_guest_proxy_target
|
||
or "http://127.0.0.1:3000"
|
||
).rstrip("/")
|
||
token = CONFIG.chatwoot_agent_bot_token
|
||
if not token:
|
||
raise RuntimeError("未設定 CHATWOOT_AGENT_BOT_TOKEN")
|
||
url = (
|
||
f"{base}/api/v1/accounts/{account_id}/conversations/"
|
||
f"{conversation_id}/messages"
|
||
)
|
||
status, body = _http_json(
|
||
"POST",
|
||
url,
|
||
headers={"api_access_token": token},
|
||
body={
|
||
"content": content,
|
||
"message_type": "outgoing",
|
||
"private": False,
|
||
},
|
||
timeout=30,
|
||
)
|
||
if status >= 300:
|
||
raise RuntimeError(f"Chatwoot 回覆失敗: {status} {body}")
|
||
|
||
|
||
def _extract_incoming(payload: dict[str, Any]) -> dict[str, Any] | None:
|
||
event = str(payload.get("event") or "")
|
||
# Agent Bot 會送 message_created / message_updated；只處理新建且來自聯絡人的訊息
|
||
if event and event not in {"message_created"}:
|
||
return None
|
||
|
||
message = payload.get("message") or payload
|
||
if not isinstance(message, dict):
|
||
return None
|
||
|
||
# 0 / incoming = 聯絡人；1 / outgoing = 客服或 bot（避免自回）
|
||
message_type = message.get("message_type")
|
||
incoming_types = {0, "0", "incoming", "Incoming"}
|
||
if message_type not in incoming_types:
|
||
return None
|
||
if message.get("private"):
|
||
return None
|
||
|
||
content = str(message.get("content") or "").strip()
|
||
if not content:
|
||
return None
|
||
|
||
conversation = message.get("conversation") or payload.get("conversation") or {}
|
||
if not isinstance(conversation, dict):
|
||
conversation = {}
|
||
conversation_id = conversation.get("id") or message.get("conversation_id")
|
||
account = payload.get("account") if isinstance(payload.get("account"), dict) else {}
|
||
account_id = (
|
||
message.get("account_id")
|
||
or conversation.get("account_id")
|
||
or account.get("id")
|
||
or CONFIG.chatwoot_account_id
|
||
)
|
||
if not conversation_id:
|
||
return None
|
||
|
||
sender = message.get("sender") or {}
|
||
username = ""
|
||
if isinstance(sender, dict):
|
||
username = str(
|
||
sender.get("name")
|
||
or sender.get("identifier")
|
||
or (sender.get("additional_attributes") or {}).get("username")
|
||
or ""
|
||
)
|
||
|
||
return {
|
||
"account_id": int(account_id),
|
||
"conversation_id": int(conversation_id),
|
||
"content": content,
|
||
"username": username,
|
||
"message_id": message.get("id"),
|
||
}
|
||
|
||
|
||
def _hard_reply_for_complaint(message: str) -> str | None:
|
||
"""客诉/提款强制话术，避免意图误判。"""
|
||
text = "".join(str(message or "").split())
|
||
if "一次不给赢" in text:
|
||
return "亲，每次游戏都会有输有赢，关键是保持好心态，接受挑战，最终会有丰厚的回报！"
|
||
# YD-CUSTOMER-018
|
||
if "回归" in text and "杀" in text:
|
||
return (
|
||
"亲，游戏输输赢赢很正常的，拼的是智力、耐心、心态和运气，"
|
||
"如果您觉得手气不好，可以稍作休息，转移一下注意力，稍后再进行投注，"
|
||
"或许下一个大奖就是您的，祝您鸿运当头，盈利多多。"
|
||
)
|
||
# YD-CUSTOMER-017
|
||
low = text.lower()
|
||
if (
|
||
"pg电子" in low
|
||
or "PG电子" in text
|
||
or "电子真是假" in text
|
||
or "假的我永远" in text
|
||
or ("电子" in text and "假" in text)
|
||
):
|
||
return (
|
||
"亲，您好！电子游戏的大奖奖池是根据机率来的哦，每一轮的结果都是随机的，"
|
||
"所以不能保证每次都能中奖。希望您能理解，这也是游戏的一部分乐趣所在！"
|
||
"我们可以一起再试试，说不定下一次就能碰上好运呢"
|
||
)
|
||
# YD-WITHDRAW-016（标准为转专人，禁止追问订单号）
|
||
if "取消提款" in text or "取消提现" in text or "申请取消提款" in text:
|
||
return (
|
||
"您好，取消提款需由专员协助核实处理，请稍候为您转接真人客服。"
|
||
"<quick_question>转专人服务</quick_question>"
|
||
)
|
||
return None
|
||
|
||
|
||
def handle_chatwoot_webhook(payload: dict[str, Any]) -> dict[str, Any]:
|
||
if not CONFIG.maxkb_bridge_enabled:
|
||
return {"ok": True, "skipped": "bridge_disabled"}
|
||
|
||
incoming = _extract_incoming(payload)
|
||
if not incoming:
|
||
return {"ok": True, "skipped": "not_incoming"}
|
||
|
||
account_id = incoming["account_id"]
|
||
conversation_id = incoming["conversation_id"]
|
||
key = f"{account_id}:{conversation_id}"
|
||
form_data = {
|
||
"account_id": str(account_id),
|
||
"conversation_id": str(conversation_id),
|
||
"bot_id": str(CONFIG.chatwoot_agent_bot_id or ""),
|
||
"username": incoming["username"] or "游客",
|
||
}
|
||
|
||
hard = _hard_reply_for_complaint(incoming["content"])
|
||
if hard:
|
||
_chatwoot_reply(account_id, conversation_id, hard)
|
||
with _SESSION_LOCK:
|
||
cur = _SESSIONS.get(key) or {}
|
||
cur["updated_at"] = int(time.time())
|
||
cur["last_message_id"] = incoming.get("message_id")
|
||
cur["hard_reply"] = True
|
||
_SESSIONS[key] = cur
|
||
_save_sessions()
|
||
return {
|
||
"ok": True,
|
||
"conversation_id": conversation_id,
|
||
"hard_reply": True,
|
||
"reply_len": len(hard),
|
||
}
|
||
|
||
sess = _get_or_create_session(key)
|
||
try:
|
||
reply = _maxkb_ask(
|
||
sess["jwt"], sess["chat_id"], incoming["content"], form_data
|
||
)
|
||
except Exception:
|
||
logger.exception("MaxKB 對話失敗，嘗試重建 session")
|
||
sess = _refresh_session(key)
|
||
reply = _maxkb_ask(
|
||
sess["jwt"], sess["chat_id"], incoming["content"], form_data
|
||
)
|
||
|
||
_chatwoot_reply(account_id, conversation_id, reply)
|
||
with _SESSION_LOCK:
|
||
cur = _SESSIONS.get(key) or {}
|
||
cur["updated_at"] = int(time.time())
|
||
cur["last_message_id"] = incoming.get("message_id")
|
||
_SESSIONS[key] = cur
|
||
_save_sessions()
|
||
|
||
return {
|
||
"ok": True,
|
||
"conversation_id": conversation_id,
|
||
"chat_id": sess["chat_id"],
|
||
"reply_len": len(reply),
|
||
}
|