patentwiz/preprocess_data.py

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