0

这是我尝试流式传输推文的项目中的部分代码，我计划将其保存到 Google App Engine 中的数据库中。能够获得坐标的纬度和经度值对我来说很重要。（我计划稍后绘制这些。）

目前的结果看起来像这样......

'tweet text'@REDACTED 不确定，说实话……作为谷歌产品，你会这么认为。他们可能在网上商店有官方扩展。\'用户名'REDACTED \'Created at't2013-09-26 08:39:45 \'Created with'tTwitter for Android \'geo't{u'type': u'Point', u'coordinates': [52.569001, -2.7846582]}\'coordinates't{u'type': u'Point', u'coordinates': [-2.7846582, 52.569001]}

我想做的是改变它所说的“如果status.coordinates不是None”来检查坐标是否在一个范围内。即 Lat 50 - 55 和 long 0 - 5。

谢谢！:)

class CustomStreamListener(tweepy.StreamListener):

    def on_status(self, status):

        if status.coordinates is not None:
            try:
                print "'tweet text'%s\n\ 'User Name't%s\n\  'Created at't%s\n\  'Created with't%s\n\ 'geo't%s\ 'coordinates't%s" % (status.text, 
                                  status.author.screen_name, 
                                  status.created_at, 
                                  status.source,
                                  status.geo,
                                  status.coordinates)
            except Exception, e:
                print >> sys.stderr, 'Encountered Exception:', e
                pass

    def on_error(self, status_code):
        print >> sys.stderr, 'Encountered error with status code:', status_code
        return True # Don't kill the stream

    def on_timeout(self):
        print >> sys.stderr, 'Timeout...'
        return True # Don't kill the stream
4

3 回答 3

2

您可能需要根据地球上两点的大圆距离来决定：

from math import *

def great_circle_distance(coordinates1, coordinates2):
  latitude1, longitude1 = coordinates1
  latitude2, longitude2 = coordinates2
  d = pi / 180  # factor to convert degrees to radians
  return acos(sin(longitude1*d) * sin(longitude2*d) +
              cos(longitude1*d) * cos(longitude2*d) *
              cos((latitude1 - latitude2) * d)) / d

def in_range(coordinates1, coordinates2, range):
  return great_circle_distance(coordinates1, coordinates2) < range

请记住，地球的 90 度传统上代表 10000 公里（AFAIK 这是米的古老定义），因此要获得 10 公里的半径，只需使用 0.09 度。

于 2013-09-26T09:35:06.890 回答
0

假设坐标以 [latitude, longitude] 格式给出，您可以按如下方式检查它们：

def check_coords(coords):
    lat = coords[0]
    lng = coords[1]
    return 50 < lat < 55 and 0 < lng < 55

如果纬度在 50 - 55 之间且经度在 0 - 5 之间，这将返回True。如果任何一个超出其定义的范围，则函数将返回False

编辑：将此添加到您的班级将使其看起来像这样：

class CustomStreamListener(tweepy.StreamListener):

    def on_status(self, status):

        if status.coordinates is not None:
            # coordinates_in_range will be either True or False
            coordinates_in_range = self.check_coords(status.coordinates['coordinates'])
            try:
                print "'tweet text'%s\n\ 'User Name't%s\n\  'Created at't%s\n\  'Created with't%s\n\ 'geo't%s\ 'coordinates't%s" % (status.text, 
                                  status.author.screen_name, 
                                  status.created_at, 
                                  status.source,
                                  status.geo,
                                  status.coordinates)
            except Exception, e:
                print >> sys.stderr, 'Encountered Exception:', e
                pass

    def on_error(self, status_code):
        print >> sys.stderr, 'Encountered error with status code:', status_code
        return True # Don't kill the stream

    def on_timeout(self):
        print >> sys.stderr, 'Timeout...'
        return True # Don't kill the stream

    def check_coords(self, coords):
        latitude, longitude = coords
        return 50 < latitude < 55 and 0 < longitude < 55
于 2013-09-26T09:22:55.297 回答
0

感谢@Alfe 和@Hellsgate 的帮助和建议。

以下代码有效（目前 Hellsgate 上面的代码返回带有地理标记的推文，但不按坐标过滤。）

在 CustomStreamListener 之上，您只需要添加导入语句和 OAuth 方法。下面的代码将只返回来自英国各地的搜索词“Rarity Diamonds,Applejack,Discord”的推文

class CustomStreamListener(tweepy.StreamListener):

def check_coords(self, status):
    if status.coordinates is not None:
        lat = status.coordinates['coordinates'][1]
        lng = status.coordinates['coordinates'][0]        
        return 49.7 < lat < 58.5 and -6 < lng < 2.1

def on_status(self, status):


    if self.check_coords(status):
        # if check_coords returns true, the following will run
        try:
            print "'tweet text'%s\n\ 'User Name't%s\n\  'Created at't%s\n\  'Created with't%s\n\ 'geo't%s\ 'coordinates't%s" % (status.text, 
                              status.author.screen_name, 
                              status.created_at, 
                              status.source,
                              status.geo,
                              status.coordinates)
        except Exception, e:
            print >> sys.stderr, 'Encountered Exception:', e
            pass

def on_error(self, status_code):
    print >> sys.stderr, 'Encountered error with status code:', status_code
    return True # Don't kill the stream

def on_timeout(self):
    print >> sys.stderr, 'Timeout...'
    return True # Don't kill the stream


streaming_api = tweepy.streaming.Stream(auth, CustomStreamListener(), timeout=60)


print >> sys.stderr, 'Filtering the public timeline for "%s"' % (' '.join(sys.argv[1:]),)
"""For track=[""], put in words to search for. Commas separate individual terms IE"Applejack, Discord",
to search for tweets containing more than one term, separate keywords with a space. IE "Rarity Diamonds" """
streaming_api.filter(follow=None, track=["Rarity Diamonds,Applejack,Discord"])
于 2013-09-27T10:40:54.740 回答