[PYTHON] Acquisition continue par l'API Twitter (Astuces)

Comme vous le savez, lorsque vous recevez des tweets avec l'API Twitter, le nombre de fois où vous les recevez sur une certaine période de temps et le nombre de tweets que vous recevez à la fois sont limités.

Afin de bien gérer cette limitation, voici quelques conseils pour l'obtenir en continu. Veuillez noter que cet article ne couvre pas le POST. (Bien que ce soit presque la même chose)

Assurez-vous de vérifier les informations principales pour les valeurs numériques. L'exemple de code est publié sur Gist. (J'ai entendu dire qu'il existe un package Twitter, mais je ne l'utilise pas)

Limites de taux de l'API Twitter (limites du nombre d'acquisitions au cours d'une certaine période)

Les limites sont les suivantes.

GET endpoints

The standard API rate limits described in this table refer to GET (read) endpoints. Note that endpoints not listed in the chart default to 15 requests per allotted user. All request windows are 15 minutes in length.  These rate limits apply to the standard API endpoints only, does not apply to premium APIs.

Extrait de Limites de taux - Développeurs Twitter

Il y a une limite au nombre d'acquisitions par unités de 15 minutes. Selon cette page, par exemple, les restrictions pour «« recherche / tweets »» sont les suivantes (oh, Standard).

Endpoint Resource family Requests / window (user auth) Requests / window (app auth)
GET search/tweets search 180 450
Extrait de [Limites de taux - Développeurs Twitter](https://developer.twitter.com/en/docs/basics/rate-limits)

Obtenir le statut de restriction

Vous pouvez obtenir l'état de la limite actuelle en tant que point de terminaison à l'adresse https://api.twitter.com/1.1/application/rate_limit_status.json. Le paramètre «ressources» est facultatif et spécifie «Famille de ressources».

Exemple d'information obtenue

user auth (OAuth v1.1)


{
  "rate_limit_context": {
    "access_token": "*******************"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 180,
        "remaining": 180,
        "reset": 1591016735
      }
    }
  }
}

app auth (OAuth v2.0)


{
  "rate_limit_context": {
    "application": "dummykey"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 450,
        "remaining": 450,
        "reset": 1591016736
      }
    }
  }
}

Chacun des nombres «reset» correspond à l'heure de l'époque, qui indique l'heure à réinitialiser.

Dans l'exemple ci-dessus, ʻuser auth (OAuth v1.1) with epoch time 1591016735=2020-06-01 22: 05: 35, ʻapp auth (OAuth v2.0) with 1591016736 = Indique qu'il sera réinitialisé à 2020-06-01 22: 05: 36.

Si vous ne respectez pas la limite, le nombre de "restant" sera "0".

Organiser les éléments

Les éléments des informations acquises sont les suivants. (Exemple de famille de recherche)

Category Family Endpoint Key Value
rate_limit_context access_token (user auth (v1.1)) Contenu du jeton d'accès
application (app auth (v2.0)) dummykey (Semble être fixe)
resources
search
/search/tweets
limit Nombre maximum de fois dans le délai
remaining Nombre restant de fois accessibles dans le délai imparti
reset Heure à laquelle la limite de temps est réinitialisée(epoch time)

Cas où la mauvaise famille de ressources est spécifiée

L'exemple ci-dessous est celui où vous spécifiez par erreur «utilisateur» pour la famille de ressources. (En fait, vous devez spécifier «utilisateurs» (avec s))

user auth


{
  "rate_limit_context": {
    "access_token": "*******************"
  }
}

app auth


{
  "rate_limit_context": {
    "application": "dummykey"
  }
}

Les deux renvoient " rate_limit_context "mais n'ont pas de" ressources".

Réponse en cas d'erreur de limite de fréquence (résultat d'acquisition)

