scraper tutorial

Recently I had to get acquainted closely with the portals of public procurement of Kazakhstan and Uzbekistan within the framework of the School of Data. We (the author of the post, the scraper developer and journalists) researched the topic of “accessible environment” (convenient infrastructure for people with disabilities) and were faced with the need to write a scraper that would download data by keywords.

The final text for our task here

Here are the links to the sources we worked with:

Tenders (and other forms) are published in Uzbekistan and corporate customers, and budgetary customers. This information can be read in the upper left corner of the pages.

While working on the topic, we learned that there are several types of public procurement – lots, announcements, tenders, tenders, direct purchases. Therefore, if you are looking for something among the purchases, be careful. Check all types of purchases. If the information is not in one category (lots), then it can be in another (in tenders).

Before starting work, you should probably familiarize yourself with the legislation and understand what and where should be. Each of these types has its own rules, and we found our objects only in specific categories. This problem got out sideways after we started searching and downloading data for Uzbekistan. There were no ramps in the “tenders”, they could be found in the “competitions”. Fortunately, the structure of the pages turned out to be the same and changes (to the already written scraper) did not take long.

The code on the github is suitable for searching for any other government purchases. Just make a keyword substitution. The number of those that you can work with is immense and limited only by your imagination. You can download all tenders where procurement for COVID was mentioned, you can analyze road repairs, construction, etc.

The technical part of the manual. Parsing the code

Code on Github

Scraper for public procurement of Uzbekistan and Kazakhstan

During development programs there were several problems:

The public procurement website of Kazakhstan has a public API, but to use it, you need to obtain a token from the registry manager. For this, it was necessary to send a letter indicating the purpose of obtaining access. We didn’t. Fortunately, the site has an internal API in the form of a GET request with parameters. By substituting the necessary parameters, you can emulate a keyword search and further consolidate the results.

The public procurement site of Kazakhstan turned out to have a rather complicated structure for scraping. So, for example, it turned out that the structure of the pages is different in cases where the lot consists of one part and several. In addition, full information about the lot was shown using Javascript when clicking on the lot, which made this information impossible to obtain when using the classic approach of scraping static pages.

The Uzbek government procurement website did not have a public API. But there was a search that worked on the same principle as in Kazakhstan. However, unlike in Kazakhstan, keyword searches did not work. Therefore, here it was necessary to search for all tenders for the maximum allowed 90-day period, and only then separately search among them for the required requests.

Scrapers are written in command line utility format. To use them, no knowledge of programming languages ​​is required. The code and detailed instructions for launching are in the repository on Github. All that is needed is a Python interpreter of version 3.6 or higher with the modules installed from the requirements.txt file. It is recommended to use a Python virtual environment to avoid dependency conflicts. The installation of the interpreter and modules may differ depending on the operating system used. On Linux, these are:

python3 -m venv venv
source venv/bin/activate
pip3 install -r requirements.txt

Having installed everything you need, start the scraper. For Uzbekistan it will look like this:

python3 uzbekistan_scraper.py tender 01.01.2021 31.01.2021

where tender is the section we are looking for. We can also search by competitions – then instead of tender we write competitive. The two dates in the dd.mm.YYYY format are the start and end of the search span. It can be no more than 90 days. If you need to search in between more, divide it into several 90 days each. In addition, there is a limit of 5,000 search results per query.

The result of the scraper’s work will be three entities:

purchase_type_date.csv – a summary table with general information about purchases. Includes fields for lot number, name, lot value, purchase region;

• purchase_typedetaileddate.csv – a summary table with detailed information for each purchase;

• separate catalogs for each purchase with a table with detailed information and an archive with tender documents.

For Kazakhstan, the launch will look like this:

python3 kazakhstan_scraper.py "доступная среда"

If your search term contains multiple words, enclose them in quotation marks. The search will be carried out by the phrase. The number of search results here is limited to 2000.

The scraper will result in three entities:

• tenders_query.csv – a summary table with basic information on the sought tenders. Includes fields of ad names, lot, quantity and amount, purchase method, its status;

• tenders_detailed_query.csv – a summary table with detailed information on the sought tenders. In addition to basic information, it includes additional characteristics of the lot, customer, unit price, and more;

• separate catalogs, each of which contains a table with information on the ad, similar to that in the pivot table.

Parsing the code

The requirements.txt file contains a list of modules required for installation with the pip package manager. If the installation of the module causes an error, try uninstalling the module version by downloading the most recent one. The log.txt file is overwritten when the program is started – the progress is recorded in it. README.md is documentation similar to the one above.

The main variables are in the settings.py files. These are CSS selectors used by the BeautifulSoup module to parse HTML pages, regular expressions to parse text, and field names in future pivot tables. If the scraper stops displaying the contents of a particular column, the first thing to look at is the selector used. By and large, minor changes to the website redesign can only be resolved by fixing the settings file.

Let’s move on to the direct logic of the program, the kazakhstan_scraper.py file. You should start reading the code with the last two lines:

if __name__ == '__main__':
	main()

means that when the script is run as a program, the main function will be called.

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.')

Since the program is conceived as a console utility, we add a basic check – if the number of arguments, including the program name, is not equal to 2, we interrupt the execution and display the startup help.

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.') #дублируем эту запись в файл

Now let’s move on to the individual functions, but first the imports:

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
 

In the scraper for Uzbekistan, the logic of work is the same, with the exception of a few points:

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.")

Further, the differences are only in the names of the fields and selectors, with the exception of saving the archive with tender documentation.

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 дней. Чем больше результатов, тем больше времени нужно на обработку запроса, учитываю загрузку тяжёлых документов. Кроме того, архивы с документами могут занимать много места на диске.

If some functionality does not work or you need to add something new, write to the Issues of the project on Github.

Similar Posts

Leave a Reply