TaurixIT/simplex-customerservicebot
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working control channel invite

Browse Source
This commit is contained in:
Guy Van Sanden 2025-08-31 21:11:36 +02:00
parent f26e875613
commit 2cbb8c1e01
1 changed files with 122 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
src/simplex_customerservicebot.py
View File

@ -5,14 +5,17 @@ import os
import re
import uuid
import websockets
from collections import defaultdict
from typing import Any, Dict, Iterable, Tuple
__version__ = "0.5.1" # parse chatItem.content + deep debug + 'join ... as <alias>'
__version__ = "0.5.5" # fix announce typo, keep robust sender + case-insensitive join
# Defaults match your setup (no env needed)
# Defaults match your setup
WS_URL = os.environ.get("SXC_WS", "ws://127.0.0.1:5225")
CONTROL_GROUP = (os.environ.get("SXC_CONTROL_GROUP") or "BotTest").strip()
ANNOUNCE_NEW = os.environ.get("SXC_ANNOUNCE", "1").strip() != "0"
DEBUG = os.environ.get("SXC_DEBUG", "1").strip() == "1"
LOG_RAW = os.environ.get("SXC_LOG_RAW", "0").strip() == "1"
INVITE_CMDS = ["/a", "a"] # try WS-style first, then CLI-style
JOIN_RE = re.compile(r"^\s*/?\s*join\s+(.+?)(?:\s+as\s+(\S+))?\s*$", re.IGNORECASE)
@ -24,8 +27,11 @@ async def say(ws, group: str, text: str):
if group:
await ws.send(cmd(f"#{group} {text}"))
# Cache: gid -> {"name": str, "customerId": str}
groups = {}
# Caches
# gid -> {"name": str, "customerId": str}
groups: Dict[int, Dict[str, str]] = {}
# gid -> { groupMemberId(int) : alias(str) }
alias_map: Dict[int, Dict[int, str]] = defaultdict(dict)
def _first(*vals):
for v in vals:
@ -33,6 +39,13 @@ def _first(*vals):
return v
return ""
def _shorten(obj, limit=1600):
try:
s = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
except Exception:
s = str(obj)
return s if len(s) <= limit else s[:limit] + " …"
def extract_business_chat(resp: dict):
"""Return (gid, name, customerId) when a business chat is reported."""
if resp.get("type") == "acceptingBusinessRequest":
@ -60,7 +73,7 @@ def _content_and_meta(it: dict):
return cont, meta
def extract_text_from_item(it: dict) -> str:
"""Be liberal: support msgContent, direct text/markdown, and meta.itemText fallback."""
"""Support msgContent, direct text/markdown, and meta.itemText fallback."""
cont, meta = _content_and_meta(it)
# 1) Standard nested message content
@ -88,11 +101,53 @@ def extract_text_from_item(it: dict) -> str:
return ""
def extract_group_text_items(resp: dict):
def deep_iter_members(obj: Any) -> Iterable[dict]:
"""Recursively yield any dicts that look like groupMember objects."""
if isinstance(obj, dict):
if ("groupMemberId" in obj) and ("memberProfile" in obj or "localDisplayName" in obj):
yield obj
for v in obj.values():
yield from deep_iter_members(v)
elif isinstance(obj, list):
for v in obj:
yield from deep_iter_members(v)
def update_alias_map_from_member(gid: int, gm: dict):
"""Capture alias against groupMemberId whenever we see a member object."""
try:
gmid = gm.get("groupMemberId")
except Exception:
gmid = None
mp = (gm.get("memberProfile") or {})
alias = _first(mp.get("localAlias"), gm.get("localDisplayName"), mp.get("displayName"))
if isinstance(gid, int) and isinstance(gmid, int) and alias:
alias_map.setdefault(gid, {})[gmid] = alias
if DEBUG:
print(f"[alias-map] gid={gid} gmid={gmid} alias='{alias}'")
def _direction_from(it: dict) -> str:
chatItem = it.get("chatItem") or {}
chatdir = it.get("chatDir") or chatItem.get("chatDir") or {}
dtyp = (chatdir.get("type") or "").lower() # 'grouprcv' or 'groupsnd'
return "rcv" if dtyp == "grouprcv" else "snd"
def _primary_group_member(it: dict) -> dict:
"""Prefer the member embedded under chatItem.chatDir.groupMember for sender."""
chatItem = it.get("chatItem") or {}
chatDir = chatItem.get("chatDir") or {}
gm = chatDir.get("groupMember")
if gm:
return gm
gm = chatItem.get("groupMember")
if gm:
return gm
gm = it.get("groupMember")
return gm or {}
def extract_group_items(resp: dict) -> Iterable[Tuple[str, int, str, str, Any, str, str, dict]]:
"""
Yield tuples for each candidate message in newChatItems:
(group_name, group_id, direction, sender_alias, text, ctype)
We still yield when text == "" so we can debug items that didn't parse.
Yield tuples:
(group_name, group_id, direction, sender_alias, sender_gmid, text, ctype, raw_item)
"""
if resp.get("type") != "newChatItems":
return
@ -106,13 +161,18 @@ def extract_group_text_items(resp: dict):
if not gname or gid is None:
continue
# chatDir may appear at item root or under chatItem
chatdir = it.get("chatDir") or (it.get("chatItem") or {}).get("chatDir") or {}
dtyp = (chatdir.get("type") or "").lower() # 'grouprcv' or 'groupsnd'
direction = "rcv" if dtyp == "grouprcv" else "snd"
direction = _direction_from(it)
# Sender alias can appear at item root or under chatItem
gm = it.get("groupMember") or (it.get("chatItem") or {}).get("groupMember") or {}
# Deep-scan to populate alias map from any member objects
for gm in deep_iter_members(it):
update_alias_map_from_member(gid, gm)
# Extract sender from preferred location
gm = _primary_group_member(it)
try:
gmid = gm.get("groupMemberId")
except Exception:
gmid = None
mp = (gm.get("memberProfile") or {})
sender_alias = _first(mp.get("localAlias"), gm.get("localDisplayName"), mp.get("displayName"))
@ -120,7 +180,7 @@ def extract_group_text_items(resp: dict):
ctype = cont.get("type")
text = extract_text_from_item(it)
yield (gname, gid, direction, sender_alias, text, ctype, it)
yield (gname, gid, direction, sender_alias, gmid, text, ctype, it)
def is_control_group(name: str) -> bool:
return name == CONTROL_GROUP
@ -135,17 +195,28 @@ def resolve_chat_name(user_supplied: str) -> str:
cand = [info["name"] for info in groups.values() if info.get("name", "").startswith(user_supplied)]
return cand[0] if len(cand) == 1 else user_supplied
async def handle_control_command(ws, group_name: str, sender_alias: str, text: str):
def resolve_sender_alias(gid: int, gmid, fallback_alias: str) -> str:
"""Use explicit alias if present; else look up by groupMemberId; else return empty."""
if fallback_alias:
return fallback_alias
if isinstance(gid, int) and isinstance(gmid, int):
alias = alias_map.get(gid, {}).get(gmid) or ""
if alias and DEBUG:
print(f"[alias-resolved] gid={gid} gmid={gmid} -> '{alias}'")
return alias
return ""
async def handle_control_command(ws, group_name: str, gid: int, sender_alias: str, sender_gmid, text: str):
m = JOIN_RE.match(text or "")
if not m:
return
# Allow 'join <chat> as <alias>' to override when the daemon doesn't expose sender alias
# Allow 'join <chat> as <alias>' to override when daemon doesn't expose alias
raw_chat = (m.group(1) or "").strip()
override_alias = (m.group(2) or "").strip()
target_alias = override_alias or sender_alias
if not target_alias:
eff_alias = override_alias or resolve_sender_alias(gid, sender_gmid, sender_alias)
if not eff_alias:
await say(ws, group_name, "Cannot resolve your contact alias. Use a token alias or type: join <chat> as <your_alias>")
return
@ -155,26 +226,21 @@ async def handle_control_command(ws, group_name: str, sender_alias: str, text: s
return
for c in INVITE_CMDS:
invite = f"{c} {chatname} {target_alias}" # no quotes
invite = f"{c} {chatname} {eff_alias}" # no quotes
if DEBUG:
print(f"[join] trying: {invite}")
await ws.send(cmd(invite))
await say(ws, group_name, f"Invited {target_alias} to {chatname}")
await say(ws, group_name, f"Invited {eff_alias} to {chatname}")
async def announce_new_chat(ws, gname: str):
if ANNOUNCE_NEW and CONTROL_GROUP:
if DEBUG:
print(f"[announce] posting new business chat '{gname}' to control group '{CONTROL_GROUP}'")
await say(ws, CONTROL_GROUP, f"New business chat: {gname} — type: join {gname}")
def _shorten(obj, limit=1400):
try:
s = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
except Exception:
s = str(obj)
return s if len(s) <= limit else s[:limit] + " …"
async def run():
print(f"[cfg] WS={WS_URL} CONTROL_GROUP={CONTROL_GROUP} ANNOUNCE={ANNOUNCE_NEW} DEBUG={DEBUG}")
print(f"[cfg] WS={WS_URL} CONTROL_GROUP={CONTROL_GROUP} ANNOUNCE={ANNOUNCE_NEW} DEBUG={DEBUG} LOG_RAW={LOG_RAW}")
print("[cfg] Auto-invite: DISABLED")
while True:
try:
@ -192,26 +258,37 @@ async def run():
resp = msg.get("resp") or {}
rtype = resp.get("type")
if LOG_RAW and rtype:
print(f"[EVT] {rtype}: {_shorten(resp)}")
# 1) Detect and announce new business chats
gid, gname, cust = extract_business_chat(resp)
if isinstance(gid, int) and gname and cust:
if gid not in groups:
groups[gid] = {"name": gname, "customerId": cust}
print(f"[chat] business chat gid={gid} name='{gname}' customerId={cust}")
await announce_new_chat(ws, gname)
# Learn aliases from join events
if rtype == "joinedGroupMember":
try:
ginfo = resp.get("groupInfo") or {}
gid = ginfo.get("groupId")
member = resp.get("member") or {}
update_alias_map_from_member(gid, member)
except Exception:
pass
# 2) Inspect all group items and log them; process control commands
# Detect and announce new business chats
gid_b, gname_b, cust = extract_business_chat(resp)
if isinstance(gid_b, int) and gname_b and cust:
if gid_b not in groups:
groups[gid_b] = {"name": gname_b, "customerId": cust} # fixed key
print(f"[chat] business chat gid={gid_b} name='{gname_b}' customerId={cust}")
await announce_new_chat(ws, gname_b)
# Inspect group items and process control commands
if rtype == "newChatItems":
for gname2, gid2, direction, sender_alias, text, ctype, raw_item in extract_group_text_items(resp):
for gname2, gid2, direction, sender_alias, sender_gmid, text, ctype, raw_item in extract_group_items(resp):
if DEBUG:
print(f"[item] group='{gname2}' gid={gid2} dir={direction} sender='{sender_alias}' ctype='{ctype}' text={text!r}")
if direction == "rcv" and is_control_group(gname2):
if not text:
# dump raw for this unparsed control message
print(f"[item] group='{gname2}' gid={gid2} dir={direction} sender='{sender_alias}' gmid={sender_gmid} ctype='{ctype}' text={text!r}")
if direction == "rcv" and gname2 == CONTROL_GROUP:
if (not sender_alias) and (sender_gmid is None) and DEBUG:
print(f"[item-raw] {_shorten(raw_item)}")
else:
await handle_control_command(ws, gname2, sender_alias, text)
if text:
await handle_control_command(ws, gname2, gid2, sender_alias, sender_gmid, text)
except Exception as e:
print(f"[ws] {e}; retrying in 2s")