Create web_page_information_extractor.py

tools/web_page_information_extractor.py

@@ -0,0 +1,106 @@
+import os
+from io import StringIO
+
+import html2text
+import pandas as pd
+import requests
+from bs4 import BeautifulSoup
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain_core.tools import tool
+from langchain_openai import ChatOpenAI
+from loguru import logger
+from pydantic import SecretStr
+
+
+@tool("web_page_information_extractor_tool", parse_docstring=True)
+def web_page_information_extractor(url: str, request: str) -> str:
+    """
+    Extracts specific information from a web page based on the user's request.
+    This function uses a language model to extract information from the content
+    of a web page specified by the URL. The user's request specifies the type of
+    information to be extracted. The function returns the extracted information as
+    a JSON string.
+    Args:
+        url (str): The URL of the web page to extract information from.
+        request (str): The user's request describing the information to extract.
+    Returns:
+        str: The extracted information in JSON format.
+    """
+    logger.info(f"use web_page_information_extractor with param: url:{url}, request:{request}")
+
+    text = _get_text_from_url(url)
+    logger.debug(f"web_page_information_extractor text: {text}")
+
+    chat = ChatOpenAI(
+        model="gpt-4o-mini",
+        temperature=0,
+        api_key=SecretStr(os.environ['OPENAI_API_KEY'])
+    )
+
+    system_message = "You are an expert information extraction system. Respond ONLY with valid JSON based on the user's request."
+    extraction_user_prompt = f"""From the text below:\n\"\"\"\n{text}\n\"\"\"\n\nExtract the following: "{request}"."""
+
+    extracted_information = chat.invoke([
+        SystemMessage(system_message),
+        HumanMessage(extraction_user_prompt)
+    ])
+    return extracted_information.content
+
+def _get_text_from_url(url: str) -> str:
+    response = requests.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
+    response.raise_for_status()  # Raises HTTPError for bad responses
+    html = response.text
+
+    soup = BeautifulSoup(html, "html.parser")
+
+    # Get tables
+    tables = soup.find_all('table', class_='wikitable')
+    tables_text = ""
+
+    for i, table in enumerate(tables, 1):
+        # Find the nearest preceding h2 or h3 header
+        header = table.find_previous(['h2', 'h3'])
+        section_title = header.get_text().strip() if header else "Untitled Section"
+        try:
+            # Convert table to pandas DataFrame using StringIO
+            table_html = str(table).replace('\n', '')  # Remove newlines for better parsing
+            df = pd.read_html(StringIO(table_html))[0]
+
+            # Format the table with section title, context, and clean layout
+            tables_text += f"\nSection: {section_title}\n"
+            tables_text += "=" * 40 + "\n"
+            tables_text += df.to_string(index=False) + "\n\n"
+        except Exception as e:
+            tables_text += f"\nError processing table in section {section_title}: {str(e)}\n"
+            continue
+
+    # Step 3: Convert HTML to Markdown
+    markdown_converter = html2text.HTML2Text()
+    markdown_converter.ignore_links = False
+    markdown_converter.bypass_tables = False
+    markdown_converter.ignore_images = True  # optional
+    markdown_converter.body_width = 0  # don't wrap lines
+
+    text = markdown_converter.handle(html)
+    if tables_text:
+        text = f'Tables:\n{tables_text}\n\nContent\n{text}'
+
+    return text
+
+if __name__ == "__main__":
+    # result = web_page_information_extractor.invoke(
+    #     {"url": "https://en.wikipedia.org/wiki/Python_(programming_language)",
+    #      "request": "What are changes introduced in Python 3.11"})
+    # print(result)
+
+    result = web_page_information_extractor.invoke(
+        {"url": "https://en.wikipedia.org/wiki/1928_Summer_Olympics",
+         "request": "List of countries and number of athletes at the 1928 Summer Olympics"})
+    print(result)
+
+    # result = web_page_information_extractor.invoke(
+    #     {"url": "https://chem.libretexts.org/Courses/Chabot_College/Introduction_to_General_Organic_and_Biochemistry/01%3A_Chemistry_in_our_Lives/1.E%3A_Exercises",
+    # "request": "What is the surname of the equine veterinarian mentioned"})
+    # print(result)
+
+    print(_get_text_from_url("https://en.wikipedia.org/wiki/Malko_Competition"))