Pull to refresh

Как проанализировать рынок фотостудий с помощью Python (2/3). База данных

Reading time 12 min
Views 4.6K
В предыдущей статье в рамках коммерческого проекта по анализу рынка фотостудий рассмотрел создание парсинга: выгрузка списка фотостудий, списка залов, данных по бронированию с момента открытия зала до последней брони.

Полученную информацию в кэше хранить нецелесообразно, необходимо использовать базу данных.

В статье рассмотрю:

  • создание простой SQLite базы данных;
  • запись информации с помощь Python;
  • чтение данных и перевод в формат DataFrame;
  • обновление парсинга с учетом данных БД.



Требования к базе данных


Основное требование к БД по проекту: хранить данные и иметь возможность их оперативно извлечь.

Нашей БД не требуется:

  • разграничивать доступ к схемам, т.к. доступ будет только у пользователя парсингом;
  • сохранять доступ 24/7, т.к. извлечение данных допустимо по мере необходимости проведения анализа;
  • создание процедур, т.к. все вычисления будут проводится в python.

Поэтому для проекта возможно использовать простую базу данных в SQLite. Хранить её можно в виде файла либо на жестком диске, либо на флешке, либо на облачном диске для доступа с других устройств.

Особенности работы с SQLite через python


Для работы с SQLite через python используем библиотеку sqlite3.

Подключаемся к базе данных простой командой:

sqlite3.connect(путь к файлу)

Если файл отсутствует, будет создана новая база.

Запросы к базе выполняются следующим образом:

conn = sqlite3.connect(путь к файлу)
cur = conn.cursor()
cur.execute(запрос)
df = cur.fetchall()

cur.fetchall() выполняется в том случае, когда в результате запроса мы хотим получить данные из БД.

В конце записи данных в БД не забывайте заканчивать транзакцию:

conn.commit()

а в конце работы с базой не забывайте её закрывать:


conn.close()

иначе база будет блокироваться на запись или на открытие.

Создание таблиц стандартное:

CREATE TABLE t1 (поле1 тип, поле2 тип...)

или более универсальный вариант, создающий таблицу в случае её отсутствия:

CREATE TABLE IF NOT EXISTS t1 (поле1 тип, поле2 тип...)

Записываем данные в таблицу, избегая повторов:

INSERT OR IGNORE INTO t1 (поле1, поле2, ...) VALUES(значение1, значение2, ...)

Обновляем данные:

UPDATE t1 SET поле1 = значение1 WHERE поле2 = значение2

Для более удобной работы с SQLite можно использовать SQLite Manager или DB Browser for SQLite.

Первая программа является расширением к браузеру и выглядит как чередование строки запроса и блока ответа:



Вторая программа — полноценное desktop-приложение:





Структура базы данных


БД будет состоять из 4 таблиц: студии, залы, 2 таблицы бронирования.

В выгружаемых данных по бронированию есть информация о будущих периодах, которая с новым парсингом может измениться. Затирать данные нежелательно (их можно использовать, например, для вычисления дня/часа, когда была сделана бронь). Поэтому, одна таблица бронирования необходима для сырых данных парсинга, вторая — для последних, актуальных.

Создаем таблицы:

def create_tables(conn, table = 'all'):
    cur = conn.cursor()
    
    if (table == 'all') or (table == 'uStudios'):
        cur.execute('''
            CREATE TABLE IF NOT EXISTS uStudios
            (studio_id INT PRIMARY KEY UNIQUE,
            name TEXT UNIQUE,
            metro TEXT,
            address TEXT,
            phone TEXT,
            email TEXT,
            established_date DATE)
            ''')
        print('Table uStudios is created.')

    if (table == 'all') or (table == 'uHalls'):
        cur.execute('''
            CREATE TABLE IF NOT EXISTS uHalls
            (hall_id INT PRIMARY KEY UNIQUE,
            studio_id INT,
            name TEXT,
            is_hall INT,
            square FLOAT,
            ceiling FLOAT,
            open_date DATE)
            ''')
        print('Table uHalls is created.')

    if (table == 'all') or (table == 'uBooking_parsing'):
        cur.execute('''
            CREATE TABLE IF NOT EXISTS uBooking_parsing
            (hall_id INT,
            date DATE,
            hour INT,
            is_working_hour INT,
            min_hours INT,
            price INTEGER,
            is_booked INT,
            duration INT,
            parsing_date DATE)
            ''')
        print ('Table uBooking_parsing is created.')

    if (table == 'all') or (table == 'uBooking'):
        cur.execute('''
            CREATE TABLE IF NOT EXISTS uBooking
            (hall_id INT,
            date DATE,
            hour INT,
            is_working_hour INT,
            min_hours INT,
            price INTEGER,
            is_booked INT,
            duration INT,
            parsing_date DATE)
            ''')
        print ('Table uBooking is created.')


