Пролог
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!
