200 lines
11 KiB
Python
200 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import tempfile
|
||
import unittest
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from sqlalchemy import func, select
|
||
|
||
from hockey_data.config import HockeySettingsStore
|
||
from hockey_data.match_details_parser import parse_players_xml
|
||
from hockey_data.models import (
|
||
Game,
|
||
GameTeamStatistic,
|
||
PlayerSeasonStatistic,
|
||
TeamSeasonStatistic,
|
||
Tournament,
|
||
TournamentStandingRow,
|
||
TournamentStatisticRow,
|
||
)
|
||
from hockey_data.service import HockeyDataService
|
||
from hockey_data.standings_parser import parse_standings_xml
|
||
from hockey_data.tournament_statistics_parser import parse_tournament_statistics_xml
|
||
from support import LocalTestDatabase
|
||
|
||
|
||
PLAYERS_XML = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||
<PlayerStatsList tournamentId="1369">
|
||
<PlayerStats id="101" pos="f" name="Ivanov Ivan" nameen="Ivanov Ivan"
|
||
clubidt="14927" clubidg="1" teamname="Team A" teamnameen="Team A" jn="17"
|
||
gp="63" g="11" a="14" pts="25" pim="24" pm="5" sog="111" sog_pct="9.9"
|
||
hits="22" bls="50" fo="100" fow="55" fo_pct="55.0" toi_avg="14:54" />
|
||
<PlayerStats id="202" pos="g" name="Petrov Pyotr" nameen="Petrov Pyotr"
|
||
clubidt="14928" clubidg="2" teamname="Team B" teamnameen="Team B" jn="30"
|
||
gp="40" w="25" l="10" sop="5" sog="1000" ga="80" sv="920" sv_pct="92.0" gaa="2.00" so="5" toi="2400:00" />
|
||
</PlayerStatsList>'''
|
||
|
||
STANDINGS_XML = '''<?xml version="1.0" encoding="UTF-8"?>
|
||
<Standings idtournament="1369">
|
||
<League name="Общая" name_en="League">
|
||
<Team id="14927" clubid="1" name="Команда А" nameen="Team A" rank="1" gp="10" w="7" otw="1" l="2" pts="17" pts_pct="85.0" gf="31" ga="18" pim="60" />
|
||
<Team id="14928" clubid="2" name="Команда Б" nameen="Team B" rank="2" gp="10" w="6" sow="1" l="3" pts="15" pts_pct="75.0" gf="29" ga="22" pim="70" />
|
||
</League>
|
||
<Conferences><Conference item="west" name="Запад" name_en="West">
|
||
<Team id="14927" clubid="1" name="Команда А" nameen="Team A" rank="1" gp="10" w="7" pts="17" gf="31" ga="18" />
|
||
</Conference></Conferences>
|
||
</Standings>'''
|
||
|
||
POWERPLAY_XML = '''<?xml version="1.0" encoding="UTF-8"?>
|
||
<PowerPlay idtournament="1369">
|
||
<Teams name="Большинство" nameen="Power play">
|
||
<Team id="14927" name="Команда А" nameen="Team A" rank="1" gp="10" ppg="8" ppo="30" pp_pct="26.7" />
|
||
<Team id="14928" name="Команда Б" nameen="Team B" rank="2" gp="10" ppg="6" ppo="31" pp_pct="19.4" />
|
||
</Teams>
|
||
</PowerPlay>'''
|
||
|
||
RANK_XML = '''<?xml version="1.0" encoding="UTF-8"?>
|
||
<Rank idtournament="1369">
|
||
<Category id="points" name="Бомбардиры" nameen="Points leaders">
|
||
<Player id="101" rank="1" number="17" name="Иванов Иван" nameen="Ivan Ivanov" team="Команда А" teamen="Team A" gp="10" g="5" a="7" pts="12" />
|
||
</Category>
|
||
</Rank>'''
|
||
|
||
|
||
class NormalizedStatisticsStorageTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.base_dir = Path(self.tmp.name)
|
||
self.database = LocalTestDatabase(self.base_dir / "test.sqlite3")
|
||
self.database.create_all()
|
||
self.settings = HockeySettingsStore(self.base_dir / "settings")
|
||
self.service = HockeyDataService(base_dir=self.base_dir, database=self.database, settings=self.settings)
|
||
with self.database.session() as session:
|
||
session.add(Tournament(external_id="1369"))
|
||
|
||
def tearDown(self) -> None:
|
||
self.database.engine.dispose()
|
||
self.tmp.cleanup()
|
||
|
||
def test_player_season_stats_are_normalized(self) -> None:
|
||
profiles = parse_players_xml(PLAYERS_XML)
|
||
now = datetime.utcnow()
|
||
with self.database.session() as session:
|
||
self.service._store_tournament_directory_profiles(
|
||
session,
|
||
tournament_external_id="1369",
|
||
player_profiles=profiles,
|
||
referee_profiles={},
|
||
coach_profiles={},
|
||
countries=None,
|
||
now=now,
|
||
)
|
||
with self.database.session() as session:
|
||
rows = list(session.scalars(select(PlayerSeasonStatistic).order_by(PlayerSeasonStatistic.player_external_id)))
|
||
self.assertEqual(len(rows), 2)
|
||
skater, goalie = rows
|
||
self.assertEqual(skater.games, 63)
|
||
self.assertEqual(skater.points, 25)
|
||
self.assertEqual(skater.faceoff_pct, 55.0)
|
||
self.assertEqual(goalie.role, "goalkeeper")
|
||
self.assertEqual(goalie.wins, 25)
|
||
self.assertEqual(goalie.saves, 920)
|
||
self.assertEqual(goalie.save_pct, 92.0)
|
||
|
||
def test_game_team_stats_are_stored_for_total_and_periods(self) -> None:
|
||
now = datetime.utcnow()
|
||
raw_teams = {
|
||
"A": {"shots": 31, "goals": 3, "fo": 50, "fow": 28, "fow_pct": "56.0", "hits": 14, "bls": 9, "pim": 6, "toa": "14:20"},
|
||
"B": {"shots": 27, "goals": 2, "fo": 50, "fow": 22, "fow_pct": "44.0", "hits": 18, "bls": 12, "pim": 8, "toa": "11:05"},
|
||
}
|
||
raw_periods = {
|
||
"A": {"1": {"shots": 10, "goals": 1}, "2": {"shots": 12, "goals": 2}},
|
||
"B": {"1": {"shots": 8, "goals": 0}, "2": {"shots": 11, "goals": 1}},
|
||
}
|
||
with self.database.session() as session:
|
||
game = Game(external_id="900001", tournament_external_id="1369", home_team_external_id="1", away_team_external_id="2")
|
||
session.add(game)
|
||
session.flush()
|
||
created = self.service._store_game_team_statistics(session, game=game, raw_teams=raw_teams, raw_periods=raw_periods, now=now)
|
||
self.assertEqual(created, 6)
|
||
with self.database.session() as session:
|
||
home_total = session.scalar(select(GameTeamStatistic).where(GameTeamStatistic.game_external_id == "900001", GameTeamStatistic.side == "home", GameTeamStatistic.segment_key == "total"))
|
||
self.assertIsNotNone(home_total)
|
||
self.assertEqual(home_total.shots, 31)
|
||
self.assertEqual(home_total.faceoffs_won, 28)
|
||
self.assertEqual(home_total.time_on_attack, "14:20")
|
||
|
||
def test_standings_powerplay_and_rank_are_normalized(self) -> None:
|
||
now = datetime.utcnow()
|
||
standings = parse_standings_xml(STANDINGS_XML, language="ru")
|
||
powerplay = parse_tournament_statistics_xml(POWERPLAY_XML, resource_type="powerplay", language="ru")
|
||
rank = parse_tournament_statistics_xml(RANK_XML, resource_type="rank", language="ru")
|
||
with self.database.session() as session:
|
||
self.service._store_normalized_standings(session, tournament_external_id="1369", parsed=standings, now=now)
|
||
self.service._store_normalized_tournament_statistic_rows(session, tournament_external_id="1369", resource_type="powerplay", parsed=powerplay, now=now)
|
||
self.service._store_normalized_tournament_statistic_rows(session, tournament_external_id="1369", resource_type="rank", parsed=rank, now=now)
|
||
with self.database.session() as session:
|
||
standing_count = session.scalar(select(func.count(TournamentStandingRow.id)))
|
||
pp_count = session.scalar(select(func.count(TournamentStatisticRow.id)).where(TournamentStatisticRow.resource_type == "powerplay"))
|
||
rank_count = session.scalar(select(func.count(TournamentStatisticRow.id)).where(TournamentStatisticRow.resource_type == "rank"))
|
||
team_a = session.scalar(select(TeamSeasonStatistic).where(TeamSeasonStatistic.tournament_external_id == "1369", TeamSeasonStatistic.team_entry_external_id == "14927"))
|
||
self.assertEqual(standing_count, 3)
|
||
self.assertEqual(pp_count, 2)
|
||
self.assertEqual(rank_count, 1)
|
||
self.assertIsNotNone(team_a)
|
||
self.assertEqual(team_a.points, 17)
|
||
self.assertEqual(team_a.power_play_goals, 8)
|
||
self.assertEqual(team_a.power_play_opportunities, 30)
|
||
self.assertAlmostEqual(team_a.power_play_pct, 26.7)
|
||
|
||
status = self.service.normalized_storage_status(tournament_external_id="1369")
|
||
self.assertEqual(status["tables"]["standings_rows"]["rows"], 3)
|
||
self.assertEqual(status["resource_rows"]["powerplay"], 2)
|
||
self.assertEqual(status["resource_rows"]["rank"], 1)
|
||
|
||
|
||
def test_repeated_powerplay_team_sections_do_not_duplicate_team_season_rows(self) -> None:
|
||
now = datetime.utcnow()
|
||
standings = parse_standings_xml(STANDINGS_XML, language="ru")
|
||
# Real Stat2TV powerplay resources can repeat the same team in several
|
||
# sections. The actual PP section uses pp/ppgf/pp_pct rather than ppo/ppg.
|
||
parsed = {
|
||
"sections": [
|
||
{
|
||
"id": "goals-for",
|
||
"names": {"ru": "Голы", "en": "Goals"},
|
||
"periods": {},
|
||
"rows": [
|
||
{"id": "1", "name": "Команда А", "names": {"ru": "Команда А", "en": "Team A"}, "team_names": {}, "team_id": "", "player_id": "", "rank": "1", "number": "", "values": {"teid": "14927", "gp": "10", "gf": "31"}, "source_tag": "Team"},
|
||
{"id": "2", "name": "Команда Б", "names": {"ru": "Команда Б", "en": "Team B"}, "team_names": {}, "team_id": "", "player_id": "", "rank": "2", "number": "", "values": {"teid": "14928", "gp": "10", "gf": "29"}, "source_tag": "Team"},
|
||
],
|
||
},
|
||
{
|
||
"id": "powerplay",
|
||
"names": {"ru": "Большинство", "en": "Power play"},
|
||
"periods": {},
|
||
"rows": [
|
||
{"id": "1", "name": "Команда А", "names": {"ru": "Команда А", "en": "Team A"}, "team_names": {}, "team_id": "", "player_id": "", "rank": "1", "number": "", "values": {"teid": "14927", "gp": "10", "pp": "30", "ppgf": "8", "pp_pct": "26.7"}, "source_tag": "Team"},
|
||
{"id": "2", "name": "Команда Б", "names": {"ru": "Команда Б", "en": "Team B"}, "team_names": {}, "team_id": "", "player_id": "", "rank": "2", "number": "", "values": {"teid": "14928", "gp": "10", "pp": "31", "ppgf": "6", "pp_pct": "19.4"}, "source_tag": "Team"},
|
||
],
|
||
},
|
||
]
|
||
}
|
||
with self.database.session() as session:
|
||
self.service._store_normalized_standings(session, tournament_external_id="1369", parsed=standings, now=now)
|
||
self.service._store_normalized_tournament_statistic_rows(session, tournament_external_id="1369", resource_type="powerplay", parsed=parsed, now=now)
|
||
# Rebuild again in the same Session to cover pending/deleted ORM rows.
|
||
self.service._rebuild_team_season_statistics(session, "1369", now=now)
|
||
with self.database.session() as session:
|
||
rows = list(session.scalars(select(TeamSeasonStatistic).where(TeamSeasonStatistic.tournament_external_id == "1369").order_by(TeamSeasonStatistic.team_key)))
|
||
self.assertEqual(len(rows), 2)
|
||
team_a = next(row for row in rows if row.team_entry_external_id == "14927")
|
||
self.assertEqual(team_a.power_play_opportunities, 30)
|
||
self.assertEqual(team_a.power_play_goals, 8)
|
||
self.assertAlmostEqual(team_a.power_play_pct, 26.7)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|