twitter API v2

tgbot-collection · Jul 23, 2023 · 3bd9b0a · 3bd9b0a
1 parent d344dfe
commit 3bd9b0a
Show file tree

Hide file tree

Showing 5 changed files with 161 additions and 188 deletions.
diff --git a/README.md b/README.md
@@ -1,18 +1,26 @@
 # TwitterBot
+
 🦉 A telegram Twitter bot that will allow you to send tweets!
 
+Supports Twitter API v2! 🎉🎉🎉🎉🎉🎉
+
+# Public Bot has been discontinued
+
 ![](assets/announcement.png)
 
 # Features
+
 All the following features rely on authorized users.
+
 * send text tweet
 * send tweet with one photo(photo and document are supported.)
 * reply bot tweet message to add more tweets to this thread
 * reply `/delete` to delete tweet
-* send any video tweet to download the video
+* ~~send any video tweet to download the video~~ If you have Basic or Access subscription, you can download video tweet.
 * send pictures as media group, it will result in multiple photos in your tweet(max photo restrictions is 4)
 
 # Commands
+
 ```
 start - Start using it today
 sign_in - Go to sign in with Twitter
@@ -23,40 +31,53 @@ delete - Delete tweet
 ```
 
 # Usage
+
 ## Screenshots
+
 The most simple way to tweet!
 
 ![](assets/tweet.png)
