标签: twitterapi-python

我正在关注使用 Twippy 库和 Twitter API 的 Twitter 机器人教程：https://realpython.com/twitter-bot-python-tweepy/

我设置了 config.py 文件，并将 Windows 环境变量设置为带有所有令牌的用户变量。但是当我运行我的文件时，它会给出错误，因为检索我的令牌时 os.getenv() 为 None

consumer_key = os.getenv("CONSUMER_KEY")
consumer_secret = os.getenv("CONSUMER_SECRET")
access_token = os.getenv("ACCESS_TOKEN")
access_token_secret = os.getenv("ACCESS_TOKEN_SECRET")

在 Windows 终端中，我打印了每个变量，它们都是正确的。我在这里缺少什么吗？任何帮助深表感谢！

我对编码还很陌生，目前正在网上学习课程，其中我们开始使用 Twitter API。我试图捕获 Twitter 帐户的 ID 并将其显示在下面，但我不断收到此消息TypeError，并且无法弄清楚如何修复它。任何帮助将不胜感激，谢谢。我的代码

一直在玩 Twitter API。事情进展顺利，直到我添加了开始和结束时间以及搜索的粒度。我相当有信心字符串值是正确的，按照 Twitter api 文档，并正确添加到参数中。但是我的参数做错了，有人可以帮助我吗？

有效的代码，IE 会产生 200 响应和推文：

keywords = 'telsa'
api, query = 'https://api.twitter.com/2', '/tweets/search/recent'
request = f'{api}{query}'
params = {'query': keywords,
          'tweet.fields': 'created_at,lang', # no space before lang
          'max_results': '10'}
header = {'authorization':f'Bearer {bearer_token}'}

response = requests.get(request, headers=header, params=params)
if response: # boolean True if response is 200,
    print(response.response.json())
else:
    print(response)
    print(params)

不起作用的代码：400 响应：添加 start_time、end_time、粒度

<snip>
header = {'authorization':f'Bearer {bearer_token}'}

**************** NEW CODE **************************
# Get datetime for now and 7 days ago, correctly formatted for …

我正在尝试从 geeksforgeeks 复制此代码片段，只不过它使用 Twitter API v1.1 的 oauth，而我使用 API v2。

# the screen name of the user
screen_name = "PracticeGfG"

# fetching the user
user = api.get_user(screen_name)

# fetching the ID
ID = user.id_str

print("The ID of the user is : " + ID)

OUTPUT:
The ID of the user is: 4802800777.

这是我的：

import os
import tweepy

API_KEY = os.getenv('API_KEY')
API_KEY_SECRET = os.getenv('API_KEY_SECRET')
BEARER_TOKEN = os.getenv('BEARER_TOKEN')
ACCESS_TOKEN = os.getenv('ACCESS_TOKEN')
ACCESS_TOKEN_SECRET = os.getenv('ACCESS_TOKEN_SECRET')

screen_name = 'nytimes'