Параметром table задаю название таблицы, которую необходимо создать. По умолчанию создает все.

В полях таблиц видны данные, которые не парсили (дата открытия студии, дата открытия зала). Вычисление этих полей опишу позже.

Взаимодействие с базой данных


Создадим 6 процедур для взаимодействия с базой данных:

  1. Запись списка фотостудий в базу данных;
  2. Выгрузка списка фотостудий из базы данных;
  3. Запись списка залов;
  4. Выгрузка списка залов;
  5. Выгрузка данных по бронированию;
  6. Запись данных по бронированию.

1. Запись списка фотостудий в базу данных


На входе в процедуру передаем параметры соединения с БД и таблицу в виде DataFrame. Записываем данные построчно, перебирая все строчки циклом. Полезным для этой операции свойством строковых данных в python является замена символов "?" элементами кортежа, указанным после.

Процедура записи списка фотостудий выглядит следующим образом:
def studios_to_db(conn, studio_list): 
    cur = conn.cursor()
    for i in studio_list.index:
        cur.execute('INSERT OR IGNORE INTO uStudios (studio_id, name, metro, address, phone, email) VALUES(?, ?, ?, ?, ?, ?)',
                   (i,
                   studio_list.loc[i, 'name'],
                   studio_list.loc[i, 'metro'],
                   studio_list.loc[i, 'address'],
                   studio_list.loc[i, 'phone'],
                   studio_list.loc[i, 'email']))


2. Выгрузка списка фотостудий из базы данных


На вход в процедуру передаем параметры соединения с БД. Выполняем select-запрос, перехватываем выгружаемые данные и записываем в DataFrame. Переводим дату основания фотостудии в формат даты.

Полностью процедура выглядит следующим образом:
def db_to_studios(conn):
    cur = conn.cursor()
    cur.execute('SELECT * FROM uStudios')
    studios = pd.DataFrame(cur.fetchall()
                           , columns=['studio_id', 'name', 'metro', 'address', 'phone', 'email', 'established_date']
                          ).set_index('studio_id')
    studios['established_date'] = pd.to_datetime(studios['established_date'])
    return studios


3. Запись списка залов в базу данных


Процедура аналогична записи списка фотостудий: передаем параметры подключения и таблицу залов, построчно записываем данные в базу.

Процедура записи списка залов в БД
def halls_to_db(conn, halls): 
    cur = conn.cursor()
    for i in halls.index:
        cur.execute('INSERT OR IGNORE INTO uHalls (hall_id, studio_id, name, is_hall, square, ceiling) VALUES(?, ?, ?, ?, ?, ?)',
                   (i,
                   halls.loc[i, 'studio_id'],
                   halls.loc[i, 'name'],
                   halls.loc[i, 'is_hall'],
                   halls.loc[i, 'square'],
                   halls.loc[i, 'ceiling']))


4. Выгрузка списка залов из базы данных


Процедура аналогична выгрузки списка фотостудий: передача параметров подключения, select-запрос, перехват, запись в DataFrame, перевод даты открытия зала в формат даты.
Единственное отличие: id студии и признак зала записались в байтовом виде. Возвращаем значение функцией:

int.from_bytes(число, 'little')

Процедура выгрузки списка залов выглядит следующим образом:
def db_to_halls(conn):
    cur = conn.cursor()
    cur.execute('SELECT * FROM uHalls')
    halls = pd.DataFrame(cur.fetchall(), columns=['hall_id', 'studio_id', 'name', 'is_hall', 'square', 'ceiling', 'open_date']).set_index('hall_id')
    for i in halls.index:
        halls.loc[i, 'studio_id'] = int.from_bytes(halls.loc[i, 'studio_id'], 'little')
        halls.loc[i, 'is_hall'] = int.from_bytes(halls.loc[i, 'is_hall'], 'little')
    halls['open_date'] = pd.to_datetime(halls['open_date'])
    return halls


5. Выгрузка информации по бронированию из базы данных


В процедуру передаем параметры подключения к БД и параметр parsing, показывающий из какой таблицы по бронированию запрашиваем информацию: 0 — из актуальной (по умолчанию), 1 — из таблицы парсинга. Далее выполняем select-запрос, перехватываем его, переводим в DataFrame. Даты переводим в формат дат, числа из байтового формата в формат чисел.

