387 lines
15 KiB
Python
387 lines
15 KiB
Python
import argparse
|
|
import logging
|
|
import pandas as pd
|
|
from openpyxl import load_workbook
|
|
from rapidfuzz import process, fuzz
|
|
|
|
# ... (TRADE_KEYWORDS, assign_trade, is_line_item, extract_xactimate_items as before) ...
|
|
TRADE_KEYWORDS = {
|
|
"Floor Protection": ["floor protection", "cardboard", "protect floor", "mask floor"],
|
|
"Insulation": ["insulation", "batt", "fiberglass", "blown-in"],
|
|
"Drywall": ["drywall", "sheetrock", "tape joint", "texture", "patch", "mud", "repair wall", "joint compound", "corner bead"],
|
|
"Painting": ["paint", "painting", "primer", "prime", "seal", "coating", "enamel", "mask wall", "caulk", "caulking", "mirror", "towel bar", "toilet paper", "tp holder"],
|
|
"Baseboards, Trim, Casing": ["baseboard", "trim", "casing", "moulding", "crown", "shoe mould", "quarter round"],
|
|
"Doors": ["door", "door stop", "interior door", "slab"],
|
|
"Shower": ["shower", "shower pan", "shower door", "shower surround"],
|
|
"Laminate": ["laminate", "pergo", "engineered wood"],
|
|
"Vinyl Flooring": ["vinyl floor", "vinyl sheet"],
|
|
"Tile Flooring": ["tile floor", "ceramic tile", "porcelain tile", "grout", "thinset", "cement board", "grout", "clean floor and prep"],
|
|
"Carpet": ["carpet", "carpeting", "pad", "broadloom", "tack stip"],
|
|
"HVAC": ["hvac", "register", "ventilation"],
|
|
"Content Manipulation": ["content manipulation", "move contents", "protect contents", "cover contents", "contents"],
|
|
"Cleaning": ["clean", "cleaning", "final clean", "clean up", "final cleanup", "final cleaning", "construction clean"],
|
|
"Debris Removal": ["debris removal", "dump", "haul debris", "remove debris", "trash out"],
|
|
"Cabinets": ["cabinet", "vanity", "base cabinet", "wall cabinet", "countertop"],
|
|
"Electrical": ["electrical", "outlet", "switch", "receptacle", "breaker", "light fixture", "light", "ceiling fan"],
|
|
"Showers, Tubs, Tile": ["tub", "bathtub", "shower", "tile", "surround", "enclosure", "shower pan", "shower door", "shower surround"],
|
|
"Plumbing, Toilets, Sinks": ["plumbing", "toilet", "sink", "faucet", "supply line", "angle stop", "drain", "p-trap"],
|
|
"Labor Minimums": ["labor minimum"],
|
|
"Mitigation": ["water extraction", "remediation", "mitigation"]
|
|
}
|
|
|
|
HEADERS = ["DESCRIPTION", "TRADE", "QUANTITY", "UNIT PRICE", "TAX", "O&P", "RCV", "DEPREC.", "ACV"]
|
|
|
|
|
|
def assign_trade(description):
|
|
desc = description.lower()
|
|
for trade, keywords in TRADE_KEYWORDS.items():
|
|
if any(keyword in desc for keyword in keywords):
|
|
return trade
|
|
return "Other"
|
|
|
|
|
|
def clean_description(desc):
|
|
# Remove leading number and period, lowercase, and strip
|
|
desc = re.sub(r"^\d+\.\s*", "", desc)
|
|
desc = desc.lower().strip()
|
|
return desc
|
|
|
|
|
|
def write_masters_to_excel(items1, items2, excel_path):
|
|
df1 = pd.DataFrame(items1[1:], columns=items1[0])
|
|
df2 = pd.DataFrame(items2[1:], columns=items2[0])
|
|
with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
|
|
df1.to_excel(writer, sheet_name="Estimate1", index=False)
|
|
df2.to_excel(writer, sheet_name="Estimate2", index=False)
|
|
# Diff will be added in the next step
|
|
|
|
|
|
def extract_xactimate_items_with_fallback(pdf_path, assign_trade, HEADERS):
|
|
"""
|
|
Try robust extraction first; if no items, try classic extraction.
|
|
"""
|
|
# --- Robust multi-line regex extractor ---
|
|
def robust_extractor():
|
|
import re
|
|
import pdfplumber
|
|
line_item_regex = re.compile(
|
|
r"^(\d+\.)\s+(.+?)\s+([\d,.]+(?:SF|DA|HR|LF|EA|SY|UN|MO|WK|DY|BD|FT|YD|IN|CM|M|MM|LB|KG|GM|L|GAL|PC|SET|SQ|BOX|ROLL)?)\s+"
|
|
r"([\d,.]+)\s+([\d,.]+)\s+([\d,.]+)\s+([\d,.]+)\s+[\(<]([\d,.]+)[\)>]\s+([\d,.]+)(?:\s|$)"
|
|
)
|
|
extracted_items = []
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page in pdf.pages:
|
|
text = page.extract_text()
|
|
if not text:
|
|
continue
|
|
lines = text.split('\n')
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if re.match(r"^\d+\.\s", line):
|
|
combined_line = line
|
|
j = i + 1
|
|
while j < len(lines) and not re.match(r"^\d+\.\s", lines[j]):
|
|
combined_line += " " + lines[j].strip()
|
|
j += 1
|
|
match = line_item_regex.match(combined_line)
|
|
if match:
|
|
description = match.group(1) + " " + match.group(2)
|
|
quantity_unit = match.group(3)
|
|
unit_price = match.group(4)
|
|
tax = match.group(5)
|
|
o_p = match.group(6)
|
|
rcv = match.group(7)
|
|
deprec = match.group(8)
|
|
acv = match.group(9)
|
|
trade = assign_trade(description)
|
|
extracted_items.append([
|
|
description, trade, quantity_unit, unit_price, tax, o_p, rcv, deprec, acv
|
|
])
|
|
i = j
|
|
else:
|
|
i += 1
|
|
return [HEADERS] + extracted_items
|
|
|
|
# --- Classic single-line extractor ---
|
|
def classic_extractor():
|
|
import re
|
|
import pdfplumber
|
|
LINE_ITEM_REGEX = re.compile(
|
|
r"^(\d+\.\s+.+?)\s+(\d+\.\d+\s+(?:SF|LF|EA|HR|DA|SY))\s+([\d,.]+)\s+([\d,.]+)\s+([\d,.]+)\s+([\d,.]+)\s+\(([\d,.]+)\)\s+([\d,.]+)"
|
|
)
|
|
extracted_items = []
|
|
|
|
def is_line_item(line):
|
|
return bool(LINE_ITEM_REGEX.match(line))
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page in pdf.pages:
|
|
text = page.extract_text()
|
|
if not text:
|
|
continue
|
|
lines = text.split('\n')
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].strip()
|
|
if is_line_item(line):
|
|
match = LINE_ITEM_REGEX.match(line)
|
|
if match:
|
|
description = match.group(1)
|
|
quantity_unit = match.group(2)
|
|
unit_price = match.group(3)
|
|
tax = match.group(4)
|
|
o_p = match.group(5)
|
|
rcv = match.group(6)
|
|
deprec = match.group(7)
|
|
acv = match.group(8)
|
|
trade = assign_trade(description)
|
|
extracted_items.append([
|
|
description, trade, quantity_unit, unit_price, tax, o_p, rcv, deprec, acv
|
|
])
|
|
else:
|
|
if i + 1 < len(lines) and not is_line_item(lines[i + 1]):
|
|
combined_line = line + " " + lines[i + 1].strip()
|
|
match = LINE_ITEM_REGEX.match(combined_line)
|
|
if match:
|
|
description = match.group(1)
|
|
quantity_unit = match.group(2)
|
|
unit_price = match.group(3)
|
|
tax = match.group(4)
|
|
o_p = match.group(5)
|
|
rcv = match.group(6)
|
|
deprec = match.group(7)
|
|
acv = match.group(8)
|
|
trade = assign_trade(description)
|
|
extracted_items.append([
|
|
description, trade, quantity_unit, unit_price, tax, o_p, rcv, deprec, acv
|
|
])
|
|
i += 1
|
|
i += 1
|
|
return [HEADERS] + extracted_items
|
|
|
|
# --- Try robust, fallback to classic if needed ---
|
|
try:
|
|
data = robust_extractor()
|
|
if len(data) > 1:
|
|
return data
|
|
except Exception as e:
|
|
print("Robust extractor failed:", e)
|
|
print("Trying classic extractor...")
|
|
try:
|
|
data = classic_extractor()
|
|
if len(data) > 1:
|
|
return data
|
|
except Exception as e:
|
|
print("Classic extractor failed:", e)
|
|
print("No items extracted from", pdf_path)
|
|
return [HEADERS] # Only headers if nothing found
|
|
|
|
|
|
def create_diff_tab(items1, items2, excel_path):
|
|
df1 = pd.DataFrame(items1[1:], columns=items1[0])
|
|
df2 = pd.DataFrame(items2[1:], columns=items2[0])
|
|
|
|
# Merge on DESCRIPTION
|
|
merged = pd.merge(
|
|
df1, df2, on="DESCRIPTION", how="outer", suffixes=('_EST1', '_EST2'), indicator=True
|
|
)
|
|
|
|
# DEBUG
|
|
print("Estimate 1 sample:")
|
|
print(df1.head())
|
|
print("Estimate 2 sample:")
|
|
print(df2.head())
|
|
|
|
# DEBUG
|
|
print("DF1 columns:", df1.columns)
|
|
print("DF2 columns:", df2.columns)
|
|
|
|
# DEBUG
|
|
print("Estimate 1 DESCRIPTIONS:")
|
|
print(df1['DESCRIPTION'].head(20))
|
|
print("Estimate 2 DESCRIPTIONS:")
|
|
print(df2['DESCRIPTION'].head(20))
|
|
|
|
# DEBUG
|
|
print("Estimate 1 TRADES:")
|
|
print(df1['TRADE'].head(20))
|
|
print("Estimate 2 TRADES:")
|
|
print(df2['TRADE'].head(20))
|
|
|
|
# Mark differences
|
|
merged['DIFF'] = merged['_merge'].map({
|
|
'left_only': 'Only in Estimate1',
|
|
'right_only': 'Only in Estimate2',
|
|
'both': ''
|
|
}).astype(str)
|
|
|
|
# For items in both, check if any fields differ
|
|
for col in ["QUANTITY", "UNIT PRICE", "TAX", "O&P", "RCV", "DEPREC.", "ACV"]:
|
|
est1_col = col + "_EST1"
|
|
est2_col = col + "_EST2"
|
|
mask = (merged['_merge'] == 'both') & (merged[est1_col] != merged[est2_col])
|
|
merged.loc[mask, 'DIFF'] = 'Changed'
|
|
|
|
# Write to Excel (append to previous file)
|
|
with pd.ExcelWriter(excel_path, engine='openpyxl', mode='a', if_sheet_exists='replace') as writer:
|
|
merged.to_excel(writer, sheet_name="Diff", index=False)
|
|
|
|
|
|
def highlight_diff_sheet(filename, diff_sheet_name="Diff"):
|
|
from openpyxl import load_workbook
|
|
from openpyxl.styles import PatternFill
|
|
|
|
wb = load_workbook(filename)
|
|
ws = wb[diff_sheet_name]
|
|
|
|
# Find the DIFF column index
|
|
diff_col = None
|
|
for idx, cell in enumerate(ws[1], 1):
|
|
if cell.value == "DIFF":
|
|
diff_col = idx
|
|
break
|
|
|
|
yellow = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid")
|
|
red = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
|
|
green = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid")
|
|
|
|
for row in ws.iter_rows(min_row=2, max_row=ws.max_row):
|
|
diff_value = row[diff_col - 1].value
|
|
if diff_value == "Changed":
|
|
for cell in row:
|
|
cell.fill = yellow
|
|
elif diff_value == "Only in Estimate1":
|
|
for cell in row:
|
|
cell.fill = red
|
|
elif diff_value == "Only in Estimate2":
|
|
for cell in row:
|
|
cell.fill = green
|
|
|
|
wb.save(filename)
|
|
|
|
|
|
def sort_and_freeze_diff(filename, diff_sheet_name="Diff"):
|
|
from openpyxl import load_workbook
|
|
|
|
wb = load_workbook(filename)
|
|
ws = wb[diff_sheet_name]
|
|
|
|
# Read all rows into a list
|
|
data = list(ws.values)
|
|
headers = data[0]
|
|
rows = data[1:]
|
|
|
|
# Find DIFF and DESCRIPTION column indexes
|
|
diff_idx = headers.index("DIFF")
|
|
desc_idx = headers.index("DESCRIPTION")
|
|
|
|
# Sort: DIFF, then DESCRIPTION
|
|
rows.sort(key=lambda x: (x[diff_idx], x[desc_idx]))
|
|
|
|
# Write back to sheet
|
|
for i, row in enumerate([headers] + rows, 1):
|
|
for j, value in enumerate(row, 1):
|
|
ws.cell(row=i, column=j, value=value)
|
|
|
|
# Freeze header row
|
|
ws.freeze_panes = ws["A2"]
|
|
|
|
wb.save(filename)
|
|
|
|
|
|
def auto_fit_excel_columns(filename):
|
|
wb = load_workbook(filename)
|
|
for ws in wb.worksheets:
|
|
for column_cells in ws.columns:
|
|
max_length = 0
|
|
column = column_cells[0].column_letter
|
|
for cell in column_cells:
|
|
try:
|
|
cell_length = len(str(cell.value))
|
|
if cell_length > max_length:
|
|
max_length = cell_length
|
|
except:
|
|
pass
|
|
adjusted_width = max_length + 2
|
|
ws.column_dimensions[column].width = adjusted_width
|
|
wb.save(filename)
|
|
|
|
|
|
def find_fuzzy_matches(df1, df2, threshold=85):
|
|
"""
|
|
For each item in df1, find the closest DESCRIPTION_CLEAN in df2.
|
|
Returns a DataFrame with best match and score.
|
|
"""
|
|
matches = []
|
|
choices = df2['DESCRIPTION_CLEAN'].tolist()
|
|
for idx, row in df1.iterrows():
|
|
desc = row['DESCRIPTION_CLEAN']
|
|
# Find best match in df2
|
|
best_match, score, match_idx = process.extractOne(
|
|
desc, choices, scorer=fuzz.token_sort_ratio)
|
|
matches.append((idx, best_match, score, match_idx))
|
|
match_df = pd.DataFrame(matches, columns=['df1_idx', 'best_match', 'score', 'df2_idx'])
|
|
return match_df
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Compare two Xactimate estimates and highlight differences in Excel."
|
|
)
|
|
parser.add_argument("estimate1", help="Path to the first PDF file")
|
|
parser.add_argument("estimate2", help="Path to the second PDF file")
|
|
parser.add_argument("excel_file", help="Path to the output Excel file")
|
|
args = parser.parse_args()
|
|
|
|
logging.info(f"Extracting line items from {args.estimate1} ...")
|
|
items1 = extract_xactimate_items_with_fallback(args.estimate1, assign_trade, HEADERS)
|
|
|
|
logging.info(f"Extracting line items from {args.estimate2} ...")
|
|
items2 = extract_xactimate_items_with_fallback(args.estimate2, assign_trade, HEADERS)
|
|
|
|
write_masters_to_excel(items1, items2, args.excel_file)
|
|
create_diff_tab(items1, items2, args.excel_file)
|
|
|
|
highlight_diff_sheet(args.excel_file)
|
|
sort_and_freeze_diff(args.excel_file)
|
|
|
|
# --- ADDED: Find missing and extra items ---
|
|
import pandas as pd
|
|
import re
|
|
|
|
df1 = pd.DataFrame(items1[1:], columns=items1[0])
|
|
df2 = pd.DataFrame(items2[1:], columns=items2[0])
|
|
|
|
def clean_description(desc):
|
|
desc = re.sub(r"^\d+\.\s*", "", desc)
|
|
desc = desc.lower().strip()
|
|
return desc
|
|
|
|
df1['DESCRIPTION_CLEAN'] = df1['DESCRIPTION'].apply(clean_description)
|
|
df2['DESCRIPTION_CLEAN'] = df2['DESCRIPTION'].apply(clean_description)
|
|
|
|
# missing = df1[~df1['DESCRIPTION_CLEAN'].isin(df2['DESCRIPTION_CLEAN'])]
|
|
# extra = df2[~df2['DESCRIPTION_CLEAN'].isin(df1['DESCRIPTION_CLEAN'])]
|
|
|
|
match_df = find_fuzzy_matches(df1, df2, threshold=85)
|
|
good_matches = match_df[match_df['score'] >= 85]
|
|
|
|
matched_df1 = set(good_matches['df1_idx'])
|
|
matched_df2 = set(good_matches['df2_idx'])
|
|
|
|
missing = df1.loc[~df1.index.isin(matched_df1)]
|
|
extra = df2.loc[~df2.index.isin(matched_df2)]
|
|
|
|
with pd.ExcelWriter(args.excel_file, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer:
|
|
missing[['DESCRIPTION', 'TRADE', 'QUANTITY', 'UNIT PRICE', 'RCV', 'ACV']
|
|
].to_excel(writer, sheet_name="MissingFromInsurance", index=False)
|
|
extra[['DESCRIPTION', 'TRADE', 'QUANTITY', 'UNIT PRICE', 'RCV', 'ACV']
|
|
].to_excel(writer, sheet_name="ExtraInInsurance", index=False)
|
|
|
|
auto_fit_excel_columns(args.excel_file)
|
|
|
|
print(f"Comparison complete! Results saved to {args.excel_file}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|