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

Made code indentation to comply with PEP8

Browse Source
This commit is contained in:
Manuel Cortez
2021-06-16 16:18:41 -05:00
parent 2aaa4eced3
commit 39e1fb017c
145 changed files with 9577 additions and 9584 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/sessions/__init__.py
View File

@@ -3,4 +3,4 @@
In TWBlue, a session module defines everything a social network needs to be used in the program."""
from __future__ import unicode_literals
# let's define a global object for storing sessions across the program.
sessions = {}
sessions = {}

178
src/sessions/base.py
View File

@@ -18,103 +18,103 @@ from . import session_exceptions as Exceptions
log = logging.getLogger("sessionmanager.session")
class baseSession(object):
""" toDo: Decorators does not seem to be working when using them in an inherited class."""
""" toDo: Decorators does not seem to be working when using them in an inherited class."""
# Decorators.
# Decorators.
def _require_login(fn):
""" Decorator for checking if the user is logged in.
Some functions may need this to avoid making unneeded calls."""
def f(self, *args, **kwargs):
if self.logged == True:
fn(self, *args, **kwargs)
else:
raise Exceptions.NotLoggedSessionError("You are not logged in yet.")
return f
def _require_login(fn):
""" Decorator for checking if the user is logged in.
Some functions may need this to avoid making unneeded calls."""
def f(self, *args, **kwargs):
if self.logged == True:
fn(self, *args, **kwargs)
else:
raise Exceptions.NotLoggedSessionError("You are not logged in yet.")
return f
def _require_configuration(fn):
""" Check if the user has a configured session."""
def f(self, *args, **kwargs):
if self.settings != None:
fn(self, *args, **kwargs)
else:
raise Exceptions.NotConfiguredSessionError("Not configured.")
return f
def _require_configuration(fn):
""" Check if the user has a configured session."""
def f(self, *args, **kwargs):
if self.settings != None:
fn(self, *args, **kwargs)
else:
raise Exceptions.NotConfiguredSessionError("Not configured.")
return f
def __init__(self, session_id):
""" session_id (str): The name of the folder inside the config directory where the session is located."""
super(baseSession, self).__init__()
self.session_id = session_id
self.logged = False
self.settings = None
self.db={}
@property
def is_logged(self):
return self.logged
def __init__(self, session_id):
""" session_id (str): The name of the folder inside the config directory where the session is located."""
super(baseSession, self).__init__()
self.session_id = session_id
self.logged = False
self.settings = None
self.db={}
def get_configuration(self):
""" Get settings for a session."""
file_ = "%s/session.conf" % (self.session_id,)
log.debug("Creating config file %s" % (file_,))
self.settings = config_utils.load_config(os.path.join(paths.config_path(), file_), os.path.join(paths.app_path(), "Conf.defaults"))
self.init_sound()
self.deshelve()
@property
def is_logged(self):
return self.logged
def init_sound(self):
try: self.sound = sound.soundSystem(self.settings["sound"])
except: pass
def get_configuration(self):
""" Get settings for a session."""
file_ = "%s/session.conf" % (self.session_id,)
log.debug("Creating config file %s" % (file_,))
self.settings = config_utils.load_config(os.path.join(paths.config_path(), file_), os.path.join(paths.app_path(), "Conf.defaults"))
self.init_sound()
self.deshelve()
@_require_configuration
def login(self, verify_credentials=True):
pass
def init_sound(self):
try: self.sound = sound.soundSystem(self.settings["sound"])
except: pass
@_require_configuration
def authorise(self):
pass
@_require_configuration
def login(self, verify_credentials=True):
pass
def shelve(self):
"""Shelve the database to allow for persistance."""
shelfname=os.path.join(paths.config_path(), str(self.session_id), "cache")
if self.settings["general"]["persist_size"] == 0:
if os.path.exists(shelfname+".dat"):
os.remove(shelfname+".dat")
return
try:
if not os.path.exists(shelfname+".dat"):
output.speak("Generating database, this might take a while.",True)
shelf=shelve.open(os.path.join(paths.config_path(), shelfname),'c')
for key, value in list(self.db.items()):
if type(key) != str and type(key) != str:
output.speak("Uh oh, while shelving the database, a key of type " + str(type(key)) + " has been found. It will be converted to type str, but this will cause all sorts of problems on deshelve. Please bring this to the attention of the " + application.name + " developers immediately. More information about the error will be written to the error log.",True)
log.error("Uh oh, " + str(key) + " is of type " + str(type(key)) + "!")
if type(value) == list and self.settings["general"]["persist_size"] != -1 and len(value) > self.settings["general"]["persist_size"]:
shelf[key]=value[self.settings["general"]["persist_size"]:]
else:
shelf[key]=value
shelf.close()
except:
output.speak("An exception occurred while shelving the " + application.name + " database. It will be deleted and rebuilt automatically. If this error persists, send the error log to the " + application.name + " developers.",True)
log.exception("Exception while shelving" + shelfname)
os.remove(shelfname)
@_require_configuration
def authorise(self):
pass
def deshelve(self):
"""Import a shelved database."""
shelfname=os.path.join(paths.config_path(), str(self.session_id)+"/cache")
if self.settings["general"]["persist_size"] == 0:
if os.path.exists(shelfname+".dat"):
os.remove(shelfname+".dat")
return
try:
shelf=shelve.open(os.path.join(paths.config_path(), shelfname),'c')
for key,value in list(shelf.items()):
self.db[key]=value
shelf.close()
except:
output.speak("An exception occurred while deshelving the " + application.name + " database. It will be deleted and rebuilt automatically. If this error persists, send the error log to the " + application.name + " developers.",True)
log.exception("Exception while deshelving" + shelfname)
try:
os.remove(shelfname)
except:
pass
def shelve(self):
"""Shelve the database to allow for persistance."""
shelfname=os.path.join(paths.config_path(), str(self.session_id), "cache")
if self.settings["general"]["persist_size"] == 0:
if os.path.exists(shelfname+".dat"):
os.remove(shelfname+".dat")
return
try:
if not os.path.exists(shelfname+".dat"):
output.speak("Generating database, this might take a while.",True)
shelf=shelve.open(os.path.join(paths.config_path(), shelfname),'c')
for key, value in list(self.db.items()):
if type(key) != str and type(key) != str:
output.speak("Uh oh, while shelving the database, a key of type " + str(type(key)) + " has been found. It will be converted to type str, but this will cause all sorts of problems on deshelve. Please bring this to the attention of the " + application.name + " developers immediately. More information about the error will be written to the error log.",True)
log.error("Uh oh, " + str(key) + " is of type " + str(type(key)) + "!")
if type(value) == list and self.settings["general"]["persist_size"] != -1 and len(value) > self.settings["general"]["persist_size"]:
shelf[key]=value[self.settings["general"]["persist_size"]:]
else:
shelf[key]=value
shelf.close()
except:
output.speak("An exception occurred while shelving the " + application.name + " database. It will be deleted and rebuilt automatically. If this error persists, send the error log to the " + application.name + " developers.",True)
log.exception("Exception while shelving" + shelfname)
os.remove(shelfname)
def deshelve(self):
"""Import a shelved database."""
shelfname=os.path.join(paths.config_path(), str(self.session_id)+"/cache")
if self.settings["general"]["persist_size"] == 0:
if os.path.exists(shelfname+".dat"):
os.remove(shelfname+".dat")
return
try:
shelf=shelve.open(os.path.join(paths.config_path(), shelfname),'c')
for key,value in list(shelf.items()):
self.db[key]=value
shelf.close()
except:
output.speak("An exception occurred while deshelving the " + application.name + " database. It will be deleted and rebuilt automatically. If this error persists, send the error log to the " + application.name + " developers.",True)
log.exception("Exception while deshelving" + shelfname)
try:
os.remove(shelfname)
except:
pass

2
src/sessions/session_exceptions.py
View File

@@ -6,4 +6,4 @@ class NonExistentSessionError(Exception): pass
class NotLoggedSessionError(BaseException): pass
class NotConfiguredSessionError(BaseException): pass
class RequireCredentialsSessionError(BaseException): pass
class AlreadyAuthorisedError(BaseException): pass
class AlreadyAuthorisedError(BaseException): pass

2
src/sessions/twitter/__init__.py
View File

@@ -1 +1 @@
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-

277
src/sessions/twitter/compose.py
View File

@@ -1,11 +1,4 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals
from future import standard_library
standard_library.install_aliases()
from builtins import str
from builtins import chr
from builtins import range
import platform
system = platform.system()
from . import utils
@@ -21,151 +14,151 @@ from .long_tweets import twishort, tweets
log = logging.getLogger("compose")
def StripChars(s):
"""Converts any html entities in s to their unicode-decoded equivalents and returns a string."""
entity_re = re.compile(r"&(#\d+|\w+);")
def matchFunc(match):
"""Nested function to handle a match object.
If we match &blah; and it's not found, &blah; will be returned.
if we match #\d+, unichr(digits) will be returned.
Else, a unicode string will be returned."""
if match.group(1).startswith('#'): return chr(int(match.group(1)[1:]))
replacement = html.entities.entitydefs.get(match.group(1), "&%s;" % match.group(1))
return replacement
return str(entity_re.sub(matchFunc, s))
"""Converts any html entities in s to their unicode-decoded equivalents and returns a string."""
entity_re = re.compile(r"&(#\d+|\w+);")
def matchFunc(match):
"""Nested function to handle a match object.
If we match &blah; and it's not found, &blah; will be returned.
if we match #\d+, unichr(digits) will be returned.
Else, a unicode string will be returned."""
if match.group(1).startswith('#'): return chr(int(match.group(1)[1:]))
replacement = html.entities.entitydefs.get(match.group(1), "&%s;" % match.group(1))
return replacement
return str(entity_re.sub(matchFunc, s))
chars = "abcdefghijklmnopqrstuvwxyz"
def compose_tweet(tweet, db, relative_times, show_screen_names=False, session=None):
""" It receives a tweet and returns a list with the user, text for the tweet or message, date and the client where user is."""
if system == "Windows":
original_date = arrow.get(tweet.created_at, locale="en")
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = tweet.created_at
if hasattr(tweet, "message"):
value = "message"
elif hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
if hasattr(tweet, "retweeted_status") and value != "message":
text = StripChars(getattr(tweet.retweeted_status, value))
else:
text = StripChars(getattr(tweet, value))
if show_screen_names:
user = tweet.user.screen_name
else:
user = tweet.user.name
source = re.sub(r"(?s)<.*?>", "", tweet.source)
if hasattr(tweet, "retweeted_status"):
if (hasattr(tweet, "message")) == False and tweet.retweeted_status.is_quote_status == False:
text = "RT @%s: %s" % (tweet.retweeted_status.user.screen_name, text)
elif tweet.retweeted_status.is_quote_status:
text = "%s" % (text)
else:
text = "RT @%s: %s" % (tweet.retweeted_status.user.screen_name, text)
if not hasattr(tweet, "message"):
""" It receives a tweet and returns a list with the user, text for the tweet or message, date and the client where user is."""
if system == "Windows":
original_date = arrow.get(tweet.created_at, locale="en")
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = tweet.created_at
if hasattr(tweet, "message"):
value = "message"
elif hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
if hasattr(tweet, "retweeted_status") and value != "message":
text = StripChars(getattr(tweet.retweeted_status, value))
else:
text = StripChars(getattr(tweet, value))
if show_screen_names:
user = tweet.user.screen_name
else:
user = tweet.user.name
source = re.sub(r"(?s)<.*?>", "", tweet.source)
if hasattr(tweet, "retweeted_status"):
if (hasattr(tweet, "message")) == False and tweet.retweeted_status.is_quote_status == False:
text = "RT @%s: %s" % (tweet.retweeted_status.user.screen_name, text)
elif tweet.retweeted_status.is_quote_status:
text = "%s" % (text)
else:
text = "RT @%s: %s" % (tweet.retweeted_status.user.screen_name, text)
if not hasattr(tweet, "message"):
if hasattr(tweet, "retweeted_status"):
text = utils.expand_urls(text, tweet.retweeted_status.entities)
else:
text = utils.expand_urls(text, tweet.entities)
if config.app['app-settings']['handle_longtweets']: pass
return [user+", ", text, ts+", ", source]
if hasattr(tweet, "retweeted_status"):
text = utils.expand_urls(text, tweet.retweeted_status.entities)
else:
text = utils.expand_urls(text, tweet.entities)
if config.app['app-settings']['handle_longtweets']: pass
return [user+", ", text, ts+", ", source]
def compose_direct_message(item, db, relative_times, show_screen_names=False, session=None):
if system == "Windows":
# Let's remove the last 3 digits in the timestamp string.
# Twitter sends their "epoch" timestamp with 3 digits for milliseconds and arrow doesn't like it.
original_date = arrow.get(int(item.created_timestamp))
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = item.created_timestamp
text = StripChars(item.message_create["message_data"]["text"])
source = "DM"
sender = session.get_user(item.message_create["sender_id"])
if db["user_name"] == sender.screen_name:
if show_screen_names:
user = _(u"Dm to %s ") % (session.get_user(item.message_create["target"]["recipient_id"]).screen_name)
else:
user = _(u"Dm to %s ") % (session.get_user(item.message_create["target"]["recipient_id"]).name)
else:
if show_screen_names:
user = sender.screen_name
else:
user = sender.name
if text[-1] in chars: text=text+"."
text = utils.expand_urls(text, item.message_create["message_data"]["entities"])
return [user+", ", text, ts+", ", source]
if system == "Windows":
# Let's remove the last 3 digits in the timestamp string.
# Twitter sends their "epoch" timestamp with 3 digits for milliseconds and arrow doesn't like it.
original_date = arrow.get(int(item.created_timestamp))
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = item.created_timestamp
text = StripChars(item.message_create["message_data"]["text"])
source = "DM"
sender = session.get_user(item.message_create["sender_id"])
if db["user_name"] == sender.screen_name:
if show_screen_names:
user = _(u"Dm to %s ") % (session.get_user(item.message_create["target"]["recipient_id"]).screen_name)
else:
user = _(u"Dm to %s ") % (session.get_user(item.message_create["target"]["recipient_id"]).name)
else:
if show_screen_names:
user = sender.screen_name
else:
user = sender.name
if text[-1] in chars: text=text+"."
text = utils.expand_urls(text, item.message_create["message_data"]["entities"])
return [user+", ", text, ts+", ", source]
def compose_quoted_tweet(quoted_tweet, original_tweet, show_screen_names=False, session=None):
""" It receives a tweet and returns a list with the user, text for the tweet or message, date and the client where user is."""
if hasattr(quoted_tweet, "retweeted_status"):
if hasattr(quoted_tweet.retweeted_status, "full_text"):
value = "full_text"
else:
value = "text"
text = StripChars(getattr(quoted_tweet.retweeted_status, value))
else:
if hasattr(quoted_tweet, "full_text"):
value = "full_text"
else:
value = "text"
text = StripChars(getattr(quoted_tweet, value))
if show_screen_names:
quoting_user = quoted_tweet.user.screen_name
else:
quoting_user = quoted_tweet.user.name
source = quoted_tweet.source
if hasattr(quoted_tweet, "retweeted_status"):
text = "rt @%s: %s" % (quoted_tweet.retweeted_status.user.screen_name, text)
if text[-1] in chars: text=text+"."
original_user = original_tweet.user.screen_name
if hasattr(original_tweet, "message"):
original_text = original_tweet.message
elif hasattr(original_tweet, "full_text"):
original_text = StripChars(original_tweet.full_text)
else:
original_text = StripChars(original_tweet.text)
quoted_tweet.message = _(u"{0}. Quoted tweet from @{1}: {2}").format( text, original_user, original_text)
quoted_tweet = tweets.clear_url(quoted_tweet)
quoted_tweet.entities["urls"].extend(original_tweet.entities["urls"])
return quoted_tweet
""" It receives a tweet and returns a list with the user, text for the tweet or message, date and the client where user is."""
if hasattr(quoted_tweet, "retweeted_status"):
if hasattr(quoted_tweet.retweeted_status, "full_text"):
value = "full_text"
else:
value = "text"
text = StripChars(getattr(quoted_tweet.retweeted_status, value))
else:
if hasattr(quoted_tweet, "full_text"):
value = "full_text"
else:
value = "text"
text = StripChars(getattr(quoted_tweet, value))
if show_screen_names:
quoting_user = quoted_tweet.user.screen_name
else:
quoting_user = quoted_tweet.user.name
source = quoted_tweet.source
if hasattr(quoted_tweet, "retweeted_status"):
text = "rt @%s: %s" % (quoted_tweet.retweeted_status.user.screen_name, text)
if text[-1] in chars: text=text+"."
original_user = original_tweet.user.screen_name
if hasattr(original_tweet, "message"):
original_text = original_tweet.message
elif hasattr(original_tweet, "full_text"):
original_text = StripChars(original_tweet.full_text)
else:
original_text = StripChars(original_tweet.text)
quoted_tweet.message = _(u"{0}. Quoted tweet from @{1}: {2}").format( text, original_user, original_text)
quoted_tweet = tweets.clear_url(quoted_tweet)
quoted_tweet.entities["urls"].extend(original_tweet.entities["urls"])
return quoted_tweet
def compose_followers_list(tweet, db, relative_times=True, show_screen_names=False, session=None):
if system == "Windows":
original_date = arrow.get(tweet.created_at, locale="en")
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = tweet.created_at
if hasattr(tweet, "status"):
if system == "Windows":
original_date2 = arrow.get(tweet.status.created_at, locale="en")
if relative_times:
ts2 = original_date2.humanize(locale=languageHandler.curLang[:2])
else:
ts2 = original_date2.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts2 = _("Unavailable")
else:
ts2 = _("Unavailable")
return [_(u"%s (@%s). %s followers, %s friends, %s tweets. Last tweeted %s. Joined Twitter %s") % (tweet.name, tweet.screen_name, tweet.followers_count, tweet.friends_count, tweet.statuses_count, ts2, ts)]
if system == "Windows":
original_date = arrow.get(tweet.created_at, locale="en")
if relative_times == True:
ts = original_date.humanize(locale=languageHandler.curLang[:2])
else:
ts = original_date.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts = tweet.created_at
if hasattr(tweet, "status"):
if system == "Windows":
original_date2 = arrow.get(tweet.status.created_at, locale="en")
if relative_times:
ts2 = original_date2.humanize(locale=languageHandler.curLang[:2])
else:
ts2 = original_date2.shift(seconds=db["utc_offset"]).format(_(u"dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
else:
ts2 = _("Unavailable")
else:
ts2 = _("Unavailable")
return [_(u"%s (@%s). %s followers, %s friends, %s tweets. Last tweeted %s. Joined Twitter %s") % (tweet.name, tweet.screen_name, tweet.followers_count, tweet.friends_count, tweet.statuses_count, ts2, ts)]
def compose_list(list):
name = list.name
if list.description == None: description = _(u"No description available")
else: description = list.description
user = list.user.name
members = str(list.member_count)
if list.mode == "private": status = _(u"private")
else: status = _(u"public")
return [name, description, user, members, status]
name = list.name
if list.description == None: description = _(u"No description available")
else: description = list.description
user = list.user.name
members = str(list.member_count)
if list.mode == "private": status = _(u"private")
else: status = _(u"public")
return [name, description, user, members, status]

2
src/sessions/twitter/long_tweets/__init__.py
View File

@@ -1,3 +1,3 @@
# -*- coding: utf-8 -*-
""" this package holds different modules to extract information regarding long tweets. A long tweet contains more than one tweet (such a quoted tweet), or is made via services like twishort."""
from __future__ import unicode_literals
from __future__ import unicode_literals

56
src/sessions/twitter/long_tweets/tweets.py
View File

@@ -20,33 +20,33 @@ from __future__ import unicode_literals
from sessions.twitter import utils
def is_long(tweet):
""" Check if the passed tweet contains a quote in its metadata.
tweet dict: a tweet dictionary.
returns True if a quote is detected, False otherwise."""
if hasattr(tweet, "quoted_status_id") and hasattr(tweet, "quoted_status"):
return tweet.quoted_status_id
elif hasattr(tweet, "retweeted_status") and hasattr(tweet.retweeted_status, "quoted_status_id") and hasattr(tweet.retweeted_status, "quoted_status"):
return tweet.retweeted_status.quoted_status_id
return False
""" Check if the passed tweet contains a quote in its metadata.
tweet dict: a tweet dictionary.
returns True if a quote is detected, False otherwise."""
if hasattr(tweet, "quoted_status_id") and hasattr(tweet, "quoted_status"):
return tweet.quoted_status_id
elif hasattr(tweet, "retweeted_status") and hasattr(tweet.retweeted_status, "quoted_status_id") and hasattr(tweet.retweeted_status, "quoted_status"):
return tweet.retweeted_status.quoted_status_id
return False
def clear_url(tweet):
""" Reads data from a quoted tweet and removes the link to the Status from the tweet's text.
tweet dict: a tweet dictionary.
returns a tweet dictionary without the URL to the status ID in its text to display."""
if hasattr(tweet, "retweeted_status"):
if hasattr(tweet.retweeted_status, "full_text"):
value = "full_text"
else:
value = "text"
urls = utils.find_urls_in_text(getattr(tweet.retweeted_status, value))
try: tweet.message = tweet.message.replace(urls[-1], "")
except IndexError: pass
else:
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
urls = utils.find_urls_in_text(getattr(tweet, value))
try: tweet.message = tweet.message.replace(urls[-1], "")
except IndexError: pass
return tweet
""" Reads data from a quoted tweet and removes the link to the Status from the tweet's text.
tweet dict: a tweet dictionary.
returns a tweet dictionary without the URL to the status ID in its text to display."""
if hasattr(tweet, "retweeted_status"):
if hasattr(tweet.retweeted_status, "full_text"):
value = "full_text"
else:
value = "text"
urls = utils.find_urls_in_text(getattr(tweet.retweeted_status, value))
try: tweet.message = tweet.message.replace(urls[-1], "")
except IndexError: pass
else:
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
urls = utils.find_urls_in_text(getattr(tweet, value))
try: tweet.message = tweet.message.replace(urls[-1], "")
except IndexError: pass
return tweet

136
src/sessions/twitter/long_tweets/twishort.py
View File

@@ -28,77 +28,77 @@ from sessions.twitter import utils
log = logging.getLogger("long_tweets.twishort")
def get_twishort_uri(url):
""" Takes A twishort URl and returns the twishort ID.
url str: an url like http://twishort.com/id.
returns a twishort ID if the URL is valid, False otherwise."""
try:
return url.split("twishort.com/")[1]
except ValueError:
return False
""" Takes A twishort URl and returns the twishort ID.
url str: an url like http://twishort.com/id.
returns a twishort ID if the URL is valid, False otherwise."""
try:
return url.split("twishort.com/")[1]
except ValueError:
return False
def is_long(tweet):
""" Check if the passed tweet is made with Twishort.
returns True if is a long tweet, False otherwise."""
long = False
for url in range(0, len(tweet.entities["urls"])):
try:
if tweet.entities["urls"][url] != None and "twishort.com" in tweet.entities["urls"][url]["expanded_url"]:
long = get_twishort_uri(tweet.entities["urls"][url]["expanded_url"])
except IndexError:
pass
# sometimes Twitter returns URL's with None objects, so let's take it.
# see https://github.com/manuelcortez/TWBlue/issues/103
except TypeError:
pass
if long == False and hasattr(tweet, "retweeted_status"):
for url in range(0, len(tweet.retweeted_status.entities["urls"])):
try:
if tweet.retweeted_status.entities["urls"][url] != None and "twishort.com" in tweet.retweeted_status.entities["urls"][url]["expanded_url"]:
long = get_twishort_uri(tweet.retweeted_status.entities["urls"][url]["expanded_url"])
except IndexError:
pass
except TypeError:
pass
return long
""" Check if the passed tweet is made with Twishort.
returns True if is a long tweet, False otherwise."""
long = False
for url in range(0, len(tweet.entities["urls"])):
try:
if tweet.entities["urls"][url] != None and "twishort.com" in tweet.entities["urls"][url]["expanded_url"]:
long = get_twishort_uri(tweet.entities["urls"][url]["expanded_url"])
except IndexError:
pass
# sometimes Twitter returns URL's with None objects, so let's take it.
# see https://github.com/manuelcortez/TWBlue/issues/103
except TypeError:
pass
if long == False and hasattr(tweet, "retweeted_status"):
for url in range(0, len(tweet.retweeted_status.entities["urls"])):
try:
if tweet.retweeted_status.entities["urls"][url] != None and "twishort.com" in tweet.retweeted_status.entities["urls"][url]["expanded_url"]:
long = get_twishort_uri(tweet.retweeted_status.entities["urls"][url]["expanded_url"])
except IndexError:
pass
except TypeError:
pass
return long
def get_full_text(uri):
""" Get Twishort's full text.
uri str: Twishort's identifier.
returns the contents of the tweet."""
try:
r = requests.get("http://api.twishort.com/1.1/get.json", params={"uri": uri, "api_key": keys.keyring.get("twishort_api_key")})
msg = r.json()["text"]
# Try to parse possible HTML entities.
from sessions.twitter.compose import StripChars
msg = StripChars(msg)
return msg
except:
return False
""" Get Twishort's full text.
uri str: Twishort's identifier.
returns the contents of the tweet."""
try:
r = requests.get("http://api.twishort.com/1.1/get.json", params={"uri": uri, "api_key": keys.keyring.get("twishort_api_key")})
msg = r.json()["text"]
# Try to parse possible HTML entities.
from sessions.twitter.compose import StripChars
msg = StripChars(msg)
return msg
except:
return False
def create_tweet(user_token, user_secret, text, media=0):
""" Send a tweet to be extended by using Twishort.
user_token, user_secret str: Twitter user access key and secret, used by TWBlue to authorise against Twitter.
text str: Tweet text, max 10000 characters.
media int: Not used currently.
Returns text to be placed in the Tweet if the post has been succeeded, 0 otherwise."""
twitter = OAuth1Session(keys.keyring.get("api_key"), client_secret=keys.keyring.get("api_secret"), resource_owner_key=user_token, resource_owner_secret=user_secret)
twishort_key=keys.keyring.get("twishort_api_key")
x_auth_service_provider = "https://api.twitter.com/1.1/account/verify_credentials.json"
twishort_post_url = "http://api.twishort.com/1.1/post.json"
twishort_update_ids_url = "http://api.twishort.com/1.1/update_ids.json"
r=requests.Request('GET', x_auth_service_provider)
prep=twitter.prepare_request(r)
resp=twitter.send(prep)
twitter.headers={
'X-Auth-Service-Provider':x_auth_service_provider,
'X-Verify-Credentials-Authorization':prep.headers['Authorization'],
}
data = {'api_key':twishort_key,
"text": text.encode("utf-8"),
"media": media}
response = twitter.post(twishort_post_url, data=data)
try:
return response.json()["text_to_tweet"]
except:
print("There was a problem creating a long tweet")
return 0
""" Send a tweet to be extended by using Twishort.
user_token, user_secret str: Twitter user access key and secret, used by TWBlue to authorise against Twitter.
text str: Tweet text, max 10000 characters.
media int: Not used currently.
Returns text to be placed in the Tweet if the post has been succeeded, 0 otherwise."""
twitter = OAuth1Session(keys.keyring.get("api_key"), client_secret=keys.keyring.get("api_secret"), resource_owner_key=user_token, resource_owner_secret=user_secret)
twishort_key=keys.keyring.get("twishort_api_key")
x_auth_service_provider = "https://api.twitter.com/1.1/account/verify_credentials.json"
twishort_post_url = "http://api.twishort.com/1.1/post.json"
twishort_update_ids_url = "http://api.twishort.com/1.1/update_ids.json"
r=requests.Request('GET', x_auth_service_provider)
prep=twitter.prepare_request(r)
resp=twitter.send(prep)
twitter.headers={
'X-Auth-Service-Provider':x_auth_service_provider,
'X-Verify-Credentials-Authorization':prep.headers['Authorization'],
}
data = {'api_key':twishort_key,
"text": text.encode("utf-8"),
"media": media}
response = twitter.post(twishort_post_url, data=data)
try:
return response.json()["text_to_tweet"]
except:
print("There was a problem creating a long tweet")
return 0

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

@@ -22,432 +22,432 @@ from .wxUI import authorisationDialog
log = logging.getLogger("sessions.twitterSession")
class Session(base.baseSession):
""" A session object where we will save configuration, the twitter object and a local storage for saving the items retrieved through the Twitter API methods"""
""" A session object where we will save configuration, the twitter object and a local storage for saving the items retrieved through the Twitter API methods"""
def order_buffer(self, name, data, ignore_older=True):
""" Put new items in the local database.
name str: The name for the buffer stored in the dictionary.
data list: A list with tweets.
ignore_older bool: if set to True, items older than the first element on the list will be ignored.
returns the number of items that have been added in this execution"""
if name == "direct_messages":
return self.order_direct_messages(data)
num = 0
last_id = None
if (name in self.db) == False:
self.db[name] = []
if ("users" in self.db) == False:
self.db["users"] = {}
if ignore_older and len(self.db[name]) > 0:
if self.settings["general"]["reverse_timelines"] == False:
last_id = self.db[name][0].id
else:
last_id = self.db[name][-1].id
for i in data:
if ignore_older and last_id != None:
if i.id < last_id:
log.error("Ignoring an older tweet... Last id: {0}, tweet id: {1}".format(last_id, i.id))
continue
if utils.find_item(i.id, self.db[name]) == None and utils.is_allowed(i, self.settings, name) == True:
i = self.check_quoted_status(i)
i = self.check_long_tweet(i)
if i == False: continue
if self.settings["general"]["reverse_timelines"] == False: self.db[name].append(i)
else: self.db[name].insert(0, i)
num = num+1
if hasattr(i, "user"):
if (i.user.id in self.db["users"]) == False:
self.db["users"][i.user.id] = i.user
return num
def order_buffer(self, name, data, ignore_older=True):
""" Put new items in the local database.
name str: The name for the buffer stored in the dictionary.
data list: A list with tweets.
ignore_older bool: if set to True, items older than the first element on the list will be ignored.
returns the number of items that have been added in this execution"""
if name == "direct_messages":
return self.order_direct_messages(data)
num = 0
last_id = None
if (name in self.db) == False:
self.db[name] = []
if ("users" in self.db) == False:
self.db["users"] = {}
if ignore_older and len(self.db[name]) > 0:
if self.settings["general"]["reverse_timelines"] == False:
last_id = self.db[name][0].id
else:
last_id = self.db[name][-1].id
for i in data:
if ignore_older and last_id != None:
if i.id < last_id:
log.error("Ignoring an older tweet... Last id: {0}, tweet id: {1}".format(last_id, i.id))
continue
if utils.find_item(i.id, self.db[name]) == None and utils.is_allowed(i, self.settings, name) == True:
i = self.check_quoted_status(i)
i = self.check_long_tweet(i)
if i == False: continue
if self.settings["general"]["reverse_timelines"] == False: self.db[name].append(i)
else: self.db[name].insert(0, i)
num = num+1
if hasattr(i, "user"):
if (i.user.id in self.db["users"]) == False:
self.db["users"][i.user.id] = i.user
return num
def order_people(self, name, data):
""" Put new items on the local database. Useful for cursored buffers (followers, friends, users of a list and searches)
name str: The name for the buffer stored in the dictionary.
data list: A list with items and some information about cursors.
returns the number of items that have been added in this execution"""
num = 0
if (name in self.db) == False:
self.db[name] = []
for i in data:
if utils.find_item(i.id, self.db[name]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db[name].append(i)
else: self.db[name].insert(0, i)
num = num+1
return num
def order_people(self, name, data):
""" Put new items on the local database. Useful for cursored buffers (followers, friends, users of a list and searches)
name str: The name for the buffer stored in the dictionary.
data list: A list with items and some information about cursors.
returns the number of items that have been added in this execution"""
num = 0
if (name in self.db) == False:
self.db[name] = []
for i in data:
if utils.find_item(i.id, self.db[name]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db[name].append(i)
else: self.db[name].insert(0, i)
num = num+1
return num
def order_direct_messages(self, data):
""" Add incoming and sent direct messages to their corresponding database items.
data list: A list of direct messages to add.
returns the number of incoming messages processed in this execution, and sends an event with data regarding amount of sent direct messages added."""
incoming = 0
sent = 0
if ("direct_messages" in self.db) == False:
self.db["direct_messages"] = []
for i in data:
# Twitter returns sender_id as str, which must be converted to int in order to match to our user_id object.
if int(i.message_create["sender_id"]) == self.db["user_id"]:
if "sent_direct_messages" in self.db and utils.find_item(i.id, self.db["sent_direct_messages"]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db["sent_direct_messages"].append(i)
else: self.db["sent_direct_messages"].insert(0, i)
sent = sent+1
else:
if utils.find_item(i.id, self.db["direct_messages"]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db["direct_messages"].append(i)
else: self.db["direct_messages"].insert(0, i)
incoming = incoming+1
pub.sendMessage("sent-dms-updated", total=sent, account=self.db["user_name"])
return incoming
def order_direct_messages(self, data):
""" Add incoming and sent direct messages to their corresponding database items.
data list: A list of direct messages to add.
returns the number of incoming messages processed in this execution, and sends an event with data regarding amount of sent direct messages added."""
incoming = 0
sent = 0
if ("direct_messages" in self.db) == False:
self.db["direct_messages"] = []
for i in data:
# Twitter returns sender_id as str, which must be converted to int in order to match to our user_id object.
if int(i.message_create["sender_id"]) == self.db["user_id"]:
if "sent_direct_messages" in self.db and utils.find_item(i.id, self.db["sent_direct_messages"]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db["sent_direct_messages"].append(i)
else: self.db["sent_direct_messages"].insert(0, i)
sent = sent+1
else:
if utils.find_item(i.id, self.db["direct_messages"]) == None:
if self.settings["general"]["reverse_timelines"] == False: self.db["direct_messages"].append(i)
else: self.db["direct_messages"].insert(0, i)
incoming = incoming+1
pub.sendMessage("sent-dms-updated", total=sent, account=self.db["user_name"])
return incoming
def __init__(self, *args, **kwargs):
super(Session, self).__init__(*args, **kwargs)
# Adds here the optional cursors objects.
cursors = dict(direct_messages=-1)
self.db["cursors"] = cursors
self.reconnection_function_active = False
self.counter = 0
self.lists = []
def __init__(self, *args, **kwargs):
super(Session, self).__init__(*args, **kwargs)
# Adds here the optional cursors objects.
cursors = dict(direct_messages=-1)
self.db["cursors"] = cursors
self.reconnection_function_active = False
self.counter = 0
self.lists = []
# @_require_configuration
def login(self, verify_credentials=True):
""" Log into twitter using credentials from settings.
if the user account isn't authorised, it needs to call self.authorise() before login."""
if self.settings["twitter"]["user_key"] != None and self.settings["twitter"]["user_secret"] != None:
try:
log.debug("Logging in to twitter...")
self.auth = tweepy.OAuthHandler(keyring.get("api_key"), keyring.get("api_secret"))
self.auth.set_access_token(self.settings["twitter"]["user_key"], self.settings["twitter"]["user_secret"])
self.twitter = tweepy.API(self.auth)
if verify_credentials == True:
self.credentials = self.twitter.verify_credentials()
self.logged = True
log.debug("Logged.")
self.counter = 0
except IOError:
log.error("The login attempt failed.")
self.logged = False
else:
self.logged = False
raise Exceptions.RequireCredentialsSessionError
def login(self, verify_credentials=True):
""" Log into twitter using credentials from settings.
if the user account isn't authorised, it needs to call self.authorise() before login."""
if self.settings["twitter"]["user_key"] != None and self.settings["twitter"]["user_secret"] != None:
try:
log.debug("Logging in to twitter...")
self.auth = tweepy.OAuthHandler(keyring.get("api_key"), keyring.get("api_secret"))
self.auth.set_access_token(self.settings["twitter"]["user_key"], self.settings["twitter"]["user_secret"])
self.twitter = tweepy.API(self.auth)
if verify_credentials == True:
self.credentials = self.twitter.verify_credentials()
self.logged = True
log.debug("Logged.")
self.counter = 0
except IOError:
log.error("The login attempt failed.")
self.logged = False
else:
self.logged = False
raise Exceptions.RequireCredentialsSessionError
# @_require_configuration
def authorise(self):
""" Authorises a Twitter account. This function needs to be called for each new session, after self.get_configuration() and before self.login()"""
if self.logged == True:
raise Exceptions.AlreadyAuthorisedError("The authorisation process is not needed at this time.")
else:
self.auth = tweepy.OAuthHandler(keyring.get("api_key"), keyring.get("api_secret"))
redirect_url = self.auth.get_authorization_url()
webbrowser.open_new_tab(redirect_url)
self.authorisation_dialog = authorisationDialog()
self.authorisation_dialog.cancel.Bind(wx.EVT_BUTTON, self.authorisation_cancelled)
self.authorisation_dialog.ok.Bind(wx.EVT_BUTTON, self.authorisation_accepted)
self.authorisation_dialog.ShowModal()
def authorise(self):
""" Authorises a Twitter account. This function needs to be called for each new session, after self.get_configuration() and before self.login()"""
if self.logged == True:
raise Exceptions.AlreadyAuthorisedError("The authorisation process is not needed at this time.")
else:
self.auth = tweepy.OAuthHandler(keyring.get("api_key"), keyring.get("api_secret"))
redirect_url = self.auth.get_authorization_url()
webbrowser.open_new_tab(redirect_url)
self.authorisation_dialog = authorisationDialog()
self.authorisation_dialog.cancel.Bind(wx.EVT_BUTTON, self.authorisation_cancelled)
self.authorisation_dialog.ok.Bind(wx.EVT_BUTTON, self.authorisation_accepted)
self.authorisation_dialog.ShowModal()
def verify_authorisation(self, pincode):
self.auth.get_access_token(pincode)
self.settings["twitter"]["user_key"] = self.auth.access_token
self.settings["twitter"]["user_secret"] = self.auth.access_token_secret
self.settings.write()
del self.auth
def verify_authorisation(self, pincode):
self.auth.get_access_token(pincode)
self.settings["twitter"]["user_key"] = self.auth.access_token
self.settings["twitter"]["user_secret"] = self.auth.access_token_secret
self.settings.write()
del self.auth
def authorisation_cancelled(self, *args, **kwargs):
""" Destroy the authorization dialog. """
self.authorisation_dialog.Destroy()
del self.authorisation_dialog
def authorisation_cancelled(self, *args, **kwargs):
""" Destroy the authorization dialog. """
self.authorisation_dialog.Destroy()
del self.authorisation_dialog
def authorisation_accepted(self, *args, **kwargs):
""" Gets the PIN code entered by user and validate it through Twitter."""
pincode = self.authorisation_dialog.text.GetValue()
self.verify_authorisation(pincode)
self.authorisation_dialog.Destroy()
def authorisation_accepted(self, *args, **kwargs):
""" Gets the PIN code entered by user and validate it through Twitter."""
pincode = self.authorisation_dialog.text.GetValue()
self.verify_authorisation(pincode)
self.authorisation_dialog.Destroy()
def get_more_items(self, update_function, users=False, dm=False, name=None, *args, **kwargs):
""" Get more items for twitter objects.
update_function str: function to call for getting more items. Must be member of self.twitter.
users, dm bool: If any of these is set to True, the function will treat items as users or dm (they need different handling).
name str: name of the database item to put new element in."""
results = []
if "cursor" in kwargs and kwargs["cursor"] == 0:
output.speak(_(u"There are no more items to retrieve in this buffer."))
return
data = getattr(self.twitter, update_function)(*args, **kwargs)
if users == True:
if type(data) == dict and "next_cursor" in data:
if "next_cursor" in data: # There are more objects to retrieve.
self.db[name]["cursor"] = data["next_cursor"]
else: # Set cursor to 0, wich means no more items available.
self.db[name]["cursor"] = 0
for i in data["users"]: results.append(i)
elif type(data) == list:
results.extend(data[1:])
elif dm == True:
if "next_cursor" in data: # There are more objects to retrieve.
self.db[name]["cursor"] = data["next_cursor"]
else: # Set cursor to 0, wich means no more items available.
self.db[name]["cursor"] = 0
for i in data["events"]: results.append(i)
else:
results.extend(data[1:])
return results
def get_more_items(self, update_function, users=False, dm=False, name=None, *args, **kwargs):
""" Get more items for twitter objects.
update_function str: function to call for getting more items. Must be member of self.twitter.
users, dm bool: If any of these is set to True, the function will treat items as users or dm (they need different handling).
name str: name of the database item to put new element in."""
results = []
if "cursor" in kwargs and kwargs["cursor"] == 0:
output.speak(_(u"There are no more items to retrieve in this buffer."))
return
data = getattr(self.twitter, update_function)(*args, **kwargs)
if users == True:
if type(data) == dict and "next_cursor" in data:
if "next_cursor" in data: # There are more objects to retrieve.
self.db[name]["cursor"] = data["next_cursor"]
else: # Set cursor to 0, wich means no more items available.
self.db[name]["cursor"] = 0
for i in data["users"]: results.append(i)
elif type(data) == list:
results.extend(data[1:])
elif dm == True:
if "next_cursor" in data: # There are more objects to retrieve.
self.db[name]["cursor"] = data["next_cursor"]
else: # Set cursor to 0, wich means no more items available.
self.db[name]["cursor"] = 0
for i in data["events"]: results.append(i)
else:
results.extend(data[1:])
return results
def api_call(self, call_name, action="", _sound=None, report_success=False, report_failure=True, preexec_message="", *args, **kwargs):
""" Make a call to the Twitter API. If there is a connectionError or another exception not related to Twitter, It will call the method again at least 25 times, waiting a while between calls. Useful for post methods.
If twitter returns an error, it will not call the method anymore.
call_name str: The method to call
action str: What you are doing on twitter, it will be reported to the user if report_success is set to True.
for example "following @tw_blue2" will be reported as "following @tw_blue2 succeeded".
_sound str: a sound to play if the call is executed properly.
report_success and report_failure bool: These are self explanatory. True or False.
preexec_message str: A message to speak to the user while the method is running, example: "trying to follow x user"."""
finished = False
tries = 0
if preexec_message:
output.speak(preexec_message, True)
while finished==False and tries < 25:
try:
val = getattr(self.twitter, call_name)(*args, **kwargs)
finished = True
except TweepError as e:
output.speak(e.reason)
val = None
if e.error_code != 403 and e.error_code != 404:
tries = tries+1
time.sleep(5)
elif report_failure and hasattr(e, 'reason'):
output.speak(_("%s failed. Reason: %s") % (action, e.reason))
finished = True
def api_call(self, call_name, action="", _sound=None, report_success=False, report_failure=True, preexec_message="", *args, **kwargs):
""" Make a call to the Twitter API. If there is a connectionError or another exception not related to Twitter, It will call the method again at least 25 times, waiting a while between calls. Useful for post methods.
If twitter returns an error, it will not call the method anymore.
call_name str: The method to call
action str: What you are doing on twitter, it will be reported to the user if report_success is set to True.
for example "following @tw_blue2" will be reported as "following @tw_blue2 succeeded".
_sound str: a sound to play if the call is executed properly.
report_success and report_failure bool: These are self explanatory. True or False.
preexec_message str: A message to speak to the user while the method is running, example: "trying to follow x user"."""
finished = False
tries = 0
if preexec_message:
output.speak(preexec_message, True)
while finished==False and tries < 25:
try:
val = getattr(self.twitter, call_name)(*args, **kwargs)
finished = True
except TweepError as e:
output.speak(e.reason)
val = None
if e.error_code != 403 and e.error_code != 404:
tries = tries+1
time.sleep(5)
elif report_failure and hasattr(e, 'reason'):
output.speak(_("%s failed. Reason: %s") % (action, e.reason))
finished = True
# except:
# tries = tries + 1
# time.sleep(5)
if report_success:
output.speak(_("%s succeeded.") % action)
if _sound != None: self.sound.play(_sound)
return val
if report_success:
output.speak(_("%s succeeded.") % action)
if _sound != None: self.sound.play(_sound)
return val
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.reverse()
return tl
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.reverse()
return tl
# @_require_login
def get_favourites_timeline(self, name, *args, **kwargs):
""" Gets favourites for the authenticated user or a friend or follower.
name str: Name for storage in the database.
args and kwargs are passed directly to the Twython function."""
tl = self.call_paged("favorites", *args, **kwargs)
return self.order_buffer(name, tl)
def get_favourites_timeline(self, name, *args, **kwargs):
""" Gets favourites for the authenticated user or a friend or follower.
name str: Name for storage in the database.
args and kwargs are passed directly to the Twython function."""
tl = self.call_paged("favorites", *args, **kwargs)
return self.order_buffer(name, tl)
def call_paged(self, update_function, *args, **kwargs):
""" Makes a call to the Twitter API methods several times. Useful for get methods.
this function is needed for retrieving more than 200 items.
update_function str: The function to call. This function must be child of self.twitter
args and kwargs are passed to update_function.
returns a list with all items retrieved."""
max = 0
results = []
data = getattr(self.twitter, update_function)(count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
results.extend(data)
for i in range(0, max):
if i == 0: max_id = results[-1].id
else: max_id = results[0].id
data = getattr(self.twitter, update_function)(max_id=max_id, count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
results.extend(data)
results.reverse()
return results
def call_paged(self, update_function, *args, **kwargs):
""" Makes a call to the Twitter API methods several times. Useful for get methods.
this function is needed for retrieving more than 200 items.
update_function str: The function to call. This function must be child of self.twitter
args and kwargs are passed to update_function.
returns a list with all items retrieved."""
max = 0
results = []
data = getattr(self.twitter, update_function)(count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
results.extend(data)
for i in range(0, max):
if i == 0: max_id = results[-1].id
else: max_id = results[0].id
data = getattr(self.twitter, update_function)(max_id=max_id, count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
results.extend(data)
results.reverse()
return results
# @_require_login
def get_user_info(self):
""" Retrieves some information required by TWBlue for setup."""
f = self.twitter.get_settings()
sn = f["screen_name"]
self.settings["twitter"]["user_name"] = sn
self.db["user_name"] = sn
self.db["user_id"] = self.twitter.get_user(screen_name=sn).id
try:
self.db["utc_offset"] = f["time_zone"]["utc_offset"]
except KeyError:
self.db["utc_offset"] = -time.timezone
# Get twitter's supported languages and save them in a global variable
#so we won't call to this method once per session.
if len(application.supported_languages) == 0:
application.supported_languages = self.twitter.supported_languages()
self.get_lists()
self.get_muted_users()
self.settings.write()
def get_user_info(self):
""" Retrieves some information required by TWBlue for setup."""
f = self.twitter.get_settings()
sn = f["screen_name"]
self.settings["twitter"]["user_name"] = sn
self.db["user_name"] = sn
self.db["user_id"] = self.twitter.get_user(screen_name=sn).id
try:
self.db["utc_offset"] = f["time_zone"]["utc_offset"]
except KeyError:
self.db["utc_offset"] = -time.timezone
# Get twitter's supported languages and save them in a global variable
#so we won't call to this method once per session.
if len(application.supported_languages) == 0:
application.supported_languages = self.twitter.supported_languages()
self.get_lists()
self.get_muted_users()
self.settings.write()
# @_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)
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)
# @_require_login
def get_muted_users(self):
""" Gets muted users (oh really?)."""
self.db["muted_users"] = self.twitter.mutes_ids()
def get_muted_users(self):
""" Gets muted users (oh really?)."""
self.db["muted_users"] = self.twitter.mutes_ids()
# @_require_login
def get_stream(self, name, function, *args, **kwargs):
""" Retrieves the items for a regular stream.
name str: Name to save items to the database.
function str: A function to get the items."""
last_id = -1
if name in self.db:
try:
if self.db[name][0]["id"] > self.db[name][-1]["id"]:
last_id = self.db[name][0]["id"]
else:
last_id = self.db[name][-1]["id"]
except IndexError:
pass
tl = self.call_paged(function, sinze_id=last_id, *args, **kwargs)
self.order_buffer(name, tl)
def get_stream(self, name, function, *args, **kwargs):
""" Retrieves the items for a regular stream.
name str: Name to save items to the database.
function str: A function to get the items."""
last_id = -1
if name in self.db:
try:
if self.db[name][0]["id"] > self.db[name][-1]["id"]:
last_id = self.db[name][0]["id"]
else:
last_id = self.db[name][-1]["id"]
except IndexError:
pass
tl = self.call_paged(function, sinze_id=last_id, *args, **kwargs)
self.order_buffer(name, tl)
def get_cursored_stream(self, name, function, items="users", get_previous=False, *args, **kwargs):
""" Gets items for API calls that require using cursors to paginate the results.
name str: Name to save it in the database.
function str: Function that provides the items.
items: When the function returns the list with results, items will tell how the order function should be look. for example get_followers_list returns a list and users are under list["users"], here the items should point to "users".
get_previous bool: wether this function will be used to get previous items in a buffer or load the buffer from scratch.
returns number of items retrieved."""
items_ = []
try:
if "cursor" in self.db[name] and get_previous:
cursor = self.db[name]["cursor"]
else:
cursor = -1
except KeyError:
cursor = -1
if cursor != -1:
tl = getattr(self.twitter, function)(cursor=cursor, count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
else:
tl = getattr(self.twitter, function)(count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
tl[items].reverse()
num = self.order_cursored_buffer(name, tl[items])
# Recently, Twitter's new endpoints have cursor if there are more results.
if "next_cursor" in tl:
self.db[name]["cursor"] = tl["next_cursor"]
else:
self.db[name]["cursor"] = 0
return num
def get_cursored_stream(self, name, function, items="users", get_previous=False, *args, **kwargs):
""" Gets items for API calls that require using cursors to paginate the results.
name str: Name to save it in the database.
function str: Function that provides the items.
items: When the function returns the list with results, items will tell how the order function should be look. for example get_followers_list returns a list and users are under list["users"], here the items should point to "users".
get_previous bool: wether this function will be used to get previous items in a buffer or load the buffer from scratch.
returns number of items retrieved."""
items_ = []
try:
if "cursor" in self.db[name] and get_previous:
cursor = self.db[name]["cursor"]
else:
cursor = -1
except KeyError:
cursor = -1
if cursor != -1:
tl = getattr(self.twitter, function)(cursor=cursor, count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
else:
tl = getattr(self.twitter, function)(count=self.settings["general"]["max_tweets_per_call"], *args, **kwargs)
tl[items].reverse()
num = self.order_cursored_buffer(name, tl[items])
# Recently, Twitter's new endpoints have cursor if there are more results.
if "next_cursor" in tl:
self.db[name]["cursor"] = tl["next_cursor"]
else:
self.db[name]["cursor"] = 0
return num
def check_connection(self):
""" Restart the Twitter object every 5 executions. It is useful for dealing with requests timeout and other oddities."""
log.debug("Executing check connection...")
self.counter += 1
if self.counter >= 4:
log.debug("Restarting connection after 5 minutes.")
del self.twitter
self.logged = False
self.login(False)
self.counter = 0
def check_connection(self):
""" Restart the Twitter object every 5 executions. It is useful for dealing with requests timeout and other oddities."""
log.debug("Executing check connection...")
self.counter += 1
if self.counter >= 4:
log.debug("Restarting connection after 5 minutes.")
del self.twitter
self.logged = False
self.login(False)
self.counter = 0
def check_quoted_status(self, tweet):
""" Helper for get_quoted_tweet. Get a quoted status inside a tweet and create a special tweet with all info available.
tweet dict: A tweet dictionary.
Returns a quoted tweet or the original tweet if is not a quote"""
status = tweets.is_long(tweet)
if status != False and config.app["app-settings"]["handle_longtweets"]:
quoted_tweet = self.get_quoted_tweet(tweet)
return quoted_tweet
return tweet
def check_quoted_status(self, tweet):
""" Helper for get_quoted_tweet. Get a quoted status inside a tweet and create a special tweet with all info available.
tweet dict: A tweet dictionary.
Returns a quoted tweet or the original tweet if is not a quote"""
status = tweets.is_long(tweet)
if status != False and config.app["app-settings"]["handle_longtweets"]:
quoted_tweet = self.get_quoted_tweet(tweet)
return quoted_tweet
return tweet
def get_quoted_tweet(self, tweet):
""" Process a tweet and extract all information related to the quote. """
quoted_tweet = tweet
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
setattr(quoted_tweet, value, utils.expand_urls(getattr(quoted_tweet, value), quoted_tweet.entities))
if quoted_tweet.is_quote_status == True and hasattr(quoted_tweet, "quoted_status"):
original_tweet = quoted_tweet.quoted_status
elif hasattr(quoted_tweet, "retweeted_status") and quoted_tweet.retweeted_status.is_quote_status == True and hasattr(quoted_tweet.retweeted_status, "quoted_status"):
original_tweet = quoted_tweet.retweeted_status.quoted_status
else:
return quoted_tweet
original_tweet = self.check_long_tweet(original_tweet)
if hasattr(original_tweet, "full_text"):
value = "full_text"
elif hasattr(original_tweet, "message"):
value = "message"
else:
value = "text"
setattr(original_tweet, value, utils.expand_urls(getattr(original_tweet, value), original_tweet.entities))
return compose.compose_quoted_tweet(quoted_tweet, original_tweet)
def get_quoted_tweet(self, tweet):
""" Process a tweet and extract all information related to the quote. """
quoted_tweet = tweet
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
setattr(quoted_tweet, value, utils.expand_urls(getattr(quoted_tweet, value), quoted_tweet.entities))
if quoted_tweet.is_quote_status == True and hasattr(quoted_tweet, "quoted_status"):
original_tweet = quoted_tweet.quoted_status
elif hasattr(quoted_tweet, "retweeted_status") and quoted_tweet.retweeted_status.is_quote_status == True and hasattr(quoted_tweet.retweeted_status, "quoted_status"):
original_tweet = quoted_tweet.retweeted_status.quoted_status
else:
return quoted_tweet
original_tweet = self.check_long_tweet(original_tweet)
if hasattr(original_tweet, "full_text"):
value = "full_text"
elif hasattr(original_tweet, "message"):
value = "message"
else:
value = "text"
setattr(original_tweet, value, utils.expand_urls(getattr(original_tweet, value), original_tweet.entities))
return compose.compose_quoted_tweet(quoted_tweet, original_tweet)
def check_long_tweet(self, tweet):
""" Process a tweet and add extra info if it's a long tweet made with Twyshort.
tweet dict: a tweet object.
returns a tweet with a new argument message, or original tweet if it's not a long tweet."""
long = twishort.is_long(tweet)
if long != False and config.app["app-settings"]["handle_longtweets"]:
message = twishort.get_full_text(long)
if hasattr(tweet, "quoted_status"):
tweet.quoted_status.message = message
if tweet.quoted_status.message == False: return False
tweet.quoted_status.twishort = True
for i in tweet.quoted_status.entities["user_mentions"]:
if "@%s" % (i["screen_name"]) not in tweet.quoted_status.message and i["screen_name"] != tweet.user.screen_name:
if hasattr(tweet.quoted_status, "retweeted_status") and tweet.retweeted_status.user.screen_name == i["screen_name"]:
continue
tweet.quoted_status.message = u"@%s %s" % (i["screen_name"], tweet.message)
else:
tweet.message = message
if tweet.message == False: return False
tweet.twishort = True
for i in tweet.entities["user_mentions"]:
if "@%s" % (i["screen_name"]) not in tweet.message and i["screen_name"] != tweet.user.screen_name:
if hasattr(tweet, "retweeted_status") and tweet.retweeted_status.user.screen_name == i["screen_name"]:
continue
return tweet
def check_long_tweet(self, tweet):
""" Process a tweet and add extra info if it's a long tweet made with Twyshort.
tweet dict: a tweet object.
returns a tweet with a new argument message, or original tweet if it's not a long tweet."""
long = twishort.is_long(tweet)
if long != False and config.app["app-settings"]["handle_longtweets"]:
message = twishort.get_full_text(long)
if hasattr(tweet, "quoted_status"):
tweet.quoted_status.message = message
if tweet.quoted_status.message == False: return False
tweet.quoted_status.twishort = True
for i in tweet.quoted_status.entities["user_mentions"]:
if "@%s" % (i["screen_name"]) not in tweet.quoted_status.message and i["screen_name"] != tweet.user.screen_name:
if hasattr(tweet.quoted_status, "retweeted_status") and tweet.retweeted_status.user.screen_name == i["screen_name"]:
continue
tweet.quoted_status.message = u"@%s %s" % (i["screen_name"], tweet.message)
else:
tweet.message = message
if tweet.message == False: return False
tweet.twishort = True
for i in tweet.entities["user_mentions"]:
if "@%s" % (i["screen_name"]) not in tweet.message and i["screen_name"] != tweet.user.screen_name:
if hasattr(tweet, "retweeted_status") and tweet.retweeted_status.user.screen_name == i["screen_name"]:
continue
return tweet
def get_user(self, id):
""" Returns an user object associated with an ID.
id str: User identifier, provided by Twitter.
returns a tweepy user object."""
if ("users" in self.db) == False or (id in self.db["users"]) == False:
try:
user = self.twitter.get_user(id=id)
except TweepError as err:
user = UserModel(None)
user.screen_name = "deleted_user"
user.id = id
user.name = _("Deleted account")
user.id_str = id
self.db["users"][user.id_str] = user
return user
else:
return self.db["users"][id]
def get_user(self, id):
""" Returns an user object associated with an ID.
id str: User identifier, provided by Twitter.
returns a tweepy user object."""
if ("users" in self.db) == False or (id in self.db["users"]) == False:
try:
user = self.twitter.get_user(id=id)
except TweepError as err:
user = UserModel(None)
user.screen_name = "deleted_user"
user.id = id
user.name = _("Deleted account")
user.id_str = id
self.db["users"][user.id_str] = user
return user
else:
return self.db["users"][id]
def get_user_by_screen_name(self, screen_name):
""" Returns an user identifier associated with a screen_name.
screen_name str: User name, such as tw_blue2, provided by Twitter.
returns an user ID."""
if ("users" in self.db) == False:
user = utils.if_user_exists(self.twitter, screen_name)
self.db["users"][user["id_str"]] = user
return user["id_str"]
else:
for i in list(self.db["users"].keys()):
if self.db["users"][i].screen_name == screen_name:
return self.db["users"][i].id_str
user = utils.if_user_exists(self.twitter, screen_name)
self.db["users"][user.id_str] = user
return user.id_str
def get_user_by_screen_name(self, screen_name):
""" Returns an user identifier associated with a screen_name.
screen_name str: User name, such as tw_blue2, provided by Twitter.
returns an user ID."""
if ("users" in self.db) == False:
user = utils.if_user_exists(self.twitter, screen_name)
self.db["users"][user["id_str"]] = user
return user["id_str"]
else:
for i in list(self.db["users"].keys()):
if self.db["users"][i].screen_name == screen_name:
return self.db["users"][i].id_str
user = utils.if_user_exists(self.twitter, screen_name)
self.db["users"][user.id_str] = user
return user.id_str
def save_users(self, user_ids):
""" Adds all new users to the users database. """
if len(user_ids) == 0:
return
log.debug("Received %d user IDS to be added in the database." % (len(user_ids)))
users_to_retrieve = [user_id for user_id in user_ids if user_id not in self.db["users"]]
# Remove duplicates
users_to_retrieve = list(dict.fromkeys(users_to_retrieve))
if len(users_to_retrieve) == 0:
return
log.debug("TWBlue will get %d new users from Twitter." % (len(users_to_retrieve)))
users = self.twitter.lookup_users(user_ids=users_to_retrieve, tweet_mode="extended")
for user in users:
self.db["users"][user.id_str] = user
log.debug("Added %d new users" % (len(users)))
def save_users(self, user_ids):
""" Adds all new users to the users database. """
if len(user_ids) == 0:
return
log.debug("Received %d user IDS to be added in the database." % (len(user_ids)))
users_to_retrieve = [user_id for user_id in user_ids if user_id not in self.db["users"]]
# Remove duplicates
users_to_retrieve = list(dict.fromkeys(users_to_retrieve))
if len(users_to_retrieve) == 0:
return
log.debug("TWBlue will get %d new users from Twitter." % (len(users_to_retrieve)))
users = self.twitter.lookup_users(user_ids=users_to_retrieve, tweet_mode="extended")
for user in users:
self.db["users"][user.id_str] = user
log.debug("Added %d new users" % (len(users)))

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

@@ -19,198 +19,198 @@ url_re2 = re.compile("(?:\w+://|www\.)[^ ,.?!#%=+][^ \\n\\t]*")
bad_chars = '\'\\\n.,[](){}:;"'
def find_urls_in_text(text):
return url_re2.findall(text)
return url_re2.findall(text)
def find_urls (tweet):
urls = []
# Let's add URLS from tweet entities.
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
for i in entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "quoted_status"):
for i in tweet.quoted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "retweeted_status"):
for i in tweet.retweeted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet["retweeted_status"], "quoted_status"):
for i in tweet.retweeted_status.quoted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "message"):
i = "message"
elif hasattr(tweet, "full_text"):
i = "full_text"
else:
i = "text"
if hasattr(tweet, "message_create"):
extracted_urls = find_urls_in_text(tweet.message_create["message_data"]["text"])
else:
extracted_urls = find_urls_in_text(getattr(tweet, i))
# Don't include t.co links (mostly they are photos or shortened versions of already added URLS).
for i in extracted_urls:
if i not in urls and "https://t.co" not in i:
urls.append(i)
return urls
urls = []
# Let's add URLS from tweet entities.
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
for i in entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "quoted_status"):
for i in tweet.quoted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "retweeted_status"):
for i in tweet.retweeted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet["retweeted_status"], "quoted_status"):
for i in tweet.retweeted_status.quoted_status.entities["urls"]:
if i["expanded_url"] not in urls:
urls.append(i["expanded_url"])
if hasattr(tweet, "message"):
i = "message"
elif hasattr(tweet, "full_text"):
i = "full_text"
else:
i = "text"
if hasattr(tweet, "message_create"):
extracted_urls = find_urls_in_text(tweet.message_create["message_data"]["text"])
else:
extracted_urls = find_urls_in_text(getattr(tweet, i))
# Don't include t.co links (mostly they are photos or shortened versions of already added URLS).
for i in extracted_urls:
if i not in urls and "https://t.co" not in i:
urls.append(i)
return urls
def find_item(id, listItem):
for i in range(0, len(listItem)):
if listItem[i].id == id: return i
return None
for i in range(0, len(listItem)):
if listItem[i].id == id: return i
return None
def find_list(name, lists):
for i in range(0, len(lists)):
if lists[i].name == name: return lists[i].id
for i in range(0, len(lists)):
if lists[i].name == name: return lists[i].id
def is_audio(tweet):
try:
if len(find_urls(tweet)) < 1:
return False
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
if len(entities["hashtags"]) > 0:
for i in entities["hashtags"]:
if i["text"] == "audio":
return True
except IndexError:
print(tweet.entities["hashtags"])
log.exception("Exception while executing is_audio hashtag algorithm")
try:
if len(find_urls(tweet)) < 1:
return False
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
if len(entities["hashtags"]) > 0:
for i in entities["hashtags"]:
if i["text"] == "audio":
return True
except IndexError:
print(tweet.entities["hashtags"])
log.exception("Exception while executing is_audio hashtag algorithm")
def is_geocoded(tweet):
if hasattr(tweet, "coordinates") and tweet.coordinates != None:
return True
if hasattr(tweet, "coordinates") and tweet.coordinates != None:
return True
def is_media(tweet):
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
if entities.get("media") == None:
return False
for i in entities["media"]:
if i.get("type") != None and i.get("type") == "photo":
return True
return False
if hasattr(tweet, "message_create"):
entities = tweet.message_create["message_data"]["entities"]
else:
entities = tweet.entities
if entities.get("media") == None:
return False
for i in entities["media"]:
if i.get("type") != None and i.get("type") == "photo":
return True
return False
def get_all_mentioned(tweet, conf, field="screen_name"):
""" Gets all users that have been mentioned."""
results = []
for i in tweet.entities["user_mentions"]:
if i["screen_name"] != conf["user_name"] and i["screen_name"] != tweet.user.screen_name:
if i.get(field) not in results:
results.append(i.get(field))
return results
""" Gets all users that have been mentioned."""
results = []
for i in tweet.entities["user_mentions"]:
if i["screen_name"] != conf["user_name"] and i["screen_name"] != tweet.user.screen_name:
if i.get(field) not in results:
results.append(i.get(field))
return results
def get_all_users(tweet, conf):
string = []
if hasattr(tweet, "retweeted_status"):
string.append(tweet.user.screen_name)
tweet = tweet.retweeted_status
if hasattr(tweet, "sender"):
string.append(tweet.sender.screen_name)
else:
if tweet.user.screen_name != conf["user_name"]:
string.append(tweet.user.screen_name)
for i in tweet.entities["user_mentions"]:
if i["screen_name"] != conf["user_name"] and i["screen_name"] != tweet.user.screen_name:
if i["screen_name"] not in string:
string.append(i["screen_name"])
if len(string) == 0:
string.append(tweet.user.screen_name)
return string
string = []
if hasattr(tweet, "retweeted_status"):
string.append(tweet.user.screen_name)
tweet = tweet.retweeted_status
if hasattr(tweet, "sender"):
string.append(tweet.sender.screen_name)
else:
if tweet.user.screen_name != conf["user_name"]:
string.append(tweet.user.screen_name)
for i in tweet.entities["user_mentions"]:
if i["screen_name"] != conf["user_name"] and i["screen_name"] != tweet.user.screen_name:
if i["screen_name"] not in string:
string.append(i["screen_name"])
if len(string) == 0:
string.append(tweet.user.screen_name)
return string
def if_user_exists(twitter, user):
try:
data = twitter.get_user(screen_name=user)
return data
except TweepError as err:
if err.api_code == 50:
return None
else:
return user
try:
data = twitter.get_user(screen_name=user)
return data
except TweepError as err:
if err.api_code == 50:
return None
else:
return user
def is_allowed(tweet, settings, buffer_name):
clients = settings["twitter"]["ignored_clients"]
if hasattr(tweet, "sender"): return True
allowed = True
tweet_data = {}
if hasattr(tweet, "retweeted_status"):
tweet_data["retweet"] = True
if tweet.in_reply_to_status_id_str != None:
tweet_data["reply"] = True
if hasattr(tweet, "quoted_status"):
tweet_data["quote"] = True
if hasattr(tweet, "retweeted_status"):
tweet = tweet.retweeted_status
source = tweet.source
for i in clients:
if i.lower() == source.lower():
return False
return filter_tweet(tweet, tweet_data, settings, buffer_name)
clients = settings["twitter"]["ignored_clients"]
if hasattr(tweet, "sender"): return True
allowed = True
tweet_data = {}
if hasattr(tweet, "retweeted_status"):
tweet_data["retweet"] = True
if tweet.in_reply_to_status_id_str != None:
tweet_data["reply"] = True
if hasattr(tweet, "quoted_status"):
tweet_data["quote"] = True
if hasattr(tweet, "retweeted_status"):
tweet = tweet.retweeted_status
source = tweet.source
for i in clients:
if i.lower() == source.lower():
return False
return filter_tweet(tweet, tweet_data, settings, buffer_name)
def filter_tweet(tweet, tweet_data, settings, buffer_name):
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
for i in settings["filters"]:
if settings["filters"][i]["in_buffer"] == buffer_name:
regexp = settings["filters"][i]["regexp"]
word = settings["filters"][i]["word"]
# Added if/else for compatibility reasons.
if "allow_rts" in settings["filters"][i]:
allow_rts = settings["filters"][i]["allow_rts"]
else:
allow_rts = "True"
if "allow_quotes" in settings["filters"][i]:
allow_quotes = settings["filters"][i]["allow_quotes"]
else:
allow_quotes = "True"
if "allow_replies" in settings["filters"][i]:
allow_replies = settings["filters"][i]["allow_replies"]
else:
allow_replies = "True"
if allow_rts == "False" and "retweet" in tweet_data:
return False
if allow_quotes == "False" and "quote" in tweet_data:
return False
if allow_replies == "False" and "reply" in tweet_data:
return False
if word != "" and settings["filters"][i]["if_word_exists"]:
if word in getattr(tweet, value):
return False
elif word != "" and settings["filters"][i]["if_word_exists"] == False:
if word not in getattr(tweet, value):
return False
if settings["filters"][i]["in_lang"] == "True":
if getattr(tweet, lang) not in settings["filters"][i]["languages"]:
return False
elif settings["filters"][i]["in_lang"] == "False":
if tweet.lang in settings["filters"][i]["languages"]:
return False
return True
if hasattr(tweet, "full_text"):
value = "full_text"
else:
value = "text"
for i in settings["filters"]:
if settings["filters"][i]["in_buffer"] == buffer_name:
regexp = settings["filters"][i]["regexp"]
word = settings["filters"][i]["word"]
# Added if/else for compatibility reasons.
if "allow_rts" in settings["filters"][i]:
allow_rts = settings["filters"][i]["allow_rts"]
else:
allow_rts = "True"
if "allow_quotes" in settings["filters"][i]:
allow_quotes = settings["filters"][i]["allow_quotes"]
else:
allow_quotes = "True"
if "allow_replies" in settings["filters"][i]:
allow_replies = settings["filters"][i]["allow_replies"]
else:
allow_replies = "True"
if allow_rts == "False" and "retweet" in tweet_data:
return False
if allow_quotes == "False" and "quote" in tweet_data:
return False
if allow_replies == "False" and "reply" in tweet_data:
return False
if word != "" and settings["filters"][i]["if_word_exists"]:
if word in getattr(tweet, value):
return False
elif word != "" and settings["filters"][i]["if_word_exists"] == False:
if word not in getattr(tweet, value):
return False
if settings["filters"][i]["in_lang"] == "True":
if getattr(tweet, lang) not in settings["filters"][i]["languages"]:
return False
elif settings["filters"][i]["in_lang"] == "False":
if tweet.lang in settings["filters"][i]["languages"]:
return False
return True
def twitter_error(error):
if error.api_code == 179:
msg = _(u"Sorry, you are not authorised to see this status.")
elif error.api_code == 144:
msg = _(u"No status found with that ID")
else:
msg = _(u"Error code {0}").format(error.api_code,)
output.speak(msg)
if error.api_code == 179:
msg = _(u"Sorry, you are not authorised to see this status.")
elif error.api_code == 144:
msg = _(u"No status found with that ID")
else:
msg = _(u"Error code {0}").format(error.api_code,)
output.speak(msg)
def expand_urls(text, entities):
""" Expand all URLS present in text with information found in entities"""
urls = find_urls_in_text(text)
for url in entities["urls"]:
if url["url"] in text:
text = text.replace(url["url"], url["expanded_url"])
return text
""" Expand all URLS present in text with information found in entities"""
urls = find_urls_in_text(text)
for url in entities["urls"]:
if url["url"] in text:
text = text.replace(url["url"], url["expanded_url"])
return text

