Анализ матчей Dota 2 с помощью ИИ: часть 1. Готовим данные для обучения ИИ.

# Инициализируем библиотеки
import time
import json
import pandas as pd
from pprint import pprint
from sqlalchemy import create_engine, Column, MetaData, func, NUMERIC, VARCHAR, BOOLEAN
from sqlalchemy.orm import aliased, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from python_opendota.apis.tags import benchmarks_api
import python_opendota
import requests
import openpyxl
# Настройки подключения к базе данных
POSTGRES_USER = os.getenv("POSTGRES_USER", "admin")
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "Init!@#$")
POSTGRES_SERVER = os.getenv("POSTGRES_SERVER", "127.0.0.1")
POSTGRES_PORT = os.getenv("POSTGRES_PORT", "5432")
POSTGRES_DB = os.getenv("POSTGRES_DB", "dota2")
DATABASE_URL = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_SERVER}:{POSTGRES_PORT}/{POSTGRES_DB}"
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
session = Session()
# Создаем нашу модель
Base = declarative_base()
class MatchInDb(Base):
    __tablename__ = "matches"
    match_id = Column(NUMERIC, primary_key=True)
    duration = Column(NUMERIC)
    start_time = Column(NUMERIC)
    radiant_team_id = Column(NUMERIC)
    radiant_name = Column(VARCHAR)
    dire_team_id = Column(NUMERIC)
    dire_name = Column(VARCHAR)
    leagueid = Column(NUMERIC)    
    league_name = Column(VARCHAR)
    series_id = Column(NUMERIC)
    series_type = Column(NUMERIC)
    radiant_score = Column(NUMERIC)
    dire_score = Column(NUMERIC)
    radiant_win = Column(BOOLEAN)
    
class TeamInDb(Base):
    __tablename__ = "teams"
    
    team_id = Column(NUMERIC, primary_key=True)
    rating = Column(NUMERIC)
    wins = Column(NUMERIC)
    losses = Column(NUMERIC)
    last_match_time = Column(NUMERIC)
    name = Column(VARCHAR)
    tag = Column(VARCHAR)
    logo_url = Column(VARCHAR)
# Объявляем переменные для request
base_url = "http://api.opendota.com/api"
teams_url = base_url + "/teams"
pro_matches_url = base_url + "/proMatches"
# Получаем минимальный match_id
less_than_match_id = session.query(func.min(MatchInDb.match_id)).scalar()
if less_than_match_id is not None:
    pro_matches_url += f"?less_than_match_id={less_than_match_id}"
number_of_matches_to_add = 10000
number_of_matches_in_request = 100
iterations = int(number_of_matches_to_add / number_of_matches_in_request)
for i in range(iterations):
    less_than_match_id = session.query(func.min(MatchInDb.match_id)).first()[0]
    pro_matches_url = base_url + "/proMatches" + "?less_than_match_id=" + str(less_than_match_id)
    pro_matches_result = requests.get(pro_matches_url)
    pro_matches_result_json = pro_matches_result.json()
    # Сохраните все матчи в одной транзакции
    matches = [MatchInDb(**pro_match_json) for pro_match_json in pro_matches_result_json]
    session.add_all(matches)
    session.commit()
    time.sleep(1)
    print("Итерация завершена: ", i)
        
    i -= 1
# Получаем список команд
team_ids_query = session.query(MatchInDb.radiant_team_id.label("team_id"))\
.outerjoin(TeamInDb, MatchInDb.radiant_team_id == TeamInDb.team_id)\
.filter(TeamInDb.team_id == None)\
.filter(MatchInDb.radiant_team_id != None)
team_ids_query = team_ids_query.union_all(
    session.query(MatchInDb.dire_team_id.label("team_id"))\
    .outerjoin(TeamInDb, MatchInDb.dire_team_id == TeamInDb.team_id)\
    .filter(TeamInDb.team_id == None)\
    .filter(MatchInDb.dire_team_id != None)
)
# Используем subquery для создания подзапроса
team_ids_subquery = team_ids_query.subquery()
# Выполняем основной запрос
team_ids = session.query(team_ids_subquery.c.team_id).all()
# Переменная для задержки
delay = 1
for team_id in team_ids:
    # Получаем URL
    team_id = team_id[0]
    team_url = teams_url + "/" + str(team_id)
    # Получаем информацию по команде
    team_result = requests.get(team_url)
    # Превращаем их в JSON
    team_result_json = json.loads(team_result.text)
    # Сохраняем в БД
    try:
        team_data = {key: value for key, value in team_result_json.items()}
        team = TeamInDb(**team_data)
        session.add(team)
        session.commit()
        session.refresh(team)
    except Exception as e:
        print(f"Ошибка при сохранении команды: {str(e)}")
        continue
    # Засыпаем
    time.sleep(delay)
# Выгружаем данные в Excel
team_radiant = aliased(TeamInDb)
team_dire = aliased(TeamInDb)
result = session.query(team_radiant.rating, team_dire.rating, MatchInDb.radiant_win)\
.join(team_radiant, MatchInDb.radiant_team_id == team_radiant.team_id)\
.join(team_dire, MatchInDb.dire_team_id == team_dire.team_id)\
.all()
# Выгружаем данные в DataFrame
df = pd.DataFrame(result, columns=['radiant_rating', 'dire_rating', 'radiant_win'])
# Сохраняем DataFrame в Excel
df.to_excel('dota2_training_data.xlsx', index=False)