from repositories.match_repository import ( get_match_by_external_id, mark_match_parsed, mark_match_parse_error, clear_match_squad_data, ) from repositories.player_repository import ( get_player_id_by_name_and_team, get_player_id_by_external_id, create_player_from_lineup, ) from repositories.coach_repository import ( get_coach_id_by_name_and_team, get_coach_id_by_external_id, create_coach_from_lineup, ) from repositories.referee_repository import get_referee_id_by_name, upsert_referee from repositories.match_lineup_repository import ( replace_match_lineups, get_match_lineup_for_editor, ) from repositories.match_coach_repository import replace_match_coaches from repositories.match_referee_repository import replace_match_referees def sync_match_page( match_external_id: str, home_starting: list[dict], away_starting: list[dict], home_bench: list[dict], away_bench: list[dict], home_coaches: list[dict], away_coaches: list[dict], referees: list[dict], ) -> None: match_row = get_match_by_external_id(match_external_id) if not match_row: raise ValueError(f"Match not found by external_id: {match_external_id}") match_id, _, home_team_id, away_team_id = match_row # Сохраняем текущий выбор капитана ДО очистки данных матча. Это важно # при повторной загрузке с сайта: на сайте капитана иногда не указывают, # и в таком случае нельзя молча терять уже сделанный оператором выбор. previous_lineup = get_match_lineup_for_editor( match_id=match_id, home_team_id=home_team_id, away_team_id=away_team_id, ) def previous_captain(side: str) -> dict | None: rows = previous_lineup.get(f"{side}_starting", []) or [] for player in rows: if bool(player.get("is_captain")): return { "player_id": player.get("player_id"), "number": str(player.get("number") or "").strip(), } return None previous_captains = { "home": previous_captain("home"), "away": previous_captain("away"), } clear_match_squad_data(match_id) created_players_count = 0 created_coaches_count = 0 def resolve_player(player: dict, team_id: int) -> tuple[int | None, str | None]: """Находит игрока из протокола или создаёт его в справочнике автоматически.""" nonlocal created_players_count player_name = str(player.get("player_name") or "").strip() player_external_id = str(player.get("player_external_id") or "").strip() number = str(player.get("number") or "").strip() position = str(player.get("position") or "").strip() player_id = None player_position = None if player_external_id: player_row = get_player_id_by_external_id(player_external_id) if player_row: player_id, player_position = player_row if player_id is None and player_name: player_id = get_player_id_by_name_and_team(player_name, team_id) if player_id is None and player_name: player_row = create_player_from_lineup( team_id=team_id, external_id=player_external_id, full_name=player_name, number=number, position=position, ) if player_row: player_id, player_position = player_row created_players_count += 1 return player_id, player_position or position def resolve_coach(coach: dict, team_id: int) -> int | None: """Находит тренера из протокола или создаёт его в справочнике автоматически.""" nonlocal created_coaches_count coach_name = str(coach.get("coach_name") or "").strip() coach_external_id = str(coach.get("coach_external_id") or "").strip() role = str(coach.get("role") or "").strip() coach_id = None if coach_external_id: coach_id = get_coach_id_by_external_id(coach_external_id) if coach_id is None and coach_name: coach_id = get_coach_id_by_name_and_team(coach_name, team_id) if coach_id is None and coach_name: coach_id = create_coach_from_lineup( team_id=team_id, external_id=coach_external_id, full_name=coach_name, role=role, ) if coach_id: created_coaches_count += 1 return coach_id lineup_rows = [] def append_players(players: list[dict], team_id: int, lineup_type: str) -> None: for player in players: player_id, player_position = resolve_player(player, team_id) lineup_rows.append( { "match_id": match_id, "team_id": team_id, "player_id": player_id, "player_name": player.get("player_name") or "", "number": player.get("number"), "position": player.get("position"), "position_full": player_position, # Капитаном может быть только игрок основного состава. # Если сайт по ошибке пометил игрока запаса, не переносим # такой флаг в матчевые данные. "is_captain": bool(player.get("is_captain")) and lineup_type == "starting", "lineup_type": lineup_type, "source": "parser", } ) append_players(home_starting, home_team_id, "starting") append_players(away_starting, away_team_id, "starting") append_players(home_bench, home_team_id, "bench") append_players(away_bench, away_team_id, "bench") def restore_previous_captain_if_missing(side: str, team_id: int) -> bool: """Возвращает старого капитана только когда сайт не указал нового. Восстановление разрешено исключительно для игрока, который всё ещё присутствует в основном составе. Если прежний капитан отсутствует или ушёл в запас, капитан остаётся не выбран — веб-интерфейс покажет обязательное предупреждение оператору. """ starting_rows = [ row for row in lineup_rows if row.get("team_id") == team_id and row.get("lineup_type") == "starting" ] imported_captains = [row for row in starting_rows if bool(row.get("is_captain"))] # Ровно один капитан с сайта — корректные данные, они имеют приоритет. if len(imported_captains) == 1: return False # Если сайт по ошибке передал больше одного капитана, не выбираем # случайного игрока. Сбрасываем конфликт и используем тот же безопасный # fallback, что и при полностью отсутствующем капитане. if len(imported_captains) > 1: for row in imported_captains: row["is_captain"] = False old_captain = previous_captains.get(side) if not old_captain: return False old_player_id = old_captain.get("player_id") old_number = str(old_captain.get("number") or "").strip() # Сначала точное совпадение по player_id. if old_player_id is not None: for row in starting_rows: if row.get("player_id") is not None and str(row.get("player_id")) == str(old_player_id): row["is_captain"] = True return True # Fallback для старых/неполных данных — по игровому номеру. if old_number: matches = [ row for row in starting_rows if str(row.get("number") or "").strip() == old_number ] if len(matches) == 1: matches[0]["is_captain"] = True return True return False restored_home_captain = restore_previous_captain_if_missing("home", home_team_id) restored_away_captain = restore_previous_captain_if_missing("away", away_team_id) if restored_home_captain or restored_away_captain: restored = [] if restored_home_captain: restored.append("home") if restored_away_captain: restored.append("away") print( f"[parser_game] website captain missing for match={match_external_id}; " f"preserved previous captain for: {', '.join(restored)}" ) coach_rows = [] def append_coaches(coaches: list[dict], team_id: int, side: str) -> None: for coach in coaches: coach_id = resolve_coach(coach, team_id) coach_rows.append( { "match_id": match_id, "team_id": team_id, "side": side, "coach_id": coach_id, "coach_name": coach.get("coach_name") or "", "role": coach.get("role"), "source": "parser", } ) append_coaches(home_coaches, home_team_id, "home") append_coaches(away_coaches, away_team_id, "away") referee_rows = [] for referee in referees: referee_name = referee["referee_name"].strip() referee_id = get_referee_id_by_name(referee_name) if referee_id is None: referee_id = upsert_referee(full_name=referee_name) referee_rows.append( { "match_id": match_id, "referee_id": referee_id, "referee_name": referee_name, "role": referee.get("role"), "source": "parser", } ) try: replace_match_lineups(match_id, lineup_rows) replace_match_coaches(match_id, coach_rows) replace_match_referees(match_id, referee_rows) mark_match_parsed(match_external_id) if created_players_count or created_coaches_count: print( f"[parser_game] auto-created for match={match_external_id}: " f"players={created_players_count}, coaches={created_coaches_count}" ) except Exception as e: mark_match_parse_error(match_external_id, str(e)) raise