File size: 8,925 Bytes
eb569a5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 |
import os
import requests
import zipfile
import xml.etree.ElementTree as ET
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 extract_patents(year, month, day, logging):
"""
This function reads a patent file in XML format, splits it into individual patents, parse 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)
description_string = " ".join(description_text)
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(year, month, day, logging=False):
"""
Download weekly patent files from the USPTO website for a specific date, extract individual
patents from the downloaded file, parse each patent's content, and save the information
as separate text files.
Parameters:
year (int): The year of the weekly patents to download and process.
month (int): The month of the weekly patents to download and process.
day (int): The day of the weekly patents to download and process.
logging (bool): The boolean to print logs
Returns:
list: A list of strings containing the names of saved patent text files.
This function first downloads the weekly patent file, then extracts the individual patents,
and finally parses each patent's content to retrieve patent_id, file_id, and full text.
It saves the extracted information for each patent as separate text files in a directory
named 'data', with the name of each file being the corresponding 'file_id'.
The function returns a list of strings containing the names of all the saved patent text files.
"""
if logging:
print("### Downloading weekly patent files...")
download_success = download_weekly_patents(year, month, day, logging)
if not download_success:
print("Failed to download the weekly patents.")
return
if logging:
print("### Extracting individual patents...")
saved_patent_names = extract_patents(year, month, day, logging)
return saved_patent_names |