使用 Python 构建 Twitter 机器人

2020-05-05 08:00:00 · 飞浪

介绍 Twitter无需介绍——每个人都知道它。这个流行的社交媒体平台使用主题标签来代表主题，任何人都有权创建自己的主题标签。每秒都会使用这些主题标签进行大量讨论。营销等行业经常使用Tw

介绍

Twitter 无需介绍——每个人都知道它。这个流行的社交媒体平台使用主题标签来代表主题，任何人都有权创建自己的主题标签。每秒都会使用这些主题标签进行大量讨论。营销等行业经常使用 Twitter 来了解人们对特定主题标签的普遍看法，并据此为客户提供服务。

需要收集推文和主题标签才能进行分析。复制每条提及特定主题标签的推文并不是一个可行的选择，因为每分钟都会发布数百条推文。需要智能计算机程序或机器人来获取热门主题标签和推文并保存它们以供进一步分析。

在本指南中，您将学习如何制作一个可以为您获取主题标签和推文的自定义 Twitter 机器人。

获取 Twitter API

要让您的机器人与 Twitter 互动，您首先必须在 Twitter 上注册为开发者。导航至此处注册并获取 API 密钥。

在本指南中，您将发出 GET 请求来获取推文和主题标签，为此，您需要“CONSUMER_KEY”、“CONSUMER_SECRET”、“ACCESS_KEY”和“ACCESS_SECRET”键。

在开始编写代码之前，了解 Twitter 请求的速率限制非常重要。Twitter 有免费和高级开发者帐户，在 15 分钟内可以发出的请求数量有所不同。速率限制可能会发生变化，因此要获取当前规范，请阅读官方文档。

所需配置和 Python 库

使用Tweepy库实现 Twitter 机器人。API文档组织良好。

在 Python 中安装Tweepy 。

      pip install tweepy

创建config.json文件并输入所有上述 API 密钥。

      {
    "CONSUMER_KEY" : "<KEY>", 
    "CONSUMER_SECRET" : "<KEY>", 
    "ACCESS_KEY" : "<KEY>", 
    "ACCESS_SECRET" : "<KEY>"
}
    

编写机器人代码

Twitter 按位置保存主题标签和推文。为了指代不同的地理位置，使用术语WOEID（地球位置标识符）。您可以在此处找到可用国家/地区的WOEID。

导入所需库

您将在整个指南中使用这些库。将它们导入到您的文件中。

      import tweepy
import json
import schedule
import time
import datetime
import os
import csv
    

启动 API

访问config.json并使用所有访问密钥启动 API。由于互联网连接、服务器响应时间等原因，与服务器建立连接有时会出现问题。处理错误总是好的，这样会更清晰。

      def initiate_api():
    try: 
        with open('config.json', 'r') as f:
            config = json.load(f)        
        auth = tweepy.OAuthHandler(config["CONSUMER_KEY"], config["CONSUMER_SECRET"])
        auth.set_access_token(config["ACCESS_KEY"], config["ACCESS_SECRET"])
        api = tweepy.API(auth)
        return api
    except:
        print("Problems with config.json")
        return None
    

仅接受英文推文

在本指南中，您将只关注英文推文。其他语言的推文（例如中文、阿拉伯语、印地语等）将不予考虑。您可以轻松地在此功能中包含其他语言，而不会影响应用程序的其他部分。

      def isEnglish(text):
    try:
        text.encode(encoding='utf-8').decode('ascii')
    except UnicodeDecodeError:
        return False
    else:
        return True
    

获取国家/地区的 WOEID

此函数将有助于获取特定位置的 WOEID。它以api对象和位置列表作为参数并返回它们的 WOEID。

      def get_woeid(api, locations):
    twitter_world = api.trends_available()
    places = {loc['name'].lower() : loc['woeid'] for loc in twitter_world};
    woeids = []
    for location in locations:
        if location in places:
            woeids.append(places[location])
        else:
            print("err: ",location," woeid does not exist in trending topics")
    return woeids
    

获取推文

