karaolidis/fbchat
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Don't mandate a specific way to handle listening events (for now)

Browse Source
This commit is contained in:
Mads Marquart
2020-03-11 12:37:34 +01:00
parent 6660fd099d
commit 6dbcb8cc47
5 changed files with 49 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
docs/intro.rst
View File

@@ -134,18 +134,19 @@ Listening & Events
Now, we are finally at the point we have all been waiting for: Creating an automatic Facebook bot!
To get started, you create a listener object::
To get started, you create the functions you want to call on certain events::
def my_function(event: fbchat.MessageEvent):
print(f"Message from {event.author.id}: {event.message.text}")
Then you create a `fbchat.Listener` object::
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
The you use that to register methods that will handle your events::
Which you can then use to receive events, and send them to your functions::
@listener.register
def on_message(event: fbchat.MessageEvent):
print(f"Message from {event.author.id}: {event.message.text}")
And then you start handling the incoming events::
listener.run()
for event in listener.listen():
if isinstance(event, fbchat.MessageEvent):
my_function(event)
View the :ref:`examples` to see some more examples illustrating the event system.

16
examples/echobot.py
View File

@@ -3,13 +3,9 @@ import fbchat
session = fbchat.Session.login("<email>", "<password>")
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
@listener.register
def on_message(event: fbchat.MessageEvent):
print(f"{event.message.text} from {event.author.id} in {event.thread.id}")
# If you're not the author, echo
if event.author.id != session.user.id:
event.thread.send_text(event.message.text)
listener.run()
for event in listener.listen():
if isinstance(event, fbchat.MessageEvent):
print(f"{event.message.text} from {event.author.id} in {event.thread.id}")
# If you're not the author, echo
if event.author.id != session.user.id:
event.thread.send_text(event.message.text)

41
examples/keepbot.py
View File

@@ -1,4 +1,7 @@
# This example uses the `blinker` library to dispatch events. See echobot.py for how
# this could be done differenly. The decision is entirely up to you!
import fbchat
import blinker
# Change this to your group id
old_thread_id = "1234567890"
@@ -14,12 +17,12 @@ old_nicknames = {
"12345678904": "User nr. 4's nickname",
}
session = fbchat.Session.login("<email>", "<password>")
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
# Create a blinker signal
events = blinker.Signal()
@listener.register
def on_color_set(event: fbchat.ColorSet):
# Register various event handlers on the signal
@events.connect_via(fbchat.ColorSet)
def on_color_set(sender, event: fbchat.ColorSet):
if old_thread_id != event.thread.id:
return
if old_color != event.color:
@@ -27,8 +30,8 @@ def on_color_set(event: fbchat.ColorSet):
event.thread.set_color(old_color)
@listener.register
def on_emoji_set(event: fbchat.EmojiSet):
@events.connect_via(fbchat.EmojiSet)
def on_emoji_set(sender, event: fbchat.EmojiSet):
if old_thread_id != event.thread.id:
return
if old_emoji != event.emoji:
@@ -36,8 +39,8 @@ def on_emoji_set(event: fbchat.EmojiSet):
event.thread.set_emoji(old_emoji)
@listener.register
def on_title_set(event: fbchat.TitleSet):
@events.connect_via(fbchat.TitleSet)
def on_title_set(sender, event: fbchat.TitleSet):
if old_thread_id != event.thread.id:
return
if old_title != event.title:
@@ -45,8 +48,8 @@ def on_title_set(event: fbchat.TitleSet):
event.thread.set_title(old_title)
@listener.register
def on_nickname_set(event: fbchat.NicknameSet):
@events.connect_via(fbchat.NicknameSet)
def on_nickname_set(sender, event: fbchat.NicknameSet):
if old_thread_id != event.thread.id:
return
old_nickname = old_nicknames.get(event.subject.id)
@@ -58,8 +61,8 @@ def on_nickname_set(event: fbchat.NicknameSet):
event.thread.set_nickname(event.subject.id, old_nickname)
@listener.register
def on_people_added(event: fbchat.PeopleAdded):
@events.connect_via(fbchat.PeopleAdded)
def on_people_added(sender, event: fbchat.PeopleAdded):
if old_thread_id != event.thread.id:
return
if event.author.id != session.user.id:
@@ -68,8 +71,8 @@ def on_people_added(event: fbchat.PeopleAdded):
event.thread.remove_participant(added.id)
@listener.register
def on_person_removed(event: fbchat.PersonRemoved):
@events.connect_via(fbchat.PersonRemoved)
def on_person_removed(sender, event: fbchat.PersonRemoved):
if old_thread_id != event.thread.id:
return
# No point in trying to add ourself
@@ -80,4 +83,10 @@ def on_person_removed(event: fbchat.PersonRemoved):
event.thread.add_participants([removed.id])
listener.run()
# Login, and start listening for events
session = fbchat.Session.login("<email>", "<password>")
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
for event in listener.listen():
# Dispatch the event to the subscribed handlers
events.send(type(event), event=event)

