mcvsoftware/twblue
1
0
Fork 0
mirror of https://github.com/MCV-Software/TWBlue.git synced 2025-07-17 21:56:07 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

Initial work to Support Tweepy 4

Browse Source
This commit is contained in:
Manuel Cortez
2021-09-26 03:58:25 -05:00
parent 0c27427843
commit d8fca3b31a
17 changed files with 124 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
src/sessions/twitter/session.py
View File

@@ -10,7 +10,7 @@ import output
import application
from pubsub import pub
import tweepy
from tweepy.error import TweepError
from tweepy.errors import TweepyException
from tweepy.models import User as UserModel
from mysc.thread_utils import call_threaded
from keys import keyring
@@ -199,7 +199,7 @@ class Session(base.baseSession):
try:
val = getattr(self.twitter, call_name)(*args, **kwargs)
finished = True
except TweepError as e:
except TweepyException as e:
output.speak(e.reason)
val = None
if e.error_code != 403 and e.error_code != 404:
@@ -218,7 +218,7 @@ class Session(base.baseSession):
def search(self, name, *args, **kwargs):
""" Search in twitter, passing args and kwargs as arguments to the Twython function."""
tl = self.twitter.search(*args, **kwargs)
tl = self.twitter.search_tweets(*args, **kwargs)
tl.reverse()
return tl
@@ -271,12 +271,12 @@ class Session(base.baseSession):
# @_require_login
def get_lists(self):
""" Gets the lists that the user is subscribed to and stores them in the database. Returns None."""
self.db["lists"] = self.twitter.lists_all(reverse=True)
self.db["lists"] = self.twitter.get_lists(reverse=True)
# @_require_login
def get_muted_users(self):
""" Gets muted users (oh really?)."""
self.db["muted_users"] = self.twitter.mutes_ids()
self.db["muted_users"] = self.twitter.get_muted_ids()
# @_require_login
def get_stream(self, name, function, *args, **kwargs):
@@ -417,7 +417,7 @@ class Session(base.baseSession):
log.debug("Requesting user id {} as it is not present in the users database.".format(id))
try:
user = self.twitter.get_user(id=id)
except TweepError as err:
except TweepyException as err:
user = UserModel(None)
user.screen_name = "deleted_user"
user.id = id
@@ -483,13 +483,13 @@ class Session(base.baseSession):
return
log.debug("TWBlue will get %d new users from Twitter." % (len(users_to_retrieve)))
try:
users = self.twitter.lookup_users(user_ids=users_to_retrieve, tweet_mode="extended")
users = self.twitter.lookup_users(user_id=users_to_retrieve, tweet_mode="extended")
users_db = self.db["users"]
for user in users:
users_db[user.id_str] = user
log.debug("Added %d new users" % (len(users)))
self.db["users"] = users_db
except TweepError as err:
except TweepyException as err:
if hasattr(err, "api_code") and err.api_code == 17: # Users not found.
log.error("The specified users {} were not found in twitter.".format(user_ids))
# Creates a deleted user object for every user_id not found here.
@@ -523,9 +523,8 @@ class Session(base.baseSession):
def start_streaming(self):
if config.app["app-settings"]["no_streaming"]:
return
self.stream_listener = streaming.StreamListener(twitter_api=self.twitter, user=self.db["user_name"], user_id=self.db["user_id"], muted_users=self.db["muted_users"])
self.stream = streaming.Stream(auth = self.auth, listener=self.stream_listener, chunk_size=1025)
self.stream_thread = call_threaded(self.stream.filter, follow=self.stream_listener.users, stall_warnings=True)
self.stream = streaming.Stream(twitter_api=self.twitter, user=self.db["user_name"], user_id=self.db["user_id"], muted_users=self.db["muted_users"], consumer_key=keyring.get("api_key"), consumer_secret=keyring.get("api_secret"), access_token=self.settings["twitter"]["user_key"], access_token_secret=self.settings["twitter"]["user_secret"], chunk_size=1025)
self.stream_thread = call_threaded(self.stream.filter, follow=self.stream.users, stall_warnings=True)
def stop_streaming(self):
if config.app["app-settings"]["no_streaming"]:
@@ -554,7 +553,7 @@ class Session(base.baseSession):
status._json = {**status._json, **status._json["extended_tweet"]}
# Sends status to database, where it will be reduced and changed according to our needs.
buffers_to_send = []
if status.user.id_str in self.stream_listener.users:
if status.user.id_str in self.stream.users:
buffers_to_send.append("home_timeline")
if status.user.id == self.db["user_id"]:
buffers_to_send.append("sent_tweets")

