From 2da8369c70deb86903142bc7af32ded50f4c311e Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Sat, 25 Jan 2020 14:34:20 +0100 Subject: [PATCH 1/6] Refactor and improve MQTT listener a bit --- fbchat/_mqtt.py | 204 +++++++++++++++++++++++++----------------------- 1 file changed, 106 insertions(+), 98 deletions(-) diff --git a/fbchat/_mqtt.py b/fbchat/_mqtt.py index 0fddeb7..5ec3010 100644 --- a/fbchat/_mqtt.py +++ b/fbchat/_mqtt.py @@ -8,6 +8,46 @@ from . import _util, _exception, _session, _graphql, _events from typing import Iterable, Optional +HOST = "edge-chat.facebook.com" + +TOPICS = [ + # Things that happen in chats (e.g. messages) + "/t_ms", + # Group typing notifications + "/thread_typing", + # Private chat typing notifications + "/orca_typing_notifications", + # Active notifications + "/orca_presence", + # Other notifications not related to chats (e.g. friend requests) + "/legacy_web", + # Facebook's continuous error reporting/logging? + "/br_sr", + # Response to /br_sr + "/sr_res", + # Data about user-to-user calls + # TODO: Investigate the response from this! (A bunch of binary data) + # "/t_rtc", + # TODO: Find out what this does! + # TODO: Investigate the response from this! (A bunch of binary data) + # "/t_p", + # TODO: Find out what this does! + "/webrtc", + # TODO: Find out what this does! + "/onevc", + # TODO: Find out what this does! + "/notify_disconnect", + # Old, no longer active topics + # These are here just in case something interesting pops up + "/inbox", + "/mercury", + "/messaging_events", + "/orca_message_notifications", + "/pp", + "/webrtc_response", +] + + def get_cookie_header(session: requests.Session, url: str) -> str: """Extract a cookie header from a requests session.""" # The cookies are extracted this way to make sure they're escaped correctly @@ -21,6 +61,24 @@ def generate_session_id() -> int: return random.randint(1, 2 ** 53) +def fetch_sequence_id(session: _session.Session) -> int: + """Fetch sequence ID.""" + params = { + "limit": 0, + "tags": ["INBOX"], + "before": None, + "includeDeliveryReceipts": False, + "includeSeqID": True, + } + log.debug("Fetching MQTT sequence ID") + # Same doc id as in `Client.fetch_threads` + (j,) = session._graphql_requests(_graphql.from_doc_id("1349387578499440", params)) + sequence_id = j["viewer"]["message_threads"]["sync_sequence_id"] + if not sequence_id: + raise _exception.NotLoggedIn("Failed fetching sequence id") + return int(sequence_id) + + @attr.s(slots=True, kw_only=kw_only, repr=False, eq=False) class Listener: """Helper, to listen for incoming Facebook events.""" @@ -33,8 +91,6 @@ class Listener: _sync_token = attr.ib(None, type=str) _tmp_events = attr.ib(None, type=Optional[Iterable[_events.Event]]) - _HOST = "edge-chat.facebook.com" - def __repr__(self) -> str: # An alternative repr, to illustrate that you can't create the class directly return "".format( @@ -64,7 +120,6 @@ class Listener: # mqtt.max_queued_messages_set(0) # Unlimited messages can be queued # mqtt.message_retry_set(20) # Retry sending for at least 20 seconds # mqtt.reconnect_delay_set(min_delay=1, max_delay=120) - # TODO: Is region (lla | atn | odn | others?) important? mqtt.tls_set() self = cls( @@ -72,7 +127,7 @@ class Listener: mqtt=mqtt, chat_on=chat_on, foreground=foreground, - sequence_id=cls._fetch_sequence_id(session), + sequence_id=fetch_sequence_id(session), ) # Configure callbacks @@ -83,7 +138,7 @@ class Listener: # Attempt to connect try: - rc = mqtt.connect(self._HOST, 443, keepalive=10) + rc = mqtt.connect(HOST, 443, keepalive=10) except ( # Taken from .loop_forever paho.mqtt.client.socket.error, @@ -99,6 +154,46 @@ class Listener: return self + def _handle_ms(self, j): + """Handle /t_ms special logic. + + Returns whether to continue parsing the message. + """ + # TODO: Merge this with the parsing in _events + + # Update sync_token when received + # This is received in the first message after we've created a messenger + # sync queue. + if "syncToken" in j and "firstDeltaSeqId" in j: + self._sync_token = j["syncToken"] + self._sequence_id = j["firstDeltaSeqId"] + return False + + if "errorCode" in j: + error = j["errorCode"] + # TODO: 'F\xfa\x84\x8c\x85\xf8\xbc-\x88 FB_PAGES_INSUFFICIENT_PERMISSION\x00' + if error in ("ERROR_QUEUE_NOT_FOUND", "ERROR_QUEUE_OVERFLOW"): + # ERROR_QUEUE_NOT_FOUND means that the queue was deleted, since too + # much time passed, or that it was simply missing + # ERROR_QUEUE_OVERFLOW means that the sequence id was too small, so + # the desired events could not be retrieved + log.error( + "The MQTT listener was disconnected for too long," + " events may have been lost" + ) + self._sync_token = None + self._sequence_id = fetch_sequence_id(self.session) + self._messenger_queue_publish() + # TODO: Signal to the user that they should reload their data! + return False + log.error("MQTT error code %s received", error) + return False + + # Update last sequence id + # Except for the two cases above, this is always received + self._sequence_id = j["lastIssuedSeqId"] + return True + def _on_message_handler(self, client, userdata, message): # Parse payload JSON try: @@ -111,40 +206,9 @@ class Listener: log.debug("MQTT payload: %s, %s", message.topic, j) if message.topic == "/t_ms": - # Update sync_token when received - # This is received in the first message after we've created a messenger - # sync queue. - if "syncToken" in j and "firstDeltaSeqId" in j: - self._sync_token = j["syncToken"] - self._sequence_id = j["firstDeltaSeqId"] + if not _handle_ms(j): return - if "errorCode" in j: - error = j["errorCode"] - # TODO: 'F\xfa\x84\x8c\x85\xf8\xbc-\x88 FB_PAGES_INSUFFICIENT_PERMISSION\x00' - if error in ("ERROR_QUEUE_NOT_FOUND", "ERROR_QUEUE_OVERFLOW"): - # ERROR_QUEUE_NOT_FOUND means that the queue was deleted, since too - # much time passed, or that it was simply missing - # ERROR_QUEUE_OVERFLOW means that the sequence id was too small, so - # the desired events could not be retrieved - log.error( - "The MQTT listener was disconnected for too long," - " events may have been lost" - ) - self._sync_token = None - self._sequence_id = self._fetch_sequence_id(self.session) - self._messenger_queue_publish() - # TODO: Signal to the user that they should reload their data! - return - log.error("MQTT error code %s received", error) - return - - # Update last sequence id when received - if "lastIssuedSeqId" in j: - self._sequence_id = j["lastIssuedSeqId"] - else: - log.error("Missing last sequence id: %s", j) - try: # TODO: Don't handle this in a callback self._tmp_events = list( @@ -153,32 +217,14 @@ class Listener: except _exception.ParseError: log.exception("Failed parsing MQTT data") - @staticmethod - def _fetch_sequence_id(session: _session.Session) -> int: - """Fetch sequence ID.""" - params = { - "limit": 0, - "tags": ["INBOX"], - "before": None, - "includeDeliveryReceipts": False, - "includeSeqID": True, - } - log.debug("Fetching MQTT sequence ID") - # Same request as in `Client.fetch_threads` - (j,) = session._graphql_requests( - _graphql.from_doc_id("1349387578499440", params) - ) - sequence_id = j["viewer"]["message_threads"]["sync_sequence_id"] - if not sequence_id: - raise _exception.NotLoggedIn("Failed fetching sequence id") - return int(sequence_id) - def _on_connect_handler(self, client, userdata, flags, rc): if rc == 21: raise _exception.FacebookError( "Failed connecting. Maybe your cookies are wrong?" ) if rc != 0: + err = paho.mqtt.client.connack_string(rc) + log.error("MQTT Connection Error: %s", err) return # Don't try to send publish if the connection failed self._messenger_queue_publish() @@ -211,43 +257,6 @@ class Listener: # Generate a new session ID on each reconnect session_id = generate_session_id() - topics = [ - # Things that happen in chats (e.g. messages) - "/t_ms", - # Group typing notifications - "/thread_typing", - # Private chat typing notifications - "/orca_typing_notifications", - # Active notifications - "/orca_presence", - # Other notifications not related to chats (e.g. friend requests) - "/legacy_web", - # Facebook's continuous error reporting/logging? - "/br_sr", - # Response to /br_sr - "/sr_res", - # Data about user-to-user calls - # TODO: Investigate the response from this! (A bunch of binary data) - # "/t_rtc", - # TODO: Find out what this does! - # TODO: Investigate the response from this! (A bunch of binary data) - # "/t_p", - # TODO: Find out what this does! - "/webrtc", - # TODO: Find out what this does! - "/onevc", - # TODO: Find out what this does! - "/notify_disconnect", - # Old, no longer active topics - # These are here just in case something interesting pops up - "/inbox", - "/mercury", - "/messaging_events", - "/orca_message_notifications", - "/pp", - "/webrtc_response", - ] - username = { # The user ID "u": self.session.user.id, @@ -262,7 +271,7 @@ class Listener: # Application ID, taken from facebook.com "aid": 219994525426954, # MQTT extension by FB, allows making a SUBSCRIBE while CONNECTing - "st": topics, + "st": TOPICS, # MQTT extension by FB, allows making a PUBLISH while CONNECTing # Using this is more efficient, but the same can be acheived with: # def on_connect(*args): @@ -288,19 +297,18 @@ class Listener: "pack": [], } - # TODO: Make this thread safe self._mqtt.username_pw_set(_util.json_minimal(username)) headers = { - # TODO: Make this access thread safe "Cookie": get_cookie_header( self.session._session, "https://edge-chat.facebook.com/chat" ), "User-Agent": self.session._session.headers["User-Agent"], "Origin": "https://www.facebook.com", - "Host": self._HOST, + "Host": HOST, } + # TODO: Is region (lla | atn | odn | others?) important? self._mqtt.ws_set_options( path="/chat?sid={}".format(session_id), headers=headers ) From f8e110f180661f5e2b5f3838b0e4aa20e3866750 Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Sat, 25 Jan 2020 14:50:15 +0100 Subject: [PATCH 2/6] Handle connecting in `Listener.listen` --- fbchat/_mqtt.py | 150 +++++++++++++++++++++++------------------------- 1 file changed, 71 insertions(+), 79 deletions(-) diff --git a/fbchat/_mqtt.py b/fbchat/_mqtt.py index 5ec3010..b82a07c 100644 --- a/fbchat/_mqtt.py +++ b/fbchat/_mqtt.py @@ -61,6 +61,24 @@ def generate_session_id() -> int: return random.randint(1, 2 ** 53) +def mqtt_factory() -> paho.mqtt.client.Client: + # Configure internal MQTT handler + mqtt = paho.mqtt.client.Client( + client_id="mqttwsclient", + clean_session=True, + protocol=paho.mqtt.client.MQTTv31, + transport="websockets", + ) + mqtt.enable_logger() + # mqtt.max_inflight_messages_set(20) # The rest will get queued + # mqtt.max_queued_messages_set(0) # Unlimited messages can be queued + # mqtt.message_retry_set(20) # Retry sending for at least 20 seconds + # mqtt.reconnect_delay_set(min_delay=1, max_delay=120) + mqtt.tls_set() + mqtt.connect_async(HOST, 443, keepalive=10) + return mqtt + + def fetch_sequence_id(session: _session.Session) -> int: """Fetch sequence ID.""" params = { @@ -84,10 +102,10 @@ class Listener: """Helper, to listen for incoming Facebook events.""" session = attr.ib(type=_session.Session) - _mqtt = attr.ib(type=paho.mqtt.client.Client) _chat_on = attr.ib(type=bool) _foreground = attr.ib(type=bool) _sequence_id = attr.ib(type=int) + _mqtt = attr.ib(factory=mqtt_factory, type=paho.mqtt.client.Client) _sync_token = attr.ib(None, type=str) _tmp_events = attr.ib(None, type=Optional[Iterable[_events.Event]]) @@ -97,6 +115,11 @@ class Listener: self.session, self._chat_on, self._foreground ) + def __attrs_post_init__(self): + # Configure callbacks + self._mqtt.on_message = self._on_message_handler + self._mqtt.on_connect = self._on_connect_handler + @classmethod def connect(cls, session: _session.Session, chat_on: bool, foreground: bool): """Initialize a connection to the Facebook MQTT service. @@ -109,51 +132,13 @@ class Listener: Example: >>> listener = fbchat.Listener.connect(session, chat_on=True, foreground=True) """ - mqtt = paho.mqtt.client.Client( - client_id="mqttwsclient", - clean_session=True, - protocol=paho.mqtt.client.MQTTv31, - transport="websockets", - ) - mqtt.enable_logger() - # mqtt.max_inflight_messages_set(20) # The rest will get queued - # mqtt.max_queued_messages_set(0) # Unlimited messages can be queued - # mqtt.message_retry_set(20) # Retry sending for at least 20 seconds - # mqtt.reconnect_delay_set(min_delay=1, max_delay=120) - mqtt.tls_set() - - self = cls( + return cls( session=session, - mqtt=mqtt, chat_on=chat_on, foreground=foreground, sequence_id=fetch_sequence_id(session), ) - # Configure callbacks - mqtt.on_message = self._on_message_handler - mqtt.on_connect = self._on_connect_handler - - self._configure_connect_options() - - # Attempt to connect - try: - rc = mqtt.connect(HOST, 443, keepalive=10) - except ( - # Taken from .loop_forever - paho.mqtt.client.socket.error, - OSError, - paho.mqtt.client.WebsocketConnectionError, - ) as e: - raise _exception.FacebookError("MQTT connection failed") from e - - # Raise error if connecting failed - if rc != paho.mqtt.client.MQTT_ERR_SUCCESS: - err = paho.mqtt.client.error_string(rc) - raise _exception.FacebookError("MQTT connection failed: {}".format(err)) - - return self - def _handle_ms(self, j): """Handle /t_ms special logic. @@ -206,7 +191,7 @@ class Listener: log.debug("MQTT payload: %s, %s", message.topic, j) if message.topic == "/t_ms": - if not _handle_ms(j): + if not self._handle_ms(j): return try: @@ -313,45 +298,21 @@ class Listener: path="/chat?sid={}".format(session_id), headers=headers ) - def _loop_once(self) -> bool: - rc = self._mqtt.loop(timeout=1.0) - - # If disconnect() has been called - # Beware, internal API, may have to change this to something more stable! - if self._mqtt._state == paho.mqtt.client.mqtt_cs_disconnecting: - return False # Stop listening - - if rc != paho.mqtt.client.MQTT_ERR_SUCCESS: - # If known/expected error - if rc == paho.mqtt.client.MQTT_ERR_CONN_LOST: - log.warning("Connection lost, retrying") - elif rc == paho.mqtt.client.MQTT_ERR_NOMEM: - # This error is wrongly classified - # See https://github.com/eclipse/paho.mqtt.python/issues/340 - log.warning("Connection error, retrying") - elif rc == paho.mqtt.client.MQTT_ERR_CONN_REFUSED: - raise _exception.NotLoggedIn("MQTT connection refused") - else: - err = paho.mqtt.client.error_string(rc) - log.error("MQTT Error: %s", err) - + def _reconnect(self): + # Try reconnecting + self._configure_connect_options() + try: + self._mqtt.reconnect() + except ( + # Taken from .loop_forever + paho.mqtt.client.socket.error, + OSError, + paho.mqtt.client.WebsocketConnectionError, + ) as e: + log.debug("MQTT reconnection failed: %s", e) # Wait before reconnecting self._mqtt._reconnect_wait() - # Try reconnecting - self._configure_connect_options() - try: - self._mqtt.reconnect() - except ( - # Taken from .loop_forever - paho.mqtt.client.socket.error, - OSError, - paho.mqtt.client.WebsocketConnectionError, - ) as e: - log.debug("MQTT reconnection failed: %s", e) - - return True # Keep listening - def listen(self) -> Iterable[_events.Event]: """Run the listening loop continually. @@ -365,10 +326,41 @@ class Listener: >>> for event in listener.listen(): ... print(event) """ - while self._loop_once(): + # Make sure we're connected + while True: + # Beware, internal API, may have to change this to something more stable! + if self._mqtt._state == paho.mqtt.client.mqtt_cs_connect_async: + self._reconnect() + else: + break + + while True: + rc = self._mqtt.loop(timeout=1.0) + + # If disconnect() has been called + # Beware, internal API, may have to change this to something more stable! + if self._mqtt._state == paho.mqtt.client.mqtt_cs_disconnecting: + break # Stop listening + + if rc != paho.mqtt.client.MQTT_ERR_SUCCESS: + # If known/expected error + if rc == paho.mqtt.client.MQTT_ERR_CONN_LOST: + log.warning("Connection lost, retrying") + elif rc == paho.mqtt.client.MQTT_ERR_NOMEM: + # This error is wrongly classified + # See https://github.com/eclipse/paho.mqtt.python/issues/340 + log.warning("Connection error, retrying") + elif rc == paho.mqtt.client.MQTT_ERR_CONN_REFUSED: + raise _exception.NotLoggedIn("MQTT connection refused") + else: + err = paho.mqtt.client.error_string(rc) + log.error("MQTT Error: %s", err) + + self._reconnect() + if self._tmp_events: yield from self._tmp_events - self._tmp_events = None + self._tmp_events = None def disconnect(self) -> None: """Disconnect the MQTT listener. From 987993701f384b98a45ec64f9750474f251f7393 Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Sat, 25 Jan 2020 15:07:04 +0100 Subject: [PATCH 3/6] Fix NotLoggedIn --- fbchat/_exception.py | 14 ++++++-------- fbchat/_session.py | 2 +- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/fbchat/_exception.py b/fbchat/_exception.py index 76aa047..b35466b 100644 --- a/fbchat/_exception.py +++ b/fbchat/_exception.py @@ -46,6 +46,11 @@ class ParseError(FacebookError): return msg.format(self.message, self.data) +@attr.s(slots=True, auto_exc=True) +class NotLoggedIn(FacebookError): + """Raised by Facebook if the client has been logged out.""" + + @attr.s(slots=True, auto_exc=True) class ExternalError(FacebookError): """Base class for errors that Facebook return.""" @@ -86,13 +91,6 @@ class InvalidParameters(ExternalError): """ -@attr.s(slots=True, auto_exc=True) -class NotLoggedIn(ExternalError): - """Raised by Facebook if the client has been logged out.""" - - code = attr.ib() - - @attr.s(slots=True, auto_exc=True) class PleaseRefresh(ExternalError): """Raised by Facebook if the client has been inactive for too long. @@ -108,7 +106,7 @@ def handle_payload_error(j): return code = j["error"] if code == 1357001: - error_cls = NotLoggedIn + raise NotLoggedIn(j["errorSummary"]) elif code == 1357004: error_cls = PleaseRefresh elif code in (1357031, 1545010, 1545003): diff --git a/fbchat/_session.py b/fbchat/_session.py index 6e34d67..2bedefc 100644 --- a/fbchat/_session.py +++ b/fbchat/_session.py @@ -308,7 +308,7 @@ class Session: # Fall back to searching with a regex res = FB_DTSG_REGEX.search(r.text) if not res: - raise ValueError("Failed loading session, could not find fb_dtsg") + raise _exception.NotLoggedIn("Could not find fb_dtsg") fb_dtsg = res.group(1) revision = int(r.text.split('"client_revision":', 1)[1].split(",", 1)[0]) From 9c03c1035b12c29770f61e64d6e273456733564a Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Sat, 25 Jan 2020 15:07:40 +0100 Subject: [PATCH 4/6] Allow initializing the MQTT Listener without making external requests --- fbchat/_mqtt.py | 49 +++++++++++++++++++------------------------------ 1 file changed, 19 insertions(+), 30 deletions(-) diff --git a/fbchat/_mqtt.py b/fbchat/_mqtt.py index b82a07c..e61c5bd 100644 --- a/fbchat/_mqtt.py +++ b/fbchat/_mqtt.py @@ -97,48 +97,34 @@ def fetch_sequence_id(session: _session.Session) -> int: return int(sequence_id) -@attr.s(slots=True, kw_only=kw_only, repr=False, eq=False) +@attr.s(slots=True, kw_only=kw_only, eq=False) class Listener: - """Helper, to listen for incoming Facebook events.""" + """Helper, to listen for incoming Facebook events. + + Initialize a connection to the Facebook MQTT service. + + Args: + session: The session to use when making requests. + chat_on: Whether ... + foreground: Whether ... + + Example: + >>> listener = fbchat.Listener(session, chat_on=True, foreground=True) + """ session = attr.ib(type=_session.Session) _chat_on = attr.ib(type=bool) _foreground = attr.ib(type=bool) - _sequence_id = attr.ib(type=int) _mqtt = attr.ib(factory=mqtt_factory, type=paho.mqtt.client.Client) - _sync_token = attr.ib(None, type=str) - _tmp_events = attr.ib(None, type=Optional[Iterable[_events.Event]]) - - def __repr__(self) -> str: - # An alternative repr, to illustrate that you can't create the class directly - return "".format( - self.session, self._chat_on, self._foreground - ) + _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]) def __attrs_post_init__(self): # Configure callbacks self._mqtt.on_message = self._on_message_handler self._mqtt.on_connect = self._on_connect_handler - @classmethod - def connect(cls, session: _session.Session, chat_on: bool, foreground: bool): - """Initialize a connection to the Facebook MQTT service. - - Args: - session: The session to use when making requests. - chat_on: Whether ... - foreground: Whether ... - - Example: - >>> listener = fbchat.Listener.connect(session, chat_on=True, foreground=True) - """ - return cls( - session=session, - chat_on=chat_on, - foreground=foreground, - sequence_id=fetch_sequence_id(session), - ) - def _handle_ms(self, j): """Handle /t_ms special logic. @@ -326,6 +312,9 @@ class Listener: >>> for event in listener.listen(): ... print(event) """ + if not self._sequence_id: + self._sequence_id = fetch_sequence_id(self.session) + # Make sure we're connected while True: # Beware, internal API, may have to change this to something more stable! From ae2bb41509cc62eaf66e65c0a80066d9ae4302bf Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Wed, 5 Feb 2020 13:50:01 +0100 Subject: [PATCH 5/6] Small listening fixes - If an error is raised in fetch_sequence_id, don't swallow it! --- fbchat/__init__.py | 2 +- fbchat/{_mqtt.py => _listen.py} | 14 +++++++++----- 2 files changed, 10 insertions(+), 6 deletions(-) rename fbchat/{_mqtt.py => _listen.py} (98%) diff --git a/fbchat/__init__.py b/fbchat/__init__.py index 9d5e8bb..8dda631 100644 --- a/fbchat/__init__.py +++ b/fbchat/__init__.py @@ -111,7 +111,7 @@ from ._events import ( FriendRequest, Presence, ) -from ._mqtt import Listener +from ._listen import Listener from ._client import Client diff --git a/fbchat/_mqtt.py b/fbchat/_listen.py similarity index 98% rename from fbchat/_mqtt.py rename to fbchat/_listen.py index e61c5bd..4ff4bed 100644 --- a/fbchat/_mqtt.py +++ b/fbchat/_listen.py @@ -153,9 +153,7 @@ class Listener: " events may have been lost" ) self._sync_token = None - self._sequence_id = fetch_sequence_id(self.session) - self._messenger_queue_publish() - # TODO: Signal to the user that they should reload their data! + self._sequence_id = None return False log.error("MQTT error code %s received", error) return False @@ -312,7 +310,7 @@ class Listener: >>> for event in listener.listen(): ... print(event) """ - if not self._sequence_id: + if self._sequence_id is None: self._sequence_id = fetch_sequence_id(self.session) # Make sure we're connected @@ -326,6 +324,12 @@ class Listener: while True: rc = self._mqtt.loop(timeout=1.0) + # The sequence ID was reset in _handle_ms + # TODO: Signal to the user that they should reload their data! + if self._sequence_id is None: + self._sequence_id = fetch_sequence_id(self.session) + self._messenger_queue_publish() + # If disconnect() has been called # Beware, internal API, may have to change this to something more stable! if self._mqtt._state == paho.mqtt.client.mqtt_cs_disconnecting: @@ -349,7 +353,7 @@ class Listener: if self._tmp_events: yield from self._tmp_events - self._tmp_events = None + self._tmp_events = [] def disconnect(self) -> None: """Disconnect the MQTT listener. From 085bbba3023e12b3edb808c147e9a5b3a3979cf5 Mon Sep 17 00:00:00 2001 From: Mads Marquart Date: Wed, 5 Feb 2020 14:27:08 +0100 Subject: [PATCH 6/6] Improve listening usability Add Listener.register and Listener.run --- docs/intro.rst | 17 ++++++------ examples/echobot.py | 10 +++---- examples/keepbot.py | 23 ++++++---------- examples/removebot.py | 8 ++---- fbchat/_listen.py | 64 +++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 86 insertions(+), 36 deletions(-) diff --git a/docs/intro.rst b/docs/intro.rst index 39a17fd..2749607 100644 --- a/docs/intro.rst +++ b/docs/intro.rst @@ -134,17 +134,18 @@ 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 your methods that will handle your events:: +To get started, you create a listener object:: - def on_message(event): + listener = fbchat.Listener(session=session, chat_on=False, foreground=False) + +The you use that to register methods that will handle your events:: + + @listener.register + def on_message(event: fbchat.MessageEvent): print(f"Message from {event.author.id}: {event.message.text}") -And then you create a listener object, and start handling the incoming events:: +And then you start handling the incoming events:: - listener = fbchat.Listener.connect(session, False, False) - - for event in listener.listen(): - if isinstance(event, fbchat.MessageEvent): - on_message(event) + listener.run() View the :ref:`examples` to see some more examples illustrating the event system. diff --git a/examples/echobot.py b/examples/echobot.py index d196621..212e291 100644 --- a/examples/echobot.py +++ b/examples/echobot.py @@ -1,17 +1,15 @@ import fbchat session = fbchat.Session.login("", "") - -listener = fbchat.Listener.connect(session, chat_on=False, foreground=False) +listener = fbchat.Listener(session=session, chat_on=False, foreground=False) -def on_message(event): +@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) -for event in listener.listen(): - if isinstance(event, fbchat.MessageEvent): - on_message(event) +listener.run() diff --git a/examples/keepbot.py b/examples/keepbot.py index 1529f90..4ef661e 100644 --- a/examples/keepbot.py +++ b/examples/keepbot.py @@ -15,10 +15,10 @@ old_nicknames = { } session = fbchat.Session.login("", "") - -listener = fbchat.Listener.connect(session, chat_on=False, foreground=False) +listener = fbchat.Listener(session=session, chat_on=False, foreground=False) +@listener.register def on_color_set(event: fbchat.ColorSet): if old_thread_id != event.thread.id: return @@ -27,6 +27,7 @@ def on_color_set(event: fbchat.ColorSet): event.thread.set_color(old_color) +@listener.register def on_emoji_set(event: fbchat.EmojiSet): if old_thread_id != event.thread.id: return @@ -35,6 +36,7 @@ def on_emoji_set(event: fbchat.EmojiSet): event.thread.set_emoji(old_emoji) +@listener.register def on_title_set(event: fbchat.TitleSet): if old_thread_id != event.thread.id: return @@ -43,6 +45,7 @@ def on_title_set(event: fbchat.TitleSet): event.thread.set_title(old_title) +@listener.register def on_nickname_set(event: fbchat.NicknameSet): if old_thread_id != event.thread.id: return @@ -55,6 +58,7 @@ 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): if old_thread_id != event.thread.id: return @@ -64,6 +68,7 @@ def on_people_added(event: fbchat.PeopleAdded): event.thread.remove_participant(added.id) +@listener.register def on_person_removed(event: fbchat.PersonRemoved): if old_thread_id != event.thread.id: return @@ -75,16 +80,4 @@ def on_person_removed(event: fbchat.PersonRemoved): event.thread.add_participants([removed.id]) -for event in listener.listen(): - if isinstance(event, fbchat.ColorSet): - on_color_set(event) - elif isinstance(event, fbchat.EmojiSet): - on_emoji_set(event) - elif isinstance(event, fbchat.TitleSet): - on_title_set(event) - elif isinstance(event, fbchat.NicknameSet): - on_nickname_set(event) - elif isinstance(event, fbchat.PeopleAdded): - on_people_added(event) - elif isinstance(event, fbchat.PersonRemoved): - on_person_removed(event) +listener.run() diff --git a/examples/removebot.py b/examples/removebot.py index 14d449a..0c83f27 100644 --- a/examples/removebot.py +++ b/examples/removebot.py @@ -1,10 +1,10 @@ import fbchat session = fbchat.Session.login("", "") - -listener = fbchat.Listener.connect(session, chat_on=False, foreground=False) +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,6 +14,4 @@ def on_message(event): event.thread.remove_participant(event.author.id) -for event in listener.listen(): - if isinstance(event, fbchat.MessageEvent): - on_message(event) +listener.run() diff --git a/fbchat/_listen.py b/fbchat/_listen.py index 4ff4bed..a7ee27a 100644 --- a/fbchat/_listen.py +++ b/fbchat/_listen.py @@ -1,11 +1,12 @@ 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 +from typing import Iterable, Optional, Mapping, Callable HOST = "edge-chat.facebook.com" @@ -97,6 +98,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. @@ -119,6 +123,7 @@ 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 @@ -302,7 +307,8 @@ class Listener: Yields events when they arrive. - This will automatically reconnect on errors. + This will automatically reconnect on errors, except if the errors are one of + `PleaseRefresh` or `NotLoggedIn`. Example: Print events continually. @@ -398,3 +404,57 @@ 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)