Процедура выгрузки информации по бронированию:
def db_to_booking(conn, parsing = 0):
    cur = conn.cursor()
    if parsing == 1:
        cur.execute('SELECT * FROM uBooking_parsing')
    else:
        cur.execute('SELECT * FROM uBooking')
    booking = pd.DataFrame(cur.fetchall(), columns=['hall_id', 
                                                     'date', 'hour', 
                                                     'is_working_hour', 
                                                     'min_hours', 
                                                     'price', 
                                                     'is_booked', 
                                                     'duration', 
                                                     'parsing_date'])
    booking['hall_id'] = [int.from_bytes(x, 'little') if not isinstance(x, int) else x for x in booking['hall_id']]
    booking['is_booked'] = [int.from_bytes(x, 'little') if not isinstance(x, int) else x for x in booking['is_booked']]
    booking['date'] = pd.DataFrame(booking['date'])
    booking['parsing_date'] = pd.DataFrame(booking['parsing_date'])
    
    return booking


6. Запись информации по бронированию в базу данных


Самая сложная функция взаимодействия с БД, т.к. она инициирует парсинг данных по бронированию. На входе передаем процедуре параметры подключения к базе данных и список id залов, который необходимо актуализировать.

Чтобы определить последнюю дату актуальных данных,

запросим из базы последнюю дату парсинга по каждому id зала:
parsing_date = db_to_booking(conn, parsing = 1).groupby('hall_id').agg(np.max)['parsing_date']


Перебираем каждый id зала с помощью цикла.

В каждом id зала первом делом определяем

количество недель, которые необходимо парсить в прошлом:
        try:
            last_day_str = parsing_date[id]
            last_day = datetime.datetime.strptime(last_day_str, '%Y-%m-%d')
            delta_days = (datetime.datetime.now() - last_day).days
            weeks_ago = delta_days // 7
        except:
            last_day_str = '2010-01-01'
            last_day = datetime.datetime.strptime(last_day_str, '%Y-%m-%d')
            weeks_ago = 500


Если id зала есть в базе, то вычисляем. Если нет, то парсим 500 недель в прошлое или останавливаемся, когда 2 месяца не было броней (ограничение описано в предыдущей статье).

Дальше выполняем процедуры парсинга:
        d = get_past_booking(id, weeks_ago = weeks_ago)        
        d.update(get_future_booking(id))
        book = hall_booking(d)


Вначале парсим информацию по бронированию из прошлого до актуальных данных, потом из будущего (до 2 месяцев, когда записей не было) и в конце переводим данные из формата json в DataFrame.

Завершающим этапом записываем данные по бронированию зала в базу и закрываем транзакцию.

Процедура записи информации по бронированию в базу данных выглядит следующим образом:
def booking_to_db(conn, halls_id):
    cur = conn.cursor()
    cur_date = pd.Timestamp(datetime.date.today())
    parsing_date = db_to_booking(conn, parsing = 1).groupby('hall_id').agg(np.max)['parsing_date']
    
    for id in halls_id:
        
        #download last parsing_date from DataBase
        try:
            last_day_str = parsing_date[id]
            last_day = datetime.datetime.strptime(last_day_str, '%Y-%m-%d')
            delta_days = (datetime.datetime.now() - last_day).days
            weeks_ago = delta_days // 7
        except:
            last_day_str = '2010-01-01'
            last_day = datetime.datetime.strptime(last_day_str, '%Y-%m-%d')
            weeks_ago = 500
        
        d = get_past_booking(id, weeks_ago = weeks_ago)        
        d.update(get_future_booking(id))
        book = hall_booking(d)
        for i in list(range(len(book))):#book.index:
            cur.execute('INSERT OR IGNORE INTO uBooking_parsing (hall_id, date, hour, is_working_hour, min_hours, price, is_booked, duration, parsing_date) VALUES(?,?,?,?,?,?,?,?,?)',
                       (book.iloc[i]['hall_id'],
                       book.iloc[i]['date'].date().isoformat(),
                       book.iloc[i]['hour'],
                       book.iloc[i]['is_working_hour'],
                       book.iloc[i]['min_hours'],
                       book.iloc[i]['price'],
                       book.iloc[i]['is_booked'],
                       book.iloc[i]['duration'],
                       cur_date.date().isoformat()))
        conn.commit()
        print('hall_id ' + str(id) + ' added. ' + str(list(halls_id).index(id) + 1) + ' from ' + str(len(halls_id)))


Обновление дней открытия студии и залов


Дата открытия зала — это самая ранняя дата бронирования зала.

Дата открытия фотостудии — это самая ранняя дата открытия зала этой студии.

Исходя из этой логики,