81
src/sessions/twitter/streaming.py
View File

@@ -12,17 +12,17 @@ from pubsub import pub
log = logging.getLogger("sessions.twitter.streaming")
class StreamListener(tweepy.StreamListener):
class Stream(tweepy.Stream):
def __init__(self, twitter_api, user, user_id, muted_users=[], *args, **kwargs):
super(StreamListener, self).__init__(*args, **kwargs)
super(Stream, self).__init__(*args, **kwargs)
log.debug("Starting streaming listener for account {}".format(user))
self.started = False
self.users = []
self.api = twitter_api
self.user = user
self.user_id = user_id
friends = self.api.friends_ids()
friends = self.api.get_friend_ids()
log.debug("Retrieved {} friends to add to the streaming listener.".format(len(friends)))
self.users.append(str(self.user_id))
log.debug("Got {} muted users.".format(len(muted_users)))
@@ -45,78 +45,3 @@ class StreamListener(tweepy.StreamListener):
return
if status.user.id_str in self.users:
pub.sendMessage("newStatus", status=status, user=self.user)
class Stream(tweepy.Stream):
def _run(self):
# Authenticate
url = "https://%s%s" % (self.host, self.url)
# Connect and process the stream
error_counter = 0
resp = None
exc_info = None
while self.running:
if self.retry_count is not None:
if error_counter > self.retry_count:
# quit if error count greater than retry count
break
try:
auth = self.auth.apply_auth()
resp = self.session.request('POST',
url,
data=self.body,
timeout=self.timeout,
stream=True,
auth=auth,
verify=self.verify,
proxies = self.proxies)
if resp.status_code != 200:
if self.listener.on_error(resp.status_code) is False:
break
error_counter += 1
if resp.status_code == 420:
self.retry_time = max(self.retry_420_start,
self.retry_time)
time.sleep(self.retry_time)
self.retry_time = min(self.retry_time * 2,
self.retry_time_cap)
else:
error_counter = 0
self.retry_time = self.retry_time_start
self.snooze_time = self.snooze_time_step
self.listener.on_connect()
self._read_loop(resp)
except (requests.ConnectionError, requests.Timeout, ssl.SSLError, urllib3.exceptions.ReadTimeoutError, urllib3.exceptions.ProtocolError) as exc:
# This is still necessary, as a SSLError can actually be
# thrown when using Requests
# If it's not time out treat it like any other exception
if isinstance(exc, ssl.SSLError):
if not (exc.args and 'timed out' in str(exc.args[0])):
exc_info = sys.exc_info()
break
if self.listener.on_timeout() is False:
break
if self.running is False:
break
time.sleep(self.snooze_time)
self.snooze_time = min(self.snooze_time + self.snooze_time_step,
self.snooze_time_cap)
except Exception as exc:
exc_info = sys.exc_info()
# any other exception is fatal, so kill loop
break
# cleanup
self.running = False
if resp:
resp.close()
self.new_session()
if exc_info:
# call a handler first so that the exception can be logged.
self.listener.on_exception(exc_info[1])
six.reraise(*exc_info)

4
src/sessions/twitter/utils.py
View File

@@ -6,7 +6,7 @@ import logging
import requests
import time
import sound
from tweepy.error import TweepError
from tweepy.errors import TweepyException
log = logging.getLogger("twitter.utils")
""" Some utilities for the twitter interface."""
@@ -159,7 +159,7 @@ def if_user_exists(twitter, user):
try:
data = twitter.get_user(screen_name=user)
return data
except TweepError as err:
except TweepyException as err:
if err.api_code == 50:
return None
else: