257 lines
8.0 KiB
Python
257 lines
8.0 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import time
|
||
import uuid
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.request import Request, urlopen
|
||
|
||
from config import CONFIG
|
||
|
||
_AUTH_CACHE: dict[str, Any] = {"headers": None, "expires_at": 0.0}
|
||
|
||
|
||
DEFAULT_GREETING = (
|
||
"您好，欢迎光临【49彩票集团】客服服务中心，请问有什么可以帮到您的呢？"
|
||
)
|
||
|
||
|
||
class ChatwootGuestError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _request(
|
||
method: str,
|
||
path: str,
|
||
headers: dict[str, str],
|
||
payload: dict[str, Any] | None = None,
|
||
) -> tuple[int, dict[str, str], Any]:
|
||
body = None
|
||
req_headers = dict(headers)
|
||
if payload is not None:
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
req_headers["Content-Type"] = "application/json"
|
||
req_headers["Accept"] = "application/json"
|
||
|
||
api_base = (
|
||
CONFIG.chatwoot_guest_api_base or CONFIG.chatwoot_base_url
|
||
).rstrip("/")
|
||
request = Request(
|
||
f"{api_base}{path}",
|
||
data=body,
|
||
headers=req_headers,
|
||
method=method,
|
||
)
|
||
try:
|
||
with urlopen(request, timeout=20) as response:
|
||
raw = response.read().decode("utf-8", "replace")
|
||
data = json.loads(raw) if raw else {}
|
||
return response.status, dict(response.headers.items()), data
|
||
except HTTPError as exc:
|
||
raw = exc.read().decode("utf-8", "replace")
|
||
try:
|
||
data = json.loads(raw) if raw else {}
|
||
except json.JSONDecodeError:
|
||
data = {"message": raw[:300]}
|
||
raise ChatwootGuestError(
|
||
f"Chatwoot {method} {path} 失敗：{exc.code} {data}"
|
||
) from exc
|
||
except URLError as exc:
|
||
raise ChatwootGuestError(f"無法連線 Chatwoot：{exc.reason}") from exc
|
||
|
||
|
||
def _login_headers(*, force: bool = False) -> dict[str, str]:
|
||
email = CONFIG.chatwoot_guest_api_email or CONFIG.chatwoot_api_email
|
||
password = CONFIG.chatwoot_guest_api_password or CONFIG.chatwoot_api_password
|
||
if not email or not password:
|
||
raise ChatwootGuestError("未設定 Chatwoot API 帳號密碼")
|
||
|
||
now = time.time()
|
||
cached = _AUTH_CACHE.get("headers")
|
||
if not force and cached and now < float(_AUTH_CACHE.get("expires_at") or 0):
|
||
return cached
|
||
|
||
status, headers, _data = _request(
|
||
"POST",
|
||
"/auth/sign_in",
|
||
{},
|
||
{
|
||
"email": email,
|
||
"password": password,
|
||
},
|
||
)
|
||
if status >= 400:
|
||
raise ChatwootGuestError("Chatwoot 登入失敗")
|
||
|
||
token = headers.get("access-token") or headers.get("Access-Token") or ""
|
||
client = headers.get("client") or headers.get("Client") or ""
|
||
uid = headers.get("uid") or headers.get("Uid") or ""
|
||
if not token or not client or not uid:
|
||
raise ChatwootGuestError("Chatwoot 登入未回傳 token")
|
||
|
||
auth_headers = {
|
||
"access-token": token,
|
||
"client": client,
|
||
"uid": uid,
|
||
"token-type": "Bearer",
|
||
}
|
||
_AUTH_CACHE["headers"] = auth_headers
|
||
_AUTH_CACHE["expires_at"] = now + 300
|
||
return auth_headers
|
||
|
||
|
||
def _dig(data: Any, *keys: str) -> Any:
|
||
current = data
|
||
for key in keys:
|
||
if not isinstance(current, dict):
|
||
return None
|
||
current = current.get(key)
|
||
return current
|
||
|
||
|
||
def _source_id(*payloads: Any) -> str:
|
||
for payload in payloads:
|
||
if not isinstance(payload, dict):
|
||
continue
|
||
body = payload.get("payload") or payload
|
||
candidates = [
|
||
_dig(body, "contact_inbox", "source_id"),
|
||
_dig(payload, "contact_inbox", "source_id"),
|
||
_dig(
|
||
payload,
|
||
"last_non_activity_message",
|
||
"conversation",
|
||
"contact_inbox",
|
||
"source_id",
|
||
),
|
||
]
|
||
messages = body.get("messages") if isinstance(body, dict) else None
|
||
if isinstance(messages, list) and messages:
|
||
candidates.append(
|
||
_dig(
|
||
messages[0],
|
||
"conversation",
|
||
"contact_inbox",
|
||
"source_id",
|
||
)
|
||
)
|
||
for value in candidates:
|
||
if value:
|
||
return str(value)
|
||
return ""
|
||
|
||
|
||
def _contact_id(payload: Any) -> int | None:
|
||
if not isinstance(payload, dict):
|
||
return None
|
||
body = payload.get("payload") or payload
|
||
if not isinstance(body, dict):
|
||
return None
|
||
contact = body.get("contact") or body
|
||
if not isinstance(contact, dict):
|
||
return None
|
||
contact_id = contact.get("id")
|
||
return int(contact_id) if contact_id is not None else None
|
||
|
||
|
||
def create_guest_conversation() -> dict[str, Any]:
|
||
"""建立新的遊客聯絡人與對話，讓 widget 直接進入聊天室。"""
|
||
identifier = f"guest-{uuid.uuid4().hex[:16]}"
|
||
headers = _login_headers()
|
||
account_id = CONFIG.chatwoot_account_id
|
||
|
||
try:
|
||
return _create_with_headers(identifier, headers, account_id)
|
||
except ChatwootGuestError as exc:
|
||
if "401" not in str(exc) and "403" not in str(exc):
|
||
raise
|
||
headers = _login_headers(force=True)
|
||
return _create_with_headers(identifier, headers, account_id)
|
||
|
||
|
||
def _create_with_headers(
|
||
identifier: str,
|
||
headers: dict[str, str],
|
||
account_id: int,
|
||
) -> dict[str, Any]:
|
||
_, _, contact_body = _request(
|
||
"POST",
|
||
f"/api/v1/accounts/{account_id}/contacts",
|
||
headers,
|
||
{
|
||
"inbox_id": CONFIG.chatwoot_guest_inbox_id or CONFIG.chatwoot_inbox_id,
|
||
"name": "遊客",
|
||
"identifier": identifier,
|
||
"custom_attributes": {"visitor_type": "guest"},
|
||
},
|
||
)
|
||
contact_id = _contact_id(contact_body)
|
||
if not contact_id:
|
||
raise ChatwootGuestError("Chatwoot 未回傳 contact id")
|
||
|
||
source_id = _source_id(contact_body)
|
||
if not source_id:
|
||
raise ChatwootGuestError("Chatwoot 未回傳 source_id")
|
||
|
||
# 必須帶同一個 source_id。只傳 contact_id + inbox_id 時，
|
||
# ContactInboxBuilder 會再產生一組 WebWidget UUID，
|
||
# 歡迎語會掛在新 session，widget JWT 對不到就變成空白對話。
|
||
# 歡迎圖不走 Chatwoot 附件，否則 widget 會顯示檔名與「下載」。
|
||
_, _, conversation = _request(
|
||
"POST",
|
||
f"/api/v1/accounts/{account_id}/conversations",
|
||
headers,
|
||
{
|
||
"inbox_id": CONFIG.chatwoot_guest_inbox_id or CONFIG.chatwoot_inbox_id,
|
||
"contact_id": contact_id,
|
||
"source_id": source_id,
|
||
"status": "open",
|
||
"message": {
|
||
"content": CONFIG.chatwoot_greeting or DEFAULT_GREETING,
|
||
"message_type": "outgoing",
|
||
},
|
||
},
|
||
)
|
||
conversation_id = conversation.get("id") if isinstance(conversation, dict) else None
|
||
|
||
return {
|
||
"identifier": identifier,
|
||
"contact_id": contact_id,
|
||
"conversation_id": conversation_id,
|
||
"source_id": source_id,
|
||
"widget_token": make_widget_token(source_id),
|
||
"name": "遊客",
|
||
}
|
||
|
||
|
||
def make_widget_token(source_id: str) -> str:
|
||
"""產生 Chatwoot widget 的 cw_conversation JWT。"""
|
||
secret = CONFIG.chatwoot_widget_jwt_secret
|
||
if not secret:
|
||
raise ChatwootGuestError("未設定 CHATWOOT_WIDGET_JWT_SECRET")
|
||
|
||
now = int(time.time())
|
||
header = _b64url(json.dumps({"alg": "HS256"}, separators=(",", ":")).encode())
|
||
payload = _b64url(
|
||
json.dumps(
|
||
{
|
||
"source_id": source_id,
|
||
"inbox_id": CONFIG.chatwoot_guest_inbox_id or CONFIG.chatwoot_inbox_id,
|
||
"iat": now,
|
||
"exp": now + 60 * 60 * 24 * 180,
|
||
},
|
||
separators=(",", ":"),
|
||
).encode()
|
||
)
|
||
signing_input = f"{header}.{payload}".encode()
|
||
signature = hmac.new(secret.encode("utf-8"), signing_input, hashlib.sha256).digest()
|
||
return f"{header}.{payload}.{_b64url(signature)}"
|
||
|
||
|
||
def _b64url(raw: bytes) -> str:
|
||
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|