Блог об аналитике, визуализации данных, data science и BI, заметки с тегом: логирование

Парсим вакансии для аналитиков из Indeed

Thu, 27 May 2021 10:10:46 +0300

В этом материале мы расскажем, как парсить вакансии с сайта Indeed. Indeed — это крупнейший в мире поисковик вакансий. Этим текстом мы начинаем большой проект по анализу и визуализации показателей оплаты труда в области Data Science в разных странах.
Подобный анализ рынка вакансий, но только в России, мы проводили в материале Анализ рынка вакансий аналитики и BI: дашборд в Tableau, когда парсили данные с сайта HeadHunter.

А еще у нас можно почитать материал Парсим данные каталога сайта, используя Beautiful Soup и Selenium

Импорт библиотек
Библиотека fake_useragent имитирует реальный User-Agent, чтобы преодолеть защиту сайта от парсинга. Таким образом мы сможем пройти проверку HTTP заголовка User-Agent.
Модуль urllib.parse разбирает URL-адрес на компоненты и записывает его как кортеж. Он пригодится для перехода на карточки вакансий. BeautifulSoup поможет разобраться в структуре html-страницы и добыть нужную нам информацию.

import requests
from datetime import timedelta, datetime
import urllib.parse
from fake_useragent import UserAgent
from bs4 import BeautifulSoup
import pandas as pd
import time
from lxml.html import fromstring
from clickhouse_driver import Client
from clickhouse_driver import errors
import numpy as np
from funcs import check_title, get_skills_row, parse_salary, get_sheetname, create_table

Создадим таблицу в Clickhouse
Данные, которые мы собираемся собрать, будем хранить в базе Clickhouse.

create_table = '''CREATE TABLE if not exists indeed.vacancies (
    row_idx UInt16,
    query_string String,
    country String,
    title String,
    company String,
    city String,
    job_added Date,
    easy_apply UInt8,
    company_rating Nullable(Float32),
    remote UInt8,
    job_id String,
    job_link String,
    sheet String,
    skills String,
    added_date Date,
    month_salary_from_USD Float64,
    month_salary_to_USD Float64,
    year_salary_from_USD Float64,
    year_salary_to_USD Float64,
)
ENGINE = ReplacingMergeTree
SETTINGS index_granularity = 8192'''

Обход блокировок
Нам нужно обойти защиту Indeed и избежать блокировки по IP. Для этого используем анонимные прокси адреса на сайте free-proxy-list.net. Как собрать свежие прокси, мы писали в нашем предыдущем тексте «Пишем парсер свежих прокси на Python для Selenium». Прокси адреса мы запишем в массив, который понадобится в момент обращения к Indeed, когда запрос будет проверять User-Agent.

Данный метод удаляет IP из списка с прокси в том случае, если ответ от Indeed через него так и не пришел.

def remove_proxy_from_list_and_update_if_required(proxy):
    global _proxies
    _proxies.remove(proxy)
    if len(_proxies) == 0:
        update_proxy_list()

Функция, используя прокси, возвращает нам страницу Indeed, из которой мы впоследствии спарсим данные.

def get_page(updated_url, session):
    proxy = get_proxy()
    proxy_dict = {"http": proxy, "https": proxy}
    logger.info(f'try with proxy: {proxy}')
    try:
        session.proxies = proxy_dict
        return session.get(updated_url, timeout=15)
    except (requests.exceptions.RequestException, requests.exceptions.ProxyError, requests.exceptions.ConnectTimeout,
            requests.exceptions.ReadTimeout, requests.exceptions.SSLError,
            requests.exceptions.ConnectionError, url_ex.MaxRetryError, ConnectionResetError,
            socket.timeout, url_ex.ReadTimeoutError):
        remove_proxy_from_list_and_update_if_required(proxy)
        logger.info(f'try with proxy {proxy}')
        return get_page(updated_url, session)

