Créez un service Twitter BOT avec GAE / P + Tweepy + RIOT API! (Partie 2)

Synopsis

Poursuivant la Partie 1, dans la deuxième partie, je voudrais en fait faire référence à diverses API et faire fonctionner le service BOT. Le code source est exploité sur GitBucket, et le produit fini est exploité sur ici.

Traitement séquentiel avec Task Queue

Puisqu'il existe des restrictions sur l'utilisation de l'API pour surveiller le record de bataille de chaque utilisateur et le publier sur Twitter en cas de mise à jour, il est nécessaire de contrôler le processus à exécuter à des intervalles de quelques secondes. Puisque Cron, qui sera décrit plus loin, ne peut être appelé qu'à un intervalle minimum de 1 minute, le traitement de chaque utilisateur utilise un mécanisme appelé Task Queue fourni par GAE. En tant qu'image de la file d'attente des tâches, les tâches sont jetées dans un conteneur appelé Bucket et les tâches sont traitées en fonction des paramètres dans l'ordre dans lequel elles ont été placées en premier. Si vous augmentez la taille du bucket, vous pouvez traiter en parallèle. Le fichier de configuration est `` queue.yaml ''.

queue.yaml


queue:
- name: tweet #Tout va bien
  rate: 0.8/s #0 par seconde.Traiter à un rythme de 8 fois
  bucket_size: 1 #Numéro à traiter en parallèle
  retry_parameters:
    task_retry_limit: 0 #Nombre de tentatives en cas d'échec du traitement

Maintenant, utilisez add () `` `pour lancer des tâches de Python dans le Bucket défini dans`` `queue.yaml. Spécifiez name défini dans `` `queue.yaml pour`` queue_name de manière appropriée. Spécifiez l'adresse du processus à appeler dans ```url correctement`. Vous pouvez également transmettre des paramètres.

launcher.py


from google.appengine.api.taskqueue import add

import webapp2

class modelTask(db.Model): #Tâches à mettre en file d'attente
    resion = db.StringProperty()
    summoner_name = db.StringProperty()
    summoner_id = db.IntegerProperty()
    latest_game = db.IntegerProperty()
    access_key = db.StringProperty()
    access_secret = db.StringProperty()
    date_success = db.DateTimeProperty(auto_now_add=True)
    date = db.DateTimeProperty(auto_now_add=True)

class mainHandler(webapp2.RequestHandler):
    def get(self):
        qs = modelTask.all().order('-date_success')
        for q in qs: #Ajouter toutes les tâches à la file d'attente
            add(queue_name='tweet', url='/tweet', params={'qid': q.key().id()})

app = webapp2.WSGIApplication([ ('/launcher', mainHandler) ])

Ensuite, implémentez le processus spécifié dans ```url de manière appropriée` qui est appelé lorsque le tour arrive dans la file d'attente des tâches. Appelé par la méthode POST à partir de la file d'attente des tâches.

tweet.py


#! -*- coding: utf-8 -*-
from google.appengine.ext import db
from google.appengine.api.urlfetch import fetch
from django.utils.simplejson import loads

import webapp2, tweepy
from datetime import datetime

from laucher import modelTask

CONSUMER_KEY = '********************'
CONSUMER_SECRET = '**************************************'
RIOT_KEY = '***********************************'

class mainHandler(webapp2.RequestHandler):
    def post(self):
        getGame(long(self.request.get('qid')))

def getGame(qid):
    q = modelTask().get_by_id(qid, parent=None)
    #Appelez l'API RIOT
    result = fetch('https://prod.api.pvp.net/api/lol/'+q.resion+'/v1.3/game/by-summoner/'+str(q.summoner_id)+'/recent?api_key='+RIOT_KEY)
    if result.status_code == 200:
        #Définir diverses valeurs obtenues à partir de l'API
        j = loads(result.content)['games'][0]
        if j['stats']['win'] == True:
            win = 'la victoire'
        else:
            win = 'défaite'
        try:
            kill = str(j['stats']['championsKilled'])
        except:
            kill = '0'
        try:
            death = str(j['stats']['numDeaths'])
        except:
            death = '0'
        try:
            assist = str(j['stats']['assists'])
        except:
            assist = '0'

        game_type = j['subType']

        #Publiez sur Twitter s'il y a une mise à jour dans le temps de la bataille finale
        if j['createDate'] > q.latest_game:
            q.latest_game = j['createDate']
            q.put()
            if tweet(q.summoner_name+'M. dernier'+game_type+'Le record de bataille est'+kill+'tuer'+death+'mort'+assist+'Aider'+win+'est . http://tol.orfx.jp #Tweet_of_Legends', q.access_key, q.access_secret):
                q.date_success = datetime.now()
                q.put()

#Traitement des publications sur Twitter
def tweet(message, access_key, access_secret):
    try:
        auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
        auth.set_access_token(access_key, access_secret)
        api = tweepy.API(auth_handler=auth, api_root='/1.1', secure=True)
        api.update_status(status=message)
        
        return True
    except:
        return False

app = webapp2.WSGIApplication([ ('/tweet', mainHandler) ])

Traitement périodique avec Cron

J'utilise Cron parce que je veux appeler le processus d'ajout de la tâche ci-dessus à la file d'attente sur une base régulière. Le fichier de configuration est cron.yaml```. Cette fois, je vais essayer de l'exécuter toutes les 10 minutes de 1 h à 18 h, et toutes les 5 minutes sinon.

