CommonCrawl_URL_Processor/warc_wat_url_processor.py

78 lines
2.6 KiB
Python

import subprocess
import os
import gzip
import zstandard as zstd
import re
import traceback
from multiprocessing import Pool
def extract_urls_from_file(file_path):
urls = []
try:
with gzip.open(file_path, 'rt', encoding='latin-1') as file:
for line in file:
url_pattern = re.compile(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+')
line_urls = re.findall(url_pattern, line)
urls.extend(line_urls)
except (gzip.BadGzipFile, EOFError) as e:
print(f"Error while reading the compressed file '{file_path}': {e}")
except Exception as e:
print(f"An unexpected error occurred while processing '{file_path}': {e}")
print("Full traceback:")
traceback.print_exc()
return urls
def process_file(file_path):
print(f"Processing file: {file_path}")
# Extract URLs from the gzipped file
urls = extract_urls_from_file(file_path)
# Create the output file path with '_urls.txt' extension
output_file_path = os.path.splitext(file_path)[0] + '_urls.txt'
print(f"Output file path: {output_file_path}")
# Write the URLs to the output file
with open(output_file_path, 'w') as output_file:
output_file.write('\n'.join(urls))
print(f"URLs written to {output_file_path}")
# Compress the output file using zstd with compression level -18
with open(output_file_path, 'rb') as input_file, open(output_file_path + '.zst', 'wb') as output_zstd_file:
cctx = zstd.ZstdCompressor(level=18)
output_zstd_file.write(cctx.compress(input_file.read()))
print(f"Compressed file saved as '{output_file_path}.zst'")
# Remove the original gzipped file
os.remove(file_path)
print(f"Original file removed: {file_path}")
def extract_urls_from_directory(directory_path):
file_list = sorted(os.listdir(directory_path))
pool = Pool(processes=7)
pool.map(process_file, [os.path.join(directory_path, filename) for filename in file_list if filename.endswith('.warc.gz')])
pool.close()
pool.join()
with open('urls_to_download.txt', 'r') as file:
urls = file.readlines()
urls = [url.strip() for url in urls]
batch_size = 48
concurrency_level = 4
batches = [urls[i:i+batch_size] for i in range(0, len(urls), batch_size)]
for batch in batches:
pool = Pool(processes=concurrency_level)
for url in batch:
command = f'axel -n 4 {url}'
pool.apply_async(subprocess.run, args=(command,), kwds={'shell': True})
pool.close()
pool.join()
extract_urls_from_directory(os.getcwd())