使用Python筛选推特评论区用户开发抽奖功能

实践
1

您在推特的经营过程中可能会遇到这样的场景——在推特评论区筛选符合要求的评论参与者并从中抽选获奖者。鉴于庞大的数据量、效率以及公正性考虑，这样的问题最好使用程序来解决。

我们今天就来谈一下如何使用Python对推特评论区用户进行筛选从而创建公正高效的抽奖功能。

领奖台
领奖台1080×581 18.8 KB

首先，请从推特开发者平台免费激活推特(Twitter)开发者账号并在此处获得相关开发者API密钥，包括：API Key、API Key Secret、Access Token 以及 Access Token Secret。

然后，请打开终端通过以下命令安装 tweepy:

pip install tweepy

安装完成以后，您就可以使用上面提到的开发者API密钥通过 tweepy 中相关 Python API 进行进行鉴权。代码列子如下：

client = tweepy.Client(bearer_token = config.BEARER_TOKEN, 
consumer_key = config.API_KEY,
consumer_secret=config.API_KEY_SECRET,
access_token=config.ACCESS_TOKEN,
access_token_secret=config.ACCESS_TOKEN_SECRET)

接下来您可以使用函数 search_recent_tweets 对包含指定关键词并包含用户评论的推文进行搜索：

query = 'from:Taylor_Swift -is:reply'
tweets = client.search_recent_tweets(query=query, tweet_fields=['context_annotations', 'created_at', 'conversation_id', 'referenced_tweets', 'author_id'], max_results=100, expansions=["in_reply_to_user_id","referenced_tweets.id", 'author_id'])

以上代码中我们使用 from:Taylor_Swift 参数来搜索包含"Taylor Swift"关键词的推文，并通过 -is:reply 筛选其中至少包括1条评论的推文。

随后我们可以从以上结果列表中选取最后一个推文：

tweet = tweets.data[0]

并获取该推文所有的评论数据：

valid_replies =[]
specific_conversation_query = f'conversation_id:{tweet.conversation_id} to:Taylor_Swift'
for reply in tweepy.Paginator(client.search_recent_tweets, query=specific_conversation_query,
                            tweet_fields=['context_annotations', 'created_at', 'conversation_id', 'referenced_tweets', 'in_reply_to_user_id', 'id', 'author_id'], 
                            expansions=["in_reply_to_user_id","referenced_tweets.id", 'author_id'],
                            max_results=100).flatten(limit=1000):
    logging.info(reply)
    print(reply)
    valid_replies.append(reply)

上述代码中我们使用了 Paginator 函数限制只对评论区前1000页内容进行抓取，并将结果存入 valid_replies 列表中。随后，我们将会从中随机挑选包含正确答案的评论，并输出该评论所属用户的推特用户名(username)。

代码如下：

def challenge1(potential_winner):
    if "ADA" in potential_winner.text:
        return True
    else:
        return False 

username = ''
selected_winner = None
while len(valid_replies)>0:
    selected_winner = random.choice(valid_replies)
    print(f"selected: {selected_winner.data}")
    username=client.get_users(ids=[selected_winner.author_id])
    if challenge1(selected_winner):
        print(f"The {username} is valid")
        logging.info(f"The {username} is valid")
        break
    else:
        logging.warn(f"The {username} is NOT valid, removing from list")
        valid_replies.remove(selected_winner)
print(selected_winner.data)

challenge1函数负责筛选评论内容包括关键词 ADA 所对应的用户名。在 9-19 行代码中，我们对列表 valid_replies 进行循环搜索，随机从中选择一条评论并使用challenge1函数进行验证，最后再将处理结果输出到文件中。

于是我们就完成了从推特筛选评论区用户创建抽奖功能的全部功能。完整代码如下：


import csv

import tweepy
import config
import logging
import random
import re

logging.basicConfig(filename=config.GIFT_LOGS, level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')
# Authentication with Twitter
client = tweepy.Client(bearer_token = config.BEARER_TOKEN, 
consumer_key = config.API_KEY,
consumer_secret=config.API_KEY_SECRET,
access_token=config.ACCESS_TOKEN,
access_token_secret=config.ACCESS_TOKEN_SECRET)

query = 'from:Taylor_Swift -is:reply'

tweets = client.search_recent_tweets(query=query, tweet_fields=['context_annotations', 'created_at', 'conversation_id', 'referenced_tweets', 'author_id'], max_results=100, expansions=["in_reply_to_user_id","referenced_tweets.id", 'author_id'])

tweet = tweets.data[0]
print(f"Id is {tweet.id}")
print(tweet)
logging.info(tweet)

logging.info("replies:")
valid_replies =[]
specific_conversation_query = f'conversation_id:{tweet.conversation_id} to:Taylor_Swift'
for reply in tweepy.Paginator(client.search_recent_tweets, query=specific_conversation_query,
                            tweet_fields=['context_annotations', 'created_at', 'conversation_id', 'referenced_tweets', 'in_reply_to_user_id', 'id', 'author_id'], 
                            expansions=["in_reply_to_user_id","referenced_tweets.id", 'author_id'],
                            max_results=100).flatten(limit=1000):
    logging.info(reply)
    valid_replies.append(reply)

def challenge1(potential_winner):
    if "ADA" in potential_winner.text:
        return True
    else:
        return False 

username = ''
selected_winner = None
while len(valid_replies)>0:
    selected_winner = random.choice(valid_replies)
    print(f"selected: {selected_winner.data}")
    username=client.get_users(ids=[selected_winner.author_id])
    if challenge1(selected_winner):
        print(f"The {username} is valid")
        logging.info(f"The {username} is valid")
        break
    else:
        logging.warn(f"The {username} is NOT valid, removing from list")
        valid_replies.remove(selected_winner)
print(selected_winner.data)

print(f"username: {username}")
logging.info(f"Send gift to {selected_winner.author_id}, tweet: {selected_winner.id}")
logging.info(f"The winner is {username.data[0].name}, @{username.data[0].username}. Congratulations!!")

使用以上代码请不要忘记填写您的推特开发者账户API信息，并更改关键词。您也可以根据具体业务场景对以上代码进行修改，实现更加复杂的功能。