cron.yaml


cron:
- description: tweet job
  url: /launcher
  schedule: every 10 minutes from 1:00 to 17:59
  timezone: Asia/Tokyo #N'oubliez pas le fuseau horaire lorsque vous spécifiez l'heure
  
- description: tweet job
  url: /launcher
  schedule: every 5 minutes from 18:00 to 0:59
  timezone: Asia/Tokyo

Distribution de ressources avec Backend

À propos, si vous utilisez ce paramètre Cron (toutes les 5 à 10 minutes), le temps de démarrage de l'instance est calculé par incréments de 15 minutes, de sorte que le temps de démarrage de l'instance de Frontend consommera certainement 24 heures ou plus. En plus de cela, le traitement de la page d'inscription, etc. sera ajouté, donc je suis inquiet pour les 28 heures du cadre libre (à partir de mars 2014). Alors, profitons de Backend. Le backend est à l'origine un mécanisme permettant d'effectuer des travaux en coulisse tels que le traitement par lots et le traitement asynchrone, mais cette fois, il est simplement utilisé pour gagner (9 heures) de temps de démarrage de l'instance. Le fichier de configuration est backend.yaml```.

backend.yaml


backends:
- name: tweet #tout va bien
  class: B1 #Minimiser les ressources de traitement
  options: dynamic #Rendre l'instance non résidente

Si vous voulez que Cron backend le processus, spécifiez cible '' Cette fois, je vais le traiter avec Backend de 18h à 1h.

cron:
- description: tweet job
  url: /launcher
  schedule: every 10 minutes from 1:00 to 17:59
  timezone: Asia/Tokyo
  
- description: tweet job
  url: /launcher
  schedule: every 5 minutes from 18:00 to 0:59
  timezone: Asia/Tokyo
  target: tweet #backend.nom yaml

TaskQueue peut également être traité par Backend en spécifiant add (target =) `` `. Utilisez `` get_backend () '' pour voir si vous utilisez Backend. De plus, lorsque Backend est démarré, `` / _ah / start est appelé de force. Puisqu'il est appelé en premier, vous pouvez décrire le processus au démarrage, mais cette fois nous décrirons le processus vide.

launcher.py


from google.appengine.api.taskqueue import add
from google.appengine.api.backends import get_backend

import webapp2

class modelTask(db.Model):
    resion = db.StringProperty()
    summoner_name = db.StringProperty()
    summoner_id = db.IntegerProperty()
    latest_game = db.IntegerProperty()
    access_key = db.StringProperty()
    access_secret = db.StringProperty()
    date_success = db.DateTimeProperty(auto_now_add=True)
    date = db.DateTimeProperty(auto_now_add=True)

class mainHandler(webapp2.RequestHandler):
    def get(self):
        qs = modelTask.all().order('-date_success')
        target = get_backend()
        if target is None: #S'il est démarré par Backend, TaskQueue sera le même.
            for q in qs:
                add(queue_name='tweet', url='/tweet', params={'qid': q.key().id()})
        else:
            for q in qs:
                add(queue_name='tweet', url='/tweet', params={'qid': q.key().id()}, target='tweet')

class startHandler(webapp2.RequestHandler): #Appel forcé au démarrage du backend
    def get(self):
        return

app = webapp2.WSGIApplication([ ('/launcher', mainHandler), ('/_ah/start', startHandler) ])

Réduisez la consommation de ressources avec Memcache

Memcache peut être utilisé pour réduire considérablement la consommation de ressources pour les processus qui parcourent fréquemment le magasin de données mais le mettent rarement à jour. Dans RIOT API game-v1.3, Champion est renvoyé sous forme d'ID, vous devez donc convertir l'ID en un nom basé sur les informations de champion-v1.1. Comme vous ne pouvez pas accéder fréquemment à l'API, les informations acquises par champion-v1.1 sont stockées une fois dans le magasin de données. Copiez ensuite les informations stockées dans Memcache. Utilisez memcache.add (clé, valeur) '' '' pour ajouter des données à Memcache.

champion.py


from google.appengine.api import memcache
from google.appengine.ext import db
from google.appengine.api.urlfetch import fetch
from django.utils.simplejson import loads

import webapp2

RIOT_KEY = '***********************************'

class modelChampion(db.Model): #Modèle de stockage d'informations Champion
    name = db.StringProperty()
    date = db.DateTimeProperty(auto_now_add=True)

