gcpdiag.queries.web

24def extract_cell_text(element: Any) -> Optional[str]:
25  """Recursively extract text from a table cell element."""
26  if isinstance(element, str):
27    return element
28  if isinstance(element, Tag):
29    return extract_cell_text(element.next)
30  return None

33def fetch_and_extract_table_data(
34  page_url: str, tag: str = None, tag_id: str = None, class_name: str = None
35) -> List[List[str]]:
36  """Fetch table from URL and return row data as list of lists of cell text strings."""
37  table = fetch_and_extract_table(page_url, tag=tag, tag_id=tag_id, class_name=class_name)
38  if not table:
39    return []
40  rows_data = []
41  tbody = table.find('tbody')
42  rows = tbody.find_all('tr') if tbody else table.find_all('tr')
43  for row in rows:
44    cols = row.find_all('td')
45    if not cols:
46      continue
47    row_cells = []
48    for col in cols:
49      val = extract_cell_text(col.next) or ''
50      row_cells.append(val.strip())
51    rows_data.append(row_cells)
52  return rows_data

55def fetch_and_extract_table(
56  page_url: str, tag: str = None, tag_id: str = None, class_name: str = None
57):
58  """Fetch the table from the given page url and return it."""
59  table = None
60  response = get(url=page_url, timeout=10)
61  response.raise_for_status()  # Raise an exception if the response is not successful
62  soup = BeautifulSoup(response.content, 'html.parser')
63  content_fetched = None
64  if tag:
65    if tag_id:
66      content_fetched = soup.find(tag, id=tag_id)
67    elif class_name:
68      content_fetched = soup.find(tag, class_=class_name)
69    else:
70      content_fetched = soup.find(tag)
71
72  if not content_fetched:
73    logging.error('tag/id/class not found for %s with tag %s', page_url, tag)
74    return table
75  if tag == 'table':
76    return content_fetched
77  table = content_fetched.find_next('table')
78  if not table:
79    logging.error('Table not found for %s with tag %s', page_url, tag)
80    return table
81
82  return table

85def fetch_all_tables(page_url: str) -> list:
86  """Fetch all tables from the given page url."""
87  response = get(url=page_url, timeout=10)
88  response.raise_for_status()
89  soup = BeautifulSoup(response.content, 'html.parser')
90  return soup.find_all('table')

 93def get(
 94  url,
 95  params=None,
 96  timeout=10,
 97  *,
 98  data=None,
 99  headers=None,
100) -> requests.Response:
101  """A wrapper around requests.get for http calls which can't use the google discovery api"""
102  return requests.get(url=url, params=params, timeout=timeout, data=data, headers=headers)

105def parse_table(table) -> list:
106  """Parse a BeautifulSoup table into a list of rows, where each row is a list of cell texts."""
107  tbody = table.find('tbody')
108  rows = tbody.find_all('tr') if tbody else table.find_all('tr')
109  parsed_rows = []
110  for row in rows:
111    cols = row.find_all(['td', 'th'])
112    parsed_rows.append([col.text.strip() for col in cols])
113  return parsed_rows

116def fetch_and_parse_all_tables(page_url: str) -> list:
117  """Fetch all tables from the given page url and parse them."""
118  tables = fetch_all_tables(page_url)
119  return [parse_table(t) for t in tables]