Методы для парсера
Искомые данные нужно будет искать по тегам и атрибутам верстки с помощью BeautifulSoup. Мы заранее собрали ключевые слова, которые нас будут интересовать в вакансиях, и подготовили с ними отдельный датасет.

В карточках вакансий нет точной даты публикации, указано лишь сколько дней назад она была опубликована. Сохраним точную дату публикации в традиционном формате с помощью timedelta.

def raw_date_to_str(raw_date):
    raw_date = raw_date.lower()
    if '+' in raw_date or "более" in raw_date:
        delta = timedelta(days=32)
        return (datetime.now() - delta).strftime("%Y-%m-%d")
    else:
        parts = raw_date.split()
        for part in parts:
            if part.isdigit():
                delta = timedelta(days=part.isdigit())
                return (datetime.now() - delta).strftime("%Y-%m-%d")
    return ""

Сохраним id вакансии в системе Indeed. Подставляя id в URL страницы, мы сможем получить доступ к полному описанию вакансий.

def get_job_id_from_card(card):
    try:
        return card['id'].split('_')[1]
    except:
        return ""

Данный метод соберет названия вакансий.

def get_title_from_card(card):
    try:
        job_title = card.find('a', {'class': 'jobtitle'}).text
        return job_title.replace('\n', '')
    except:
        return ''

Аналогичным образом напишем методы, которые будут собирать данные о названии компании, времени публикации объявления, местоположении работодателя и рейтинге работодателя на портале.

URL сайта Indeed пишется для разных стран по-разному. Для США это будет просто indeed.com, а локализации для других стран получают префиксом xx.indeed.com. Список с префиксами мы собрали в массив заранее из https://opensource.indeedeng.io/api-documentation/docs/supported-countries/ списка Indeed.

def get_link_from_card(card, card_country):
    try:
        if card_country == 'us':
            return f"https://indeed.com{card.find('a', {'class': 'jobtitle'})['href']}"
        else:
            return f"https://{card_country}.indeed.com{card.find('a', {'class': 'jobtitle'})['href']}"
    except:
        return ""

Спарсим описание вакансии, которое можно найти по тегу ’summary’. Именно там содержатся требования, которые предъявляют к кандидату.

def get_summary_from_card_and_transform_to_skills(card):
    try:
        smr = card.find('div', {'class': 'summary'}).text
        return get_skills_row(smr)
    except:
        return ""
Необходимые hard-skills из описания вакансий будем сверять со списком 'skills'. 
skills = ["python", "tableau", "etl", "power bi", "d3.js", "qlik", "qlikview", "qliksense",
          "redash", "metabase", "numpy", "pandas", "congos", "superset", "matplotlib", "plotly",
          "airflow", "spark", "luigi", "machine learning", "amplitude", "sql", "nosql", "clickhouse",
          'sas', "hadoop", "pytorch", "tensorflow", "bash", "scala", "git", "aws", "docker",
          "linux", "kafka", "nifi", "ozzie", "ssas", "ssis", "redis", 'olap', ' r ', 'bigquery', 'api', 'excel']

Эта функция разобьет ’summary’ на слова пробелом и проверит их на соответствие нашему списку. В датасет будут возвращаться совпадения с нашим списком hard-skills.

def get_skills_row(summary):
    summary = summary.lower()
    row = []
    for sk in skills:
        if sk in summary:
            row.append(sk)
    return ','.join(row)

На выходе мы получим таблицу с примерно 30 тысячами строк.

Полный код проекта можно посмотреть в нашем репозитории на GitHub.

Эффективное логирование в Python

Tue, 30 Mar 2021 13:51:48 +0300

В Python существует встроенный модуль logging, который позволяет журналировать этапы выполнения программы. Логирование полезно когда, например, нужно оставить большой скрипт сбора / обработки данных на длительное время, а в случае возникновения непредвиденных ошибок выяснить, с чем они могут быть связаны. Анализ логов позволяет быстро и эффективно выявлять проблемные места в коде, но для удобного использования модуля следует написать несколько функций по взаимодействию с ним и вынести их в отдельный файл — сегодня мы этим и займёмся.

