[Python] Deduplication by Text Blocks with filtering

I am trying to create a python script that will remove consecutive duplicate text blocks from a text file, ie do something like this:

1. Locate text blocks that are the same, based on # of rows (with a max block size X that determines the initial max rows per text block, then reduces to 1 row at a time)
2. If there are consecutive X row text block duplicates, remove them and replace the duplicates with a simple "...has an additional X similar entries..." after the initial text block that has duplicates following it and replaces those duplicates.

I also want to ensure when compariing that

1. any 2+ spaces in the text rows are repalced with 1 space for the coparison
2. any digits are removed (including 'words that start with a number and end with a numbner or period" - which will remove dates eg: XX/XX/XX, comma/decimal numbers eg: 3,444.22, and points eg: 1.) from each row
3. Skips empty lines with strip()'d for text block comparing (ie ignored)

The result will be a new txt file that has the duplicate text blocks removed.

I have the row clean function for the comparison like this:

def clean_for_comparison(line):
# Remove dollar amounts (e.g., $1,234.56) - additional example of cleaning row
line = re.sub(r'\$[\d,]+(?:\.\d+)?', '', line)
# Remove numbers with commas and decimals (e.g., 1,234.56)
line = re.sub(r'[\d,]+(?:\.\d+)?', '', line)
# Remove date-like patterns (e.g., 17/01/2020 or 17-01-2020)
line = re.sub(r'\d{1,2}[-/.\s]?\d{1,2}[-/.\s]?\d{2,4}', '', line)
# Remove words starting and ending with digits (e.g., 123abc456)
line = re.sub(r'\b\d+[a-zA-Z]+\d+\b', '', line)
# Normalize spaces and strip
line = re.sub(r'\s+', ' ', line)
line = line.strip()
return line


I am trying to use a hash method for the text block compare by constructing the block using X rows, but am having issues with how it should work effiicently.

I use these helper functions:

def get_block_hash(block):
cleaned_lines = [clean_for_comparison(line) for line in block]
non_blank_cleaned_lines = [line for line in cleaned_lines if line]
cleaned_block_content = ''.join(non_blank_cleaned_lines)
return hashlib.md5(cleaned_block_content.encode()).hexdigest()

def get_block(lines, start_index, max_size):
block = []
i = start_index
row_count = 0

while i < len(lines) and row_count < max_size:
line = lines.strip()
if line:
block.append(lines)
row_count += 1
i += 1

if i >= len(lines):
break
if lines[i - 1].strip() == "" and lines.strip() == "":
break

return block, i


but the it gets confusing trying to do it properly/efficient....so far I was looking at this approach, but it's not quite right...

def process_file(file_path, max_consecutive_rows_to_check=4):
with open(file_path, 'r') as file:
lines = file.readlines()

filtered_lines = []
i = 0
removed_count = 0

while i < len(lines):
block, j = get_block(lines, i, max_consecutive_rows_to_check)
if not block: # Skip processing.
i = j # Increment
continue # Skip empty block

# Sliding window comparison
for size in range(min(len(block), max_consecutive_rows_to_check), 0, -1):
block_to_compare = block[:size]
position_row = initial_block_start

while True:
compare_start = position_row +1 # use the next row outside the intial block to compare?
compare_block, next_k = get_block(lines, compare_start, size)
compare_line = compare_block

if len(compare_block) != size:
break # Done with comapre - no more
if get_block_hash(block_to_compare) == get_block_hash(compare_block):
print("DUPE FOUND!")
found_dupes = True
total_dupes += 1 # dupe counter
position_row = compare_start #set next dupe position to compare consecutively with
i += compare_start
#continue #continue if it finds?
else: #Did not find anything
print("Not a dupe.")
break

# If duplicates were found:
if found_dupes:
print(f"Total duplicates found: {total_dupes}")
filtered_lines.extend(block)
filtered_lines.append(f"....<with an additional {total_dupes} entries found>...\n")
removed_count += total_dupes

i = position_row+max_consecutive_rows_to_check #+1 + size?

# If no duplicates were found:
else:
print("No duplicates found, adding to filtered lines.")
filtered_lines.extend(block)

# Advance to next block
i =+max_consecutive_rows_to_check

# Output to new file
output_filename = file_path
if output_filename.endswith(".txt"):
output_filename = output_filename[:-4]
if not output_filename.endswith("-truncated"):
output_filename = f"{output_filename}-truncated.txt"

with open(output_filename, "w") as output_file:
output_file.writelines(filtered_lines)

print(f"Processed {len(lines)} lines from {file_path}")
print(f"Output saved to {output_filename}")
if removed_count > 0:
print(f"Removed {removed_count} duplicate blocks.")


....etc..

Any help in logic flow, efficenice or if anyone knows of any exist lib that de-dupes text files based on consecutive text blocks?

Thanks

Continue reading...