Python Twitter Tool 同步搜索和流式传输

Question

我正在尝试创建一个 Python 服务器，我可以从其他应用程序调用它来请求 Twitter 数据。我通常使用 Python 作为脚本语言，所以如果有人在我的代码中看到任何危险信号，我会全力以赴！

这基本上是我目前所拥有的，当我 ping 服务器时，它运行良好，它从我的时间线获取 10 条推文并将它们发送回我的其他应用程序。我的主要问题是我想将流媒体和搜索结合起来。这样我就可以为特定的哈希标签打开流，我想实时发送到我的其他应用程序，但是我会定期搜索其他不需要的东西我是实时的。

我已经成功地分别使用了两者，但如果我想同时实现这两个，我不确定从哪里开始，在这种情况下，我想将流功能引入其中。

我正在使用 Python Twitter 工具 1.10.2 - http://mike.verdone.ca/twitter/ 和 Python 3.3

代码如下，谢谢！

编辑：通过在 if data == "SEARCH_NOW" if 语句之后添加 twitter 流连接，我能够更进一步。但这提出了我遇到的原始问题。推特流打开后，代码似乎就在那里等着。如果我把它放在时间线查找之前，那么我永远不能调用时间线查找。更新了代码以反映。

编辑 2：将搜索请求放入 twitter 流循环中会更接近一些。我现在可以打开流，每次收到与搜索词匹配的推文时，我也可以发出请求。但仍然不是独立...

文件：network_settings.py

#!/usr/bin/env python

#network settings
import socket

#set server variables
TCP_IP = '127.0.0.1'
TCP_PORT = 7001
BUFFER_SIZE  = 20

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((TCP_IP, TCP_PORT))
s.listen(1)

conn, addr = s.accept()

#print connection address when someone connects
print ('Connection address:', addr)

文件：twitter_settings.py

from twitter import *
import re

OAUTH_TOKEN = ''
OAUTH_SECRET = ''
CONSUMER_KEY = ''
CONSUMER_SECRET = ''

t = Twitter(auth=OAuth(OAUTH_TOKEN, OAUTH_SECRET, CONSUMER_KEY, CONSUMER_SECRET))

auth = OAuth(OAUTH_TOKEN, OAUTH_SECRET, CONSUMER_KEY, CONSUMER_SECRET)
stream = TwitterStream(auth = auth, secure = True)

文件：python_server.py

#python server 

import json 

from network_settings import *
from twitter_settings import *

search_term = 'test'

while 1:

    tweet_iter = stream.statuses.filter(track = search_term)

    for tweet in tweet_iter:
        # check whether this is a valid tweet
        if tweet.get('text'):

            userName = tweet["user"]["screen_name"]
            userTweet = tweet["text"]

            # now print our tweet
            print ('user: ', userName)
            print ('tweet: ', userTweet)

            #send data back
            delivery1 = json.dumps({'type':'showdown','userName':userName,'userTweet':userTweet})
            conn.send(delivery1.encode('utf-8'))
            data = conn.recv(BUFFER_SIZE)
            data = data.decode('utf-8')

            if data == "SEARCH_NOW": 
                 print ('request newest IDS tweets')

                 x = t.statuses.home_timeline(count=10)

                for i in range(10):
                    try:
                        #print(x[i])
                        userName = x[i]['entities']['user_mentions'][0]['screen_name']
                        userTweet = x[i]['text']
                        print('username: ', userName)
                        print('tweet: ', userTweet)
                        delivery = json.dumps({'type':'display','userName':userName,'userTweet':userTweet})
                        conn.send(delivery.encode('utf-8'))
                    except:
                        print('not valid tweet')

conn.close()

Answer 1

所以终于找到了解决方案。我最终使用线程在它自己的线程中运行流，然后每次搜索时我都会打开另一个线程。不确定我是否需要关闭每个线程，或者返回是否负责。如果有人有任何他们可以改进的东西，我会全力以赴！

代码如下：

#!/usr/bin/env python
#python server 

import json 
import threading
import time
import socket
from twitter import *
import re

#get thread lock ready
thread_lock = threading.Lock()

#set server variables
TCP_IP = '127.0.0.1'
TCP_PORT = 7001
BUFFER_SIZE  = 20


s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((TCP_IP, TCP_PORT))
s.listen(1)

conn, addr = s.accept()

#print connection address when someone connects
print ('Connection address:', addr)

#fill these in your app!
#twitter auth keys
OAUTH_TOKEN = ''
OAUTH_SECRET = ''
CONSUMER_KEY = ''
CONSUMER_SECRET = ''

t = Twitter(auth=OAuth(OAUTH_TOKEN, OAUTH_SECRET, CONSUMER_KEY, CONSUMER_SECRET))

auth = OAuth(OAUTH_TOKEN, OAUTH_SECRET, CONSUMER_KEY, CONSUMER_SECRET)
stream = TwitterStream(auth = auth, secure = True)


#twitter functions
def pythonSearch():
    #lock thread to not interrupt search results
    thread_lock.acquire()

    print ('request newest tweets')

    #get 10 things from timeline
    x = t.statuses.home_timeline(count=10)

    for i in range(10):
        try:
            #get username and tweet
            userName = x[i]['entities']['user_mentions'][0]['screen_name']
            userTweet = x[i]['text']

            #print out values
            print('username: ', userName)
            print('tweet: ', userTweet)

            #send json back
            delivery = json.dumps({'type':'display','userName':userName,'userTweet':userTweet})
            conn.send(delivery.encode('utf-8'))
        except:
            #not a retweet
            print('not valid tweet')

    #unlock thread when finished
    thread_lock.release()

    return


def pythonStream():
    #open stream looking for search_term
    search_term = 'TESTING'
    tweet_iter = stream.statuses.filter(track = search_term)

    for tweet in tweet_iter:
        # check whether this is a valid tweet
        if tweet.get('text'):

            #get username and tweet
            userName = tweet["user"]["screen_name"]
            userTweet = tweet["text"]

            # now print our tweet
            print ('user: ', userName)
            print ('tweet: ', userTweet)

            #send json back
            delivery1 = json.dumps({'type':'showdown','userName':userName,'userTweet':userTweet})
            conn.send(delivery1.encode('utf-8'))


#start main loop
while 1:
    #listen for calls
    data = conn.recv(BUFFER_SIZE)
    data = data.decode('utf-8')

    #if someone calls search, do a search
    if data == 'SEARCH':
        threading.Thread(target = pythonSearch).start()


    if data == 'STREAM':
        threading.Thread(target = pythonStream).start()


conn.close()