""" This module contains functions for scraping data from various sources, including a generic scraper class for basic scraping functionality of a provided URL. """ from csv import DictReader from functools import lru_cache from io import StringIO from random import choice as random_choice from typing import Callable from urllib.parse import quote_plus, unquote_plus import httpx from bs4 import BeautifulSoup from .. import logger from ..config import FAKE_HEADERS from ..typings import HTML, URL, ArtistName logger = logger.new_logger(__name__) def _require_artist_name(func: Callable) -> Callable: """Decorator to ensure that the artist name is not empty. May be used on class methods that require an artist name as a first argument. """ async def wrapper(self, artist_name: ArtistName, *args, **kwargs): if not artist_name.strip(): raise ValueError("Artist name cannot be empty.") return await func(self, artist_name, *args, **kwargs) return wrapper class BaseScraper: """Base scraper class. Can scrape HTML content from a URL, or file bytes if the URL ends with a file extension. This class is also the base class for all other scrapers. """ _new_client_retries: int = 5 _timeout: int = 30 def __init__(self, client: httpx.AsyncClient = None): """ Initialize the scraper with an optional HTTP client. Args: client: An optional HTTP client. If not provided, a new client will be created for each request, with N retries set. Providing a client can be useful for reusing connections and settings. """ self._client = client or httpx.AsyncClient( transport=httpx.AsyncHTTPTransport(retries=self._new_client_retries), ) @property def client(self) -> httpx.AsyncClient: """Get the HTTP client.""" return self._client async def _get(self, url: URL) -> HTML | bytes: """Make HTTP request and return HTML content or file bytes, depending on the URL provided. Args: url: The URL to request. Returns: HTML content or file bytes, depending on the URL provided (if it ends with a file extension or not). """ async def fetch(_client: httpx.AsyncClient): logger.debug(f"Scraping URL: {url}") # Fake user agent to avoid being blocked and to emulate a real browser. headers = random_choice(FAKE_HEADERS) return await _client.get( url, follow_redirects=True, headers=headers, timeout=self._timeout ) response = await fetch(self._client) response.raise_for_status() data = response.content if "." not in url.split("/")[-1]: # Handle content as HTML data = data.decode() return data class GenericScraper(BaseScraper): """Generic scraper class. Can scrape HTML content from a URL, or file bytes if the URL ends with a file extension. """ async def get(self, url: URL) -> HTML | bytes: """Make HTTP request and return HTML content or file bytes, depending on the URL provided. Args: url: The URL to request. Returns: HTML content or file bytes, depending on the URL provided (if it ends with a file extension or not). """ return await self._get(url) class CSVScraper(BaseScraper): """CSV scraper class. Can scrape CSV content from a URL.""" async def get(self, url: URL) -> DictReader: """Make HTTP request and return CSV content as a DictReader. Args: url: The URL to request. Must end with .csv. Returns: CSV content as a DictReader. To access the rows, iterate over the DictReader and access the values by key. """ if not url.lower().endswith(".csv"): raise ValueError("URL must end with .csv to scrape CSV content.") return DictReader(StringIO((await self._get(url)).decode("utf-8"))) class LastFM(BaseScraper): """Last.fm scraper. Uses lrucache for increased performance and to avoid unnecessary requests. """ base_url: URL = "https://www.last.fm/" @lru_cache(maxsize=1024) @_require_artist_name async def artist_search(self, artist_name: str) -> list[str]: """Search artists on Last.fm and return a list of artist names that match the search query (fuzzy search), ordered by number of listeners, descending. Args: artist_name: The name of the artist / search term. Returns: A list of artist names that fuzzy match the search query. """ # Make exception for colons in artist names, which will not be encoded # (as they're valid in the URL in this case). This is for finding artist # names with colons in them. formatted_artist_name = quote_plus(artist_name).replace(":", "%3A") url = f"{self.base_url}search/artists?q={formatted_artist_name}" html = await self._dispatch(url) # Get the first ul with class "artist-results" artist_ul = BeautifulSoup(html, "html.parser").find( "ul", class_="artist-results" ) if artist_ul is None: # If no results are found, the ul will not exist, thus return an empty list. return [] def parse_listeners_text_to_digit(string: str) -> int: return int("".join(c for c in string if c.isdigit()) or 0) artist_urls = (a["href"] for a in artist_ul.select("li h4 a:first-of-type")) artist_listeners = ( parse_listeners_text_to_digit(x.text) for x in artist_ul.select("li " "p.artist-result-listeners:first-of-type") ) artist_urls_sorted = ( artist for listeners, artist in sorted( zip(artist_listeners, artist_urls), reverse=True ) ) artist_names = list( unquote_plus(name) for name in (url.rsplit("/", maxsplit=1)[1] for url in artist_urls_sorted) ) return artist_names @lru_cache(maxsize=1024) @_require_artist_name async def get_artist_photos( self, artist_name: str, limit: int = 10, size: int = 300 ) -> list[str]: """Get artist photos from Last.fm. All photos are resized to a square of the specified size (in px). Args: artist_name: The name of the artist. Must be an exact match. limit: The maximum number of photos to return. Defaults to 10. size: The size of the square in pixels. Defaults to 300. Returns: A list of URLs to the artist's photos. """ max_limit: int = 12 # This is a hard limit until pagination is implemented. if limit > max_limit: raise NotImplementedError( "Limit cannot be greater than 12. Consider " "implementing pagination for higher limits." ) limit = max(1, min(limit, max_limit)) try: html = await self._dispatch( f"{self.base_url}music/{quote_plus(artist_name)}/+images" ) except httpx.HTTPStatusError as ex: if ex.response.status_code == 404: # No artist found return [] raise image_srcs = ( image["src"] for image in BeautifulSoup(html, "html.parser") .find("ul", class_="image-list") .find_all("img", limit=limit) ) image_ids = (src.rsplit("/", maxsplit=1)[1] for src in image_srcs) base_url = f"https://lastfm.freetls.fastly.net/i/u/{size}x{size}/" return [f"{base_url}{image_id}.jpg" for image_id in image_ids] async def _dispatch(self, url: URL) -> HTML: """Wrap the _dispatch method to ensure that the URL is properly encoded for Last.fm's search functionality. """ safe_url = quote_plus(url, safe=":/?&=,+") return await super()._get(safe_url)