выгружаем самые ранние даты бронирования каждого зала из базы
halls = db_to_booking(conn).groupby('hall_id').agg(min)['date']


Затем построчно обновляем данные открытия залов:
    for i in list(range(len(halls))):
        cur.execute('''UPDATE uHalls SET open_date = '{1}' WHERE hall_id = {0}'''
                    .format(halls.index[i], str(halls.iloc[i])))


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

Процедура обновления дат открытия:
def update_open_dates(conn):
    
    cur = conn.cursor()
    
    #update open date in uHalls
    halls = db_to_booking(conn).groupby('hall_id').agg(min)['date']
    
    for i in list(range(len(halls))):
        cur.execute('''UPDATE uHalls SET open_date = '{1}' WHERE hall_id = {0}'''
                    .format(halls.index[i], str(halls.iloc[i])))

    #update open date in uStudios
    studios = db_to_halls(conn)
    studios['open_date'] = pd.to_datetime(studios['open_date'])
    studios = studios.groupby('studio_id').agg(min)['open_date']
    for i in list(range(len(studios))):
        cur.execute('''UPDATE uStudios SET established_date = '{1}' WHERE  studio_id = {0}'''
                    .format(studios.index[i], str(studios.iloc[i])))
    
    conn.commit()


Обновление парсинга


Все процедуры в этой и предыдущей статьях мы объединим в данной процедуре. Её можно запускать как при первом парсинге, так и при обновлении данных.

Процедура выглядит следующим образом:
def update_parsing(directory = './/', is_manual = 0):
    
    start_time = time.time()
    
    #is DataBase exists?
    if not os.path.exists(directory + 'photostudios_moscow1.sqlite'):
        if is_manual == 1:
            print('Data base is not exists. Do you want to create DataBase (y/n)? ')
            answer = input().lower()
        else: 
            answer == 'y'
        
        if answer == 'y':
            conn = sqlite3.connect(directory + 'photostudios_moscow1.sqlite')
            conn.close()
            print('DataBase is created')
        elif answer != 'n':
            print('Error in input!')
            return list()
    
    print('DataBase is exists')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
        
    #connect to DataBase
    conn = sqlite3.connect(directory + 'photostudios_moscow1.sqlite')
    cur = conn.cursor()       

    #has DataBase 4 tables?
    tables = [x[0] for x in list(cur.execute('SELECT name FROM sqlite_master WHERE type="table"'))]
    if not ('uStudios' in tables) & ('uHalls' in tables) & ('uBooking_parsing' in tables) & ('uBooking' in tables):
        if is_manual == 1:
            print('Do you want to create missing tables (y/n)? ')
            answer = input().lower()
        else:
            answer = 'y'
        
        if anwer == 'y':
            if not ('uStudios' in tables):
                create_tables(conn, table = 'uStudios')

            if not ('uHalls' in tables):
                create_tables(conn, table = 'uHalls')

            if not ('uBooking_parsing' in tables):
                create_tables(conn, table = 'uBooking_parsing')

            if not ('uBooking' in tables):
                create_tables(conn, table = 'uBooking')

        elif answer != 'n':
            print('Error in input!')
            return list()

    conn.commit()
    print(str(tables) + ' are exist in DataBase')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
    
    #update uStudios
    studios = studio_list()
    new_studios = studios[[x not in list(db_to_studios(conn).index) for x in list(studios.index)]]
    if len(new_studios) > 0:
        print(str(len(new_studios)) + ' new studios detected: \n' + str(list(new_studios['name'])))
        studios_to_db(conn, new_studios)
    
    conn.commit()
    print('Studio list update was successful')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
    
    #update uHalls
    halls = hall_list(list(studios.index)).sort_index()
    new_halls = halls[[x not in list(db_to_halls(conn).index) for x in list(halls.index)]]
    if len(new_halls) > 0:
        halls_to_db(conn, new_halls)
    
    conn.commit()
    print('Halls list update was successful')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
        
    #update uBooking_parsing
    booking_to_db(conn, halls.index)   
    
    conn.commit()
    print('Booking_parsing update was successful')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
    
    #update uBooking from uBooking_parsing
    cur.execute('DELETE FROM uBooking')
    cur.execute('''
        insert into uBooking (hall_id, date, hour, is_working_hour, min_hours, price, is_booked, duration, parsing_date) 
        select hall_id, date, hour, is_working_hour, min_hours, price, is_booked, duration, parsing_date
        from
        (
            select *, row_number() over(partition by hall_id, date, hour order by parsing_date desc) rn 
            from uBooking_parsing
        ) t
        where rn = 1
    ''')
    conn.commit()
    print('Booking update was successful')
    print("--- %s seconds ---" % (time.time() - start_time))
    start_time = time.time()
    
    update_open_dates(conn)
    conn.commit()
    print('Open date update was successful')
    print("--- %s seconds ---" % (time.time() - start_time))
    
    conn.close()