此函数将获取给定主题标签的推文。api对象将获取给定查询的推文。此代码仅获取英文推文，但您也可以操纵它来获取其他语言的推文。

      '''
Getting Tweets for the given hashtag with max of 1000 popular tweets with english dialect
'''
def get_tweets(api, query):
    tweets = []
    for status in tweepy.Cursor(api.search,
                       q=query,
                       count=1000,
                       result_type='popular',
                       include_entities=True,
                       monitor_rate_limit=True, 
                       wait_on_rate_limit=True,
                       lang="en").items():
     
        # Getting only tweets which has english dialects
        if isEnglish(status.text) == True:
            tweets.append([status.id_str, query, status.created_at.strftime('%d-%m-%Y %H:%M'), status.user.screen_name, status.text])
    return tweets
    

它将返回给定查询的推文列表。推文将包括 ID、主题标签、创建时间、用户名和推文正文。

获取热门标签

在此功能中，您将获取给定位置的热门主题标签。

注意：如果您使用的是免费开发者帐户，那么您的请求数量将非常有限。此代码的编程方式是，如果您的每小时请求已用尽，此机器人将等待一小时，然后恢复。

      def get_trending_hashtags(api, location):
    woeids = get_woeid(api, location)
    trending = set()
    for woeid in woeids:
        try:
            trends = api.trends_place(woeid)
        except:
            print("API limit exceeded. Waiting for next hour")
            #time.sleep(3605) # change to 5 for testing
            trends = api.trends_place(woeid)
        # Checking for English dialect Hashtags and storing text without #
        topics = [trend['name'][1:] for trend in trends[0]['trends'] if (trend['name'].find('#') == 0 and isEnglish(trend['name']) == True)]
        trending.update(topics)
    
    return trending
    

把所有东西放在一起

此函数将所有函数整合在一起。所有获取的推文都将保存在trending_tweets目录中。每次机器人运行时，它都会将热门标签和推文保存在带有时间戳的不同csv文件中。

      def twitter_bot(api, locations):
    today = datetime.datetime.today().strftime("%d-%m-%Y-%s")
    if not os.path.exists("trending_tweets"):
        os.makedirs("trending_tweets")
    file_tweets = open("trending_tweets/"+today+"-tweets.csv", "a+")
    file_hashtags = open("trending_tweets/"+today+"-hashtags.csv", "w+")
    writer = csv.writer(file_tweets)
    
    hashtags = get_trending_hashtags(api, locations)
    file_hashtags.write("\n".join(hashtags))
    print("Hashtags written to file.")
    file_hashtags.close()
    
    for hashtag in hashtags:
        try:
            print("Getting Tweets for the hashtag: ", hashtag)
            tweets = get_tweets(api, "#"+hashtag)
        except:
            print("API limit exceeded. Waiting for next hour")
            #time.sleep(3605) # change to 0.2 sec for testing
            tweets = get_tweets(api, "#"+hashtag)
        for tweet in tweets:
            writer.writerow(tweet)
    
    file_tweets.close()
    

有一个注释代码，time.sleep()。如果你想尝试一下文件保存过程，请使用它。

主要功能

最后，主函数会立即调用机器人。由于请求较少，我只使用位置列表中的一个位置。您可以将任意数量的国家/地区放入其中，此代码将处理所有事情。

计划包用于保持程序始终运行。目前，机器人将在每天 00:00 获取数据；但是，您可以根据需要更改计划。有一个注释部分，每 10 秒安排机器人运行一次。实际上，您需要大量的请求和非常好的硬件基础设施来每 10 秒处理一次请求。

      def main():
    ''' 
    Use location = [] list for getting trending tags from different countries. 
    I have limited number of request hence I am using only 1 location
    '''
    #locations = ['new york', 'los angeles', 'philadelphia', 'barcelona', 'canada', 'united kingdom', 'india']        
    locations = ['new york']
    api = initiate_api()
    
    schedule.every().day.at("00:00").do(twitter_bot, api, locations)
    #schedule.every(10).seconds.do(twitter_bot, api, locations)
    while True:
        schedule.run_pending()
        time.sleep(1)
        
if __name__ == "__main__":
    main()
    

结论

