131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
from dataclasses import dataclass
|
|
from typing import ClassVar, Any
|
|
from urllib.parse import urlencode
|
|
from time import sleep
|
|
|
|
from requests import get
|
|
|
|
@dataclass
|
|
class GroupMe:
|
|
"""
|
|
The class implementing a GroupMe api client.
|
|
|
|
A very thin wrapper at best
|
|
"""
|
|
|
|
http_api_url: ClassVar[str] = "https://api.groupme.com/v3"
|
|
api_token: str
|
|
latest_message_ids = {}
|
|
iteration_to_return = []
|
|
|
|
def _get(self, url: str, params: dict[str, str] = {}):
|
|
token_qs = {"token": self.api_token}
|
|
params = {**token_qs, **params} # combine the dicts
|
|
|
|
url = "{}{}?{}".format(self.http_api_url, url, urlencode(params))
|
|
|
|
print(url)
|
|
|
|
return get(url)
|
|
|
|
def _update_latest(self, group_id: str, request: list) -> None:
|
|
"""
|
|
Update the latest messages dictionary with a list of JSON-encoded
|
|
message objects. It's the stuff you get from the /groups/{}/messages
|
|
endpoint.
|
|
"""
|
|
|
|
try:
|
|
messages = request["messages"]
|
|
except KeyError:
|
|
return
|
|
|
|
latest_message = 0
|
|
latest_message_id = None
|
|
for message in messages:
|
|
created_at = int(message["created_at"])
|
|
|
|
if created_at > latest_message:
|
|
latest_message = created_at
|
|
latest_message_id = message["id"]
|
|
|
|
if latest_message_id:
|
|
self.latest_message_ids[group_id] = latest_message_id
|
|
|
|
@property
|
|
def groups(self):
|
|
r = self._get("/groups")
|
|
return r.json()["response"]
|
|
|
|
@property
|
|
def chats(self):
|
|
r = self._get("/chats")
|
|
return r.json()["response"]
|
|
|
|
@property
|
|
def grouplikes(self):
|
|
return self.groups + self.chats
|
|
|
|
def get_messages_from_group(self, group_id: str):
|
|
r = self._get("/groups/{}/messages".format(group_id))
|
|
r = r.json()["response"]
|
|
|
|
self._update_latest(group_id, r)
|
|
return r
|
|
|
|
def get_messages_from_group_after_message_id(self, group_id: str, message_id: str):
|
|
if not message_id:
|
|
return self.get_messages_from_group(group_id)
|
|
|
|
r = self._get("/groups/{}/messages".format(group_id), {"since_id": message_id})
|
|
if r.status_code == 304: # none updated yet
|
|
return {
|
|
"messages": []
|
|
}
|
|
r = r.json()["response"]
|
|
|
|
self._update_latest(group_id, r)
|
|
return r
|
|
|
|
def get_new_messages_from_group(self, group_id: str):
|
|
return self.get_messages_from_group_after_message_id(
|
|
group_id,
|
|
self.latest_message_ids[group_id]
|
|
)
|
|
|
|
def get_all_new_messages(self):
|
|
messages = []
|
|
for i in self.groups:
|
|
try:
|
|
last_message_id = self.latest_message_ids[i["id"]]
|
|
except KeyError:
|
|
last_message_id = None
|
|
|
|
messages_to_append = self.get_messages_from_group_after_message_id(i["id"], last_message_id)["messages"]
|
|
for index, message in enumerate(messages_to_append):
|
|
messages_to_append[index]["group_name"] = i["name"]
|
|
|
|
messages += messages_to_append
|
|
|
|
return messages
|
|
|
|
def __iter__(self): # iterate to get new messages
|
|
for group in self.groups:
|
|
# we don't want to return past messages
|
|
self.get_messages_from_group(group["id"])
|
|
|
|
return self
|
|
|
|
def __next__(self):
|
|
# check if there's any messages on the stack
|
|
if self.iteration_to_return:
|
|
return self.iteration_to_return.pop()
|
|
|
|
# else scan for new messages
|
|
new = []
|
|
while not new:
|
|
sleep(10)
|
|
new = self.get_all_new_messages()
|
|
|
|
self.iteration_to_return = new
|
|
return self.iteration_to_return.pop() |