98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
import requests
|
|
from bs4 import BeautifulSoup
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from services.teams_service import sync_teams
|
|
from parsers.parser_sources import get_parser_source, source_absolute_url
|
|
|
|
|
|
def fetch_html(url: str) -> str:
|
|
headers = {"User-Agent": "Mozilla/5.0", "Accept": "*/*"}
|
|
response = requests.get(url, timeout=30, headers=headers)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
|
|
def get_links(html: str, source: dict) -> list[str]:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
links: list[str] = []
|
|
|
|
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
|
|
|
|
if "team_id=" in href:
|
|
team_external_id = href.split("team_id=")[-1].strip()
|
|
links.append(source_absolute_url(source, "/team/" + team_external_id))
|
|
else:
|
|
links.append(source_absolute_url(source, href))
|
|
|
|
return links
|
|
|
|
|
|
def get_url_teams(source_key: str | None = None) -> list[dict]:
|
|
source = get_parser_source(source_key)
|
|
html = fetch_html(source["teams_url"])
|
|
links = get_links(html, source)
|
|
teams_data: list[dict] = []
|
|
|
|
with ThreadPoolExecutor() as pool:
|
|
responses = [pool.submit(fetch_html, link) for link in links]
|
|
for result in responses:
|
|
try:
|
|
html = result.result()
|
|
team_data = parse_teams_html(html)
|
|
teams_data.append(team_data)
|
|
except Exception as e:
|
|
print(f"[parser_teams] Error fetching team data: {e}")
|
|
|
|
return teams_data
|
|
|
|
|
|
def parse_teams_html(html: str) -> dict:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
name = soup.find("a", class_="team-promo__team-name").text.strip()
|
|
external_id = soup.find("a", class_="team-promo__logo").get("href").split("/")[-1]
|
|
stat_info = soup.find("ul", class_="stats-info").find_all(
|
|
"div", class_="stats-info__number"
|
|
)
|
|
logo_url = soup.find("img", class_="team-promo__img").get("src")
|
|
games = stat_info[0].text.strip()
|
|
wins = stat_info[1].text.strip()
|
|
goals = stat_info[2].text.strip()
|
|
tournaments = stat_info[3].text.strip()
|
|
|
|
return {
|
|
"external_id": str(external_id),
|
|
"name": name,
|
|
"logo_url": logo_url,
|
|
"games": games,
|
|
"wins": wins,
|
|
"goals": goals,
|
|
"tournaments": tournaments,
|
|
}
|
|
|
|
|
|
def run_parser_teams(source_key: str | None = None) -> None:
|
|
source = get_parser_source(source_key)
|
|
print(f"[parser_teams] Источник: {source['title']}")
|
|
print(f"[parser_teams] URL: {source['teams_url']}")
|
|
|
|
teams_data = get_url_teams(source_key)
|
|
if teams_data:
|
|
sync_teams(teams_data)
|
|
print(f"[parser_teams] Teams synced: {len(teams_data)}")
|
|
else:
|
|
print("[parser_teams] Команды не найдены")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_parser_teams()
|