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.
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) ])
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
À 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) ])
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()
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é!