Update mylab/attempt1/preprocess_data.py

mylab/attempt1/preprocess_data.py

@@ -0,0 +1,281 @@
+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