我如何使用python和tweepy以便从两个给定日期之间的Twitter收集推文？
有没有办法从...直到...值传递给搜索API？
注意：我需要能够进行搜索，但不限于特定用户
我正在使用python，我知道代码应该像这样，但是我需要帮助才能使其正常工作。
身份验证= tweepy.OAuthHandler（consumer_key，consumer_secret）
auth.set_access_token（access_token_key，access_token_secret）
api = tweepy.API（身份验证）
集合= []
用于tweepy.Cursor（api.search，???????）。items（）中的推文：
集合[tweet.id] = tweet._json
使用 Python 中的 tweepy，我正在寻找一种方法来列出一个帐户中的所有关注者，以及用户名和关注者数量。现在我可以通过这种方式获取所有 id 的列表：
ids = []
for page in tweepy.Cursor(api.followers_ids, screen_name="username").pages():
ids.extend(page)
time.sleep(1)
但是有了这个 id 列表，我无法获得每个 id 的用户名和关注者数量，因为速率限制超过了......我如何完成这段代码？
谢谢你们！
我有一个包含推文ID列表的文件,我想要检索这些推文.该文件包含超过100000条推文,而twitter API仅允许检索100条.
api = tweepy.API(auth)
good_tweet_ids = [i for i in por.TweetID[0:100]]
tweets = api.statuses_lookup(good_tweet_ids)
for tweet in tweets:
print(tweet.text)
有没有办法检索更多的推文说1000或2000,我不想采取数据样本并将结果保存到文件并每次更改推文ID的索引,所以有办法做到这一点!？
我正在使用 Tweepy 的 Stream Listener 并希望检索有关英国当前政治辩论的推文。不幸的是，在 RT 和回复的情况下，我只会收到截断的推文。
如：-
RT @ZaidJilani：Chuck Schumer（antiBDS 法案的发起人）说我们应该扼杀加沙。Jeremy Corbyn 说压迫他们会……
当全文应该是：-
查克舒默（antiBDS 法案的发起人）说我们应该扼杀加沙。Jeremy Corbyn 说压迫他们只会使人们变得激进。
我已经看到有一种方法可以将 `tweet_mode=extended 与常规 Twitter.API 一起使用。但是我找不到与 Streaming API 类似的东西。有没有人解决这个问题？我的代码如下：-
from tweepy import Stream
from tweepy import OAuthHandler
from tweepy.streaming import StreamListener
from redis import Redis
from rq import Queue
import requests
import time
import io
import os
import json
import threading
import multiprocessing
from datetime import datetime, timedelta
import _credentials
# twitter OAuth
ckey = _credentials.ckey
consumer_secret = _credentials.consumer_secret
access_token_key = …因此，正如问题所暗示的那样，我正试图弄清楚如何获取推文 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")
这段代码工作正常，只是试图了解我将添加什么以获得所需的数据。
import tweepy
ACCESS_KEY = 'XXXXXXXXXXXXXXXXXX'
ACCESS_SECRET = 'XXXXXXXXXXXXXXXXXXXXXXX'
CONSUMER_KEY = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
CONSUMER_SECRET = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
api = tweepy.Client(bearer_token='XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX',
access_token=ACCESS_KEY,
access_token_secret=ACCESS_SECRET,
consumer_key=CONSUMER_KEY,
consumer_secret=CONSUMER_SECRET)
api.media_upload(file.jpg)
我无法上传照片并获取media_ids
我无权访问 Twitter API V1
我正在尝试通过 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 …我尝试使用以下代码通过 tweepy API 检索推文，但检索到的 json 字典有错误。
代码：
import tweepy
from tweepy import OAuthHandler
from tweepy import Stream
import socket
import json
consumer_key="****"
consumer_secret="****"
access_token="****"
access_secret="****"
class TweetListener(Stream):
def __init__(self, *args, csocket):
super().__init__(*args)
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):
twtr_stream = TweetListener(
consumer_key, consumer_secret,
access_token, access_secret,
csocket=c_socket
)
twtr_stream.filter(track=['ETH'])
s = socket.socket()
host = "127.0.0.1"
port = …