Разберем ее работу по порядку.

На входе в процедуру передаем 2 параметра: адрес папки, откуда брать базу данных или куда ее установить (по умолчанию берем папку с python-документов), и необязательный параметр is_manual, который при значении «1» будет запрашивать необходимость создания базы данных или таблиц в случае их отсутствия.

Вначале проверяем есть ли база данных. Если нет, создаём:
    if not os.path.exists(directory + 'photostudios_moscow1.sqlite'):
        if is_manual == 1:
            print('Data base is not exists. Do you want to create DataBase (y/n)? ')
            answer = input().lower()
        else: 
            answer == 'y'
        
        if answer == 'y':
            conn = sqlite3.connect(directory + 'photostudios_moscow1.sqlite')
            conn.close()
            print('DataBase is created')
        elif answer != 'n':
            print('Error in input!')
            return list()


Присоединяемся к БД и сохраняем функцию курсок отдельной переменной:
    conn = sqlite3.connect(directory + 'photostudios_moscow1.sqlite')
    cur = conn.cursor() 


Следующим действием проверяем, все ли таблицы созданы. Если нет, создаем недостающие. В конце изменений завершаем транзакцию:
    tables = [x[0] for x in list(cur.execute('SELECT name FROM sqlite_master WHERE type="table"'))]
    if not ('uStudios' in tables) & ('uHalls' in tables) & ('uBooking_parsing' in tables) & ('uBooking' in tables):
        if is_manual == 1:
            print('Do you want to create missing tables (y/n)? ')
            answer = input().lower()
        else:
            answer = 'y'
        
        if anwer == 'y':
            if not ('uStudios' in tables):
                create_tables(conn, table = 'uStudios')

            if not ('uHalls' in tables):
                create_tables(conn, table = 'uHalls')

            if not ('uBooking_parsing' in tables):
                create_tables(conn, table = 'uBooking_parsing')

            if not ('uBooking' in tables):
                create_tables(conn, table = 'uBooking')

        elif answer != 'n':
            print('Error in input!')
            return list()

    conn.commit()


Обновляем список фотостудий. Сравниваем с данными БД и выводим количество и список новых фотостудий:
    studios = studio_list()
    new_studios = studios[[x not in list(db_to_studios(conn).index) for x in list(studios.index)]]
    if len(new_studios) > 0:
        print(str(len(new_studios)) + ' new studios detected: \n' + str(list(new_studios['name'])))
        studios_to_db(conn, new_studios)


conn.commit()

Обновляем список залов и выводим название новых:
    halls = hall_list(list(studios.index)).sort_index()
    new_halls = halls[[x not in list(db_to_halls(conn).index) for x in list(halls.index)]]
    if len(new_halls) > 0:
        halls_to_db(conn, new_halls)
    
    conn.commit()


Обновляем информацию по бронированию в таблице uBooking_parsing. Скрипт получился простым, т.к. всю сложную работу мы сделали в самой процедуре booking_to_db
    booking_to_db(conn, halls.index)   
    
    conn.commit()


Обновляем актуальную информацию по бронированию в таблице uBooking. Для этого удаляем старую версию uBooking и записываем данные из таблицы uBooking_parsing с последними (для каждого зала, даты и часа брони) датами парсинга:
    cur.execute('DELETE FROM uBooking')
    cur.execute('''
        insert into uBooking (hall_id, date, hour, is_working_hour, min_hours, price, is_booked, duration, parsing_date) 
        select hall_id, date, hour, is_working_hour, min_hours, price, is_booked, duration, parsing_date
        from
        (
            select *, row_number() over(partition by hall_id, date, hour order by parsing_date desc) rn 
            from uBooking_parsing
        ) t
        where rn = 1
    ''')
    conn.commit()


Обновляем даты открытия студий и залов:
    update_open_dates(conn)
    conn.commit()


И закрываем базу
    conn.close()


Парсинг с сохранением данных в БД настроен успешно!

Инициируем парсинг/обновление следующей процедурой:
update_parsing()


Итог


В данной и предыдущей статьях мы рассмотрели алгоритм работы парсинга открытой информацией по фотостудиям. Полученные данные собрали в базу данных.

В следующей статье рассмотрим примеры анализа полученных данных.

Готовый проект вы можете найти на моей странице в github.
Tags:
Hubs:
+3
Comments 3
Comments Comments 3

Articles