Source code for cowidev.testing.incremental.albania

import re

from bs4 import BeautifulSoup, element
import pandas as pd

from cowidev.utils.web import get_soup
from cowidev.utils.clean import clean_count, clean_date
from cowidev.testing.utils.incremental import increment


import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


[docs]class Albania: location = "Albania" units = "tests performed" source_label = "Ministry of Health and Social Protection" notes = "" _base_url = "https://shendetesia.gov.al" _url_subdirectory = "category/lajme/page" _num_max_pages = 3 regex = { "title": r"COVID19/ Ministria e Shëndetësisë:", "date": r"(\d+\/\d+\/\d+)", "count": r"(\d+) testime nga të cilat kanë rezultuar pozitivë me COVID19", }
[docs] def read(self) -> pd.Series: data = [] for cnt in range(1, self._num_max_pages + 1): url = f"{self._base_url}/{self._url_subdirectory}/{cnt}/" soup = get_soup(url, verify=False) data, proceed = self._parse_data(soup) if not proceed: break return pd.Series(data)
[docs] def _parse_data(self, soup: BeautifulSoup) -> tuple: """Get data from the source page.""" # Get relevant element elem = self._get_relevant_element(soup) if not elem: return None, True # Extract url and date from element url, date = self._get_link_and_date_from_element(elem) # Extract text from url text = self._get_text_from_url(url) daily_change = self._parse_metrics(text) record = { "source_url": url, "date": date, "daily_change": daily_change, } return record, False
[docs] def _get_relevant_element(self, soup: BeautifulSoup) -> element.Tag: """Get the relevant element in news feed.""" news_list = soup.find_all("h2") url_idx = [ i for i, news in enumerate(news_list) if re.search(self.regex["title"], news.text) ] if not url_idx: return None return news_list[url_idx[0]]
[docs] def _get_text_from_url(self, url: str) -> str: """Extract text from the url.""" soup = get_soup(url, verify=False) text = soup.find("div", class_="pageDescription").get_text(strip=True).replace(",", "") return text
[docs] def _parse_date_from_element(self, elem: element.Tag) -> str: """Get date from relevant element.""" date_tag = elem.findNextSibling("div", class_="dateDetalils") date = re.search(self.regex["date"], date_tag.text).group() return clean_date(date, "%d/%m/%Y")
[docs] def _parse_metrics(self, text: str) -> int: """Get metrics from news text.""" count = int(re.search(self.regex["count"], text).group(1)) return clean_count(count)
[docs] def export(self): data = self.read() increment( sheet_name=self.location, country=self.location, units=self.units, date=data["date"], source_url=data["source_url"], source_label=self.source_label, daily_change=data["daily_change"], )
[docs]def main(): Albania().export()