client = tweepy.Client(consumer_key=API_KEY, consumer_secret=API_KEY_SECRET, access_token=ACCESS_TOKEN, …

我有一个 python 函数，它使用请求库和 BeautifulSoup 来抓取特定用户的推文。

import requests
from bs4 import BeautifulSoup

contents = requests.get("https://twitter.com/user")
soup = BeautifulSoup(contents.text, "html.parser")

当请求库访问 Twitter 时，它使用 Twitter 的旧版本。但是，由于 Twitter 最近放弃了对其旧版本的支持，请求库不再有效并返回 html 代码，指出此版本的 Twitter 已过时。

有没有办法让请求库访问较新版本的 Twitter？

我正在使用以下网址来获取用户推文。

https://api.twitter.com/2/users/{user_twitter_id}/tweets

以下是我的请求参数

query_params = {
    'max_results': max_results,
    'expansions': 'attachments.media_keys',
    'tweet.fields': 'id,created_at,text,author_id,in_reply_to_user_id,referenced_tweets,attachments,withheld,geo,entities,public_metrics,possibly_sensitive,source,lang,context_annotations,conversation_id,reply_settings',
    'media.fields': 'media_key,duration_ms,height,preview_image_url,type,url,width,public_metrics,non_public_metrics,organic_metrics,promoted_metrics,alt_text'
}

以下是我在“包含”中收到的回复，其中包含媒体列表

"includes": {
    "media": [
        {
            "height": 750,
            "media_key": "3_1489397927281840131",
            "type": "photo",
            "url": "https://pbs.twimg.com/media/FKtnhhDWUAMtyBi.jpg",
            "width": 1125
        },
        {
            "height": 750,
            "media_key": "3_1489397930452783110",
            "type": "photo",
            "url": "https://pbs.twimg.com/media/FKtnhs3XEAYeMkP.jpg",
            "width": 1125
        },
        {
            "height": 750,
            "media_key": "3_1489397944214302727",
            "type": "photo",
            "url": "https://pbs.twimg.com/media/FKtnigIXMAcTO6t.jpg",
            "width": 1125
        },
        {
            "duration_ms": 242784,
            "height": 1080,
            "media_key": "13_1489018359819771906",
            "preview_image_url": "https://pbs.twimg.com/media/FKoOePDWYAA1ZZB.jpg",
            "public_metrics": {
                "view_count": 275300
            },
            "type": "video",
            "width": 1920
        },
        {
            "height": 2400,
            "media_key": "3_1488933061307809794", …

即使我的帐户处于活动状态，我也无法使用我的 Twitter 开发者帐户进行身份验证

import tweepy\nconsumer_key= 'XX1'\nconsumer_secret= 'XX2'\naccess_token= 'XX3'\naccess_token_secret= 'XX4'\nauth = tweepy.OAuthHandler(consumer_key, consumer_secret)\nauth.set_access_token(access_token, access_token_secret)\napi = tweepy.API(auth)\napi.update_status("Hello Tweepy")\n

我收到错误：

Forbidden: 403 Forbidden\n\n453 - You currently have Essential access which includes access to Twitter API v2 endpoints only. If you need access to this endpoint, you\xe2\x80\x99ll need to apply for Elevated access via the Developer Portal. You can learn more here: https://developer.twitter.com/en/docs/twitter-api/getting-started/about-twitter-api#v2-access-leve\n

开发者门户上没有从 Essential 升级到 Elevated 的选项。有什么建议吗？

我的挑战是获取拥有超过 3000 万粉丝的帐户的所有粉丝列表。

目前，我正在 Twitter 的 REST API 上使用GET followers/list端点，但是，由于免费 API 的速率限制，这需要很多天才能实现。

我愿意为高级 API 访问付费，但是我找不到任何数据表明高级 API 具有必要的端点和足够的速率限制来帮助最多在几个小时内解决此任务。

对此事有任何见解将不胜感激......

我正在尝试通过 twitter 文档中给出的 curl 命令将 url 注册为 twitter 上的 webhook。我认为 Twitter 文档有点过时或不正确。

 curl --request POST --url "https://api.twitter.com/1.1/account_activity/all/prod/webhooks.json?url=https%3A%2F%2FXXXX.com%2Fwebhook%2Ftwitter" --header "authorization: OAuth oauth_consumer_key='XXXXXXXXXXXXXXXXXXXXXX', oauth_nonce='ODgyNjc5NjQ0MTM3NzI4NTcwMjY4NDQ0', oauth_signature='7daMyzB1JClE4xv8hXNCimWpGtA%3D', oauth_signature_method='HMAC-SHA1', oauth_timestamp='1568620293', oauth_token='XXXXXXXXXXXXXXXXXXXXX', oauth_version='1.0'"

运行此代码后，我收到错误

{"errors":[{"code":215,"message":"Bad Authentication data."}]}

下面给出了我用来生成时间戳、oauth_nonce、oauth_signature 的代码。这些代码在python3中。我不太确定他们是否给出了正确的输出。

oauth_timestamp = str(int(time.time()))


#oauth_nonce
def get_nonce():
    nonce = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits)for x in range(32))
return nonce


#oauth_signature

def escape(s):
"""Percent Encode the passed in string"""
    return urllib.parse.quote_plus(s, safe='~')


def create_auth_header(parameters):
"""For all collected parameters, order them and create auth header"""
    ordered_parameters = {}
    ordered_parameters = …

当有人用我的@handle回复该推文时，我尝试使用下面编写的代码来获取推文中视频的媒体网址：

import tweepy
from tweepy import Stream
from tweepy.streaming import StreamListener
import json

class StdOutListener(StreamListener):
    def on_data(self, data):
        clean_data = json.loads(data)
        tweetId = clean_data['id']
        tweet_name = clean_data['user']['screen_name']
auth = tweepy.OAuthHandler("consumer_key", "consumer_secret")
            auth.set_access_token("access_token", "access_token_secret")
            api = tweepy.API(auth)
            mediaupload_id = clean_data['in_reply_to_status_id']
            origtweet_media = api.get_status(mediaupload_id)
            print()
            print(origtweet_media)
            native_media = origtweet_media['extended_entities']['media'][0]['video_info']
            tweet_media = native_media['variants'][0]['url']
            print(tweet_media)

def setUpAuth():
    auth = tweepy.OAuthHandler("consumer_key", "consumer_secret")
    auth.set_access_token("access_token", "access_token_secret")
    api = tweepy.API(auth)
    return api, auth

def followStream():
    api, auth = setUpAuth()
    listener = StdOutListener()
    stream = Stream(auth, listener)
    stream.filter(track=["@YOUR_TWITTER_HANDLE"], is_async=True) …

希望你一切都好。

我正在尝试使用 tweepy 库长时间连续提取推文。为此，我有一个无限循环运行的 python 脚本。唯一的问题是，每次连接失败时，由于任何可能的原因，我的脚本都会崩溃，我将不得不重新启动它。所以我想让它特别针对连接错误，为此我使用 try... except 条件，但它似乎不起作用。我已经尝试了网上找到的所有内容，但没有任何效果，有人可以帮助我吗？

这是我的代码：

consumer_key = ""
consumer_secret = ""
access_token = ""
access_token_secret = ""

auth = tweepy.OAuth1UserHandler(
   consumer_key, consumer_secret,
   access_token, access_token_secret
)
api = tweepy.API(auth)

city = 'LX'
country = 'PT'
coord = '38.736946,-9.142685'
radius = '25km'

while True:
    today = datetime.now()

    if len(str(today.month)) == 1:
        month = str(0) + str(today.month)
    else:
        month = str(today.month)    
    n=10
    tweets=tweepy.Cursor(api.search_tweets,q="",geocode=coord+','+radius,tweet_mode="extended").items(n)
 
    fname = f'{country}_{city}_{today.day}{month}{str(today.year)[2:]}.json'

    with open(fname, 'a') as outfile:
        for tweet in tweets:
            try: 
                outfile.write(json.dumps(tweet._json))
                outfile.write('\n') …

因此，正如问题所暗示的那样，我正试图弄清楚如何获取推文 ID # 或整个 url。使用下面的代码，我可以获得我想要的推文并将它们打印出来。但是，我不想打印它们，而是想获取 url 或至少是推文 id（以便我可以制作自己的 url），以便稍后在我的代码中使用。有什么建议吗？在 tweepy 文档中没有看到关于此的任何内容，但也许我缺少某些东西或我可以使用其他东西来实现这一目标？

import tweepy

# Authenticate to Twitter
auth = tweepy.OAuthHandler("*******", "*****")
auth.set_access_token("******", "*****")
api = tweepy.API(auth)

# Function to extract tweets 
def get_tweets(username): 

        tweets = api.user_timeline(screen_name=username,count = 1) 

        tmp=[]

        tweets_for_csv = [tweet.text for tweet in tweets]
        for t in tweets_for_csv:
            tmp.append(t)
        print(tmp)



if __name__ == '__main__':
    get_tweets("@username")

这段代码工作正常，只是试图了解我将添加什么以获得所需的数据。

我正在尝试通过 tweepy API 从 Twitter 获取推文，为此，我编写了以下代码，其中包含用于连接到 pyspark 的套接字，以便进一步分析正在获取的推文。代码：

import tweepy
from tweepy import OAuthHandler
from tweepy import Stream
import socket
import json

consumer_key="IvWeydhuRZoHbigNvX9wl8FbX"
consumer_secret="hv9NnWV64f4ECnK3eqJIhwQiu0SoDMtL1b9ImPJJFPwSSKleOg"
access_token="3945967939-cqy1y3J6cPcKi2fohK3qr5QSB2Z3lrOcs863tPI"
access_secret="oJIXDC37NsRU1FNP4jznCyIVwRiBjaHF9L3qsyxybkykS"

class TweetListener(Stream):
    def __init__(self, csocket):
        self.client_socket=csocket
    def on_data(self, data):
        try:
            msg = json.loads(data)
            print(msg('text').encode('utf-8'))
            self.client_socket.send(msg('text').encode('utf-8'))
            return True
        except BaseException as e:
            print('Error %s'%str(e))
        return True
    def on_error(self, status):
        print(status)
        return True

def send_data(c_socket):
    auth = OAuthHandler(consumer_key,consumer_secret)
    auth.set_access_token(access_token,access_secret)
    twtr_stream = Stream(auth, TweetListener(c_socket))
    twtr_stream.filter(track=['ETH'])

s = socket.socket()
host = "127.0.0.1"
port = 5000
s.bind((host,port))
print("Active port …