diff options
Diffstat (limited to 'twitter.py')
| -rwxr-xr-x | twitter.py | 193 |
1 files changed, 0 insertions, 193 deletions
diff --git a/twitter.py b/twitter.py deleted file mode 100755 index 673fce7..0000000 --- a/twitter.py +++ /dev/null @@ -1,193 +0,0 @@ -#!/usr/bin/env python3 - -from urllib.error import HTTPError -from urllib.request import urlopen, Request -import logging - -# from requests_oauthlib import OAuth1Session -from datetime import datetime -import sys -import json - -bearer = None - - -def getBearer(): - global bearer - if bearer: - return bearer - headers = { - "Authorization": "Basic Zzl1MXI2SFpYTXg0SXU5UGs5VlNvTzFUdzpmeTIyQjN2QVRRNUI2eGthb1BFdFFRUmtuUGQ1WGZBbnBKVG5hc0ZRa3NyUm5qaVNsaw==", - "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8", - } - data = b"grant_type=client_credentials" - url = "https://api.twitter.com/oauth2/token" - - res = urlopen(Request(url, headers=headers, data=data, method="POST")) - response = json.loads(res.read().decode("UTF-8")) - bearer = response["access_token"] - - return bearer - - -def unshorten_urls(title, description, urls): - for url in urls: - shorted_url = url["url"] - long_url = url["expanded_url"] - - if "images" in url: - img = url["images"][0]["url"] - long_url_html = '<a href="' + long_url + '"><img src="' + img + '"/></a>' - else: - long_url_html = '<a href="' + long_url + '">' + long_url + "</a>" - - description = description.replace(shorted_url, long_url_html) - title = title.replace(shorted_url, long_url) - return title, description - - -def twitter(user): - # 500.000 Tweets per month - # API KEY = g9u1r6HZXMx4Iu9Pk9VSoO1Tw - # API SECRET KEY = fy22B3vATQ5B6xkaoPEtQQRknPd5XfAnpJTnasFQksrRnjiSlk - - headers = {"authorization": "Bearer " + getBearer()} - - # Recent = last 7 days - url = ( - "https://api.twitter.com/2/tweets/search/recent?query=from:" - + user - + "&tweet.fields=created_at,author_id,lang,source,public_metrics,entities&expansions=referenced_tweets.id,attachments.media_keys&media.fields=url" - ) - - try: - res = urlopen(Request(url, headers=headers)) - response = json.loads(res.read().decode("UTF-8")) - except Exception as exc: - logging.error("Request to twitter failed.", exc_info=exc) - return None - - feed = { - "title": "Twitter: " + user, - "url": "https://twitter.com/" + user, - "description": "The latest entries of the twitter account of " + user, - "content": [], - } - - if not response["meta"]["result_count"]: - return feed - - feed["content"] = [ - parse_tweet( - user, - tweet, - response.get("includes", {}).get("tweets", []), - response.get("includes", {}).get("media", []), - headers, - ) - for tweet in response["data"] - ] - - return feed - - -def parse_tweet(user, tweet, included_tweets, included_media, headers): - title = description = tweet["text"] - link = "https://twitter.com/" + user + "/status/" + str(tweet["id"]) - - # Check included re-tweets / replace by Retweet - ref_enclosures = [] - for rt in tweet.get("referenced_tweets", []): - if rt["type"] == "retweeted": - rt_info = title[: title.index(":") + 2] - ref_tweet = next(t for t in included_tweets if t["id"] == rt["id"]) - title = rt_info + ref_tweet["text"] - description = rt_info + ref_tweet["text"] - title, description = unshorten_urls( - title, description, ref_tweet.get("entities", {}).get("urls", []) - ) - elif rt["type"] == "replied_to": - description += "<br/>This was a reply to: " + rt["id"] - text, enclosures = fetch_single_tweet(rt["id"], headers) - description += text - ref_enclosures.extend(enclosures) - elif rt["type"] == "quoted": - description += "<br/>Quoted tweet: " + rt["id"] - text, enclosures = fetch_single_tweet(rt["id"], headers) - description += text - ref_enclosures.extend(enclosures) - else: - description += f"<br/><br/>Unknown reference type: {rt['type']}" - - title, description = unshorten_urls( - title, description, tweet.get("entities", {}).get("urls", []) - ) - - # Attach media - enclosures = [] - included_media_keys = tweet.get("attachments", {}).get("media_keys", []) - for included_media_key in included_media_keys: - ref_media = next( - t for t in included_media if t["media_key"] == included_media_key - ) - if "url" not in ref_media: - continue - if ref_media.get("type", "") == "photo": - description += '<br/><img src="' + ref_media["url"] + '" />' - else: - enclosures.append(ref_media["url"]) - enclosures.extend(ref_enclosures) - - # Append Retweets etc - description += "<br/><br/>" - description += str(tweet["public_metrics"]["retweet_count"]) + " Retweets, " - description += str(tweet["public_metrics"]["like_count"]) + " Likes, " - description += str(tweet["public_metrics"]["reply_count"]) + " Replies, " - description += str(tweet["public_metrics"]["quote_count"]) + " Quotes" - description += "<br/>" - description += "Source: " + tweet["source"] - - date = datetime.strptime(tweet["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ") - - return { - "title": title, - "url": link, - "content": description, - "date": date, - "enclosures": enclosures, - } - - -def fetch_single_tweet(id, headers): - url = f"https://api.twitter.com/2/tweets/{id}?tweet.fields=entities&expansions=attachments.media_keys&media.fields=url" - try: - res = urlopen(Request(url, headers=headers)) - response = json.loads(res.read().decode("UTF-8")) - except Exception as exc: - logging.error("Request to twitter failed (single tweet).", exc_info=exc) - return None - - text = response['data'].get('text', 'no text') - - enclosures = [] - for media in response['data'].get('includes', {}).get('media', []): - if "url" not in media: - continue - if media.get("type", "") == "photo": - text += '<br/><img src="' + media["url"] + '" />' - else: - enclosures.append(media["url"]) - - return text, enclosures - -def main(channel): - print(twitter(channel)) - - -if __name__ == "__main__": - if len(sys.argv) != 2: - print("Usage:", sys.argv[0], "<twitter channel>") - sys.exit(1) - main(sys.argv[1]) - # twitter('rheinbahn_intim') - # twitter('realDonaldTrump') |