在本指南中，您学习了如何制作一个 Twitter 机器人，它可以获取不同地理位置的热门主题标签和推文。只需对此代码进行一些修改，您就可以将其用于任何业务用例。

Twitter 是研究人类对某个主题的行为和态度的最流行方式之一。要研究推文，我们必须首先收集大量推文，而自动获取推文始终是最好的解决方案。在下一个指南中，我们将构建一个机器学习模型来对这些推文进行情绪分析。单击此处阅读Python 中的 Twitter 情绪分析。

_{免责声明：本内容来源于第三方作者授权、网友推荐或互联网整理，旨在为广大用户提供学习与参考之用。所有文本和图片版权归原创网站或作者本人所有，其观点并不代表本站立场。如有任何版权侵犯或转载不当之情况，请与我们取得联系，我们将尽快进行相关处理与修改。感谢您的理解与支持！}

_查看原文

python

技术指南

阅读全文

使用 Python 构建 Twitter 机器人

杭州电子商务研究院

4年前 · 面向社会、服务行业、政产学研结合、整合资源、和谐发展

def initiate_api(): try: with open('config.json', 'r') as f: config = json.load(f) auth = tweepy.OAuthHandler(config["CONSUMER_KEY"], config["CONSUMER_SECRET"]) auth.set_access_token(config["ACCESS_KEY"], config["ACCESS_SECRET"]) api = tweepy.API(auth) return api except: print("Problems with config.json") return None

def get_woeid(api, locations): twitter_world = api.trends_available() places = {loc['name'].lower() : loc['woeid'] for loc in twitter_world}; woeids = [] for location in locations: if location in places: woeids.append(places[location]) else: print("err: ",location," woeid does not exist in trending topics") return woeids

''' Getting Tweets for the given hashtag with max of 1000 popular tweets with english dialect ''' def get_tweets(api, query): tweets = [] for status in tweepy.Cursor(api.search, q=query, count=1000, result_type='popular', include_entities=True, monitor_rate_limit=True, wait_on_rate_limit=True, lang="en").items(): # Getting only tweets which has english dialects if isEnglish(status.text) == True: tweets.append([status.id_str, query, status.created_at.strftime('%d-%m-%Y %H:%M'), status.user.screen_name, status.text]) return tweets

def get_trending_hashtags(api, location): woeids = get_woeid(api, location) trending = set() for woeid in woeids: try: trends = api.trends_place(woeid) except: print("API limit exceeded. Waiting for next hour") #time.sleep(3605) # change to 5 for testing trends = api.trends_place(woeid) # Checking for English dialect Hashtags and storing text without # topics = [trend['name'][1:] for trend in trends[0]['trends'] if (trend['name'].find('#') == 0 and isEnglish(trend['name']) == True)] trending.update(topics) return trending

def twitter_bot(api, locations): today = datetime.datetime.today().strftime("%d-%m-%Y-%s") if not os.path.exists("trending_tweets"): os.makedirs("trending_tweets") file_tweets = open("trending_tweets/"+today+"-tweets.csv", "a+") file_hashtags = open("trending_tweets/"+today+"-hashtags.csv", "w+") writer = csv.writer(file_tweets) hashtags = get_trending_hashtags(api, locations) file_hashtags.write("\n".join(hashtags)) print("Hashtags written to file.") file_hashtags.close() for hashtag in hashtags: try: print("Getting Tweets for the hashtag: ", hashtag) tweets = get_tweets(api, "#"+hashtag) except: print("API limit exceeded. Waiting for next hour") #time.sleep(3605) # change to 0.2 sec for testing tweets = get_tweets(api, "#"+hashtag) for tweet in tweets: writer.writerow(tweet) file_tweets.close()

def main(): ''' Use location = [] list for getting trending tags from different countries. I have limited number of request hence I am using only 1 location ''' #locations = ['new york', 'los angeles', 'philadelphia', 'barcelona', 'canada', 'united kingdom', 'india'] locations = ['new york'] api = initiate_api() schedule.every().day.at("00:00").do(twitter_bot, api, locations) #schedule.every(10).seconds.do(twitter_bot, api, locations) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": main()