Python для gambling'a. Часть 1 — Сбор данных

Пролог

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

На примере Франции я возьму 3 чемпионата - Певрую и Вторую лигу у мужчин (LNB и Pro B), и Первую лигу у женщин (LFB Women)

На примере Франции я возьму 3 чемпионата — Певрую и Вторую лигу у мужчин (LNB и Pro B), и Первую лигу у женщин (LFB Women)

В итоге я увлекся и насобирал 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, итераций мало — так что все равно).

Вышло так:

таблица по 35 строку - champioships

таблица по 35 строку — champioships

{
    "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"
        }
...
}

Поскольку бд в теории планируется держать актуаьной, а ссылки и названия чемпионатов имеют свойсво меняться по разным причинам (напрмер, сменился главный спонсор), нужно иногда проверять актуальность ссылки.

вот например чемпионат Щвеции поменял название в сезоне 2018/2019

вот например чемпионат Щвеции поменял название в сезоне 2018/2019

Для такой задачи я решил написать небольшой тестик с помощью 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.

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

Иногда на не очень популярных лига может не быть вкладок ODDS

Иногда на не очень популярных лига может не быть вкладок ODDS

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

#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, и отошёл на несколько дней от ноута подальше))

Результат вышел таким:

таблица matches

таблица matches

таблица details

таблица details

Глянул — сколько всего матчей обработалось:

select count(*) from matches;

получилось примерно по 1000 матчей на чемпионат

получилось примерно по 1000 матчей на чемпионат

Вообщем-то все, получился довольно гибкий парсер, который можно подстроить под любой вид спорта и регулировать количество сезонов для обработки

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;

еще 1114 матчей ровно попало в значения тотала

еще 1114 матчей ровно попало в значения тотала

Он показывает, сколько матчей закончилось с тоталом больше, чем был дан букмекером, и с тоталом меньше. И показывает, на дистанции 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!

© Habrahabr.ru