如何识别statuses/filter流式API返回推文对应的跟踪关键词

要解决这个问题，你可以在获取到推文文本后，和预设的跟踪关键词列表做匹配，提取出当前推文命中的所有关键词，同时你原有代码存在缩进错误，也需要同步修正：

核心改动点 #

• 把跟踪关键词列表单独定义为公共变量，避免重复编写
• 新增大小写不敏感的关键词匹配逻辑，支持单条推文命中多个关键词的场景
• 修正原有代码的缩进错误（on_error方法位置、文件写入缩进、流初始化位置等问题）

修改后完整代码 #

import tweepy
from tweepy import Stream, StreamListener
import json
import csv
import nltk
nltk.download('punkt')

# 单独定义跟踪关键词列表，统一维护
TRACK_KEYWORDS = ['Trump', 'ocean', 'Democrat', 'Republican', 'gardening','BMCSoftware', 'school', 'CT', '#metoo', 'governor', 'Iran']

class MyListener(StreamListener):
    def on_data(self, data):
        raw_tweets = json.loads(data)
        try:
            created_at = raw_tweets['created_at']
            tweet_text = raw_tweets['text']
            tweet_id = raw_tweets['id']
            screen_name = raw_tweets['user']['screen_name']
            hashtags = raw_tweets['entities']['hashtags']
            description = raw_tweets['user']['description']
            retweet_count = raw_tweets['retweet_count']
            
            # 新增：匹配命中的关键词，大小写不敏感
            matched_keywords = []
            lower_text = tweet_text.lower()
            for kw in TRACK_KEYWORDS:
                if kw.lower() in lower_text:
                    matched_keywords.append(kw)
            # 把命中的关键词转成字符串方便存入csv
            matched_kw_str = ','.join(matched_keywords)

            with open('Stream_data.csv', 'a', encoding='utf-8') as myFile:
                writer = csv.writer(myFile)
                # 新增命中关键词字段存入csv
                writer.writerow([created_at, tweet_id, screen_name, hashtags, description, tweet_text, retweet_count, matched_kw_str])
            
            print("created_at: ", created_at)
            print("id: ", tweet_id)
            print("screen_name: ", screen_name)
            print("hashtag: ", hashtags)
            print("description: ", description)
            print("text: ", tweet_text)
            print("retweet: ", retweet_count)
            print("命中关键词: ", matched_kw_str)

        except Exception as e:
            # 打印错误信息方便排查
            print("处理出错: ", e)

    def on_error(self, status):
        print("接口报错: ", status)

# 鉴权和流初始化逻辑放在类外，替换成你自己的密钥即可
consumer_key = '你的consumer_key'
consumer_secret = '你的consumer_secret'
access_token = '你的access_token'
access_token_secret = '你的access_token_secret'

auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

twitter_stream = Stream(auth, MyListener())
twitter_stream.filter(track=TRACK_KEYWORDS)

补充说明 #

如果你需要更精确的匹配（比如避免短词被误命中，比如CT可能出现在其他单词里），可以改成正则全词匹配，把匹配逻辑替换为：

import re
for kw in TRACK_KEYWORDS:
    # 构造全词匹配的正则，忽略大小写
    pattern = re.compile(r'\b' + re.escape(kw) + r'\b', re.IGNORECASE)
    if pattern.search(tweet_text):
        matched_keywords.append(kw)

存入csv时新增的encoding='utf-8'参数可以避免特殊字符乱码。

内容的提问来源于stack exchange，提问作者NOMAN AHMED