Version 2.0.0 (#40)

* Refactor base utilities

* TransfermarktClubPlayers

* TransfermarktClubProfile

* TransfermarktClubSearch

* TransfermarktCompetitionClubs

* TransfermarktCompetitionSearch

* TransfermarktPlayerProfile

* TransfermarktPlayerTransfers

* TransfermarktPlayerMarketValue

* TransfermarktPlayerStats

* TransfermarktPlayerSearch

* Refactor utils

* Add updatedAt datetime

* Refactor unit tests

* Update version and dependencies

Author: felipeall

app/services/base.py

@@ -0,0 +1,120 @@
+from dataclasses import dataclass, field
+from typing import Optional
+from xml.etree import ElementTree
+
+import requests
+from bs4 import BeautifulSoup
+from fastapi import HTTPException
+from lxml import etree
+from requests import Response, TooManyRedirects
+
+from app.utils.utils import trim
+from app.utils.xpath import Commons
+
+
+@dataclass
+class TransfermarktBase:
+    URL: str
+    page: ElementTree = field(default_factory=lambda: None, init=False)
+    response: dict = field(default_factory=lambda: {}, init=False)
+
+    def make_request(self, url: Optional[str] = None) -> Response:
+        url = self.URL if not url else url
+        try:
+            response: Response = requests.get(
+                url=url,
+                headers={
+                    "User-Agent": (
+                        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                        "AppleWebKit/537.36 (KHTML, like Gecko) "
+                        "Chrome/113.0.0.0 "
+                        "Safari/537.36"
+                    ),
+                },
+            )
+        except TooManyRedirects:
+            raise HTTPException(status_code=404, detail=f"Not found for url: {url}")
+        if 400 <= response.status_code < 500:
+            raise HTTPException(
+                status_code=response.status_code,
+                detail=f"Client Error. {response.reason} for url: {url}",
+            )
+        elif 500 <= response.status_code < 600:
+            raise HTTPException(
+                status_code=response.status_code,
+                detail=f"Server Error. {response.reason} for url: {url}",
+            )
+        return response
+
+    def request_url_bsoup(self) -> BeautifulSoup:
+        response: Response = self.make_request()
+        return BeautifulSoup(markup=response.content, features="html.parser")
+
+    @staticmethod
+    def convert_bsoup_to_page(bsoup: BeautifulSoup) -> ElementTree:
+        return etree.HTML(str(bsoup))
+
+    def request_url_page(self) -> ElementTree:
+        bsoup: BeautifulSoup = self.request_url_bsoup()
+        return self.convert_bsoup_to_page(bsoup=bsoup)
+
+    def raise_exception_if_not_found(self, xpath: str):
+        if not self.get_text_by_xpath(xpath):
+            raise HTTPException(status_code=404, detail=f"Invalid request (url: {self.URL})")
+
+    def get_list_by_xpath(self, xpath: str, remove_empty: Optional[bool] = True) -> Optional[list]:
+        elements: list = self.page.xpath(xpath)
+        if remove_empty:
+            elements_valid: list = [trim(e) for e in elements if trim(e)]
+        else:
+            elements_valid: list = [trim(e) for e in elements]
+        return elements_valid or []
+
+    def get_text_by_xpath(
+        self,
+        xpath: str,
+        pos: int = 0,
+        iloc: Optional[int] = None,
+        iloc_from: Optional[int] = None,
+        iloc_to: Optional[int] = None,
+        join_str: Optional[str] = None,
+    ) -> Optional[str]:
+        element = self.page.xpath(xpath)
+
+        if not element:
+            return None
+
+        if isinstance(element, list):
+            element = [trim(e) for e in element if trim(e)]
+
+        if isinstance(iloc, int):
+            element = element[iloc]
+
+        if isinstance(iloc_from, int) and isinstance(iloc_to, int):
+            element = element[iloc_from:iloc_to]
+
+        if isinstance(iloc_to, int):
+            element = element[:iloc_to]
+
+        if isinstance(iloc_from, int):
+            element = element[iloc_from:]
+
+        if isinstance(join_str, str):
+            return join_str.join([trim(e) for e in element])
+
+        try:
+            return trim(element[pos])
+        except IndexError:
+            return None
+
+    def get_search_last_page_number(self, xpath_base: str) -> int:
+        url_page_number_last = self.get_text_by_xpath(xpath_base + Commons.Search.PAGE_NUMBER_LAST)
+        url_page_number_active = self.get_text_by_xpath(xpath_base + Commons.Search.PAGE_NUMBER_ACTIVE)
+
+        if url_page_number_last:
+            return int(url_page_number_last.split("=")[-1])
+
+        if url_page_number_active:
+            return int(url_page_number_active.split("=")[-1])
+
+        return 1

app/services/clubs/players.py

@@ -1,100 +1,68 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
+from datetime import datetime
 
-from fastapi import HTTPException
-
-from app.utils.utils import (
-    clean_response,
-    extract_from_url,
-    get_list_by_xpath,
-    get_text_by_xpath,
-    request_url_page,
-    safe_regex,
-)
+from app.services.base import TransfermarktBase
+from app.utils.regex import REGEX_DOB
+from app.utils.utils import clean_response, extract_from_url, safe_regex
 from app.utils.xpath import Clubs
 
 
 @dataclass
-class TransfermarktClubPlayers:
-    club_id: str
+class TransfermarktClubPlayers(TransfermarktBase):
+    club_id: str = None
     season_id: str = None
-    club_players: dict = field(default_factory=lambda: {})
-
-    def get_club_players(self) -> dict:
-        self._request_page()
-        self._update_season_id()
-        self._update_club_in_past()
-
-        self.club_players["id"] = self.club_id
-        self.club_players["clubName"] = get_text_by_xpath(self, Clubs.Players.CLUB_NAME)
-
-        self._check_club_found()
-
-        self.club_players["seasonYear"] = self.season_id
-        self.club_players["players"] = self._parse_club_players()
-
-        return clean_response(self.club_players)
+    URL: str = "https://www.transfermarkt.com/-/kader/verein/{club_id}/saison_id/{season_id}/plus/1"
 
-    def _request_page(self) -> None:
-        club_players_url = (
-            f"https://www.transfermarkt.com/-/kader/verein/{self.club_id}/saison_id/{self.season_id}/plus/1"
-        )
-        self.page = request_url_page(url=club_players_url)
+    def __post_init__(self):
+        self.URL = self.URL.format(club_id=self.club_id, season_id=self.season_id)
+        self.page = self.request_url_page()
+        self.raise_exception_if_not_found(xpath=Clubs.Players.CLUB_NAME)
+        self.__update_season_id()
+        self.__update_past_flag()
 
-    def _update_season_id(self):
+    def __update_season_id(self):
         if self.season_id is None:
-            self.season_id = extract_from_url(get_text_by_xpath(self, Clubs.Players.CLUB_URL), "season_id")
+            self.season_id = extract_from_url(self.get_text_by_xpath(Clubs.Players.CLUB_URL), "season_id")
 
-    def _update_club_in_past(self):
-        self.past = "Current club" in get_list_by_xpath(self, Clubs.Players.PAST_FLAG)
+    def __update_past_flag(self):
+        self.past = "Current club" in self.get_list_by_xpath(Clubs.Players.PAST_FLAG)
 
-    def _parse_club_players(self) -> list:
+    def __parse_club_players(self) -> list:
         page_nationalities = self.page.xpath(Clubs.Players.PAGE_NATIONALITIES)
         page_players_infos = self.page.xpath(Clubs.Players.PAGE_INFOS)
         page_players_signed_from = self.page.xpath(
             Clubs.Players.Past.PAGE_SIGNED_FROM if self.past else Clubs.Players.Present.PAGE_SIGNED_FROM,
         )
-
-        players_urls: list = get_list_by_xpath(self, Clubs.Players.URLS)
-        players_ids: list = [extract_from_url(url) for url in players_urls]
-
-        players_names: list = get_list_by_xpath(self, Clubs.Players.NAMES)
-
-        players_positions: list = get_list_by_xpath(self, Clubs.Players.POSITIONS)
-        players_dobs: list = [
-            safe_regex(dob_age, r"^(?P<dob>.*)\s\((?P<age>\d*)\)", "dob")
-            for dob_age in get_list_by_xpath(self, Clubs.Players.DOB_AGE)
+        players_ids = [extract_from_url(url) for url in self.get_list_by_xpath(Clubs.Players.URLS)]
+        players_names = self.get_list_by_xpath(Clubs.Players.NAMES)
+        players_positions = self.get_list_by_xpath(Clubs.Players.POSITIONS)
+        players_dobs = [
+            safe_regex(dob_age, REGEX_DOB, "dob") for dob_age in self.get_list_by_xpath(Clubs.Players.DOB_AGE)
         ]
-        players_ages: list = [
-            safe_regex(dob_age, r"^(?P<dob>.*)\s\((?P<age>\d*)\)", "age")
-            for dob_age in get_list_by_xpath(self, Clubs.Players.DOB_AGE)
-        ]
-        players_nationalities: list = [
-            nationality.xpath(Clubs.Players.NATIONALITIES) for nationality in page_nationalities
+        players_ages = [
+            safe_regex(dob_age, REGEX_DOB, "age") for dob_age in self.get_list_by_xpath(Clubs.Players.DOB_AGE)
         ]
+        players_nationalities = [nationality.xpath(Clubs.Players.NATIONALITIES) for nationality in page_nationalities]
         players_current_club = (
-            get_list_by_xpath(self, Clubs.Players.Past.CURRENT_CLUB) if self.past else [None] * len(players_ids)
+            self.get_list_by_xpath(Clubs.Players.Past.CURRENT_CLUB) if self.past else [None] * len(players_ids)
         )
-        players_heights: list = get_list_by_xpath(
-            self,
+        players_heights = self.get_list_by_xpath(
             Clubs.Players.Past.HEIGHTS if self.past else Clubs.Players.Present.HEIGHTS,
         )
-        players_foots: list = get_list_by_xpath(
-            self,
+        players_foots = self.get_list_by_xpath(
             Clubs.Players.Past.FOOTS if self.past else Clubs.Players.Present.FOOTS,
             remove_empty=False,
         )
-        players_joined_on: list = get_list_by_xpath(
-            self,
+        players_joined_on = self.get_list_by_xpath(
             Clubs.Players.Past.JOINED_ON if self.past else Clubs.Players.Present.JOINED_ON,
         )
-        players_joined: list = ["; ".join(e.xpath(Clubs.Players.JOINED)) for e in page_players_infos]
-        players_signed_from: list = ["; ".join(e.xpath(Clubs.Players.SIGNED_FROM)) for e in page_players_signed_from]
-        players_contracts: list = get_list_by_xpath(
-            self,
-            Clubs.Players.Past.CONTRACTS if self.past else Clubs.Players.Present.CONTRACTS,
+        players_joined = ["; ".join(e.xpath(Clubs.Players.JOINED)) for e in page_players_infos]
+        players_signed_from = ["; ".join(e.xpath(Clubs.Players.SIGNED_FROM)) for e in page_players_signed_from]
+        players_contracts = (
+            [None] * len(players_ids) if self.past else self.get_list_by_xpath(Clubs.Players.Past.CONTRACTS)
         )
-        players_marketvalues: list = get_list_by_xpath(self, Clubs.Players.MARKET_VALUES)
-        players_statuses: list = ["; ".join(e.xpath(Clubs.Players.STATUSES)) for e in page_players_infos]
+        players_marketvalues = self.get_list_by_xpath(Clubs.Players.MARKET_VALUES)
+        players_statuses = ["; ".join(e.xpath(Clubs.Players.STATUSES)) for e in page_players_infos]
 
         return [
             {
@@ -133,6 +101,9 @@ def _parse_club_players(self) -> list:
             )
         ]
 
-    def _check_club_found(self) -> None:
-        if not self.club_players["clubName"]:
-            raise HTTPException(status_code=404, detail=f"Club Players not found for id: {self.club_id}")
+    def get_club_players(self) -> dict:
+        self.response["id"] = self.club_id
+        self.response["players"] = self.__parse_club_players()
+        self.response["updatedAt"] = datetime.now()
+
+        return clean_response(self.response)