+
 ## Sign in
 
 Chat with [this bot](https://t.me/tele_tweetbot), and go to oauth by its instruction:
 ![](assets/intro.png)
 
 Copy and paste the auth code to this bot. And you're good to go!
+
 ## Tweet
+
 ### New thread tweet
+
 Send any text message, photo/photo as file with caption will send tweet with photos.
 
 ![](assets/tweet_web.png)
+
 ### Reply to this thread
+
 Reply to bot's tweet message,
 
 ![](assets/thread-bot.png)
 
 and you'll consult in a series of thread.
 
 ![](assets/thread-web.png)
+
 ## Delete tweet
+
 Reply command `/delete` to bot's message and it will delete this tweet for you.
 
 ![](assets/delete.png)
 
 ## Download tweet video
-You could just send any video tweet, such as this one 
+
+**ONLY WORKS IF YOU HAVE BASIC OR PRO SUBSCRIOTION**
+
+You could just send any video tweet, such as this one
 
 https://twitter.com/BennyThinks/status/1306081739660427264?s=20
 
-The bot will first determine if this is a video tweet, if the answer is yes, 
+The bot will first determine if this is a video tweet, if the answer is yes,
 it will ask what do you want to do with it:
 
 ![](assets/download_or_tweet.png)
@@ -66,33 +87,36 @@ Depending on your choice, this bot will download this video or retweet.
 ![](assets/video_choice.png)
 
 ## multiple photo mode
+
 **Group feature requires a newer version of Telegram**
 Just send a group of photos, add some captions if you want to, and hit send.
 ![](assets/multi_photo.jpg)
 
 **Warning: if you send more than four photos in a group, this bot will only send first four photos in a tweet**
 
-
 # General deployment
+
 This bot use oauth, so you need to apply an app, setup callback url.
-More info could be seen [here](https://github.com/twitterdev/twauth-web).
 
-## Bot
-```shell script
-git clone https://github.com/tgbot-collection/TeleTweet/
-cd TeleTweet
-pip3 install -r requirements.txt
-vim teletweet/config.py
-# setup bot token, consumer key, consumer secret, app id, app secret(https://core.telegram.org/)
-python3 teletweet/bot.py
-```
-If you want the bot to be only available to you, please setup `ALLOW_USER`, supplied with your own Telegram User ID.
-Use `,` to separate.
+## Setup Twitter Developer
+
+1. Apply a Twitter Developer account https://developer.twitter.com/en/portal/dashboard
+2. Create a project
+3. Create an app and attach it to your project. Remember API key and API secret key!
+4. User authentication settings, Click Edit and enable OAuth
+5. App permissions Read/Write, Web App, Automated App or Bot, Callback `http://127.0.0.1:8888/callback`
+
+## Setup Telegram APP ID
+
+Get it from https://core.telegram.org/
+
+## Run Web server
 
-## Web server
 ```shell script
 vim twauth.py
 # change this three lines to your own
+# CONSUMER_KEY ---> API key
+# CONSUMER_SECRET ---> API secret key
 APP_CONSUMER_KEY = os.environ.get("CONSUMER_KEY") or '1'
 APP_CONSUMER_SECRET = os.environ.get("CONSUMER_SECRET") or '2'
 callback_url = os.environ.get("CALLBACK_URL") or "http://127.0.0.1:8888/callback"
@@ -101,8 +125,28 @@ callback_url = os.environ.get("CALLBACK_URL") or "http://127.0.0.1:8888/callback
 python3 twauth.py
 ```
 
+Open `http://127.0.0.1:8888` in your browser and follow the instructions.
+
+## Run Bot
+
+```shell script
+git clone https://github.com/tgbot-collection/TeleTweet/
+cd TeleTweet
+pip3 install -r requirements.txt
+vim teletweet/config.py
+# setup bot token, consumer key, consumer secret, app id, app secret
+python3 teletweet/bot.py
+```
+
+If you want the bot to be only available to you, please setup `ALLOW_USER` with your own Telegram User ID.
+Use `,` to separate.
+
+Send `/sign_in` to bot,paste your auth code, and you're good to go!
+
 # Docker
+
 You can run/develop using docker. You can refer to `docker-compose.yml`
+
 ```shell
 echo "{}"> env/auth.json
 vim env/teletweet.env
@@ -119,15 +163,17 @@ docker-compose up -d
 ```
 
 # Plans
+
 - [x] support multi-user, based on oauth
 - [x] help
 - [x] about
 - [x] start
 - [x] multi photo
-
 
 # Credits
+
 * [twauth-web](https://github.com/twitterdev/twauth-web)
- 
+
 # License
-GPL 2.0
+
+GPL 2.0__
diff --git a/requirements.txt b/requirements.txt
@@ -6,3 +6,4 @@ requests==2.31.0
 tgbot-ping==1.0.7
 tornado==6.3.2
 tgcrypto==1.2.5
+tweepy==4.14.0
diff --git a/teletweet/tweet.py b/teletweet/tweet.py
@@ -9,118 +9,62 @@
 
 import logging
 import re
-import time
 import traceback
+from typing import Union
 
-import filetype
-import twitter
-from twitter.api import CHARACTER_LIMIT
-from twitter.error import TwitterError
-from twitter.twitter_utils import calc_expected_status_length
+import tweepy
 
 from config import CONSUMER_KEY, CONSUMER_SECRET
 from helper import get_auth_data
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(filename)s [%(levelname)s]: %(message)s")
 
 
-class NewApi(twitter.Api):
-    def PostUpdate(
-        self,
-        status,
-        media=None,
-        media_additional_owners=None,
-        media_category=None,
-        in_reply_to_status_id=None,
-        auto_populate_reply_metadata=False,
-        exclude_reply_user_ids=None,
-        latitude=None,
-        longitude=None,
-        place_id=None,
-        display_coordinates=False,
-        trim_user=False,
-        verify_status_length=True,
-        attachment_url=None,
-    ):
-        # if this is [file], single photo or media
-        if media and len(media) == 1:
-            media_type = filetype.guess(media[0]).mime
-            if media_type and "video" in media_type:
-                # we'll first try the ordinary one, if that fails, execute new method in exception
-                try:
-                    return super(NewApi, self).PostUpdate(
-                        status,
-                        media[0],
-                        media_additional_owners,
-                        media_category,
-                        in_reply_to_status_id,
-                        auto_populate_reply_metadata,
-                        exclude_reply_user_ids,
-                        latitude,
-                        longitude,
-                        place_id,
-                        display_coordinates,
-                        trim_user,
-                        verify_status_length,
-                        attachment_url,
-                    )
-                except TwitterError:
-                    logging.warning("possible long video, trying new method...")
-                    video_id = self.UploadMediaChunked(media=media[0], media_category="tweet_video")
-                    logging.info("video id is %s, status is %s", video_id, status)
-                    time.sleep(20)
-                    # Waits until the async processing of the uploaded media finishes and `video_id` becomes valid.
-                    status = super(NewApi, self).PostUpdate(
-                        status=status, media=video_id, in_reply_to_status_id=in_reply_to_status_id
-                    )
-                    return status
-
-        return super(NewApi, self).PostUpdate(
-            status,
-            media,
-            media_additional_owners,
-            media_category,
-            in_reply_to_status_id,
-            auto_populate_reply_metadata,
-            exclude_reply_user_ids,
-            latitude,
-            longitude,
-            place_id,
-            display_coordinates,
-            trim_user,
-            verify_status_length,
-            attachment_url,
-        )
-
-
 def __connect_twitter(chat_id: int):
     auth_data = get_auth_data(chat_id)
     logging.info("Connecting to twitter api...")
-    return NewApi(
+    client = tweepy.Client(
         consumer_key=CONSUMER_KEY,
         consumer_secret=CONSUMER_SECRET,
-        access_token_key=auth_data["ACCESS_KEY"],
+        access_token=auth_data["ACCESS_KEY"],
         access_token_secret=auth_data["ACCESS_SECRET"],
-        sleep_on_rate_limit=True,
     )
+    api = tweepy.API(
+        tweepy.OAuth1UserHandler(
+            consumer_key=CONSUMER_KEY,
+            consumer_secret=CONSUMER_SECRET,
+            access_token=auth_data["ACCESS_KEY"],
+            access_token_secret=auth_data["ACCESS_SECRET"],
+        )
+    )
+    return client, api
 
 
-# database format
+def upload_media(api, pic) -> Union[list, None]:
+    if pic is None:
+        return None
+    ids = []
+    for item in pic:
+        item.seek(0)
+        mid = api.media_upload(item.name, file=item)
+        ids.append(mid)
+    return [i.media_id for i in ids]
 
 
-def send_tweet(message, pic=None) -> dict:
+def send_tweet(message, pics: Union[list, None] = None) -> dict:
     logging.info("Preparing tweet for...")
     chat_id = message.chat.id
     text = message.text or message.caption
     if not text:
         text = ""
     tweet_id = __get_tweet_id_from_reply(message)
     try:
-        api = __connect_twitter(chat_id)
+        client, api = __connect_twitter(chat_id)
         logging.info("Tweeting...")
-        status = api.PostUpdate(text, media=pic, in_reply_to_status_id=tweet_id)
+        ids = upload_media(api, pics)
+        status = client.create_tweet(text=text, media_ids=ids, in_reply_to_tweet_id=tweet_id)
         logging.info("Tweeted")
-        response = status.AsDict()
+        response = status.data
     except Exception as e:
         logging.error(traceback.format_exc())
         response = {"error": str(e)}
@@ -131,9 +75,10 @@ def send_tweet(message, pic=None) -> dict:
 def get_me(chat_id) -> str:
     logging.info("Get me!")
     try:
-        api = __connect_twitter(chat_id)
-        name = api.VerifyCredentials().name
-        user_id = api.VerifyCredentials().screen_name
+        client, api = __connect_twitter(chat_id)
+        me = client.get_me().data
+        name = me["name"]
+        user_id = me["username"]
         response = f"[{name}](https://twitter.com/{user_id})"
     except Exception as e:
         logging.error(traceback.format_exc())
@@ -150,10 +95,9 @@ def delete_tweet(message) -> dict:
         return {"error": "Which tweet do you want to delete? This does not seem like a valid tweet message."}
 
     try:
-        api = __connect_twitter(chat_id)
+        client, api = __connect_twitter(chat_id)
         logging.info("Deleting......")
-        status = api.DestroyStatus(tweet_id)
-        response = status.AsDict()
+        response = client.delete_tweet(tweet_id).data
     except Exception as e:
         logging.error(traceback.format_exc())
         response = {"error": str(e)}
@@ -181,21 +125,22 @@ def __get_tweet_id_from_url(url) -> int:
 
 
 def get_video_download_link(chat_id, tweet_id):
-    api = __connect_twitter(chat_id)
+    client, api = __connect_twitter(chat_id)
     logging.info("Getting video tweets......")
-    status = api.GetStatus(tweet_id)
+    status = client.get_tweet(tweet_id).data
     return __get_video_url(status.AsDict())
 
 
 def is_video_tweet(chat_id, text) -> str:
     # will return an id
     tweet_id = __get_tweet_id_from_url(text)
     logging.info("tweet id is %s", tweet_id)
-    api = __connect_twitter(chat_id)
+    client, api = __connect_twitter(chat_id)
     logging.info("Getting video tweets......")
     try:
-        status = api.GetStatus(tweet_id)
-        if __get_video_url(status.AsDict()):
+        # TODO I don't have permission to this. Thank you Elon Musk.
+        status = client.get_tweet(tweet_id)
+        if __get_video_url(status.data):
             return str(tweet_id)
     except Exception as e:
         logging.debug(traceback.format_exc())