28
src/sessions/twitter/wxUI.py
View File

@@ -1,18 +1,18 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import wx
class authorisationDialog(wx.Dialog):
def __init__(self):
super(authorisationDialog, self).__init__(parent=None, title=_(u"Authorising account..."))
panel = wx.Panel(self)
sizer = wx.BoxSizer(wx.VERTICAL)
static = wx.StaticText(panel, wx.NewId(), _(u"Enter your PIN code here"))
self.text = wx.TextCtrl(panel, -1)
self.ok = wx.Button(panel, wx.ID_OK)
self.cancel = wx.Button(panel, wx.ID_CANCEL)
sizer.Add(self.text, 0, wx.ALL, 5)
sizer.Add(self.cancel, 0, wx.ALL, 5)
panel.SetSizer(sizer)
min = sizer.CalcMin()
self.SetClientSize(min)
def __init__(self):
super(authorisationDialog, self).__init__(parent=None, title=_(u"Authorising account..."))
panel = wx.Panel(self)
sizer = wx.BoxSizer(wx.VERTICAL)
static = wx.StaticText(panel, wx.NewId(), _(u"Enter your PIN code here"))
self.text = wx.TextCtrl(panel, -1)
self.ok = wx.Button(panel, wx.ID_OK)
self.cancel = wx.Button(panel, wx.ID_CANCEL)
sizer.Add(self.text, 0, wx.ALL, 5)
sizer.Add(self.cancel, 0, wx.ALL, 5)
panel.SetSizer(sizer)
min = sizer.CalcMin()
self.SetClientSize(min)