import requests from bs4 import BeautifulSoup from concurrent.futures import ThreadPoolExecutor from services.players_service import sync_team_roster from parsers.parser_sources import get_parser_source, source_absolute_url AMPLUA_FULL = { "Пз.": "Полузащитник", "Вр.": "Вратарь", "Зщ.": "Защитник", "Нп.": "Нападающий", "": "", } def fetch_html(url: str) -> str: headers = {"User-Agent": "Mozilla/5.0", "Accept": "*/*"} r = requests.get(url, headers=headers, timeout=20) r.raise_for_status() return r.text def get_links(html: str, source: dict) -> list[dict]: soup = BeautifulSoup(html, "html.parser") links: list[dict] = [] teams_list = soup.find("ul", class_="teams__list") if not teams_list: return links items = teams_list.find_all("li") for item in items: link_el = item.find("a", class_="teams__link") href = link_el.get("href") if link_el else "" if not href: continue team_external_id = href.split("team_id=")[-1].strip() if "team_id=" in href else "" links.append( { "team_external_id": team_external_id, "url": source_absolute_url(source, href), } ) return links def parse_team(html: str) -> dict: soup = BeautifulSoup(html, "html.parser") team_el = soup.find("a", class_="team-promo__team-name") team_name = team_el.get_text(strip=True) if team_el else None tabs = soup.find("div", class_="tabs__content") tbody = tabs.find("tbody") if tabs else None row_players = tbody.find_all("tr", class_="table__row") if tbody else [] coaches_root = soup.find("ul", class_="composition-list") row_coaches = ( coaches_root.find_all("div", class_="composition-list__item-back") if coaches_root else [] ) players = [] for row in row_players: player_td = row.find("td", class_="table__cell table__cell--player") player_a = player_td.find("a") if player_td else None href = player_a.get("href") if player_a else None player_id = href.split("/")[-1] if href else None vars_td = row.find_all("td", class_="table__cell table__cell--variable") games = vars_td[0].get_text(strip=True) if len(vars_td) > 0 else "0" goals_raw = vars_td[1].get_text(strip=True) if len(vars_td) > 1 else "" goals_parts = goals_raw.split() goals = goals_parts[0] if len(goals_parts) > 0 else "0" penaltys = ( goals_parts[1].replace("(", "").replace(")", "") if len(goals_parts) > 1 else "0" ) assists = vars_td[2].get_text(strip=True) if len(vars_td) > 2 else "0" yellows = vars_td[3].get_text(strip=True) if len(vars_td) > 3 else "0" reds = vars_td[4].get_text(strip=True) if len(vars_td) > 4 else "0" number_td = row.find("td", class_="table__cell table__cell--number") pos_td = row.find( "td", class_="table__cell table__cell--amplua table__cell--amplua" ) name_p = row.find("p", class_="table__player-name") born_td = row.find("td", class_="table__cell table__cell--middle table__cell--birth mobile-hide") full_player = name_p.get_text(strip=True) if name_p else "" parts = full_player.split() pos_short = pos_td.get_text(strip=True) if pos_td else "" players.append( { "player_id": player_id or "", "number": number_td.get_text(strip=True) if number_td else "", "pos": pos_short, "amplua": AMPLUA_FULL.get(pos_short, pos_short), "player": full_player, "lastname": parts[0] if len(parts) >= 1 else "", "name": parts[-1] if len(parts) >= 2 else "", "born": born_td.get_text(strip=True).split(",")[0] if born_td else "", "games": games, "goals": goals, "penaltys": penaltys, "assists": assists, "yellows": yellows, "reds": reds, } ) coaches = [] for coach in row_coaches: coach_link = coach.find("a", class_="composition-list__player") href = coach_link.get("href") if coach_link else "" coach_id = href.split("/")[-1] if href else "" name = coach.find("span", class_="composition-list__player-first-name") lastname = coach.find("span", class_="composition-list__player-last-name") born = coach.find("span", class_="composition-list__player-birth-date") amplua = coach.find("span", class_="composition-list__player-games-text") first_name = name.get_text(strip=True) if name else "" last_name = lastname.get_text(strip=True) if lastname else "" coaches.append( { "coach_id": coach_id, "name": first_name, "lastname": last_name, "player": f"{last_name} {first_name}".strip(), "born": born.get_text(strip=True).replace(",", "") if born else "", "amplua": amplua.get_text(strip=True).replace(".", "") if amplua else "", } ) return { "team_name": team_name, "players": players, "coaches": coaches, } def run_parser_players(source_key: str | None = None) -> None: source = get_parser_source(source_key) print(f"[parser_players] Источник: {source['title']}") print(f"[parser_players] URL: {source['teams_url']}") html = fetch_html(source["teams_url"]) links = get_links(html, source) with ThreadPoolExecutor(max_workers=8) as pool: futures = {pool.submit(fetch_html, item["url"]): item for item in links} for future, item in futures.items(): try: html = future.result() team_data = parse_team(html) sync_team_roster( team_external_id=item["team_external_id"], team_data=team_data, ) print( f"[parser_players] team={item['team_external_id']} " f"players={len(team_data.get('players') or [])} " f"coaches={len(team_data.get('coaches') or [])}" ) except Exception as e: print(f"[parser_players] error team={item['team_external_id']}: {e}") if not links: print("[parser_players] Команды для парсинга игроков не найдены") if __name__ == "__main__": run_parser_players()