class mainHandler(webapp2.RequestHandler):
    def get(self):
        #Acquisition d'informations de champion
        result = fetch('https://prod.api.pvp.net/api/lol/na/v1.1/champion?api_key='+RIOT_KEY)
        if result.status_code == 200:
            js = loads(result.content)['champions']
            for j in js:
                #Stockage d'informations Champion
                modelchampion = modelChampion().get_or_insert(str(j['id']))
                modelchampion.name = j['name']
                modelchampion.put()

                #Copier les informations du champion dans Memcache
                memcache.add("champion_"+str(j['id']), j['name'], 86399)

app = webapp2.WSGIApplication([ ('/champion', mainHandler) ], debug=True)

Utilisez `` memcache.get (key) '' pour référencer les données ajoutées à Memcache. Étant donné que les données ajoutées à Memcache peuvent être perdues, il est nécessaire de décrire le traitement à ce moment-là. Maintenant, ajoutons le nom du champion au contenu publié dans tweet.py.

tweet.py


from google.appengine.api import memcache
from champion import modelChampion

def getGame(qid):
    q = modelQueue().get_by_id(qid, parent=None)
    result = fetch('https://prod.api.pvp.net/api/lol/'+q.resion+'/v1.3/game/by-summoner/'+str(q.summoner_id)+'/recent?api_key='+RIOT_KEY)
    if result.status_code == 200:
        j = loads(result.content)['games'][0]
        if j['stats']['win'] == True:
            win = 'la victoire'
        else:
            win = 'défaite'
        try:
            kill = str(j['stats']['championsKilled'])
        except:
            kill = '0'
        try:
            death = str(j['stats']['numDeaths'])
        except:
            death = '0'
        try:
            assist = str(j['stats']['assists'])
        except:
            assist = '0'

        #Obtenez des informations sur les champions de Memcache
        champion = memcache.get("champion_"+str(j['championId']))
        if champion is None: #Si vous ne pouvez pas obtenir d'informations sur les champions de Memcache, à partir du magasin de données
            champion = modelChampion.get_by_key_name(str(j['championId'])).name

        game_type = j['subType']
        
        if j['createDate'] > q.latest_game:
            q.latest_game = j['createDate']
            q.put()
            if tweet(q.summoner_name+'M. dernier'+game_type+'Le record de bataille est'+champion+'alors'+kill+'tuer'+death+'mort'+assist+'Aider'+win+'alorsす 。 http://tol.orfx.jp #Tweet_of_Legends', q.access_key, q.access_secret):
                q.date_success = datetime.now()
                q.put()

Résumé

Il s'agit du premier mémorandum d'application créé avec GAE. Une fois que vous avez compris, il est très facile de créer une application à partir de la prochaine fois, c'est donc recommandé!

Recommended Posts

Créez un service Twitter BOT avec GAE / P + Tweepy + RIOT API! (Partie 1)
Créez un service Twitter BOT avec GAE / P + Tweepy + RIOT API! (Partie 2)
Étapes pour créer un bot Twitter avec Python
J'ai fait un Twitter BOT avec GAE (python) (avec une référence)
Créez un robot de réponse automatique en temps réel à l'aide de l'API Twitter Streaming
Créez un Twitter BOT avec le SDK GoogleAppEngine pour Python
J'ai créé un bot Twitter avec Go x Qiita API x Lambda
[LINE Messaging API] Créez un BOT qui se connecte à quelqu'un avec Python
Créez un bot qui stimule les tendances Twitter
Créer un service Web avec Docker + Flask
Faisons un bot Twitter avec Python!
Tweet les prévisions météo avec le bot Partie 2
Créer un bot Twitter Trend avec heroku + Python
Créer un LINE BOT avec Minette pour Python
[Life hack] Bot de support Anna pour femmes avec API Twitter
Créez une carte thermique de tweet avec l'API Google Maps
[LINE Messaging API] Créer un BOT de retour de perroquet avec Python
Créer un bot slack
J'ai écrit un script pour créer rapidement un environnement de développement pour Twitter Bot avec AWS Lambda + Python 2.7
Créez une application d'apprentissage automatique avec ABEJA Platform + LINE Bot
Créez un bot Mastodon avec une fonction pour répondre automatiquement avec Python
Tornado - Créons une API Web qui renvoie facilement JSON avec JSON
Créez une API Web capable de fournir des images avec Django
Créez une API d'intégration sociale pour les applications smartphone avec Django
Créer une API avec Django
Utiliser l'API Twitter avec Python
Créer une page d'accueil avec django
Soutenez-vous avec l'API Twitter
Créer un répertoire avec python
Update_with_media réussi avec l'API Twitter
[AWS Hands-on] Créons un service d'identification des célébrités avec une architecture sans serveur!
Créez un nuage de mots avec uniquement des mots positifs / négatifs sur Twitter
Essayez de créer un article de Qiita avec l'API REST [Préparation environnementale]
Créer une API REST pour faire fonctionner dynamodb avec le Framework Django REST
Comment créer une API de machine learning sans serveur avec AWS Lambda
Les débutants créeront un bot de trading automatique Bitcoin visant beaucoup d'argent! Partie 2 [Transaction avec API]