Недавно пришлось познакомиться тесно с порталами государственных закупок Казахстана и Узбекистана в рамках Школы Данных. Мы (авторка поста, разработчик скрепера и журналисты) исследовали тему "доступной среды" (удобная инфраструктура для людей с инвалидностью) и столкнулись с необходимиостью написать скрепер, которые бы скачивал данные по ключевым словам.
Итоговый текст по нашей задаче тут.
Вот ссылки на источники, с которыми мы работали:
Сайт госзакупок Узбекистана (две ссылки ниже)
В Узбекистане публикуются тендеры (и другие формы) и корпоративных заказчиков, и бюджетных заказчиков. Эту информацию можно прочитать в левом верхнем угле страниц.
В ходе работы над темой мы узнали, что есть несколько видов государственных закупок - лоты, объявления, тендеры, конкурсы, прямые закупки Поэтому если будете что-то искать среди закупок - будьте бдительны. Проверьте все виды закупок. Если информации нет в одной категории (лоты), то она может быть в другой (в тендерах).
Перед работой стоит наверняка ознакомиться с законодательством и понять, что и где должно быть. В каждом из этих своих видов свои правила, и наши предметы мы находили только в конкретных категориях. Эта проблемы вылезла боком после того, как мы начали искать и скачивать данные для Узбекистана. Пандусов в “тендерах” не оказалось, их можно было найти в “конкурсах”. К счастью, структура страниц оказалась одинаковой и изменения (в уже написанный скрепер) много времени не заняли.
Код на гитхабе подойдет для поиска любых других госзакупок. Просто сделайте замену ключевых слов. Количетсво тем, с которыми можно работать, необъятное и ограничивается только фантазией. Можно скачать все тендеры, где упоминались закупки по COVID, можно анализировать ремонт дорог, строительство и прочее.
Техническая часть мануала. Разбор кода
Скрепер для госзакупок Узбекистана и Казахстана
В ходе разработки программы возникло несколько проблем:
У сайта госзакупок Казахстана существует публичное API, но для его использования нужно получить токен у распорядителя реестра. Для этого, нужно было отправить письмо с указанием цели получения доступа. Мы этого не делали. К счастью, у сайта существует внутреннее API в формате GET-запроса с параметрами. Подставляя необходимые параметры, можно эмулировать поиск по ключевому слову и дальше скрепить результаты.
У сайта госзакупок Казахстана оказалась достаточно сложная для скрепинга структура. Так, например, оказалось, что структура страниц отличается в случаях когда лот состоит из одной части и из нескольких. Кроме того, полная информация о лоте показывалась с помощью Javascript при нажатии на лот, что делало эту информацию невозможной к получению при использовании классического подхода скрепинга статических страниц.
У сайта госзакупок Узбекистана публичное API отсутствовало. Но был поиск, работающий по такому же принципу, как и у Казахстана. Однако, в отличии от Казахстана, поиск по ключевым словам не работал. Поэтому здесь пришлось искать все тендеры за максимально разрешённый 90-дневный период, а уже затем отдельно искать среди них искомые запросы.
Скреперы написаны в формате утилит командной строки. Для их использования не требуется знания языков программирования. Код и подробная инструкция по запуска есть в репозитории на Github. Всё, что необходимо — интерпретатор Python версии не ниже 3.6 с установленными модулями из файла requirements.txt. Рекомендуется использовать виртуальное окружение Python во избежания конфликта зависимостей. Установка интерпретатора и модулей может отличаться в зависимости от используемой операционной системы. На Linux это:
python3 -m venv venv
source venv/bin/activate
pip3 install -r requirements.txtУстановив все необходимое, запускаем скрепер. Для Узбекистана это будет выглядеть так:
python3 uzbekistan_scraper.py tender 01.01.2021 31.01.2021где tender - раздел, по которому ищем. Ещё можем искать по конкурсам – тогда вместо tender пишем competitive. Две даты в формате dd.mm.YYYY – это начало и конец промежутка поиска. Он может быть не больше 90 дней. Если нужно искать в промежутке больше, разделите его на несколько по 90 дней. Кроме того, есть ограничение в 5000 результатов поиска на запрос.
Результатом работы скрепера будет три сущности:
• purchase_type_date.csv — сводная таблица с общей информацией о закупках. Включает в себя поля номера лота, названия, стоимости лота, региона проведения закупки;
• purchase_typedetaileddate.csv — сводная таблица с подробной информацией по каждой закупке;
• отдельные каталоги для каждой закупки с таблицей с детальной информацией и архивом с тендерными документами.
Для Казахстана запуск будет выглядеть так:
python3 kazakhstan_scraper.py "доступная среда"Если поисковый запрос состоит из нескольких слов, возьмите их в кавычки. Поиск будет происходить по словосочетанию. Количество результатов поиска здесь имеет ограничение в 2000 единиц.
Результатом работы скрепера будут три сущности:
• tenders_query.csv — сводная таблица с основной информацией по искомым тендерам. Включает в себя поля наименований объявления, лота, количество и сумму, способ закупки, её статус;
• tenders_detailed_query.csv — сводная таблица с детальной информацией по искомым тендерам. Кроме основной информации включает в себя дополнительную характеристику лота, заказчика, цену за единицу и другое;
• отдельные каталоги, в каждом из которых находится таблица с информацией по объявлению, аналогичной той, что в сводной таблице.
Разбор кода
В файле requirements.txt находится список модулей, необходимых для установки пакетным менеджером pip. Если установка модуля вызывает ошибку, попробуйте удалить версию модуля, скачав таким образом, самую последнюю. Файл log.txt перезаписывается при запуске программы — в него записывается прогресс выполнения. README.md – это документация, аналогичная той, что выше.
Основные переменные находятся в файлах settings.py. Это CSS-селекторы, используемые модулем BeautifulSoup для парсинга HTML-страниц, регулярные выражения для парсинга текста и названия полей в будущих сводных таблицах. Если скрепер перестанет выдавать содержимое той или иной колонки, в первую очередь стоит смотреть именно на используемый селектор. По большому счёту, мелкие изменения редизайна сайта могут быть решены только за счёт исправлений в файле настроек.
Перейдём к непосредственной логике программы, файл kazakhstan_scraper.py. Чтение кода стоит начать с последних двух строк:
if __name__ == '__main__':
main()значит, что при запуске скрипта как программы, будет вызвана функция main.
def main(): # объявляем функцию
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) #отключаем показ ошибки о проверку сертификатов сайта. Просто чтобы не раздражало.
logging.basicConfig(level=logging.INFO, format='%(levelname)s:%(message)s', filename='log.txt', filemode='w') #включаем логгирование в сответствующий файл с перезаписью
if len(sys.argv) != 2:
raise ValueError('Usage: python3 kazakhstan_tenders.py [search_word]. If your search query consists of two and more words, take them into quotes.')Так как программа задумана как консольная утилита, добавляем базову проверку — если число аргументов включая название программы не равно 2, прерываем выполнение с показом справки по запуску.
search_word = sys.argv[1] #присваем переменной значения введённого посикового запроса
start_time = time() #активируем таймер чтобы знать, сколько времени заняло выполнение
print('Start downloading') #выводим сообщение о начале работы
logging.info('Start downloading') #пишем его в лог
url_list = get_general_table(search_word) #запускаем первую функцию для получения общей таблицы, её результат записываем
tenders_info = get_detailed_table(url_list)# запускаем вторую функцию для получения подробных таблиц, её результат записываем
get_csv_with_tenders_info(tenders_info, search_word)# формируем и сохраняем финальную сводную таблицу
end_time = time() #останавливаем таймер
print(f'Download for {search_word} completed in {end_time - start_time:.2f} seconds.') #выводим сообщение об успешно завершённой загрузке за время
logging.info(f'Download for {search_word} completed in {end_time - start_time:.2f} seconds.') #дублируем эту запись в файлперейдём теперь к отдельным функциям, но сначала импорты:
import logging #модуль логирования
import os #модуль работы с системой
import pandas #модуль для работы с данными и таблицами
import requests #модуль работы с веб-страницами
import sys #модуль для работы с аргументами командной строки
import urllib3 #нужен только для отключения записи об ошибке сертификата
from bs4 import BeautifulSoup #главный модуль для парсинга веб-страниц
from collections import OrderedDict #сортированный словарь, который мы используем для создания двумерного массива. Обычный словарь в Python расставит колонки в случайном порядке, нам это не подходит.
from datetime import datetime #модуль работы с датой
from time import time #модуль таймера
from kazakhstan_settings import * #загружаем все переменные из файла настроек
def get_general_table(search_word): #объявляем функцию
start_time = time() #таймер
kazakhstan_entrypoint = f'{ENTRY_POINT}/ru/search/lots?filter%5Bname5D={search_word}&count_record=2000&search' \
f'=&filter%5Bnumber%5D' \
f'=&filter%5Bnumber_anno%5D=&filter%5Benstru%5D=&filter%5Bcustomer%5D=&filter' \
f'%5Bamount_from%5D=&filter%5Bamount_to%5D=&filter%5Btrade_type%5D=&filter%5Bmonth%5D' \
f'=&filter%5Bplan_number%5D=&filter%5Bend_date_from%5D=&filter%5Bend_date_to%5D=&filter' \
f'%5Bstart_date_to%5D=&filter%5Byear%5D=&filter%5Bitogi_date_from%5D=&filter' \
f'%5Bitogi_date_to%5D=&filter%5Bstart_date_from%5D=&filter%5Bmore%5D=' #формируем адрес поискового запроса, который мы будем совершать, подставляя в словосочетание, которое мы ищем. Ещё может быть интересен параметр count_record — количество получаемых результатов. 2000 должно хватить в большинстве случаев.
response = requests.get(kazakhstan_entrypoint, headers=HEADERS, verify=False) #получаем HTML-страницу, подставляя заголовки, чтобы выглядеть как реальный человек и игнорировать проверку сертификата. Казахстанская цензура требует всех устанавливать государственные сертфикаты, блокируемые разработчиками браузеров.
soup = BeautifulSoup(response.content, 'html.parser') #парсим полученную страницу
urls_list = [f"{ENTRY_POINT}{url['href']}?tab=lots" for url in soup.select(SELECT_GENERAL_URLS)] #получаем список всех ссылок на лоты
general_table = list() #создаём список для будущей общей таблицы
general_table_dict = OrderedDict() #создаём сортированный словарь для будущей общей таблицы
general_table_dict[LOT_ID] = [soup.select(SELECT_LOT_ID)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_LOT_ID))] #создаём колонку с название из переменной LOT_ID и содержимым в виде списка значений содержимого селектора. Тут и дальше стоит отметить, что мы получаем список, а для прохода по списку используем функцию enumerate. Она подходит лучше обычного for в ситуациях, когда нам нужно получать и индекс итерируемого объекта и его значение.
general_table_dict[ANNOUNCE_NAME] = [soup.select(SELECT_ANNOUNCE_NAME)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_ANNOUNCE_NAME))] #создаём колонку с название из переменной ANNOUNCE_NAME и содержимым в виде списка значений содержимого селектора
general_table_dict[LOT_CUSTOMER] = [idx.next_sibling.strip() for idx in soup.find_all(SELECT_LOT_CUSTOMER, text=SELECT_LOT_CUSTOMER_SIBLING)] #в отличии от других колонок, здесь мы не можем получить необходимый текст только селектором, поэтому мы ищем соседний текст, а уже затем получаем соседний ему и нужный нам
general_table_dict[LOT_NAME] = [soup.select(SELECT_LOT_NAME)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_LOT_NAME))]
general_table_dict[LOT_NUMBER] = [soup.select(SELECT_LOT_NUMBER)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_LOT_NUMBER))]
general_table_dict[LOT_PRICE] = [soup.select(SELECT_LOT_PRICE)[idx].get_text().strip().replace(' ', '') for idx, _ in enumerate(soup.select(SELECT_LOT_PRICE))] #здесь, вдобавок к обычным действиям, производим автозамену запятой на ничего
general_table_dict[LOT_PURCHASE_METHOD] = [soup.select(SELECT_LOT_PURCHASE_METHOD)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_LOT_PURCHASE_METHOD))]
general_table_dict[LOT_STATUS] = [soup.select(SELECT_LOT_STATUS)[idx].get_text().strip() for idx, _ in enumerate(soup.select(SELECT_LOT_STATUS))]
record_general_table(general_table_dict, search_word) #вызываем функцию для сохранения общей таблицы
general_table.append(pandas.DataFrame(general_table_dict)) #превращаем наш словарь в объект DataFrame и добавляем его в список общей таблицы. Тем самым мы дописываем в неё ещё одну строку со всеми колонками.
end_time = time() #останавливаем таймер
print(f'Urls from general table for {search_word} and the table have been downloaded in {end_time - start_time:.2f} seconds.') # сообщаем о завершенной работе
logging.info(f'Urls from general table for {search_word} and the table have been downloaded in {end_time - start_time:.2f} seconds.') #и записываем это в файл
return urls_list# возвращаем список с ссылками на страницы с объявлениями
def record_general_table(general_table_dict, search_word): #передаём функции упорядоченный словарь и поисковое словосочетание
if not os.path.exists(PATH_TO_LOCATION):
os.makedirs(PATH_TO_LOCATION) #проверяем, существует ли каталог kazakgstan/, если нет, создаём его
pandas.DataFrame(general_table_dict).to_csv(PATH_TO_LOCATION + f'tenders_{search_word}.csv') #записываем в него общую таблицу
def get_detailed_table(urls_list): #передаём функции список с загружёнными ссылками на объявления
detailed_table = list() #создаём список, будущую сводную таблицу
for url in urls_list: #будем обходить этот список по ссылке за раз
try: # здесь мы делаем всё по тому же алгоритму, что и в функции выше
start_time = time() старт таймера
response = requests.get(url, headers=HEADERS, verify=False)
soup = BeautifulSoup(response.content, 'html.parser')
detailed_table_dict = OrderedDict()
detailed_table_dict[ANNOUNCE_ID] = soup.find_all(SELECT_ANNOUNCE_HEADER)[0]['value']
detailed_table_dict[ANNOUNCE_NAME] = soup.find_all(SELECT_ANNOUNCE_HEADER)[1]['value']
detailed_table_dict[ANNOUNCE_STATUS] = soup.find_all(SELECT_ANNOUNCE_HEADER)[2]['value']
detailed_table_dict[ANNOUNCE_PUBLICATION_DATE] = soup.find_all(SELECT_ANNOUNCE_HEADER)[3]['value']
detailed_table_dict[ANNOUNCE_START_DATE] = soup.find_all(SELECT_ANNOUNCE_HEADER)[4]['value']
detailed_table_dict[ANNOUNCE_END_DATE] = soup.find_all(SELECT_ANNOUNCE_HEADER)[5]['value']
try: #внешний вид страницы может отличаться в зависимости от количества лотов в объявлении. Мы обрабатываем этот случай, делая поиск не по соседнему тексту, а используя селектор.
detailed_table_dict[LOT_ID] = soup.find(SELECT_LOT_HEADER,
text=SELECT_LOT_ID_DETAILED_SIBLING).next_sibling.strip()
detailed_table_dict[LOT_NAME] = soup.find(SELECT_LOT_HEADER,
text=SELECT_LOT_NAME_DETAILED_SIBLING).next_sibling.strip()
detailed_table_dict[LOT_DESCRIPTION] = soup.find(SELECT_LOT_HEADER,
text=SELECT_LOT_DESCRIPTION_SIBLING).next_sibling.strip()
detailed_table_dict[LOT_DESCRIPTION_DETAILED] = soup.find(SELECT_LOT_HEADER,
text=SELECT_LOT_DESCRIPTION_DETAILED_SIBLING).next_sibling.strip()
except AttributeError:
detailed_table_dict[LOT_ID] = [soup.select(SELECT_LOT_ID_DETAILED)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_ID_DETAILED))]
detailed_table_dict[LOT_NAME] = [soup.select(SELECT_LOT_NAME_DETAILED)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_NAME_DETAILED))]
detailed_table_dict[LOT_DESCRIPTION] = ''
detailed_table_dict[LOT_DESCRIPTION_DETAILED] = ''
detailed_table_dict[LOT_CUSTOMER_NAME] = [soup.select(SELECT_LOT_CUSTOMER_NAME)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_CUSTOMER_NAME))]
detailed_table_dict[LOT_CHARACTERISTICS_FULL] = [
soup.select(SELECT_LOT_CHARACTERISTICS_FULL)[idx].get_text().strip() for idx, _ in
enumerate(soup.select(SELECT_LOT_CHARACTERISTICS_FULL))]
detailed_table_dict[LOT_PRICE_PER_ONE] = [
soup.select(SELECT_LOT_PRICE_PER_ONE)[idx].get_text().strip().replace(' ', '') for
idx, _ in
enumerate(soup.select(SELECT_LOT_PRICE_PER_ONE))]
detailed_table_dict[LOT_NUMBER] = [soup.select(SELECT_LOT_NUMBER_DETAILED)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_NUMBER_DETAILED))]
detailed_table_dict[LOT_MEASUREMENT] = [soup.select(SELECT_LOT_MEASUREMENT)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_MEASUREMENT))]
detailed_table_dict[LOT_PLANNED_TOTAL] = [
soup.select(SELECT_LOT_PLANNED_TOTAL)[idx].get_text().strip().replace(' ', '') for
idx, _ in
enumerate(soup.select(SELECT_LOT_PLANNED_TOTAL))]
detailed_table_dict[LOT_TOTAL_1_YEAR] = [soup.select(SELECT_LOT_TOTAL_1_YEAR)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_TOTAL_1_YEAR))]
detailed_table_dict[LOT_TOTAL_2_YEAR] = [soup.select(SELECT_LOT_TOTAL_2_YEAR)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_TOTAL_2_YEAR))]
detailed_table_dict[LOT_TOTAL_3_YEAR] = [soup.select(SELECT_LOT_TOTAL_3_YEAR)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_TOTAL_3_YEAR))]
detailed_table_dict[LOT_STATUS] = [soup.select(SELECT_LOT_STATUS_DETAILED)[idx].get_text().strip() for
idx, _ in
enumerate(soup.select(SELECT_LOT_STATUS_DETAILED))]
record_detailed_table(detailed_table_dict, detailed_table_dict[ANNOUNCE_ID]) #записываем подробную таблицу для каждого объявления
detailed_table.append(pandas.DataFrame(detailed_table_dict)) #превращаем наш словарь в объект DataFrame и добавляем его в список общей таблицы. Тем самым мы дописываем в неё ещё одну строку со всеми колонками.
end_time = time() # конец таймера
print(f'Lot {detailed_table_dict[ANNOUNCE_ID]} info has been downloaded in {end_time - start_time:.2f} seconds.')# выводим за сколько секунд была загружена информация об объявлении
logging.info(
f'Lot {detailed_table_dict[ANNOUNCE_ID]} info has been downloaded in {end_time - start_time:.2f} seconds.') # дублируем её в файл
except IndexError: обрабатываем ошибку когда ссылка на объявление есть, а такой страницы нету
print('Page is not found. Probably, it wad deleted.') #выводим сообщение об ошибке
logging.error('Page is not found. Probably, it was deleted.') #пишем в лог
continue #переходим к следующей ссылке
return detailed_table
В скрепере для Узбекистана логика работы такая же, за исключением нескольких моментов:
def main():
if len(sys.argv) != 4:# здесь мы проверяем наличие уже четырёх аргументов
raise ValueError('Usage: python3 uzbekistan_scraper.py [tender|competitive] [start_date] [end_date]')
purchase_type = verify_purchase_type(sys.argv[1]) #для Узбекистана мы можем искать по конкурсам или по тендерам. Здесь мы проверяем, какой вариант был задан.
start_date = sys.argv[2] #присваеваем первую временную границу
end_date = sys.argv[3]# присваиваем вторую временную границу
verify_date(start_date, end_date) #проверяем, что промежуток между первой и второй датами не превышает 90 дней
def verify_purchase_type(purchase_type):
if purchase_type == 'tender': #если при запуске ввели это значение, то
purchase_type = 'tender2' #подставляем в будущий запрос это
elif purchase_type == 'competitive':# если такое, то это
pass
else:# иначе выдаём ошибку
logging.error('Purchase type can be only tender or competitive.')
raise ValueError('Purchase type can be only tender or competitive.')
return purchase_type
def verify_date(start_date, end_date):
start_date = datetime.strptime(start_date, '%d.%m.%Y') #превращаем строку с датой в специальный формат
end_date = datetime.strptime(end_date, '%d.%m.%Y')
if abs((end_date - start_date).days) > 90: #сравниваем их и выдаём ошибку, если промежуток составляет больше 90 дней
logging.error("Difference between dates shouldn't be more than 90 days.")
raise ValueError("Difference between dates shouldn't be more than 90 days.")
Дальше отличия заключаются только в названиях полей и селекторах, за исключением сохранения архива с тендерной документацией.
def record_detailed_table(detailed_table_dict, lot_id, lot_documents_url):
lot_location = f'{PATH_TO_LOCATION}/{lot_id}/'
if not os.path.exists(lot_location):
os.makedirs(lot_location)
pandas.DataFrame(detailed_table_dict).to_csv(lot_location + lot_id + '.csv')
response = requests.get(lot_documents_url, headers=HEADERS, verify=False) #кроме того, что мы сохраняем таблицу, мы ещё сохраняем архив с документацией
try:
with open(f"{lot_location}{lot_id}.{lot_documents_url.split('.')[-1]}", 'wb') as f: #название архива на сайте было сгенерировано автоматически, мы переназываем его соответственно к номеру закупки
f.write(response.content)
except FileNotFoundError:# если архива на странице нету, выдаём об этом предупреждение и продолжаем
logging.error("Archive with purchase documentation hasn't been found on the page.")
return
uzbekistan_entrypoint = f'{ENTRY_POINT}/ru/ajax/filter?LotID=&PriceMin=&PriceMax=&RegionID=&TypeID=&DistrictID=&INN=&CategoryID=&EndDate={end_date}&PageSize=5000&Src=AllMarkets&PageIndex=1&Type={purchase_type}&Tnved=&StartDate={start_date}' поисковый запрос тоже, конечно, будет другим. Здесь мы подставляем две даты, тип раздела, по которому ищем. Интерес может представлять параметр PageSize. На сайте максимальное количество результатов 2000, в запросе можно подставить любое число. 5000 должно быть достаточно, учитывая временное ограничение в 90 дней. Чем больше результатов, тем больше времени нужно на обработку запроса, учитываю загрузку тяжёлых документов. Кроме того, архивы с документами могут занимать много места на диске.
Если какой-то функционал не работает или нужно добавить что-то новое, пишите в Issues проекта на Github.
zoldaten
добавьте в теги «парсер», ибо это парсинг называется.