Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
bg = Image.new('RGBA', (w, h), "#FFFFFF")
draw = ImageDraw.Draw(bg)
y_text = 10
for line in lines:
if line:
width, height = font.getsize(line)
else:
width, height = font.getsize('A')
draw.text(((w - width) / 2, y_text), line, font=font, fill=FOREGROUND)
y_text += height + 5
with open(fn, 'w') as f:
bg.save(f)
return fn
parser = ttp.Parser()
def get_tweet_components(text):
result = parser.parse(text)
urls = ' '.join(result.urls)
tags = ' '.join(['#' + tag for tag in result.tags])
if result.reply:
result.users.remove(result.reply)
result.reply = '@' + result.reply
else:
result.reply = ''
users = ' '.join(['@' + user for user in result.users])
return ' '.join([result.reply, urls, users, tags]).strip()
def get_tweets(request):
datas = []
p = ttp.Parser()
try:
api = twitter.Api(
consumer_key=settings.TWITTER_CONSUMER_KEY,
consumer_secret=settings.TWITTER_CONSUMER_SECRET,
access_token_key=settings.TWITTER_ACCESS_TOKEN,
access_token_secret=settings.TWITTER_ACCESS_TOKEN_SECRET
)
tweets = api.GetUserTimeline(screen_name='kodlaco')
for tweet in tweets:
datas.append({
#'text': p.parse(tweet.text).html,
'text': tweet.text,
'id_str': tweet.id_str
})
# Make t.co URLs into their original URLs, clickable.
if 'entities' in json_data and 'description' in json_data['entities']:
entities = json_data['entities']['description']
if 'urls' in entities:
for entity in entities['urls']:
start, end = entity['indices'][0], entity['indices'][1]
shown_url = entity['display_url']
link_url = entity['expanded_url']
url_html = '<a rel="external" href="%s">%s</a>'
desc = desc.replace(json_data['description'][start:end],
url_html % (link_url, shown_url))
# Make #hashtags and @usernames clickable.
parser = ttp.Parser()
parsed = parser.parse(desc)
return parsed.html
def link_urls(text):
"""
Enriches urls in the comment text with an anchor.
"""
parser = ttp.Parser(max_url_length=40)
parser._urls = []
return ttp.URL_REGEX.sub(parser._parse_urls, text)
]
logging.basicConfig(filename='logger.log',
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Twitter client
auth = tweepy.OAuthHandler(config.twitter['key'], config.twitter['secret'])
auth.set_access_token(config.twitter['access_token'],
config.twitter['access_token_secret'])
api = tweepy.API(auth)
# Tweet parser
parser = ttp.Parser()
# backoff time
backoff = BACKOFF
def strip_symbol(term):
return term.strip('$')
def now_str():
now = datetime.datetime.now()
return now.strftime('%Y-%m-%d-%H-%M-%S-%f')
def get_out_filename(symbol, time_span, quote):
r = requests.get(CHART_API % (symbol, time_span), stream=True)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Extract information from streaming tweet set')
parser.add_argument('--json-raw', nargs="*", help='Input to analyse e.g. tweets.json')
parser.add_argument('--output', "-o", help="Output to write (else stdout) e.g. -o pycon.json")
parser.add_argument('--json-cleaned', help='Cleaned input json')
parser.add_argument('--remove-nodes', nargs="*", default=[], help='Remove named nodes e.g. "--remove-nodes #pycon @pycon"')
parser.add_argument('--draw-networkx', action="store_true", help='Draw the graph using networkX')
parser.add_argument('--write-graphml', help='Filename for graphml output')
parser.add_argument('--remove-usernames-below', type=int, default=50, help='Remove usernames who are mentioned less than n times e.g. "--remove-usernames-below 50"')
parser.add_argument('--remove-hashtags-below', type=int, default=2, help='Remove hashtags that are mentioned less than n times e.g. "--remove-hashtagss-below 2"')
parser.add_argument('--remove-phrases-below', type=int, default=10, help='Remove phrases (>1 word) that are mentioned less than n times e.g. "--remove-phrases-below 10"')
args = parser.parse_args()
if args.json_raw:
tweet_parser = ttp.Parser()
# stream through a list of user-provided filenames
all_json_lines = files(args.json_raw)
tweets = get_tweets(all_json_lines)
# get tweets (ignore rubbish from streaming api), extract useful info
stream = get_tweet_body(tweets)
stream = get_useful_information(tweet_parser, stream)
if args.output:
output = open(args.output, 'w')
else:
output = sys.stdout # use stdout if no file specified
items = []
for item in stream:
outstr = json.dumps(item)
output.write("%s\n" % (outstr))