Kris
7215418610
新增企业微信、微博、WhatsApp、Workplace 四个渠道扩展。 企业微信渠道扩展主要模块：config, gateway, webhook, webhook_bot, outbound, streaming, pairing, security, crypto, dedupe, persistent_dedupe, card, directory, events, externalcontact, media, mentions, menu, message, oauth, status 微博渠道扩展主要模块：config, gateway, webhook, outbound, streaming, pairing, security, dedupe, passive_reply, broadcast, message, menu, media, subscription, template, status WhatsApp 渠道扩展主要模块：config, gateway, webhook, outbound, streaming, pairing, security, dedupe, actions, monitor, status Workplace 渠道扩展主要模块：config, gateway, webhook, outbound, streaming, pairing, security, dedupe, actions, challenge, groups, media, mentions, menu, monitor, persona, quick_reply, signature, subscriptions, template, threading, users, status
81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
import hashlib
|
|
import hmac
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _compute_appsecret_proof(access_token: str, app_secret: str) -> str:
|
|
return hmac.new(
|
|
app_secret.encode("utf-8"),
|
|
access_token.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
class WorkplacePersonaClient:
|
|
def __init__(self, page_id: str, access_token: str, api_version: str = "v24.0", app_secret: str = ""):
|
|
self._page_id = page_id
|
|
self._access_token = access_token
|
|
self._api_version = api_version
|
|
self._app_secret = app_secret
|
|
|
|
async def list_personas(self) -> list[dict]:
|
|
url = f"https://graph.facebook.com/{self._api_version}/{self._page_id}/personas"
|
|
params = {"access_token": self._access_token}
|
|
if self._app_secret:
|
|
params["appsecret_proof"] = _compute_appsecret_proof(self._access_token, self._app_secret)
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
url,
|
|
params=params,
|
|
)
|
|
if resp.status_code == 200:
|
|
return resp.json().get("data", [])
|
|
logger.error("persona list failed: %d", resp.status_code)
|
|
return []
|
|
except Exception:
|
|
logger.exception("persona list error")
|
|
return []
|
|
|
|
async def create_persona(self, name: str, profile_picture_url: str) -> str | None:
|
|
url = f"https://graph.facebook.com/{self._api_version}/{self._page_id}/personas"
|
|
params = {"access_token": self._access_token}
|
|
if self._app_secret:
|
|
params["appsecret_proof"] = _compute_appsecret_proof(self._access_token, self._app_secret)
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
url,
|
|
params=params,
|
|
json={"name": name, "profile_picture_url": profile_picture_url},
|
|
)
|
|
if resp.status_code == 200:
|
|
persona_id = resp.json().get("id", "")
|
|
logger.info("persona created: %s", persona_id)
|
|
return persona_id
|
|
logger.error("persona create failed: %d %s", resp.status_code, resp.text[:200])
|
|
return None
|
|
except Exception:
|
|
logger.exception("persona create error")
|
|
return None
|
|
|
|
async def delete_persona(self, persona_id: str) -> bool:
|
|
url = f"https://graph.facebook.com/{self._api_version}/{self._page_id}/personas/{persona_id}"
|
|
params = {"access_token": self._access_token}
|
|
if self._app_secret:
|
|
params["appsecret_proof"] = _compute_appsecret_proof(self._access_token, self._app_secret)
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.delete(
|
|
url,
|
|
params=params,
|
|
)
|
|
return resp.status_code == 200
|
|
except Exception:
|
|
logger.exception("persona delete error")
|
|
return False
|