Create preprocess_data.py

mylab/attempt2/preprocess_data.py

@@ -0,0 +1,345 @@
+import os
+import requests
+import zipfile
+import streamlit as st 
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta
+import tempfile
+import pickle
+
+
+def download_weekly_patents(year, month, day, logging):
+    """
+    Download weekly patent files from the USPTO website based on a specific date.
+    Parameters:
+    year (int): The year of the patent.
+    month (int): The month of the patent.
+    day (int): The day of the patent.
+    logging (bool): The boolean to print logs
+    Returns:
+    bool: True if the download is successful, False otherwise.
+    """
+
+    # Check if the "data" folder exists and create one if it doesn't
+    data_folder = os.path.join(os.getcwd(), "data")
+    if not os.path.exists(data_folder):
+        if logging:
+            print("Data folder not found. Creating a new 'data' folder.")
+        os.makedirs(data_folder)
+
+    directory = os.path.join(
+        os.getcwd(), "data", "ipa" + str(year)[2:] + f"{month:02d}" + f"{day:02d}"
+    )
+
+    if os.path.exists(directory):
+        print(f"File {directory} already exists. Skipping download.")
+        return True
+
+    if logging:
+        print("Building the URL...")
+    base_url = "https://bulkdata.uspto.gov/data/patent/application/redbook/fulltext"
+    file_url = (
+        base_url
+        + "/"
+        + str(year)
+        + "/ipa"
+        + str(year)[2:]
+        + f"{month:02d}"
+        + f"{day:02d}"
+        + ".zip"
+    )
+
+    if logging:
+        print(f"URL constructed: {file_url}")
+    r = requests.get(file_url, stream=True)
+
+    if logging:
+        print("Requesting the file...")
+    if r.status_code == 200:
+        if logging:
+            print("File retrieved successfully. Starting download...")
+        local_path = os.path.join(os.getcwd(), "data", "patents.zip")
+
+        with open(local_path, "wb") as f:
+            for chunk in r.iter_content(chunk_size=1024):
+                if chunk:
+                    f.write(chunk)
+        if logging:
+            print("File downloaded successfully. Starting extraction...")
+        with zipfile.ZipFile(local_path, "r") as zip_ref:
+            zip_ref.extractall(os.path.join(os.getcwd(), "data"))
+
+        if logging:
+            print("File extracted successfully.")
+        # Deleting the ZIP file after extraction
+        os.remove(local_path)
+        if logging:
+            print(f"ZIP file {local_path} deleted after extraction.")
+
+        return True
+    else:
+        print(
+            "File could not be downloaded. Please make sure the year, month, and day are correct."
+        )
+        return False
+
+def filter_rf_patents(patents, keywords=None, fields=None):
+    """
+    Filters patents based on keywords and specified fields, with parsing for raw patent files.
+    """
+    import streamlit as st  # Use Streamlit for debugging
+
+    if keywords is None:
+        keywords = ["Radio Frequency", "Antenna", "UAV", "Wireless Charging"]  # Default keywords
+    if fields is None:
+        fields = ["Title", "Abstract", "Summary", "Claims", "Detailed Description"]  # Default fields
+
+    # Standardize field names
+    FIELD_NAME_MAPPING = {
+        "abstract": "Abstract",
+        "ABSTRACT": "Abstract",
+        "summary": "Summary",
+        "SUMMARY": "Summary",
+        "claims": "Claims",
+        "CLAIMS": "Claims",
+        "detailed description": "Detailed Description",
+        "DETAILED DESCRIPTION": "Detailed Description",
+        "title": "Title",
+        "TITLE": "Title",
+    }
+
+    def parse_patent(file_path):
+        """
+        Parses an XML patent file into a structured dictionary.
+        """
+        try:
+            tree = ET.parse(file_path)
+            root = tree.getroot()
+
+            # Extract fields from XML (adjust based on actual XML structure)
+            patent_data = {
+                "Title": root.findtext(".//title", default=""),
+                "Abstract": root.findtext(".//abstract", default=""),
+                "Summary": root.findtext(".//summary", default=""),
+                "Claims": root.findtext(".//claims", default=""),
+                "Detailed Description": root.findtext(".//detailedDescription", default=""),
+            }
+
+            # Normalize field names
+            normalized_patent = {}
+            for field, content in patent_data.items():
+                normalized_field = FIELD_NAME_MAPPING.get(field, field)
+                normalized_patent[normalized_field] = content.strip() if content else ""
+
+            return normalized_patent
+        except Exception as e:
+            st.write(f"Error parsing patent {file_path}: {e}")
+            return None
+
+    filtered_patents = []
+
+    # Display first 5 patents for inspection (before parsing)
+    st.write("Debugging: First 5 raw patents for inspection")
+    for patent in patents[:5]:
+        st.write(patent)  # Display raw data
+
+    for patent in patents:
+        if isinstance(patent, str): 
+            parsed_patent = parse_patent(patent)
+            if not parsed_patent:
+                continue
+        elif isinstance(patent, dict):  
+            parsed_patent = patent
+        else:
+            st.write(f"Unknown patent format: {type(patent)}")
+            continue
+
+        # Field-specific matching
+        matched = False
+        for field in fields:
+            field_content = parsed_patent.get(field, "")
+            st.write(f"Checking field '{field}': {field_content}")
+            if field_content and any(keyword.lower() in field_content.lower() for keyword in keywords):
+                st.write(f"Match found in field '{field}'")
+                filtered_patents.append(parsed_patent)
+                matched = True
+                break
+
+        # Global fallback if no fields match
+        if not matched:
+            full_text = " ".join(parsed_patent.values())  # Combine all fields
+            if any(keyword.lower() in full_text.lower() for keyword in keywords):
+                st.write("Match found in global fallback search!")
+                filtered_patents.append(parsed_patent)
+
+    st.write(f"Total filtered patents: {len(filtered_patents)}")
+    return filtered_patents
+
+
+
+
+def extract_patents(year, month, day, logging):
+    """
+    This function reads a patent file in XML format, splits it into individual patents, parses each
+    XML file, and saves each patent as a separate txt file in a directory named 'data'.
+    Parameters:
+    year (int): The year of the patent file to process.
+    month (int): The month of the patent file to process.
+    day (int): The day of the patent file to process.
+    logging (bool): The boolean to print logs
+    Returns:
+    None
+    The function creates a separate XML file for each patent and stores these files in
+    a directory. The directory is named based on the year, month, and day provided.
+    If the directory does not exist, the function creates it. The function also prints
+    the total number of patents found.
+    """
+
+    directory = os.path.join(
+        os.getcwd(), "data", "ipa" + str(year)[2:] + f"{month:02d}" + f"{day:02d}"
+    )
+    saved_patent_names_path = os.path.join(directory, 'saved_patent_names.pkl')
+
+    if os.path.exists(directory):
+        print(f"File {directory} already exists. Skipping extract.")
+        
+        # Load saved_patent_names from file
+        with open(saved_patent_names_path, 'rb') as f:
+            saved_patent_names = pickle.load(f)
+            
+        return saved_patent_names
+    else:
+        os.mkdir(directory)
+
+    if logging:
+        print("Locating the patent file...")
+    file_path = os.path.join(
+        os.getcwd(),
+        "data",
+        "ipa" + str(year)[2:] + f"{month:02d}" + f"{day:02d}" + ".xml",
+    )
+
+    if logging:
+        print("Reading the patent file...")
+    with open(file_path, "r") as f:
+        contents = f.read()
+
+    if logging:
+        print("Splitting the XML file into individual XMLs...")
+    temp = contents.split('<?xml version="1.0" encoding="UTF-8"?>')
+    allXmls = [
+        '<?xml version="1.0" encoding="UTF-8"?>' + s.replace("\n", "") for s in temp
+    ]
+
+    # saving only the XMLs that contain a patent
+    patents = []
+    for xml_string in allXmls:
+        start_index = xml_string.find("<!DOCTYPE")
+        end_index = xml_string.find(">", start_index)
+
+        if start_index != -1 and end_index != -1:
+            doctype_declaration = xml_string[start_index : end_index + 1]
+            # Extract only the name of the DOCTYPE
+            doctype_name = doctype_declaration.split()[1]
+            if doctype_name == "us-patent-application":
+                patents.append(xml_string)
+
+    if logging:
+        print(f"Total patents found: {len(patents)}")
+        print("Writing individual patents to separate txt files...")
+    
+    saved_patent_names = []
+    for patent in patents:
+        try:
+            root = ET.fromstring(patent)
+
+            patent_id = root.find(
+                ".//publication-reference/document-id/doc-number"
+            ).text
+            file_id = root.attrib["file"]
+
+            ipcr_classifications = root.findall(".//classification-ipcr")
+
+            if any(ipcr.find("./section").text == "C" for ipcr in ipcr_classifications):
+                description_element = root.find(".//description")
+                description_text = get_full_text(description_element)
+
+                # Filter RF-relevant content
+                filtered_description = filter_rf_patents(description_text)
+                if filtered_description:
+                    description_string = " ".join(filtered_description)
+                    output_file_path = os.path.join(directory, f"{file_id}.txt")
+                    with open(output_file_path, "w") as f:
+                        f.write(description_string)
+                    saved_patent_names.append(f"{file_id}.txt")
+
+            elif logging:
+                print(
+                    f"Patent {patent_id} does not belong to section 'C'. Skipping this patent."
+                )
+        except ET.ParseError as e:
+            print(f"Error while parsing patent: {patent_id}. Skipping this patent.")
+            print(f"Error message: {e}")
+
+    # Save saved_patent_names to file
+    with open(saved_patent_names_path, 'wb') as f:
+        pickle.dump(saved_patent_names, f)
+
+    if logging:
+        print("Patent extraction complete.")
+
+    # Deleting the main XML file after extraction
+    os.remove(file_path)
+
+    if logging:
+        print(f"Main XML file {file_path} deleted after extraction.")
+    return saved_patent_names
+
+
+def get_full_text(element):
+    """
+    Recursively parse XML elements and retrieve the full text from the XML tree.
+    Parameters:
+        element (xml.etree.ElementTree.Element): The root XML element to start parsing.
+    Returns:
+        list: A list of strings containing the full text from the XML element and its children.
+    """
+
+    text = []
+    if element.text is not None and element.text.strip():
+        text.append(element.text.strip())
+    for child in element:
+        text.extend(get_full_text(child))
+        if child.tail is not None and child.tail.strip():
+            text.append(child.tail.strip())
+    return text
+
+
+def parse_and_save_patents(start_date, end_date, logging=False):
+    """
+    Download weekly patent files from the USPTO website for a range of dates, extract individual
+    patents from the downloaded file, parse each patent's content, and save the information
+    as separate text files.
+    Parameters:
+        start_date (datetime): The start date of the range.
+        end_date (datetime): The end date of the range.
+        logging (bool): The boolean to print logs
+    Returns:
+        list: A list of strings containing the names of saved patent text files.
+    """
+    all_saved_patent_names = []
+
+    current_date = start_date
+    while current_date <= end_date:
+        year, month, day = current_date.year, current_date.month, current_date.day
+        if logging:
+            print(f"Processing patents for {current_date.strftime('%Y-%m-%d')}...")
+
+        download_success = download_weekly_patents(year, month, day, logging)
+        if download_success:
+            saved_patent_names = extract_patents(year, month, day, logging)
+            all_saved_patent_names.extend(saved_patent_names)
+
+        current_date += timedelta(days=7)  # USPTO weekly files are organized by week
+
+    return all_saved_patent_names