[PYTHON] Obtenez tous les tweets en direct du baseball professionnel

introduction

Je rassemblais des tweets en direct sur le baseball dans mes recherches universitaires, alors je les ai résumés. J'écris principalement sur le raclage d'histoires et l'obtention de nombreux tweets en utilisant tweepy.

Tweet en direct sur le baseball professionnel 2019 (NPB)

Nous publierons les résultats de la poursuite de l'acquisition de tweets en direct du baseball professionnel 2019 (NPB) avec des balises de hachage chaque jour. J'ai exécuté le fichier Python avec cron tous les jours et je l'ai obtenu pendant un an.

• Parfois, il n'est pas possible de l'obtenir en raison d'un bogue, ou il est interrompu parce que le jeu a été annulé. Veuillez noter que

Les balises de hachage recherchées sont les suivantes. Il se peut que des balises de hachage actives n'aient pas encore été trouvées. Dans Hanshin, est-ce "#tiger van"?

Vue d'ensemble de l'acquisition de tweet en direct

1. Obtenez la carte de match / l'heure du match du jour cible

Obtenu en grattant des sites Web qui fournissent des nouvelles sportives de dernière heure.

• Destination de grattage
  • SPORTS BULL（https://sportsbull.jp/stats/npb/） --Sports Navi (par Yahoo! JAPAN) (https://baseball.yahoo.co.jp/npb/schedule/)

2. Obtenez des tweets en direct à partir de hashtags en spécifiant l'heure du match

ID et balise de hachage de chaque équipe (requête de recherche)

Objet dictionnaire Hashtag (tag_list) key: team_id défini par moi-même item: Balise de hachage (requête lors de la recherche de tweets)

tag_list = {0: '#kyojin OR #giants', 1: '#dragons',\
            2: '#carp', 3: '#swallows OR #yakultswallows',4: '#hanshin OR #tigers',\
            5: '#baystars', 6: '#lovefighters', 7: '#sbhawks',8: '#rakuteneagles',\
            9: '#seibulions', 10: '#chibalotte', 11: '#Orix_Buffaloes'}

la mise en oeuvre

Publié sur github → ici

Bibliothèque de spécifications (Python)

La bibliothèque utilisée cette fois est la suivante. Veuillez installer selon le cas.

getLiveTweet_NPB.py

from urllib.request import urlopen
import tweepy
from datetime import timedelta
import time
import sqlite3
from contextlib import closing
import datetime
from bs4 import BeautifulSoup
import urllib.request as req

Obtenez une carte de match

SPORTS BULL（https://sportsbull.jp/stats/npb/） Gratter le match joué à la date et à l'heure spécifiées par. Vous pouvez l'obtenir auprès de Sponavi, mais celui-ci a une structure HTML plus simple.

getLiveTweet_NPB.py

def get_gameteamId(gamedate):
    url = 'https://sportsbull.jp/stats/npb/home/index/' + gamedate
    print(url)
    res = req.urlopen(url)
    soup = BeautifulSoup(res, 'html.parser')
    q = soup.select('.game-block a')
    gameId_list = []
    flag_list = [1 for i in range(12)]
    i = 0
    for p in q:
        urls = p.get('href')
        #Traitement en cas d'annulation
        p_ = p.select('.st-03')
        for p__ in p_:
            if 'Annuler' in str(p__.text):
                print('Annuler')
                flag_list[i] = 0
                flag_list[i+1] = 0
        if flag_list[i] == 1:
            print(urls[-10:])
            gameId_list.append(urls[-10:])
        i += 2
    print('flag_list: ',flag_list)
    q = soup.select('.game-block .play-box01 dt')
    teamId_list = []
    teamId_dict = {'Géant': 0, 'Chunichi': 1, 'Hiroshima': 2, 'Yakult': 3, 'Hanshin': 4, 'ＤｅＮＡ': 5,
                  'Jambon japonais': 6, 'Softbank': 7, 'Rakuten': 8, 'Seibu': 9, 'Lotte': 10, 'ORIX': 11}
    i = 0
    for p in q:
        if flag_list[i] == 1:
            teamId_list.append(teamId_dict[p.text])
        i += 1
    return gameId_list, teamId_list


#Date
def get_date(days_ago):
	date = datetime.date.today()
	date -= datetime.timedelta(days=days_ago)
	date_str = str(date)
	date_str = date_str[:4]+date_str[5:7]+date_str[8:10]
	return date_str


#Exemple--------------------------
n = 1
game_date = get_date(n) #Automatique(Obtenir des données il y a n jours)
game_date = '20200401' #Entrée manuelle
print(game_date,'Obtenez des données pour,')
# -----------------------------

#Liste des identifiants de jeu et des identifiants d'équipe
gameteamId_list = get_gameteamId(game_date)
gameId_list = gameteamId_list[0]
teamId_list = gameteamId_list[1]
print('gameId_list:',gameId_list)
print('teamId_list:',teamId_list)

Exemple de résultat d'exécution

Obtenir des données pour 20200401
https://sportsbull.jp/stats/npb/home/index/20200401
flag_list: [1,1,1,1,0,0,0,0,0,0,0,0]
gameId_list: [2020040101,2020040102]
teamId_list: [0,1,2,3]

dans ce cas, Géant (domicile) vs Chine (extérieur) à gameId = 2019040101 Hiroshima (domicile) vs Yakult (extérieur) à gameId = 2019040102 Le match a eu lieu

Obtenez les heures de début et de fin du match