Пишем логгер

Создадим файл loggers.py. Для начала импортируем модули и задаём пару значений по умолчанию — директорию для файла с логом и наименование конфигурационного файла, содержащего шаблоны логирования. Его мы опишем следом.

import os
import json
import logging
import logging.config

FOLDER_LOG = "log"
LOGGING_CONFIG_FILE = 'loggers.json'

Опишем функцию для создания папки с логом: она принимает наименование для папки, но по умолчанию будет называть её «log». Директорию создаём при помощи модуля os и только в том случае, если такой директории ещё не существует.

def create_log_folder(folder=FOLDER_LOG):
    if not os.path.exists(folder):
        os.mkdir(folder)

Теперь опишем функцию создания нового логгера по заданному шаблону. Функция должна создать директорию для логирования, открыть конфигурационный файл и достать нужный шаблон. Затем по шаблону при помощи модуля logging создаём новый логгер:

def get_logger(name, template='default'):
    create_log_folder()
    with open(LOGGING_CONFIG_FILE, "r") as f:
        dict_config = json.load(f)
        dict_config["loggers"][name] = dict_config["loggers"][template]
    logging.config.dictConfig(dict_config)
    return logging.getLogger(name)

Для удобства опишем ещё одну функцию — получение стандартного лога. Она ничего не принимает и нужна только для инициализации лога с шаблоном default:

def get_default_logger():
    create_log_folder()
    with open(LOGGING_CONFIG_FILE, "r") as f:
        logging.config.dictConfig(json.load(f))

    return logging.getLogger("default")

Описываем конфигурационный файл

Создадим по соседству файл loggers.json — он будет содержать настройки логгера. Внутри указываем такие настройки, как версию логгера, форматы логирования для разных уровней, наименование выходного файла и его максимальный размер:

{
    "version": 1,
    "disable_existing_loggers": false,
    "formatters": {
        "default": {
            "format": "%(asctime)s - %(processName)-10s - %(name)-10s - %(levelname)-8s - %(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "default"
        },
        "rotating_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "default",
            "filename": "log/main.log",
            "maxBytes": 10485760,
            "backupCount": 20
        }
    },
    "loggers": {
        "default": {
            "handlers": ["console", "rotating_file"],
            "level": "DEBUG"
        }
    }
}

Использование логгера

Теперь давайте представим, что вы выгружаете данные по API и складываете их в базу данных на примере нашего материала про транзакции в SQLAlchemy. Рассмотрим заключительную часть кода: добавим строку с инициализацией стандартного логгера и изменим код так, чтобы сначала в лог выводился offset, затем в случае успеха предложение «Successfully inserted data», а в случае ошибки выводилась сама ошибка и предложение: «Error: tried to insert data but got an error».

logger = get_logger('main')

offset = 0
subs_count = get_subs_count(group_id)

while offset < subs_count:
    with engine.connect() as conn:
        transaction = conn.begin()
        try:
            logger.info(f"{offset} / {subs_count}")
            df = get_subs_info(group_id, offset)
            df.to_sql('subscribers', con=conn, if_exists='append', index=False)
            if offset == 10:
                raise(ValueError("This is a test errror"))
            transaction.commit()
            logger.info(f"Successfully inserted data")
        except Exception as E:
            transaction.rollback()
            logger.error(f"Error: tried to insert {df} but got an error: {E}")
    time.sleep(1)
    offset += 10

Теперь во время работы программы будет отображаться такой вывод, который также будет записан в файл main.log папки log в директории проекта. После завершения работы программы можно исследовать логи, посмотреть, на каких offset возникли проблемы, какие данные не удалось вставить и прочитать текст ошибки: