patentwiz/preprocess_data.py

import os
import requests
import zipfile
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.
    Parameters:
        patents (list): List of patent texts (as strings or structured data).
        keywords (list): Keywords to filter patents.
        fields (list): Fields to search for keywords (e.g., Title, Abstract, Claims).
    Returns:
        list: Filtered patents.
    """
    if keywords is None:
        keywords = ["Radio Frequency", "Antenna", "UAV", "Wireless Charging"]  # Default keywords
    if fields is None:
        fields = ["Title", "Abstract"]  # Default fields

    filtered_patents = []
    for patent in patents:
        # If patent is a string, search for keywords in the entire text
        if isinstance(patent, str):
            if any(keyword.lower() in patent.lower() for keyword in keywords):
                filtered_patents.append(patent)
        # If patent is structured (e.g., dictionary), search within fields
        elif isinstance(patent, dict):
            for field in fields:
                field_content = patent.get(field.lower(), "")
                if any(keyword.lower() in field_content.lower() for keyword in keywords):
                    filtered_patents.append(patent)
                    break
    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