Je voulais non seulement résoudre le problème avec paiza et AtCoder, mais aussi commencer à construire le programme. J'ai également été inspiré par cet article (la plupart de ce que vous devez faire pour programmer un service Web a été appris en grattant). Cette fois, je voudrais gratter les entreprises listées dans Rikunabi Direct et les enregistrer dans la DB.
Rikunabi Direct est un service qui affiche et présente chaque semaine plusieurs entreprises que Rikunabi Direct a décidé de faire correspondre au demandeur d'emploi de l'industrie de choix inscrit par le demandeur d'emploi. Les recherches du côté des demandeurs d'emploi ne sont pas possibles. Pour le moment, une liste de sociétés cotées est fournie, mais il est très fastidieux ou imprudent d'examiner un total de plus de 16 000 sociétés à partir de la fin. Par conséquent, j'ai pensé à le rendre consultable en grattant les informations de l'entreprise et en les enregistrant dans la base de données.
Cela prend trop de temps à gratter (2,5 secondes par cas), j'ai donc démarré deux PhantomJS et essayé d'accélérer par un traitement parallèle.
main.py
import utils
import os
import MySQLdb
import time
import selenium
import settings
import numpy as np
from concurrent.futures import ThreadPoolExecutor
from selenium import webdriver
utils.py
import sys
import requests
import csv
import os
import MySQLdb
import settings
import time
import selenium
from selenium import webdriver
main.py
NUMBER_OF_BROWSER = 2
browser = utils.generate_browser()
browser_2 = utils.generate_browser()
browser_list = [browser, browser_2]
browsers = []
for i in range(NUMBER_OF_BROWSERS):
browsers.append([browser_list[i], USER_IDs[i], PASS_WORDs[i]])
utils.py
PHANTOMJS_PATH = '/usr/local/bin/phantomjs'
def generate_browser():
browser = webdriver.PhantomJS(executable_path=PHANTOMJS_PATH)
print('PhantomJS initializing')
browser.implicitly_wait(3)
return browser
Attendre implicitly_wait () pour attendre le démarrage de PhantomJS.
main.py
TIME_TO_WAIT = 5
USER_IDs = [USER_ID, USER_ID_2]
PASS_WORDs = [PASS_WORD, PASS_WORD_2]
browser = utils.generate_browser()
browser_2 = utils.generate_browser()
browser_list = [browser, browser_2]
for browser_param in browsers:
utils.login(browser_param[1], browser_param[2], browser_param[0])
utils.set_wait_time(TIME_TO_WAIT, browser_param[0])
utils.check_current_url(browser_param[0])
utils.py
def generate_browser():
browser = webdriver.PhantomJS(executable_path=PHANTOMJS_PATH)
print('PhantomJS initializing')
browser.implicitly_wait(3)
return browser
def set_wait_time(time, browser):
browser.set_page_load_timeout(time)
def login(user, pass_word, browser):
#Accédez à la page de connexion
url_login = 'https://rikunabi-direct.jp/2020/login/'
browser.get(url_login)
#Jugement d'accès. Quitter si inaccessible
test = requests.get(url_login)
status_code = test.status_code
if status_code == 200:
print('HTTP status code ' + str(status_code) + ':Accédez à la page de connexion')
else:
print('HTTP status code ' + str(test.status_code) + ':Impossible d'accéder à la page de connexion')
sys.exit()
time.sleep(5)
#Entrez l'utilisateur et le mot de passe
#user
element = browser.find_element_by_name('accountId')
element.clear()
element.send_keys(user)
print('J'ai saisi l'utilisateur')
#mot de passe
element = browser.find_element_by_name('password')
element.clear()
element.send_keys(pass_word)
print('J'ai entré le mot de passe')
#Envoyer
submit = browser.find_element_by_xpath("//img[@alt='S'identifier']")
submit.click()
print('J'ai appuyé sur le bouton de connexion')
Utilisez parcourir.get (url) pour accéder à la page url et utilisez browser.find_element_by_name () pour spécifier la valeur de l'attribut name afin de trouver l'élément souhaité. Les éléments que vous souhaitez rechercher sont les suivants
<input type="text" name="accountId" autocomplete="off" value="" ...
Obtenez l'élément avec browser.find_element_by_name ('accountID'). Effacez la zone de texte de l'élément acquis avec clear () et entrez l'ID utilisateur avec send_keys (). Saisissez le mot de passe de la même manière. Le bouton de soumission a été obtenu en le spécifiant avec xpath. xpath provient de la validation Google Chrome, faites un clic droit> Copier> Copier XPath et copiez-le (au début, je ne connaissais pas cette manière pratique et je l'ai spécifié comme un chemin absolu comme / html / body / ... J'ai eu beaucoup de problèmes.) Pour find_element_by_, je me suis référé à ici.
main.py
#Si csv n'existe pas, stockez toutes les URL dans un tableau et exportez-les au format csv
if os.path.exists(URL_PATH) == False:
utils.move_to_company_list(browsers[0][0])
url_arr = utils.get_url(NUMBER_OF_COMPANY, browsers[0][0])
utils.export_csv(url_arr, URL_PATH)
utils.browser_close(browsers[0][0])
else:
#Si csv existe, lisez csv et url_Magasin à arr
url_arr = utils.import_csv(URL_PATH)
utils.py
def move_to_company_list(browser):
#Aller à la page de toutes les sociétés cotées
element = browser.find_element_by_link_text('Toutes les sociétés cotées')
element.click()
#Il s'ouvrira dans un autre onglet, alors passez au deuxième onglet
browser.switch_to_window(browser.window_handles[1])
#Obtenez les URL de toutes les sociétés cotées
def get_url(number_of_company, browser):
url_arr = []
for i in range(2, number_of_company):
url_xpath = '/html/body/div/div/table/tbody/tr[{0}]/td/ul/li/a'.format(i)
element = browser.find_element_by_xpath(url_xpath)
url = element.get_attribute('href')
url_arr.append(url)
print(str(i))
print(url)
return url_arr
#Exporter le tableau au format CSV
def export_csv(arr, csv_path):
with open(csv_path, 'w') as f:
writer = csv.writer(f, lineterminator='\n')
writer.writerow(arr)
#Fermer l'onglet actuel et revenir au premier onglet
def browser_close(browser):
browser.close()
browser.switch_to_window(browser.window_handles[0])
def import_csv(csv_path):
if os.path.exists(csv_path) == True:
with open(csv_path, 'r') as f:
data = list(csv.reader(f))#Tableau bidimensionnel.Le 0e élément est un tableau d'URL.
return data[0]
else:
print('csv n'existe pas')
sys.exit()
Il faut du temps pour obtenir l'URL à chaque fois, donc une fois que vous obtenez l'URL, écrivez-la dans csv et enregistrez-la.
main.py
url_arrs = list(np.array_split(url_arr, NUMBER_OF_BROWSERS))
for i in range(NUMBER_OF_BROWSERS):
print('length of array{0} : '.format(i) + str(len(url_arrs[i])))
main.py
connector = MySQLdb.connect(
unix_socket = DB_UNIX_SOCKET,
host=DB_HOST, user=DB_USER, passwd=DB_PASS_WORD, db=DB_NAME
)
corsor = connector.cursor()
main.py
#Effectuer un traitement de scraping dans chaque navigateur (traitement parallèle))
with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
for i in range(NUMBER_OF_BROWSERS):
executor.submit(utils.scraping_process, browsers[i][0], url_arrs[i], corsor, connector)
utils.py
def open_new_page(url, browser):
try:
browser.execute_script('window.open()')
browser.switch_to_window(browser.window_handles[1])
browser.get(url)
except selenium.common.exceptions.TimeoutException:
browser_close(browser)
print('connection timeout')
print('retrying ...')
open_new_page(url, browser)
def content_scraping(corsor, connector, browser):
#Trouver une cible de scraping
name_element = browser.find_element_by_class_name('companyDetail-companyName')
position_element = browser.find_element_by_xpath('//div[@class="companyDetail-sectionBody"]/p[1]')
job_description_element = browser.find_element_by_xpath('//div[@class="companyDetail-sectionBody"]/p[2]')
company_name = name_element.text
position = position_element.text
job_description = job_description_element.text
url = browser.current_url
casual_flag = is_exist_casual(browser)
#----------Procédure d'enregistrement DB ci-dessous----------#
#INSERT
corsor.execute('INSERT INTO company_data_2 SET name="{0}", url="{1}", position="{2}", description="{3}", is_casual="{4}"'.format(company_name, url, position, job_description, casual_flag))
connector.commit()
def scraping_process(browser, url_arr, corsor, connector):
count = 0
for url in url_arr:
open_new_page(url, browser)
print('{0} scraping start'.format(count))
check_current_url(browser)
try:
content_scraping(corsor, connector, browser)
except selenium.common.exceptions.NoSuchElementException:
print('Entreprises qui ne sont actuellement pas répertoriées')
except MySQLdb._exceptions.ProgrammingError:
print('SQL programming Error')
browser_close(browser)
print('{0} scraping process end.'.format(count))
count += 1
Le texte décrit dans l'élément récupéré peut être utilisé dans element.text. Pour oprn_new_page (), si une exception attendue (TimeoutExecption dans ce cas) se produit, attendez un peu et implémentez un processus de nouvelle tentative récursive pour se rappeler et essayer de vous connecter. J'ai appris les fonctions récursives dans le processus de résolution de problèmes avec AtCoder et d'autres. Je ne savais pas où l'utiliser dans le traitement proprement dit, mais cette fois, j'ai pu créer moi-même un exemple d'utilisation en créant moi-même un programme.
Il existe cinq types d'informations que vous souhaitez récupérer: le nom de l'entreprise, l'URL, le type de poste, la description du poste et la possibilité de travailler en civil. Obtient l'élément par la valeur des attributs Xpath et name, respectivement.
Certaines des sociétés cotées ont cessé de publier. Dans ce cas, find_element_by essaiera d'obtenir l'élément qui n'existe pas, et à ce moment, NoSuchElementException se produira. Dans ce cas, attrapez ceci. Si job_description etc. contient des guillemets simples ou doubles, MySQLdb renvoie une erreur de programmation. Quelqu'un me dit comment faire quelque chose comme l'instruction de préparation PDO de PHP ~~~~~~~~~~
Scraping_process () qui exécute réellement le processus est le flux d'ouverture de la page cible> spécification de l'élément et obtention de celui-ci> enregistrement DB. La partie la plus longue de ce processus se situe entre le processus d'ouverture de la première page et l'acquisition de l'élément suivant. En effet, l'affichage de la page prend beaucoup de temps après son ouverture. Afin d'améliorer le ralentissement dû à cette partie, un traitement parallèle est effectué avec ThreadPoolExecutor de concurrent.futures. Je pense que browser2 est en état de marche pendant que browser1 attend que la page soit affichée. Cela le rend beaucoup plus rapide que le traitement avec un seul navigateur.
Nous avons réussi à gratter près de 16 000 entreprises. J'ai beaucoup appris en créant moi-même une fonction récursive, en étudiant la structure du HTML pour spécifier xpath et en essayant de paralléliser le dernier processus en le regroupant dans une fonction. C'était.
* 1 Liaisons Selenium Python 4. Rechercher des éléments [Python] Comment utiliser Selenium Résumé de la sélection d'éléments avec Selenium API Selenium (reverse pull) Vérifier l'existence du fichier avec python Prendre et vérifier XPath dans Chrome Exécution de tâches parallèles à l'aide de concurrent.futures en Python J'ai étudié en profondeur le traitement parallèle et le traitement parallèle de Python J'ai exploité MySQL à partir de Python 3 sur Mac
Recommended Posts