97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
import requests
|
|
from bs4 import BeautifulSoup
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from services.teams_service import sync_teams
|
|
|
|
|
|
TEAMS_URL = "https://wfl.rfs.ru/tournament/1061879/teams"
|
|
|
|
|
|
def fetch_html(url: str) -> str:
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0",
|
|
"Accept": "*/*"
|
|
}
|
|
|
|
try:
|
|
r = requests.get(url, headers=headers, timeout=20)
|
|
r.raise_for_status()
|
|
return r.text
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return f"Сайт недоступен: {e}"
|
|
|
|
|
|
def get_links(html) -> list:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
links: list[dict] = []
|
|
items = soup.find("ul", class_="teams__list").find_all("li")
|
|
for i in items:
|
|
links.append(
|
|
"https://wfl.rfs.ru/team/"
|
|
+ i.find("a", class_="teams__link").get("href").split("team_id=")[-1]
|
|
)
|
|
return links
|
|
|
|
|
|
def get_url_teams() -> list[dict]:
|
|
html = fetch_html(TEAMS_URL)
|
|
links = get_links(html)
|
|
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"Error fetching team data: {e}")
|
|
|
|
return teams_data
|
|
|
|
|
|
def parse_teams_html(html: str) -> dict:
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
teams_data: dict = {}
|
|
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()
|
|
|
|
teams_data = {
|
|
"external_id": str(external_id),
|
|
"name": name,
|
|
"logo_url": logo_url,
|
|
"games": games,
|
|
"wins": wins,
|
|
"goals": goals,
|
|
"tournaments": tournaments,
|
|
}
|
|
|
|
return teams_data
|
|
|
|
|
|
def run_parser_teams() -> None:
|
|
teams_data = get_url_teams()
|
|
sync_teams(teams_data)
|
|
print(f"Teams synced: {len(teams_data)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_parser_teams()
|