Class Twitter

pattern/web/__init__.py:1410–1524

TWITTER_RETWEET = re.compile(r"(\s|^RT )(@[a-z0-9_\-]+)", re.I) # Word starts with "RT @".

class Twitter(SearchEngine):

    def __init__(self, license=None, throttle=0.5, language=None):
        SearchEngine.__init__(self, license or TWITTER_LICENSE, throttle, language)
        self._pagination = {}

    def _authenticate(self, url):
        url.query.update({
            "oauth_version": "1.0",
            "oauth_nonce": oauth.nonce(),
            "oauth_timestamp": oauth.timestamp(),
            "oauth_consumer_key": self.license[0],
            "oauth_token": self.license[2][0],
            "oauth_signature_method": "HMAC-SHA1"
        })
        url.query["oauth_signature"] = oauth.sign(url.string.split("?")[0], url.query,
            method = GET,
            secret = self.license[1],
             token = self.license[2][1]
        )
        return url

    def search(self, query, type=SEARCH, start=1, count=10, sort=RELEVANCY, size=None, cached=False, **kwargs):
        """ Returns a list of results from Twitter for the given query.
            - type : SEARCH,
            - start: Result.id or int,
            - count: maximum 100.
            There is a limit of 150+ queries per 15 minutes.
        """
        if type != SEARCH:
            raise SearchEngineTypeError
        if not query or count < 1 or (isinstance(start, (int, float)) and start < 1):
            return Results(TWITTER, query, type)
        if isinstance(start, (int, float)) and start < 10000:
            id = (query, kwargs.get("geo"), int(start), count)
            id = self._pagination.pop(id, "")
        else:
            id = start or ""
        # 1) Construct request URL.
        url = URL(TWITTER + "search/tweets.json?", method=GET)
        url.query = {
               "q": query,
          "max_id": id,
           "count": min(count, 100)
        }
        # 2) Restrict location with geo=(latitude, longitude, radius).
        #    It can also be a (latitude, longitude)-tuple with default radius "10km".
        if "geo" in kwargs:
            url.query["geocode"] = ",".join((map(str, kwargs.pop("geo")) + ["10km"])[:3])
        # 3) Restrict most recent with date="YYYY-MM-DD".
        #    Only older tweets are returned.
        if "date" in kwargs:
            url.query["until"] = kwargs.pop("date")
        # 4) Restrict language.
        url.query["lang"] = self.language or ""
        # 5) Authenticate.
        url = self._authenticate(url)
        # 6) Parse JSON response.