Si une erreur de limite de débit se produit, 429 est renvoyé dans res.status_code (code d'état HTTP). (420 peut être retourné [^ 1].)

[^ 1]: Généralement, 429 " Too Many Requests: Renvoyé lorsqu'une demande ne peut pas être servie en raison de l'épuisement de la limite de débit de l'application pour la ressource . "est renvoyé, mais très rarement 420" ʻAméliorer votre Calme: renvoyé lorsqu'une application est soumise à une limitation de débit pour avoir effectué trop de demandes. "" Peut être retourné. Ce dernier peut se produire si vous faites accidentellement plusieurs demandes en même temps (non vérifiées). → Il y avait une explication dans "Connexion à un point de terminaison de streaming - Développeurs Twitter", donc je l'ai ajoutée au texte. (03/06/2020).

Code Text Description
420 Enhance Your Calm Returned when an app is being rate limited for making too many requests.
429 Too Many Requests Returned when a request cannot be served due to the app's rate limit having been exhausted for the resource. See Rate Limiting.
Extrait de [Codes de réponse - Développeurs Twitter](https://developer.twitter.com/en/docs/basics/response-codes)

[Mis à jour le 06/03/2020] Il y avait une explication détaillée de 420 ci-dessous.

420 Rate Limited

The client has connected too frequently. For example, an endpoint returns this status if:

  • A client makes too many login attempts in a short period of time.
  • Too many copies of an application attempt to authenticate with the same credentials.
Extrait de [Connexion à un point de terminaison de streaming - Développeurs Twitter](https://developer.twitter.com/en/docs/tweets/filter-realtime/guides/connecting)

88 est entré dans le code d'erreur JSON.

{
  "errors": [
    {
      "code": 88,
      "message": "Rate limit exceeded"
    }
  ]
}
Code Text Description
88 Rate limit exceeded Corresponds with HTTP 429. The request limit for this resource has been reached for the current rate limit window.
Extrait de [Codes de réponse - Développeurs Twitter](https://developer.twitter.com/en/docs/basics/response-codes)

Pour les exceptions telles que les «demandes», veuillez consulter chaque site.

Flux de processus

Le flux de traitement considère-t-il la limite de débit comme suit?

while True:
    try:
        res =Demande à l'API get/post
        res.raise_for_status()
    except requests.exceptions.HTTPError:
        #429 lorsque la limite de débit est atteinte/420 est retourné
        if res.status_code in (420, 429):
Obtenez des informations sur la limite de taux ← Ici
Attendez tranquillement jusqu'à l'heure de réinitialisation
            continue
        420/Traitement des exceptions autre que 429
    except OtherException:
Gestion des exceptions

Traitement lorsqu'il peut être acquis avec succès
casser ou retourner ou céder etc.

Ce qui suit est une mesure concrète pour la partie «Obtenir des informations sur la limite de débit».

Exemple d'obtention d'informations sur les restrictions

Classe GetTweetStatus

En tant qu'échantillon d'acquisition d'informations, il n'y a pas beaucoup de mérite à l'implémenter dans une classe, mais compte tenu de l'incorporer réellement dans un programme, je pense qu'il serait préférable de l'écrire sous une forme facile à modulariser, alors j'en ai fait une classe appelée GetTweetStatus. Il y a. (Il y a aussi une volonté d'éviter autant que possible l'accès de l'extérieur comme apikey et Bearer ...)

class GetTweetStatus


    def __init__(self, apikey, apisec, access_token="", access_secret=""):
        self._apikey = apikey
        self._apisec = apisec
        self._access_token = access_token
        self._access_secret = access_secret
        self._access_token_mask = re.compile(r'(?P<access_token>"access_token":)\s*".*"')

La dernière ligne, re.compile (), sert à masquer l'affichage du ʻaccess_token` reçu.

Partie acquisition

user auth (OAuth v1.1)

GetTweetStatus.get_limit_status_v1(&nbsp;)


    def get_limit_status_v1(self, resource_family="search"):
        """OAuth v1.Obtenir le statut en utilisant 1"""

        #Utilisez OAuth1Session car OAuth est compliqué
        oauth1 = OAuth1Session(self._apikey, self._apisec, self._access_token, self._access_secret)

        params = {
            'resources': resource_family  # help, users, search, statuses etc.
        }

        try:
            res = oauth1.get(STATUS_ENDPOINT, params=params, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise requests.ConnectionError("Cannot get Limit Status")
        except Exception:
            raise Exception("Cannot get Limit Status")
        return res.json()

app auth (OAuth v2.0)

GetTweetStatus.get_limit_status_v2(&nbsp;)


    def get_limit_status_v2(self, resource_family="search"):
        """OAuth v2.0 (Bearer)Obtenir le statut en utilisant"""
        bearer = self._get_bearer() #Obtenir le porteur

        headers = {
            'Authorization':'Bearer {}'.format(bearer),
            'User-Agent': USER_AGENT
        }
        params = {
            'resources': resource_family  # help, users, search, statuses etc.
        }

        try:
            res = requests.get(STATUS_ENDPOINT, headers=headers, params=params, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise requests.ConnectionError("Cannot get Limit Status")
        except Exception:
            raise Exception("Cannot get Limit Status")
        return res.json()
  1. Obtenez le jeton porteur au début et
  2. Définissez ceci dans l'en-tête.
  3. Définir la famille de ressources comme paramètre
  4. Envoi à STATUS_ENDPOINT.

Génération au porteur

bearer = self._get_bearer () C'est la partie _get_bearer () appelée par #Get Bearer.

GetTweetStatus._get_bearer(&nbsp;),&nbsp;_get_credential(&nbsp;)


    def _get_bearer(self):
        """Obtenir le porteur"""
        cred = self._get_credential()
        headers = {
            'Authorization': 'Basic ' + cred,
            'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
            'User-Agent': USER_AGENT
            }
        data = {
            'grant_type': 'client_credentials',
            }

        try:
            res = requests.post(TOKEN_ENDPOINT, data=data, headers=headers, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise Exception("Cannot get Bearer")
        except requests.exceptions.HTTPError:
            if res.status_code == 403:
                raise requests.exceptions.HTTPError("Auth Error")
            raise requests.exceptions.HTTPError("Other Exception")
        except Exception:
            raise Exception("Cannot get Bearer")
        rjson = res.json()
        return rjson['access_token']

    def _get_credential(self):
        """Générer des informations d'identification"""
        pair = self._apikey + ':' + self._apisec
        bcred = b64encode(pair.encode('utf-8'))
        return bcred.decode()
  1. Combinez APIKEY et APISEC pour l'encodage Base64
  2. Définir dans l'en-tête
  3. Définissez les données utiles sur grant_type =" client_credentials "
  4. Demande (POST) à Endpoint.
  5. Le jeton du porteur est défini dans "ʻaccess_token`" du JSON renvoyé, alors récupérez-le.

Pièce d'affichage

Il est implémenté comme une méthode. Est-ce un endroit pour implémenter quelque chose qui renvoie "reset" en utilisation réelle?

GetTweetStatus.disp_limit_status(&nbsp;)


    def disp_limit_status(self, version=2, resource_family="search"):
        """Afficher la limite de débit par version"""
        if version == 2:
            resj = self.get_limit_status_v2(resource_family=resource_family)
        elif version == 1:
            resj = self.get_limit_status_v1(resource_family=resource_family)
        else:
            raise Exception("Version error: {version}")

        #Afficher JSON
        print(self._access_token_mask.sub(r'\g<access_token> "*******************"',
                                          json.dumps(resj, indent=2, ensure_ascii=False)))
        #Affichage démonté(remain/Exemple de réinitialisation)
        print("resources:")
        if 'resources' in resj:
            resources = resj['resources']
            for family in resources:
                print(f"  family: {family}")
                endpoints = resources[family]
                for endpoint in endpoints:
                    items = endpoints[endpoint]
                    print(f"    endpoint: {endpoint}")
                    limit = items['limit']
                    remaining = items['remaining']
                    reset = items['reset']
                    e2d = epoch2datetime(reset)
                    duration = get_delta(reset)
                    print(f"      limit: {limit}")
                    print(f"      remaining: {remaining}")
                    print(f"      reset: {reset}")         #← En fait un formulaire qui renvoie ceci
                    print(f"      reset(epoch2datetime): {e2d}")
                    print(f"      duration: {duration} sec")
        else:
            print("  Not Available")

Utilitaires et en-têtes liés au temps

Il s'agit de l'utilitaire de manipulation de l'heure et du début du fichier.

getTwitterStatus.py


#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Exemple d'acquisition d'informations sur la limite de débit Twitter"""

import os
import sys
import json
from base64 import b64encode
import datetime
import time
import re
import argparse

#!pip install requests
import requests
#!pip install requests_oauthlib
from requests_oauthlib import OAuth1Session


USER_AGENT = "Get Twitter Staus Application/1.0"
TOKEN_ENDPOINT = 'https://api.twitter.com/oauth2/token'
STATUS_ENDPOINT = 'https://api.twitter.com/1.1/application/rate_limit_status.json'


def epoch2datetime(epoch):
    """Heure de l'époque(Heure UNIX)Datetime(localtime)Convertir en"""
    return datetime.datetime(*(time.localtime(epoch)[:6]))


def datetime2epoch(d_utc):
    """datetime (UTC)Heure de l'époque(Heure UNIX)Convertir en"""
    #Convertir UTC en heure locale
    date_localtime = \
        d_utc.replace(tzinfo=datetime.tzinfo.tz.tzutc()).astimezone(datetime.tzinfo.tz.tzlocal())
    return int(time.mktime(date_localtime.timetuple()))


def get_delta(target_epoch_time):
    """target_epoch_Renvoie la différence entre l'heure et l'heure actuelle"""
    return target_epoch_time - int(round(time.time(), 0))

partie principale

Comme c'est un gros problème, j'ai essayé de rendre possible la spécification de la version OAuth et de la famille de ressources avec l'argument de la ligne de commande.

main(&nbsp;)


def main():
    """main()"""
    # API_KEY, API_Confirmation des variables d'environnement telles que SEC
    apikey = os.getenv('API_KEY', default="")
    apisec = os.getenv('API_SEC', default="")
    access_token = os.getenv('ACCESS_TOKEN', default="")
    access_secret = os.getenv('ACCESS_SECRET', default="")

    if apikey == "" or apisec == "":    #Si la variable d'environnement ne peut pas être obtenue
        print("API de variable d'environnement_CLÉ et API_Veuillez régler SEC.", file=sys.stderr)
        print("OAuth v1.Variable d'environnement ACCESS lors de l'utilisation de 1_TOKEN et ACCÈS_Définissez également SECRET.",
              file=sys.stderr)
        sys.exit(255)

    #Réglage de l'argument
    parser = argparse.ArgumentParser()
    parser.add_argument('-a', '--oauthversion', type=int, default=0,
                        metavar='N', choices=(0, 1, 2),
                        help=u'Spécification de la version OAuth[1|2]')
    parser.add_argument('-f', '--family', type=str, default='search',
                        metavar='Family',
                        help=u'Spécification de la famille d'API. Séparé par des virgules pour plusieurs')

    args = parser.parse_args()
    oauthversion = args.oauthversion
    family = args.family

    #Objet GetTweetStatus
    gts = GetTweetStatus(apikey, apisec, access_token=access_token, access_secret=access_secret)

    # User Auth (OAuth v1.1)Acquisition et affichage de la limite de débit par
    if (oauthversion in (0, 1)) and (access_token != "" and access_secret != ""):
        print("<<user auth (OAuth v1)>>")
        gts.disp_limit_status(version=1, resource_family=family)

    # App Auth (OAuth v2.0)Acquisition et affichage de la limite de débit par
    if oauthversion in (0, 2):
        print("<<app auth (OAuth v2)>>")
        gts.disp_limit_status(version=2, resource_family=family)

if __name__ == "__main__":
    main()

Tout le code getTwitterStatus.py

[^ 2]: J'ai essayé d'utiliser Gist pour la première fois. Je m'inquiète de savoir si l'utilisation est correcte.

Résultat d'exécution

$ python3 getTwitterStatus.py
<<user auth (OAuth v1)>>
{
  "rate_limit_context": {
    "access_token": "*******************"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 180,
        "remaining": 180,
        "reset": 1591016735
      }
    }
  }
}
resources:
  family: search
    endpoint: /search/tweets
      limit: 180
      remaining: 180
      reset: 1591016735
      reset(epoch2datetime): 2020-06-01 22:05:35
      duration: 899 sec
<<app auth (OAuth v2)>>
{
  "rate_limit_context": {
    "application": "dummykey"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 450,
        "remaining": 450,
        "reset": 1591016736
      }
    }
  }
}
resources:
  family: search
    endpoint: /search/tweets
      limit: 450
      remaining: 450
      reset: 1591016736
      reset(epoch2datetime): 2020-06-01 22:05:36
      duration: 900 sec
$

Limitation du nombre de tweets acquis

Il y a une limite au nombre de fois où vous pouvez l'obtenir en même temps, quelle que soit la limite de temps.

200 ($ count \ leq200 ) pour `statuses / user_timeline`, 100 ( count \ leq100 $) pour search / tweets. Il existe diverses autres restrictions, mais dans le cas de search / tweets, l'élément next_results sera inclus dans search_metadata afin qu'il puisse être récupéré en continu.

Pour recherche / tweets

{
  "statuses": [
  ...
  ],
  "search_metadata": {
    "completed_in": 0.047,
    "max_id": 1125490788736032770,
    "max_id_str": "1125490788736032770",
    "next_results": "?max_id=1124690280777699327&q=from%3Atwitterdev&count=2&include_entities=1&result_type=mixed",
    "query": "from%3Atwitterdev",
    "refresh_url": "?since_id=1125490788736032770&q=from%3Atwitterdev&result_type=mixed&include_entities=1",
    "count": 2,
    "since_id": 0,
    "since_id_str": "0"
  }
}

Il y a next_results dans search_metadata, donc si vous demandez ceci comme nouveau paramètre, vous pouvez également obtenir le reste des résultats de la recherche (dans les unités spécifiées dans count).

Tant que vous n'atteignez pas la limite de temps, vous pouvez vous y référer et répéter pour obtenir les résultats en continu. Autrement dit, vous pouvez obtenir $ count $ (jusqu'à 100) $ × limite $ (180 pour l'authentification de l'utilisateur) $ = 18 000 Tweet $ dans la limite de taux.

Dans le cas de l'exemple ci-dessus, $ count = 2 $, donc si vous continuez tel quel, vous pouvez obtenir $ count (2) tweets / time x limit (180) times / 15 minutes = 360 tweets / 15 minutes $, et vous serez limité. Reach (bien sûr, si vous demandez un cancer).

Lorsque tous les résultats de la recherche ont été obtenus, next_results disparaît de search_metadata.

De plus, parfois, si vous le réacquérir, next_results peut être restauré, vous voudrez peut-être attendre un moment et réessayer.

S'il n'y a pas de métadonnées

Dans le cas de statuses / user_timeline etc., * _metadata n'est pas inclus, alors faites bon usage de la spécification de max_id et générez quelque chose d'équivalent à next_results de recherche par vous-même. est nécessaire. (En fait, je ne l'ai pas utilisé pour autre chose que la recherche, donc je ne suis pas sûr, mais je pense que ce n'est pas si loin.)

Dans le cas de la recherche, les 7 derniers jours sont ciblés, mais comme ʻuser_timeline` correspond aux dernières 24 heures, je pense que l'objectif est différent en premier lieu ...

Résumé

Référence (basée sur ce qui a été référencé dans cet article)

Recommended Posts

Acquisition continue par l'API Twitter (Astuces)
Utiliser l'API Twitter (enregistrement de compte API et acquisition de tweet)
API Web EXE par Python
Utiliser l'API Twitter avec Python
Essayez d'utiliser l'API Twitter
Essayez d'utiliser l'API Twitter
Conseils d'acquisition de données de cours de bourse
Conseils relatifs aux API Google Drive (Python)
Update_with_media réussi avec l'API Twitter