Yahoo! Sponavi (https://baseball.yahoo.co.jp/npb/schedule/) Chaque page de match https://baseball.yahoo.co.jp/npb/game/[game_id]/top Puisque l'heure de début et l'heure du match peuvent être extraites, additionnez-les pour obtenir l'heure de début et l'heure de fin.

getLiveTweet_NPB.py

#Obtenez les heures de début et de fin des matchs en grattant
def gametime(game_id):
    url = 'https://baseball.yahoo.co.jp/npb/game/' + game_id + '/top'
    res = req.urlopen(url)
    soup = BeautifulSoup(res, 'html.parser')
    time = []

    #Heure de début
    css_select = '#gm_match .gamecard .column-center .stadium'
    q = soup.select(css_select)
    time.append(q[0].text[-6:-4])
    time.append(q[0].text[-3:-1])

    #heure de fin
    minutes = []
    while True:
        try:
            css_select = '#yjSNLiveDetaildata td'
            q = soup.select(css_select)
            minutes = q[1].text.split('temps')
            minutes[1] = minutes[1][:-1]
            break
        except:
            continue
    time = time + minutes
    return time

↑ Sortie de cette fonction

#Heure de début 18:00, lorsque la durée du match est de 3 heures et 15 minutes
[18,0,21,15]

Recherche par API Twitter

Obtenez tous les tweets à temps en utilisant la recherche sur l'API Twitter. Puisque 100 tweets peuvent être acquis avec une seule demande, répétez-la. Si vous êtes bloqué dans la limite de l'API, faites une pause et attendez 15 minutes.

La cible est les tweets du début de la partie jusqu'à 5 minutes après la fin de la partie.

getLiveTweet_NPB.py

# TwitterAPI
APIK = 'consumer_key'
APIS = 'consumer_secret'
AT = 'access_token'
AS = 'access_token_secret'
auth = tweepy.OAuthHandler(APIK, APIS)
auth.set_access_token(AT, AS)
api = tweepy.API(auth)

#Recherche API Twitter
def search_livetweet(team_num, api, game_id, query):
    print(query)    #Obtenez des derniers tweets
    print('Page de recherche: 1')
    try:
        tweet_data = api.search(q=query, count=1)
    except tweepy.TweepError as e:
        print('Erreur: attendez 15 minutes')
        time.sleep(60 * 15)

    tweet_data = api.search(q=query, count=100)
    table_name = 'team' + str(team_num)
    #Cette fonction permet d'enregistrer dans la base de données
    saveDB_tweet(table_name, 0, tweet_data, game_id)
    print('************************************************\n')
    next_max_id = tweet_data[-1].id

    page = 1
    while True:
        page += 1
        print('Page de recherche:' + str(page))
        try:
            tweet_data = api.search(q=query, count=100, max_id=next_max_id - 1)
            if len(tweet_data) == 0:
                break
            else:
                next_max_id = tweet_data[-1].id
                #Cette fonction permet d'enregistrer dans la base de données
                saveDB_tweet(table_name, page - 1, tweet_data, game_id)
        except tweepy.TweepError as e:
            print('Erreur: attendez 15 minutes')
            print(datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S"))
            print(e.reason)
            time.sleep(60 * 15)
            continue
        print('*'*40 + '\n')


#Spécifiez l'heure → Créer une requête → Fonction de recherche de Tweet (recherche)_livetweet())
def get_livetweet(team_id, game_id):
    date = game_id[:4] + '-' + game_id[4:6] + '-' + game_id[6:8]
    time = gametime(game_id)
    sh, sm = time[0], time[1]
    eh = int(time[0]) + int(time[2])
    em = int(time[1]) + int(time[3]) + 5  #5 minutes après la fin
    if em >= 60:
        em -= 60
        eh += 1
    eh = '{0:02d}'.format(eh)
    em = '{0:02d}'.format(em)
    
    print(date, sh, sm, eh, em)
    tag_list = {0: '#kyojin OR #giants', 1: '#dragons',\
            2: '#carp', 3: '#swallows OR #yakultswallows',4: '#hanshin OR #tigers',\
            5: '#baystars', 6: '#lovefighters', 7: '#sbhawks',8: '#rakuteneagles',\
            9: '#seibulions', 10: '#chibalotte', 11: '#Orix_Buffaloes'}
    tag = tag_list[team_num]
    query = tag + ' exclude:retweets exclude:replies\
            since:' + date + '_' + sh + ':' + sm + ':00_JST \
            until:' + date + '_' + eh + ':' + em + ':59_JST lang:ja'
    search_livetweet(team_id, api, game_id, query)

Exécutez maintenant la fonction ci-dessus

Obtenez des tweets de deux équipes pour chaque match à partir des listes gameId_list et teamId_list créées ci-dessus.

getLiveTweet_NPB.py

for i in range(len(gameId_list)):
    game_id = gameId_list[i]

    #away
    team_id = teamId_list[2*i+1]
    get_livetweet(team_id, game_id)
    print('='*60 + '\n')

    #home
    team_id = teamId_list[2*i]
    get_livetweet(team_id, game_id)
    print('='*60 + '\n')

en conclusion

Si le match est interrompu par la pluie, il ne sera peut-être pas possible d'obtenir tous les tweets. Il semble qu'une amélioration à ce sujet soit nécessaire.

La partie pour obtenir des tweets en spécifiant l'heure peut être utilisée dans n'importe quel domaine, j'espère donc que cela vous sera utile.