10
examples/removebot.py
View File

@@ -1,10 +1,6 @@
import fbchat
session = fbchat.Session.login("<email>", "<password>")
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
@listener.register
def on_message(event):
# We can only kick people from group chats, so no need to try if it's a user chat
if not isinstance(event.thread, fbchat.Group):
@@ -14,4 +10,8 @@ def on_message(event):
event.thread.remove_participant(event.author.id)
listener.run()
session = fbchat.Session.login("<email>", "<password>")
listener = fbchat.Listener(session=session, chat_on=False, foreground=False)
for event in listener.listen():
if isinstance(event, fbchat.MessageEvent):
on_message(event)

65
fbchat/_listen.py
View File

@@ -1,12 +1,11 @@
import attr
import inspect
import random
import paho.mqtt.client
import requests
from ._common import log, kw_only
from . import _util, _exception, _session, _graphql, _events
from typing import Iterable, Optional, Mapping, Callable
from typing import Iterable, Optional, Mapping
HOST = "edge-chat.facebook.com"
@@ -98,12 +97,9 @@ def fetch_sequence_id(session: _session.Session) -> int:
return int(sequence_id)
HandlerT = Callable[[_events.Event], None]
@attr.s(slots=True, kw_only=kw_only, eq=False)
class Listener:
"""Helper, to listen for incoming Facebook events.
"""Listen to incoming Facebook events.
Initialize a connection to the Facebook MQTT service.
@@ -123,7 +119,6 @@ class Listener:
_sync_token = attr.ib(None, type=Optional[str])
_sequence_id = attr.ib(None, type=Optional[int])
_tmp_events = attr.ib(factory=list, type=Iterable[_events.Event])
_handlers = attr.ib(factory=dict, type=Mapping[HandlerT, _events.Event])
def __attrs_post_init__(self):
# Configure callbacks
@@ -305,7 +300,7 @@ class Listener:
def listen(self) -> Iterable[_events.Event]:
"""Run the listening loop continually.
Yields events when they arrive.
This is a blocking call, that will yield events as they arrive.
This will automatically reconnect on errors, except if the errors are one of
`PleaseRefresh` or `NotLoggedIn`.
@@ -404,57 +399,3 @@ class Listener:
#
# def browser_close(self):
# info = self._mqtt.publish("/browser_close", payload=b"{}", qos=1)
def register(func: HandlerT) -> HandlerT:
"""Register a function that will be called when .run is called.
The input function must take a single annotated argument.
Should be used as a function decorator.
Example:
>>> @listener.register
>>> def my_handler(event: fbchat.Event):
... print(f"New event: {event}")
"""
try:
parameter = next(iter(inspect.signature(func).parameters.values()))
except Exception as e: # TODO: More precise exceptions
raise ValueError("Invalid function. Must have at least an argument") from e
if parameter.annotation is parameter.empty:
raise ValueError("Invalid function. Must be annotated")
if not issubclass(parameter.annotation, _events.Event):
raise ValueError("Invalid function. Annotation must be an event class")
# TODO: More error checks, e.g. kw_only parameters
self._handlers[func] = parameter.annotation
def unregister(func: HandlerT):
"""Unregister a previously registered function."""
try:
self._handlers.pop(func)
except KeyError:
raise ValueError("Tried to unregister a function that was not registered")
def run(self) -> None:
"""Run the listening loop, and dispatch incoming events to registered handlers.
This uses `.listen`, which reconnect on errors, except if the errors are one of
`PleaseRefresh` or `NotLoggedIn`.
Example:
Print incoming messages.
>>> @listener.register
>>> def print_msg(event: fbchat.MessageEvent):
... print(event.message.text)
...
>>> listener.run()
"""
for event in self.listen():
for handler, event_cls in self._handlers.items():
if isinstance(event, event_cls):
handler(event)