Пролог
Hello, ladies and gentlemen!
Решил немного поупражняться в публицистике, так сказать, и написать свою первую статью. Так что не судите строго.
Последние несколько месяцев львиную долю свободного времени я тратил на поиски работы/стажировки тестеровщиком. Дело это как выяснилось довольно утомительное и неблагодарное, к тому же за праздным времяпрепровождением начинаешь потихоньку забывать базу, так что решил с этим делом притормозить, слегка развеяться и кое-что повспоминать.
Понятие "база", конечно, растяжимое. Для себя я в данном случае взял Python и SQL.
Поскольку речь в статье пойдет о сборе данных, а именно о парсинге/скраппинге (кто как предпочитает), хранении и простом анализе этих данных, в контексте Python будут упомянуты такие популярные библиотеки, как Playwright, Requests, bs4, и psycopg2(надо же куда-то складывать).
Этот незамысловатый инструментарий будет направлен на создание небольшой базы данных спортивной статистики по баскетболу, а точнее трех таблиц с которыми можно по всякому поиграться.
Начало
Мне с точки зрения UI понравился сервис flashscore.com. Они не особо парятся по поводу защиты от парсинга, поэтому не нужно корячиться со всякими selenium-stealth и fakeuseragent-ами. У него есть отдельный клон только для баскетбола - basketball24.com.
Для пользы дела потребуется:
Python
Playwright
requests + bs4 (для экономии времени, можно в общем-то без них)
psycopg2 (поскольку я использую postgresql)
Несколько родных модулей из серии json/os/sys и т.д.
Итак, буду делать три таблицы такого рода:1.championships:
id (оригинального для каждой лиги)
country (страна лиги)
gender (мужской и женский)
league (название лиги)
link (ссылка на архив со статистикой)
2.matches:
match_id (оригинального для каждого матча)
league_id (id из таблицы championships)
match_date (дата встречи)
start_time (время начала встречи, у меня UTC +01:00)
team_home (название домашней команды)
team_away (название команды на выезде)
league_name (то же что и league из championships)
stage (этап турнира)
home_score (очки домашней команды, включая овертайм)
away_score (очки гостевой команды, включая овертайм)
home_score_ft (очки домашней команды в основное время)
away_score_ft (очки гостевой команды в основное время)
total_ft (очки обеих команд в основное время)
3.details:
match_id (оригинального для каждого матча, из таблицы matches)
home_q1 ... away_ot (10 столбцов с набранными очками каждой командой в каждой четверти, а также в овертайме, если он был)
home_win (коэффициент на победу домашней команды)
away_win (коэффициент на победу гостевой команды)
total (значение среднего тотала на равные коэффициенты*)
handicap (значение средней форы/гандикапа на равные коэффициенты* по итогу матча)
hc_q1 (значение средней форы/гандикапа на равные коэффициенты* по итогу первой четверти)
*равные коэффициенты, напримере объясню что имеется ввиду - на матч дают средний тотал 165.5 , то есть на Тотал Меньше 165.5(ТМ) дают коэффициент 1.87 и на Тотал Больше 165.5(ТБ) - коэффициент 1.87, то же и с форами, иногда можно встретить разные варианты - 1.86 и 1.88, 1.89 и 1.85 , 1.9 и 1.87 соответственно(звисят от маржи букмекера и величины рынка, главное чтобы были близки по значению на противоположные исходы).
championships id | country | gender | league | link ----+---------+--------+-----------+------------------------------------------------
matches match_id | league_id | match_date | start_time | team_home | team_away | league_name | stage | home_score | away_score | home_score_ft | away_score_ft | total_ft ----------+-----------+------------+------------+-----------+-----------+-------------+-------+------------+------------+---------------+---------------+----------
details match_id | home_q1 | away_q1 | home_q2 | away_q2 | home_q3 | away_q3 | home_q4 | away_q4 | home_ot | away_ot | home_win | away_win | total | handicap | hc_q1 ----------+---------+---------+---------+---------+---------+---------+---------+---------+---------+---------+----------+----------+-------+----------+-------
Наработка №1
Начнем с первой таблицы - championships.
На basketball24.com представлено огромное количество чемпионатов из всевозможных стран, поэтому вручную придется выбрать наиболее интересные
В итоге я увлекся и насобирал 101 мужской чемпионат и 48 женских. Это первые и вторые национальные лиги и европейские межклубные первенства. Я скопировал ссылку на каждый чемпионат и положил в виде двух списков: отдельно мужских и женских
#bb_links.py links_men = [ 'https://www.basketball24.com/albania/superliga/#/QyhUb9xl/table/overall', 'https://www.basketball24.com/argentina/liga-a/#/KbRL3M3T/table/overall', 'https://www.basketball24.com/australia/nbl/#/YZEnCJej/table/overall', ... ] links_women = [ 'https://www.basketball24.com/argentina/liga-femenina-women/#/UgCeKslA/draw', 'https://www.basketball24.com/australia/wnbl-women/#/IBGTDowC/table/overall', 'https://www.basketball24.com/austria/superliga-women/#/Sx8xPi88/table/overall', ... ]
Фреймворк, при переходе с главной страницы, дорисовывает всякую ерудну (по типу каких-то эндпоинтов, вроде "#/Sx8xPi88/table/overall") - ничего страшного, потом это дело исправим.
Можно приступать к созданию первой таблицы championships:
Примерная структура всего проекта (так сказать) выглядит так:
.
├── championships.py
├── link_collector.py
├── main_run.py
├── failed_run.py
├── details
│ ├── all_champs.json
│ ├── bb_links.py
│ └── main_selectors.py
│ ├── txt_links
│ │ └── austria-superliga.txt
│ │ └── failed
│ │ └── japan-b-league.txt
├── sport_handlers
│ ├── basketball_handler.py
│ └── main_handler.py
└── tests
├── test_leagues_actuality.py
└── test_slector_validity.py
Если кому-то понадобится ссылка на всю репу - тут.
#championships.py import os import json import psycopg2 from .details.bb_links import links_men, links_women ''' Class create json and sql table with basketball championships(and link for them to basketball24.com choosen by yourself manually before ''' class Championships: def __init__(self, sport=None): self.all_leagues = {} self.sport = sport print(self.sport) def add_league(self, data, gender): for l in data: country = l.split('/')[3].replace('-', ' ').title() league = l.split('/')[4].replace('-', ' ').upper() link = 'https:/' + '/'.join(l.split('/')[1:5]) if country not in self.all_leagues: self.all_leagues[country] = {'Men': {}, 'Women': {}} if league not in self.all_leagues[country][gender]: self.all_leagues[country][gender][league] = link def process_data(self, data_men, data_women): self.add_league(data_men, 'Men') self.add_league(data_women, 'Women') def save_to_json(self, filename='all_champs.json'): folder_path = 'details' file_path = os.path.join(folder_path, filename) json_data = json.dumps(self.all_leagues, indent=4) with open(file_path, 'w') as file: file.write(json_data) def connect_to_db(self, dbname): return psycopg2.connect( host="127.0.0.1", user="postgres", password="123456er", port="5432", dbname=dbname ) def create_postgresql_db(self, dbname): conn = self.connect_to_db(dbname) cur = conn.cursor() # Проверка - есть база или нет cur.execute(f"SELECT 1 FROM pg_catalog.pg_database WHERE datname = '{dbname}'") database_exists = cur.fetchone() # Если базы данных нет, создаю новую if not database_exists: cur.execute(f"CREATE DATABASE {dbname}") print(f"Database '{dbname}' created successfully") conn.close() # Подключаюсь к созданной или существующей базе conn = self.connect_to_db(dbname) cur = conn.cursor() # Проверка таблицы в базе cur.execute("SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'championships')") table_exists = cur.fetchone()[0] # Если таблицы нет, создаю новую if not table_exists: create_table_query = ''' CREATE TABLE championships ( id SERIAL PRIMARY KEY, country VARCHAR(100), gender VARCHAR(10), league VARCHAR(100), link VARCHAR(255), CONSTRAINT unique_country_gender_league UNIQUE (country, gender, league) ) ''' cur.execute(create_table_query) conn.commit() print("Table created successfully") for country, genders in self.all_leagues.items(): for gender, leagues in genders.items(): for league, link in leagues.items(): insert_query = f''' INSERT INTO championships (country, gender, league, link) VALUES ('{country}', '{gender}', '{league}', '{link}') ON CONFLICT (country, gender, league) DO NOTHING ''' cur.execute(insert_query) conn.commit() conn.close() championships = Championships() championships.process_data(links_men, links_women) championships.save_to_json() championships.create_postgresql_db(dbname="sportdb")
В этом модуле один класс, который призван обработать все ранее собранные руками ссылки, распределить их по чемпионатам, сохранить в JSON (он пригождается иногда подергать линки вручную) и создать (если не создана заранее) базу данных sportdb, а также создать таблицу championships (пришлось заколхозить троекратным for, итераций мало - так что все равно).
Вышло так:
{ "Albania": { "Men": { "SUPERLIGA": "https://www.basketball24.com/albania/superliga" }, "Women": {} }, "Argentina": { "Men": { "LIGA A": "https://www.basketball24.com/argentina/liga-a" }, "Women": { "LIGA FEMENINA WOMEN": "https://www.basketball24.com/argentina/liga-femenina-women" } }, "Australia": { "Men": { "NBL": "https://www.basketball24.com/australia/nbl" }, "Women": { "WNBL WOMEN": "https://www.basketball24.com/australia/wnbl-women" } ... }
Поскольку бд в теории планируется держать актуальной, а ссылки и названия чемпионатов имеют свойсво меняться по разным причинам(напрмер, сменился главный спонсор), нужно иногда проверять актуальность ссылки.
Для такой задачи я решил написать небольшой тестик с помощью pytest и requests+bs4. И зараядить это дело на github actions, выполняться раз в неделю сутра по понедельникам. Поскольку связать базу данных на ноутбуке с actions - дело хлопотное, оформил в список все, что хочу проверять.
#bb_to_test.py leagues = [ ('SUPERLIGA', 'https://www.basketball24.com/albania/superliga'), ('LIGA A', 'https://www.basketball24.com/argentina/liga-a'), ('LIGA FEMENINA WOMEN', 'https://www.basketball24.com/argentina/liga-femenina-women'), ... ]
#test_leagues_actuality.py import requests from bs4 import BeautifulSoup import pytest from details import bb_to_test # Функция для сравнения строк, игнорируя символы, кроме букв и цифр def clean_string(s): return ''.join(filter(str.isalnum, s)).lower() # Проверка что ссылка на лигу актуальна и назваеие лиги не поменяли @pytest.mark.parametrize("league, link", bb_to_test.leagues, ids=lambda item: item) def test_links_match(league, link): response = requests.get(link) assert response.status_code == 200 soup = BeautifulSoup(response.text, 'html.parser') element = soup.find(class_='heading__name').text.strip().lower() assert clean_string(element) == clean_string(league)
Проверяю, что ответ успешный (200), значит, ссылка жива, и пытаюсь проверить, что чемпионат не поменялся. Сравниваю название лиги из ссылки и её реальное название на странице (да, такое возможно, что различаются).
Шлепнут, для начала, у себя в терминале, разумеется, - посмотреть, как там дела:
pytest .\test_leagues_actuality.py -v .... test_leagues_actuality.py::test_links_match[SUPERLIGA-https://www.basketball24.com/albania/superliga] PASSED [ 0%] test_leagues_actuality.py::test_links_match[LIGA A-https://www.basketball24.com/argentina/liga-a] PASSED [ 1%] test_leagues_actuality.py::test_links_match[LIGA FEMENINA WOMEN-https://www.basketball24.com/argentina/liga-femenina-women] PASSED ...
name: Run Tests on: push: branches: - master schedule: - cron: '0 7 * * 1' #запуск каждый понедельник в 7 утра jobs: test: runs-on: windows-latest #на убунте тесты проходят несколько быстрее steps: - name: Checkout repository uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: 3.10.7 - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - name: Set up Playwright run: | python -m playwright install - name: Run tests run: pytest
В дальнейшем будет добавлен еще один тест для проверки актуальности селекторов (которые будут нужны при парсинге), поэтому тут в YAML-файле есть строка python -m playwright install.
Playwright
Дальше по ссылкам уже придется переходить и работать с открытыми страницами в браузере. Самый популярный инструмент для этого - Selenium, это действительно отличная штука, но меня воротит от постоянного обновления webdriver-а, а метод для обновления его автоматом, например, на Ubuntu, я так и не постиг, чтобы это было быстро. Поэтому мне ближе Playwright, который создает свой браузер. В идеале использовать TypeScript с этой библиотекой, но мне проще Python. Я буду писать код в слегка маргинальном стиле, без изысков, оптимизаций, аннотаций типов и т.д., за это - сорри.
По хорошему, следует использовать Playwright в асинхронном стиле, это ускоряет и парсинг, и разработку, но у меня древний Lenovo B590 с 4 ГБ оперативки и очень нестабильный интернет, поэтому преследовать скорость для меня лишено логики, и все будет написано в sync.
Для начала нам снова нужно собрать ссылки, но теперь для каждого чемпионата по каждому сезону. Я взял текущий сезон и четыре предыдущих.
#link_collector.py import requests import os from bs4 import BeautifulSoup from playwright.sync_api import sync_playwright, TimeoutError from details.main_selectors import Selectors import sys class SeasonsCollector: def __init__(self, link): self.champ = link self.all_links = None self.seasons = 5 def get_links(self): archive_link = f'{self.champ}/archive' response = requests.get(archive_link) soup = BeautifulSoup(response.text, 'html.parser') pre_link = f"{'/'.join(self.champ.split('/')[:-1])}/" self.all_links = [pre_link + i.get('href').split('/')[2] for cnt, i in enumerate(soup.select('.archive__season a')) if cnt < self.seasons] return self.all_links class SeasonsHandler: def __init__(self, links): self.links = links self.browser = sync_playwright().start().chromium.launch(headless=False) self.context = self.browser.new_context() self.pages = [] def open_pages(self): for link in self.links: page = self.context.new_page() page.goto(link) self.pages.append(page) print(self.pages) def click_to_bottom(self): for page in self.pages: while True: try: elements = page.query_selector_all(Selectors.show_more) if len(elements) == 0: break elements[0].click() page.wait_for_load_state('networkidle', timeout=300000) except TimeoutError: continue import time time.sleep(1) def get_links_to_matches(self, sport_selector): all_links = [] for page in self.pages: all_matches = page.query_selector_all(sport_selector) for match in all_matches: match_id = match.get_attribute('id') href = '/'.join(f'{page.url}'.split('/')[:3]) + '/match/' + match_id[4:] all_links.append(href) print(len(all_links)) return all_links def close_all(self): for page in self.pages: page.close() self.context.close() self.browser.close() print('browser has been closed') def link_collerctor(url): links_champ = SeasonsCollector(url).get_links() handler = SeasonsHandler(links_champ) handler.open_pages() handler.click_to_bottom() matches_links = handler.get_links_to_matches(Selectors.bb_all_matches) handler.close_all() file_name = '-'.join(url.split('/')[-2:]) with open(f"details\\txt_links\\{file_name}.txt", "w") as file: file.write("\n".join(matches_links)) if __name__ == "__main__": if len(sys.argv) != 2: print('Usage: python link_collector.py [url]') sys.exit(1) arg = sys.argv[1] link_collerctor(arg)
В этом модуле класс SeasonsCollector собирает ссылки на последние 5 (self.seasons = 5) сезонов - текущий и 4 прошлых. Класс SeasonsHandler собирает ссылки на каждый матч в каждом сезоне. Основная проблема в том, что матчи каждого сезона сразу отображаются не полностью, а порционно, и нужно прокликать несколько раз на кнопку "Show more matches", чтобы дойти до начала сезона.
В модуле используются готовые селекторы, я их поискал заранее и все необходимые сложил в модуль main_selectors.py
#main_selectors.py class Selectors: bb_all_matches = "[id^='g_3']" # scoreline = '.smh__template' team_home = 'div.participant__participantName:nth-child(2)' team_away = 'div.participant__participantName:nth-child(1)' tournament = '.tournamentHeader__country' date_and_time = '.duelParticipant__startTime' final_score = '.detailScore__wrapper' show_more = '.event__more.event__more--static' fulltime_score = '.detailScore__fullTime' class CustomIndexedList(list): #Pardon my french ''' TAKE CARE !!! To ensure that quarters in basketball, periods in hockey, and halves in football and handball correspond to their actual values rather than the standard indexing, an offset of 1 is applied. ''' def __getitem__(self, index): if 1 <= index <= 5: # Adjust the index to start from 1 instead of 0 return super().__getitem__(index - 1) else: raise IndexError("Index must be between 1 and 5 inclusive.") home_part = CustomIndexedList(map(lambda i: f'.smh__part.smh__home.smh__part--{i}', range(1, 6))) away_part = CustomIndexedList(map(lambda i: f'.smh__part.smh__away.smh__part--{i}', range(1, 6))) total_button = '[title="Over/Under"]' handicap_button = '[title="Asian handicap"]' home_away = '[title="Home/Away"]' coef_box = '.ui-table__body' # This selectors often change by site owners odds_on_bar = 'button._tab_33oei_5:has-text("Odds")' handicap_quarter1 = 'button._tab_33oei_5:has-text("1st Qrt")'
Поскольку индексация четвертей в баскетболе начинается с первой, для наглядности изменил способ индексации для данных по четвертям (что в итоге не особо пригодилось и вообще наверное лишено смысла).
Селектор odds_on_bar, опять-таки, имеет свойство меняться с течением времени, поэтому проверяем тестом:
#test_selector_validity.py import pytest from playwright.sync_api import sync_playwright odds_on_bar = 'button._tab_33oei_5:has-text("Odds")' @pytest.mark.parametrize("link", [ 'https://www.basketball24.com/match/v5QYCKw8', 'https://www.basketball24.com/match/Aeu3mByg', 'https://www.basketball24.com/match/WWG7ksfA' ]) def test_selector_presence(link): with sync_playwright() as p: browser = p.chromium.launch() context = browser.new_context() page = context.new_page() page.goto(link) selector_exists = page.query_selector(odds_on_bar) is not None assert selector_exists, f"Selector {odd_on_bar} not found on the page {link}"
Функция link_collector создает необходимые объекты, которые собирают все нужные ссылки. В конце закрывают все вкладки и браузер, а затем сохраняют построчно ссылки в txt-файлик. link_collector.py будет вызываться как внешний процесс из другого мод��ля, с аргументом в виде URL.
В результате у нас есть ссылки на все матчи последних пяти сезонов одного из выбранных чемпионатов, идем дальше. Каждая из этих ссылок имеет такой вид:
Для обработкаи ссылок и записи данных создаем два модуля:
#main_handler.py import psycopg2 from playwright.sync_api import sync_playwright from abc import ABC, abstractmethod from details.main_selectors import Selectors class MatchHandler: def __init__(self, links, url): self.links = links self.__url = url self.match_title = None self.scoreline = None self.coefs = None self.league_id = None self.league_name = None self.browser = None self.context = None self.page = None if links != 1: self.get_league_id() def reset_data(self): self.match_title = None self.scoreline = None self.coefs = None def is_match_exists(self, match_data): conn = psycopg2.connect( host="127.0.0.1", user="postgres", password="123456er", port="5432", dbname= ] ) cur = conn.cursor() select_match_query = """ SELECT match_id FROM matches WHERE league_id = %s AND match_date = %s AND team_home = %s AND team_away = %s """ cur.execute(select_match_query, match_data[:4]) match_id = cur.fetchone() conn.close() return match_id is not None def open_browser_and_process_links(self): with sync_playwright() as p: self.browser = p.chromium.launch(headless=False) self.context = self.browser.new_context() self.page_empty = self.context.new_page() self.create_match_tables() for link in self.links: try: self.page = self.context.new_page() self.reset_data() self.page.goto(link) self.page.wait_for_selector(Selectors.scoreline) self.process_title() self.process_scoreline() self.process_coefs() match_data = (self.league_id, self.match_title[1], self.match_title[3], self.match_title[4]) if self.is_match_exists(match_data): print(f"Match already exists: {self.match_title}") continue #Продолжаем -> continue, прекращаем обработку всех оставшихся ссылок(когда обновляю уже накачанную базу) -> break else: print('-----') print(self.match_title) print(self.scoreline) print(self.coefs) print('-----') self.page.close() self.save_to_database(self.match_title, self.scoreline, self.coefs) except Exception as e: print(e) self.save_failed_link(link) print('ok8') self.page.close() continue def save_failed_link(self, link): file_name = '-'.join(self.__url.split('/')[-2:]) with open(f"details\\txt_links\\failed\\{file_name}.txt", 'a') as file: file.write(link + "\n") print('Failed link has been saved') @abstractmethod def create_match_tables(self): pass @abstractmethod def process_scoreline(self): pass @abstractmethod def process_coefs(self): pass @abstractmethod def save_to_database(self, title, scores, coefs): pass def process_title(self): team_home_element = self.page.query_selector(Selectors.team_home) team_away_element = self.page.query_selector(Selectors.team_away) tournament_header = self.page.query_selector(Selectors.tournament) date_header = self.page.query_selector(Selectors.date_and_time) final_score_header = self.page.query_selector(Selectors.final_score) full_time_score = self.page.query_selector(Selectors.fulltime_score) team_home = team_home_element.text_content().strip() if team_home_element else None team_away = team_away_element.text_content().strip() if team_away_element else None tournament = tournament_header.text_content() if tournament_header else None date_and_time = date_header.text_content() if date_header else None final_score = final_score_header.text_content() if final_score_header else None score_ft = full_time_score.text_content() if full_time_score else None stage = self.extract_stage(tournament) date, start_time = self.extract_date_and_time(date_and_time) home_score, away_score, home_score_ft, away_score_ft = self.extract_scores(final_score, score_ft) total_result = home_score_ft + away_score_ft match_data = (self.league_id, date, start_time, team_home, team_away, self.league_name, stage, home_score, away_score, home_score_ft, away_score_ft, total_result) self.match_title = match_data def extract_stage(self, tournament): if tournament: print(tournament) colon_index = tournament.find(":") if colon_index != -1: stage_part = tournament[colon_index + 1:].strip() if '-' in stage_part: stage = stage_part.split('-', 1)[1].strip().upper() if 'ALL' in stage: stage = 'ALL STARS' if "SEMI-FINALS" in stage or "QUARTER-FINALS" in stage or "1/8-FINALS" in stage or "PROMOTION" in stage: stage = "PLAY OFFS" elif "FINAL" in stage: stage = "FINAL" elif "ROUND" in stage or "NBA" in stage: stage = "MAIN" return stage return "MAIN" def extract_date_and_time(self, date_and_time): if date_and_time: date = date_and_time.split()[0] start_time = date_and_time.split()[1] return date, start_time return None, None def extract_scores(self, final_score, score_ft): if final_score: home_score = int(final_score.split('-')[0]) away_score = int(final_score.split('-')[1]) else: home_score, away_score = None, None if score_ft is None: home_score_ft, away_score_ft = home_score, away_score else: home_score_ft = int(score_ft.split('-')[0].replace('(','')) away_score_ft = int(score_ft.split('-')[1].replace(')','')) return home_score, away_score, home_score_ft, away_score_ft def get_league_id(self): conn = psycopg2.connect( host="127.0.0.1", user="postgres", password="123456er", port="5432" ) cur = conn.cursor() cur.execute(f"SELECT id, league FROM championships WHERE link = '{self.__url}'") data = cur.fetchall() conn.close() print(data) self.league_id = data[0][0] self.league_name = data[0][1]
И
#basketball_handler.py import psycopg2 from sport_handlers.main_handler import MatchHandler from details.main_selectors import Selectors import time class Basketball(MatchHandler): def __init__(self, links, url): super().__init__(links, url) def create_match_tables(self): conn = psycopg2.connect( host="127.0.0.1", user="postgres", password="123456er", port="5432" ) cur = conn.cursor() # SQL-запрос для создания таблицы matches create_matches_table_query = """ CREATE TABLE IF NOT EXISTS matches ( match_id SERIAL PRIMARY KEY, league_id INTEGER, match_date DATE, start_time TIME, team_home VARCHAR(255), team_away VARCHAR(255), league_name VARCHAR(255), stage VARCHAR(255), home_score INTEGER, away_score INTEGER, home_score_ft INTEGER, away_score_ft INTEGER, total_ft INTEGER ); """ cur.execute(create_matches_table_query) # SQL-запрос для создания таблицы match_details create_match_details_table_query = """ CREATE TABLE IF NOT EXISTS details ( match_id INTEGER PRIMARY KEY REFERENCES matches(match_id), home_q1 INTEGER, away_q1 INTEGER, home_q2 INTEGER, away_q2 INTEGER, home_q3 INTEGER, away_q3 INTEGER, home_q4 INTEGER, away_q4 INTEGER, home_ot INTEGER, away_ot INTEGER, home_win REAL, away_win REAL, total REAL, handicap REAL, hc_q1 REAL ); """ cur.execute(create_match_details_table_query) try: add_constraint_query = """ ALTER TABLE matches ADD CONSTRAINT unique_match_constraint UNIQUE (league_id, match_date, start_time, team_home, team_away); """ cur.execute(add_constraint_query) except psycopg2.errors.DuplicateTable: pass conn.commit() conn.close() def save_to_database(self, title, scores, coefs): if not title or not scores or not coefs: return conn = psycopg2.connect( host="127.0.0.1", user="postgres", password="123456er", port="5432", ) cur = conn.cursor() try: # SQL запрос для вставки данных в таблицу matches insert_match_query = """ INSERT INTO matches (league_id, match_date, start_time, team_home, team_away, league_name, stage, home_score, away_score, home_score_ft, away_score_ft, total_ft) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING match_id """ cur.execute(insert_match_query, title) match_id = cur.fetchone()[0] # SQL-запрос для вставки данных в таблицу match_details insert_details_query = """ INSERT INTO details (match_id, home_q1, away_q1, home_q2, away_q2, home_q3, away_q3, home_q4, away_q4, home_ot, away_ot, home_win, away_win, total, handicap, hc_q1) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cur.execute(insert_details_query, (match_id,) + scores + coefs) conn.commit() except Exception as e: print('Error with saving') print(e) conn.close() def process_scoreline(self): home_parts = Selectors.home_part away_parts = Selectors.away_part try: home_scores = [self.get_int_score(self.page.query_selector(part)) for part in home_parts] away_scores = [self.get_int_score(self.page.query_selector(part)) for part in away_parts] scores_data = [item for pair in zip(home_scores, away_scores) for item in pair] except Exception as e: scores_data = [None]*10 print('Erorr with scoreline handling', e.with_traceback()) self.scoreline = tuple(scores_data) print(scores_data) def process_coefs(self): home_win = None away_win = None average_total = None average_handicap = None average_handicap_q1 = None def process_coef_button(selector): self.page.wait_for_selector(selector, timeout=3000) element = self.page.query_selector(selector) if element: element.click() self.page.wait_for_selector(Selectors.coef_box) all_coefs = [i.inner_text() for i in self.page.query_selector_all(Selectors.coef_box)] return self.find_average_value(all_coefs) return None try: self.page.query_selector(Selectors.odds_on_bar).click() self.page.wait_for_selector(Selectors.coef_box) except: print('No odds section') pass try: if self.page.query_selector(Selectors.home_away): coefs_text = self.page.query_selector(Selectors.coef_box).inner_text().split()[:2] home_win = float(coefs_text[0]) except: home_win = 1.0 try: if self.page.query_selector(Selectors.home_away): coefs_text = self.page.query_selector(Selectors.coef_box).inner_text().split()[:2] away_win = float(coefs_text[1]) except: away_win = 1.0 try: average_total = process_coef_button(Selectors.total_button) except Exception as e: print(f"Error processing total button: {e}") average_total = None try: average_handicap = process_coef_button(Selectors.handicap_button) except Exception as e: print(f"Error processing handicap button: {e}") average_handicap = None try: average_handicap_q1 = process_coef_button(Selectors.handicap_quarter1) except Exception as e: print(f"Error processing handicap quarter 1 button: {e}") average_handicap_q1 = None coefs = (home_win, away_win, average_total, average_handicap, average_handicap_q1) self.coefs = coefs print(self.coefs, '<-') @staticmethod def find_average_value(coefline): k = None value = None min_diff = 100 for case in coefline: current_total = case.split()[0] k1, k2 = list(map(float, case.split()[1:3])) diff = abs(k1 - k2) if diff < min_diff: min_diff = diff value = current_total print(f"The total/handicap with the smallest diff is: {value}") return float(value) @staticmethod def get_int_score(element): if element: content = element.text_content() try: return int(content) if content else None except ValueError: return None else: return None
В первом модуле основной класс MatchHandler содержит логику обработки матчей и в себе имеет метод process_title. Так как в целом можно парсить не только баскетбол, но и другие виды спорта, а обработка заголовка матча везде одинаковая. Абстрактные же методы для обработки и сохранения данных о счете и коэффициентах для баскетбола реализованы в классе Basketball. Также в MatchHandler есть функция save_failed_link, которая сохраняет в папку failed ссылки, по какой-то причине которые не удалось обработать. Их нужно будет попытаться обработать повторно. В противном случае нарушится хронология событий(матчей). При повторной неудаче, если она произойдет, необходимо будет внести данные вручную в бд.
Запускается все это дело так:
#main_run.py import subprocess from sport_handlers.basketball_handler import Basketball from sport_handlers.main_handler import MatchHandler #добавляем сюда чемпионаты ,которые хотим обработать, жадничать не стоит - #один чемпионат обрабатывается примерно 2-3 часа urls = [ "https://www.basketball24.com/iceland/premier-league", ] for url in urls: subprocess.run(['python', 'link_collector.py', url], check=True) file_name = '-'.join(url.split('/')[-2:]) print(file_name) with open(f"details\\txt_links\\{file_name}.txt", 'r') as file: links = [line.strip() for line in file.readlines()] basketball_handler = Basketball(links, url) basketball_handler.open_browser_and_process_links()
Проблемные ссылки обрабатываем так:
#failed_run.py import subprocess from sport_handlers.basketball_handler import Basketball from sport_handlers.main_handler import MatchHandler urls = [ "https://www.basketball24.com/south-korea/kbl", "https://www.basketball24.com/switzerland/sb-league" ] for url in urls: file_name = '-'.join(url.split('/')[-2:]) print(file_name) with open(f"details\\txt_links\\failed\\{file_name}.txt", 'r') as file: links = [line.strip() for line in file.readlines()] basketball_handler = Basketball(links, url) basketball_handler.open_browser_and_process_links()
Следует учесть что для обработки ошибок, в urls должен быть передан тот чемпионат , ошибки в котором возникли при обработки матчей.
Итого имеем
Я запустил шарманку с рандомно выбранными чемпионатами, коих оказалось 47, и отошёл на несколько дней от ноута подальше))
Результат вышел таким:
Глянул - сколько всего матчей обработалось:
select count(*) from matches;
Вообщем-то все, получился довольно гибкий парсер, который можно подстроить под любой вид спорта и регулировать количество сезонов для обработки
PS
Для затравки, так сказать, на следующую часть, если кому-то, конечно, будет интересно, вот, например,банальный SQL-запрос, который мне показался интересным и наглядным:
SELECT COUNT(CASE WHEN m.total_ft > d.total THEN 1 END) AS total_ft_greater, COUNT(CASE WHEN m.total_ft < d.total THEN 1 END) AS total_ft_less FROM matches m JOIN details d ON m.match_id = d.match_id;
Он показывает, сколько матчей закончилось с тоталом больше, чем был дан букмекером, и с тоталом меньше. И показывает, на дистанции 47386 матчей, насколько филигранно точно букмекер в среднем дает это значение. То есть почти одинаковое число матчей закончилось больше и меньше данного значения.
По приколу расчет: если взять букмекерскую маржу 5%, а это будут коэффициенты 1.90 / 1.90 на тотал больше и тотал меньше соответственно, и проставить по 100 рублей на тотал больше или тотал меньше:
47 386 * 100 = 4 738 600 — ушло на все ставки
1114 * 100 = 111 400 — вернулось возвратом, когда тотал ровно совпал
23 084*100*1.9 = 4 385 960 — ваш выигрыш, при ставке на тотал больше
23 188 * 100 * 1.9 = 4 405 720 — ваш выигрыш, при ставке на тотал меньше
Ваш доход:
4 385 960 + 111 400 — 4 738 600 = -241 240 руб (при ставках на тотал больше)
4 405 720 + 111 400 — 4 738 600 = -221 480 руб (при ставках на тотал меньше)
"Доходы", надо сказать ,скромные).Так что будьте аккуратны при работе со ставками и лучше пытаться практиковаться на верификаторах ставок, для лайва я рекомендую пользоваться bet-hub.com, а для прематч expari.com. Так ваши доходы останутся на достойном значении == 0.
Ну и не все так плохо, как кажется на первый взгляд. SQL в следующей части поможет нам провести эксперименты на исторических данных, чтобы увидеть в графе доход положительные натуральные числа.
Спасибо за внимание, берегите себя, свои нервы и свой кэш! Ciao!
