Spaces:
Runtime error
Runtime error
File size: 4,872 Bytes
46252cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """Groups resource — WhatsApp group management.
Backed by ``src/modules/group/group.controller.ts``.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, TypedDict
from .._http import quote_segment
from ..types import (
CreateGroupRequest,
GroupInfo,
GroupSettings,
GroupSummary,
InviteCodeResponse,
JoinGroupRequest,
JoinGroupResponse,
SuccessResult,
)
if TYPE_CHECKING:
from .._http import HttpExecutor
class ListGroupsQuery(TypedDict, total=False):
limit: int
offset: int
class GroupsResource:
def __init__(self, http: "HttpExecutor") -> None:
self._http = http
def list(self, session_id: str, query: ListGroupsQuery | None = None) -> list[GroupSummary]:
return self._http.request("GET", f"/api/sessions/{quote_segment(session_id)}/groups", query=query)
def get(self, session_id: str, group_id: str) -> GroupInfo:
return self._http.request("GET", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}")
def create(self, session_id: str, body: CreateGroupRequest) -> GroupInfo:
return self._http.request("POST", f"/api/sessions/{quote_segment(session_id)}/groups", body=body)
def add_participants(self, session_id: str, group_id: str, participants: list[str]) -> SuccessResult:
return self._http.request(
"POST", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/participants",
body={"participants": participants},
)
def remove_participants(self, session_id: str, group_id: str, participants: list[str]) -> SuccessResult:
return self._http.request(
"DELETE", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/participants",
body={"participants": participants},
)
def promote_participants(self, session_id: str, group_id: str, participants: list[str]) -> SuccessResult:
return self._http.request(
"POST", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/participants/promote",
body={"participants": participants},
)
def demote_participants(self, session_id: str, group_id: str, participants: list[str]) -> SuccessResult:
return self._http.request(
"POST", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/participants/demote",
body={"participants": participants},
)
def set_subject(self, session_id: str, group_id: str, subject: str) -> SuccessResult:
return self._http.request(
"PUT", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/subject", body={"subject": subject}
)
def set_description(self, session_id: str, group_id: str, description: str) -> SuccessResult:
return self._http.request(
"PUT", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/description",
body={"description": description},
)
def leave(self, session_id: str, group_id: str) -> SuccessResult:
return self._http.request("POST", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/leave")
def invite_code(self, session_id: str, group_id: str) -> InviteCodeResponse:
return self._http.request(
"GET", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/invite-code"
)
def revoke_invite_code(self, session_id: str, group_id: str) -> InviteCodeResponse:
return self._http.request(
"POST", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/invite-code/revoke"
)
def join_group(self, session_id: str, body: JoinGroupRequest) -> JoinGroupResponse:
"""Join a group via an invite code. Requires an OPERATOR-level key."""
return self._http.request("POST", f"/api/sessions/{quote_segment(session_id)}/groups/join", body=body)
def get_group_settings(self, session_id: str, group_id: str) -> GroupSettings:
"""Read the group's announce/locked/ephemeral settings."""
return self._http.request(
"GET", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/settings"
)
def update_group_settings(self, session_id: str, group_id: str, body: GroupSettings) -> SuccessResult:
"""Update group settings — at least one of announce/locked/ephemeralSeconds is required.
Requires an OPERATOR-level key. ``ephemeralSeconds`` is unsupported on the
whatsapp-web.js engine (the request then fails with 501).
"""
return self._http.request(
"PUT", f"/api/sessions/{quote_segment(session_id)}/groups/{quote_segment(group_id)}/settings", body=body
)
|