from collections import defaultdict import pandas as pd import random import re import io import pypdfium2 as pdfium import fitz from PIL import Image, ImageDraw from PyPDF2 import PdfReader, PdfWriter from PyPDF2.generic import TextStringObject, NameObject, ArrayObject, FloatObject from PyPDF2.generic import NameObject, TextStringObject, DictionaryObject, FloatObject, ArrayObject from PyPDF2 import PdfReader from PyPDF2.generic import TextStringObject import numpy as np import cv2 def convert2img(path): pdf = pdfium.PdfDocument(path) page = pdf.get_page(0) pil_image = page.render().to_pil() pl1=np.array(pil_image) img = cv2.cvtColor(pl1, cv2.COLOR_RGB2BGR) return img def convert2pillow(path): pdf = pdfium.PdfDocument(path) page = pdf.get_page(0) pil_image = page.render().to_pil() return pil_image def calculate_midpoint(x1,y1,x2,y2): xm = int((x1 + x2) / 2) ym = int((y1 + y2) / 2) return (xm, ym) def read_text(input_pdf_path): pdf_document = fitz.open('pdf',input_pdf_path) for page_num in range(pdf_document.page_count): page = pdf_document[page_num] text_instances = page.get_text("words") page.apply_redactions() return text_instances def normalize_text(text): """ Normalize text by removing all whitespace characters and converting to lowercase. """ if not isinstance(text, str): return "" # Remove all whitespace characters (spaces, tabs, newlines) text = re.sub(r'\s+', '', text) return text.lower() def build_flexible_regex(term): """ Match the full string, allowing whitespace or light punctuation between words, but not allowing extra words or partial matches. """ words = normalize_text(term).split() pattern = r'[\s\.\:\-]*'.join(map(re.escape, words)) full_pattern = rf'^{pattern}$' return re.compile(full_pattern, re.IGNORECASE) def flexible_search(df, search_terms): """ Search for terms in column names and top N rows. Returns matched column indices and cell positions. """ normalized_columns = [normalize_text(col) for col in df.columns] results = {term: {"col_matches": [], "cell_matches": []} for term in search_terms} for term in search_terms: regex = build_flexible_regex(term) # Search in column names for col_idx, col_text in enumerate(df.columns): norm_col = normalize_text(col_text) if regex.search(norm_col): results[term]["col_matches"].append(col_idx) # Search in top N rows for row_idx in range(min(3, len(df))): for col_idx in range(len(df.columns)): cell_text = normalize_text(df.iat[row_idx, col_idx]) if regex.search(cell_text): results[term]["cell_matches"].append((row_idx, col_idx)) return results def generate_current_table_without_cropping(clm_idx, clmn_name, df): selected_df = df.iloc[:, clm_idx] print("hello I generated the selected columns table without cropping") selected_df.columns = clmn_name return selected_df def crop_rename_table(indices, clmn_name, clmn_idx,df): #crop_at = (max(set(indices), key=indices.count)) + 1 crop_at = max(indices) + 1 df = df.iloc[crop_at:] # Starts from row index 5 (zero-based index) df.reset_index(drop=True, inplace=True) # Reset index after cropping slctd_clms = df.iloc[:, clmn_idx] # Select columns by index slctd_clms.columns = clmn_name # Rename selected columns return slctd_clms def clean_column_row(row): return [re.sub(r'^\d+-\s*', '', str(cell)) for cell in row] def details_in_another_table(clmn_name, clmn_idx, current_dfs, dfs): matching_dfs = [ dff for dff in dfs if dff is not current_dfs and current_dfs.shape[1] == dff.shape[1] ] if not matching_dfs: return None updated_dfs = [] for dff in matching_dfs: selected_dff = dff.iloc[:, clmn_idx].copy() # Clean the column names and make them a row cleaned_header = clean_column_row(selected_dff.columns.tolist()) col_names_as_row = pd.DataFrame([cleaned_header]) # Rename columns selected_dff.columns = clmn_name col_names_as_row.columns = clmn_name # Combine the cleaned row with data temp_df = pd.concat([col_names_as_row, selected_dff], ignore_index=True) updated_dfs.append(temp_df) combined_df = pd.concat(updated_dfs, ignore_index=True) return combined_df def map_user_input_to_standard_labels(user_inputs): patterns = { 'door_id': r'\b(?:door\s*)?(?:id|no|number)\b|\bdoor\s*name\b', 'door_type': r'\b(?:\S+\s+)?door\s*type\b|\btype(?:\s+\w+)?\b', 'structural_opening': r'\bstructural\s+opening\b', 'width': r'\bwidth\b', 'height': r'\bheight\b', } def normalize(text): return re.sub(r'\s+', ' ', text.strip(), flags=re.MULTILINE).lower() mapped = {} for item in user_inputs: normalized_item = normalize(item) matched = False for label, pattern in patterns.items(): if label not in mapped and re.search(pattern, normalized_item, re.IGNORECASE): mapped[label] = item matched = True break #if not matched: # mapped[normalized_item] = None return mapped def analyse_cell_columns(cell_columns_appearance): cell_matches = [] col_matches = [] for key in cell_columns_appearance.keys(): if len(cell_columns_appearance[key]['cell_matches']) >0: cell_matches.append(cell_columns_appearance[key]['cell_matches'][0]) if len(cell_columns_appearance[key]['col_matches']) >0: col_matches.append(cell_columns_appearance[key]['col_matches'][0]) return cell_matches, col_matches # when column names are located in the cells def get_row_column_indices(cell_clmn_indx): row_index = [] column_index = [] for t in cell_clmn_indx: row_index.append(t[0]) column_index.append(t[1]) return row_index, column_index # when column names are located in the coulmns itself def get_column_index(col_matches): idx = [] for t in col_matches: idx.append(t) return idx def extract_tables(schedule): doc = fitz.open("pdf",schedule) for page in doc: tabs = page.find_tables() dfs = [] for tab in tabs: df = tab.to_pandas() dfs.append(df) return dfs def get_selected_columns(dfs, user_patterns): selected_columns = [] selected_columns_new = None # Initialize selected_columns_new to None for i in range(len(dfs)): cell_columns_appearance = flexible_search(dfs[i], user_patterns) cell_matches, col_matches = analyse_cell_columns(cell_columns_appearance) if len(user_patterns) == 2: clmn_name = ["door_id", "door_type"] if len(user_patterns) == 4: clmn_name = ["door_id", "door_type", "width", "height"] if len(user_patterns) == 3: clmn_name = ["door_id", "door_type", "structural opening"] if len(cell_matches) == 0 and len(col_matches) == 0: print(f"this is df {i}, SEARCH IN ANOTHER DF") else: #IN COLUMNS if len(col_matches) == len(user_patterns): column_index_list = get_column_index(col_matches) print(f"this is df {i} mawgooda fel columns, check el df length 3ashan law el details fe table tany") print(column_index_list) if len(dfs[i]) <10: selected_columns_new = details_in_another_table(clmn_name, column_index_list, dfs[i], dfs) #details in the same table if len(dfs[i]) >10: selected_columns_new = generate_current_table_without_cropping(column_index_list,dfs[i]) #break #IN CELLS if len(cell_matches) == len(user_patterns): row_index_list, column_index_list = get_row_column_indices(cell_matches) print(f"this is df {i} mawgooda fel cells, check el df length 3ashan law el details fe table tany") #details in another table if len(dfs[i]) <10: #selected_columns_new = details_in_another_table(clmn_name, clmn_idx, dfs[i], dfs) selected_columns_new = details_in_another_table(clmn_name, column_index_list, dfs[i], dfs) break #details in the same table if len(dfs[i]) >10: print(f"this is df {i} call crop_rename_table(indices, clmn_name, clmn_idx,df)") selected_columns_new = crop_rename_table(row_index_list, clmn_name, column_index_list,dfs[i]) break return selected_columns_new # 3ayz akhaleehaa te search fel selected_columns column names nafsaha # 7ab2a 3ayz a3raf bardo maktooba ezay fel df el 7a2e2ya (akeed za ma el user medakhalha bezabt) def get_st_op_pattern(selected_columns, user_input): target = 'structural opening' if target in selected_columns.columns: name = user_input[2] return name return None def get_similar_colors(selected_columns_new): def generate_rgb(): return (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) # RGB tuple unique_keys = selected_columns_new['door_type'].unique() key_colors = {key: generate_rgb() for key in unique_keys} # Assign a unique RGB color to each key # Create dictionary storing values, colors, and widths if 'structural_opening' in selected_columns_new.columns: col_dict = defaultdict(lambda: {'values': [], 'color': None, 'widths': []}) else: col_dict = defaultdict(lambda: {'values': [], 'color': None, 'widths': [], 'heights': []}) if selected_columns_new.shape[1] == 2: col_dict = defaultdict(lambda: {'values': [], 'color': None}) for _, row in selected_columns_new.iterrows(): key = row['door_type'] col_dict[key]['values'].append(row['door_id']) if 'structural_opening' in selected_columns_new.columns: col_dict[key]['widths'].append(row['structural_opening']) # Add structural opening else: if selected_columns_new.shape[1] > 2: col_dict[key]['widths'].append(row['width']) # Assuming 'widht' is a typo for 'width' col_dict[key]['heights'].append(row['height']) col_dict[key]['color'] = key_colors[key] # Assign the unique RGB color # Convert defaultdict to a normal dictionary col_dict = dict(col_dict) return col_dict def get_flattened_tuples_list(col_dict): tuples_list = [] for key, values_dict in col_dict.items(): if 'heights' in values_dict and 'widths' in values_dict: # Case: Both widths and heights present tuples_list.append([ (value, width, height, values_dict["color"]) for value, width, height in zip(values_dict['values'], values_dict['widths'], values_dict['heights']) ]) elif 'widths' in values_dict: # Case: Only widths present tuples_list.append([ (value, width, values_dict["color"]) for value, width in zip(values_dict['values'], values_dict['widths']) ]) else: # Case: Neither widths nor heights tuples_list.append([ (value, values_dict["color"]) for value in values_dict['values'] ]) # Flatten the list of lists flattened_list = [item for sublist in tuples_list for item in sublist] return flattened_list def find_text_in_plan(label, x): substring_coordinates = [] words = [] point_list = [] #None, None, None for tpl in x: if tpl[4] == label: substring_coordinates.append(calculate_midpoint(tpl[0],tpl[1],tpl[2],tpl[3]))# for pdf point_list.append(calculate_midpoint(tpl[1],tpl[0],tpl[3],tpl[2]))# for rotated words.append(tpl[4]) return substring_coordinates, words, point_list def get_word_locations_plan(flattened_list, plan_texts): locations = [] not_found = [] if len(flattened_list[0]) == 2: for lbl, clr in flattened_list: location,worz, txt_pt = find_text_in_plan(lbl, plan_texts) if len(location) ==0: not_found.append(lbl) locations.append((location, lbl, clr)) if len(flattened_list[0]) == 3: for lbl, w, clr in flattened_list: location,worz, txt_pt = find_text_in_plan(lbl, plan_texts) if len(location) ==0: not_found.append(lbl) locations.append((location, lbl, clr, w)) if len(flattened_list[0]) == 4: for lbl, w, h, clr in flattened_list: location,worz, txt_pt = find_text_in_plan(lbl, plan_texts) if len(location) ==0: not_found.append(lbl) locations.append((location, lbl, clr, w, h)) return locations, not_found def get_repeated_labels(locations): seen_labels = set() repeated_labels = set() for item in locations: label = item[1] if label in seen_labels: repeated_labels.add(label) else: seen_labels.add(label) return repeated_labels def get_cleaned_data(locations): processed = defaultdict(int) new_data = [] if len(locations[0]) == 3: for coords, label, color in locations: if len(coords)>1: index = processed[label] % len(coords) # Round-robin indexing new_coord = [coords[index]] # Pick the correct coordinate new_data.append((new_coord, label, color)) processed[label] += 1 # Move to the next coordinate for this label if len(coords)==1: new_data.append((coords, label, color)) if len(locations[0]) == 4: for coords, label, color, w in locations: if len(coords)>1: index = processed[label] % len(coords) # Round-robin indexing new_coord = [coords[index]] # Pick the correct coordinate new_data.append((new_coord, label, color, w)) processed[label] += 1 # Move to the next coordinate for this label if len(coords)==1: new_data.append((coords, label, color, w)) if len(locations[0]) == 5: for coords, label, color, w, h in locations: if len(coords)>1: index = processed[label] % len(coords) # Round-robin indexing new_coord = [coords[index]] # Pick the correct coordinate new_data.append((new_coord, label, color, w, h)) processed[label] += 1 # Move to the next coordinate for this label if len(coords)==1: new_data.append((coords, label, color, w, h)) return new_data def get_width_info_tobeprinted(new_data): width_info_tobeprinted = [] if len(new_data[0]) == 4: for _,_,_, w in new_data: w = re.sub(r",", "", w) w = int(float(w)) width_info_tobeprinted.append(w) if len(new_data[0]) == 5: for _,_,_, w,h in new_data: w = re.sub(r",", "", w) h = re.sub(r",", "", h) w = int(float(w)) h = int(float(h)) width_info_tobeprinted.append(f"{w} mm wide x {h} mm high") return width_info_tobeprinted def clean_dimensions(text): # Remove commas and "mm" text = re.sub(r'[,\s]*mm', '', text) # Remove "mm" with optional spaces or commas before it text = text.replace(",", "") # Remove remaining commas if any return text def get_cleaned_width(width_info_tobeprinted): cleaned_width = [] for w in width_info_tobeprinted: cleaned_width.append(clean_dimensions(w)) return cleaned_width def get_widths_bb_format(cleaned_width, kelma): pattern = r"\bW(?:idth)?\s*[×x]\s*H(?:eight)?\b" match = re.search(pattern, kelma) widths = [] for widthaa in cleaned_width: index = max(widthaa.find("x"), widthaa.find("×"), widthaa.find("x"), widthaa.find("X"), widthaa.find("x")) width_name = widthaa[:index] height_name = widthaa[index+1:] width_name = int(float(width_name)) height_name = int(float(height_name)) if match: full_text = f"{width_name}mm wide x {height_name}mm high" else: full_text = f"{height_name}mm wide x {width_name}mm high" widths.append(full_text) return widths import fitz # PyMuPDF import PyPDF2 import io from PyPDF2.generic import TextStringObject # ✅ Required for setting string values def add_bluebeam_count_annotations(pdf_bytes, locations): pdf_stream = io.BytesIO(pdf_bytes) # Load PDF from bytes pdf_document = fitz.open("pdf", pdf_stream.read()) # Open PDF in memory page = pdf_document[0] # First page if len(locations[0]) == 3: for loc in locations: coor, lbl, clr = loc clr = (clr[0] / 255, clr[1] / 255, clr[2] / 255) for cor in coor: #Create a Circle annotation (Count Markup) annot = page.add_circle_annot( fitz.Rect(cor[0] - 10, cor[1] - 10, cor[0] + 10, cor[1] + 10) # Small circle ) #Assign required Bluebeam metadata annot.set_colors(stroke=clr, fill=(1, 1, 1)) # Set stroke color and fill white annot.set_border(width=2) # Border thickness annot.set_opacity(1) # Fully visible #Set annotation properties for Bluebeam Count detection annot.set_info("name", lbl) # Unique name for each count annot.set_info("subject", "Count") #Bluebeam uses "Count" for Count markups annot.set_info("title", lbl) # Optional annot.update() # Apply changes if len(locations[0]) == 4: for loc in locations: coor, lbl, clr,w = loc clr = (clr[0] / 255, clr[1] / 255, clr[2] / 255) for cor in coor: #Create a Circle annotation (Count Markup) annot = page.add_circle_annot( fitz.Rect(cor[0] - 10, cor[1] - 10, cor[0] + 10, cor[1] + 10) # Small circle ) #Assign required Bluebeam metadata annot.set_colors(stroke=clr, fill=(1, 1, 1)) # Set stroke color and fill white annot.set_border(width=2) # Border thickness annot.set_opacity(1) # Fully visible #Set annotation properties for Bluebeam Count detection annot.set_info("name", lbl) # Unique name for each count annot.set_info("subject", "Count") #Bluebeam uses "Count" for Count markups annot.set_info("title", lbl) # Optional annot.update() # Apply changes if len(locations[0]) == 5: for loc in locations: coor, lbl, clr,w,h = loc clr = (clr[0] / 255, clr[1] / 255, clr[2] / 255) for cor in coor: #Create a Circle annotation (Count Markup) annot = page.add_circle_annot( fitz.Rect(cor[0] - 10, cor[1] - 10, cor[0] + 10, cor[1] + 10) # Small circle ) #Assign required Bluebeam metadata annot.set_colors(stroke=clr, fill=(1, 1, 1)) # Set stroke color and fill white annot.set_border(width=2) # Border thickness annot.set_opacity(1) # Fully visible #Set annotation properties for Bluebeam Count detection annot.set_info("name", lbl) # Unique name for each count annot.set_info("subject", "Count") #Bluebeam uses "Count" for Count markups annot.set_info("title", lbl) # Optional annot.update() # Apply changes #Save modified PDF to a variable instead of a file output_stream = io.BytesIO() pdf_document.save(output_stream) pdf_document.close() return output_stream.getvalue() # Return the modified PDF as bytes def get_user_input(user_words): user_input = [] for item in user_words: user_input.append(item[0]) return user_input def modify_author_in_pypdf2(pdf_bytes, new_authors): pdf_stream = io.BytesIO(pdf_bytes) # Load PDF from bytes reader = PyPDF2.PdfReader(pdf_stream) writer = PyPDF2.PdfWriter() author_index = 0 # Track author assignment for page in reader.pages: if "/Annots" in page: #Check if annotations exist for annot in page["/Annots"]: annot_obj = annot.get_object() # Assign each annotation a unique author if len(new_authors) == 0: break if author_index < len(new_authors): annot_obj.update({"/T": TextStringObject(new_authors[author_index])})#Convert to PdfString author_index += 1 # Move to next author # If authors list is exhausted, keep the last one else: annot_obj.update({"/T": TextStringObject(new_authors[-1])}) writer.add_page(page) #Save the modified PDF to a variable output_stream = io.BytesIO() writer.write(output_stream) output_stream.seek(0) return output_stream.read() # return output_stream.getvalue() # Return modified PDF as bytes from PyPDF2 import PdfReader, PdfWriter def merge_pdf_bytes_list(pdfs): writer = PdfWriter() for pdf_bytes in pdfs: pdf_stream = io.BytesIO(pdf_bytes) reader = PdfReader(pdf_stream) for page in reader.pages: writer.add_page(page) output_stream = io.BytesIO() writer.write(output_stream) output_stream.seek(0) return output_stream.read() def process_pdf(input_pdf_path, output_pdf_path, locations, new_authors): #Load original PDF # with open(input_pdf_path, "rb") as file: # original_pdf_bytes = file.read() #Add Bluebeam-compatible count annotations annotated_pdf_bytes = add_bluebeam_count_annotations(input_pdf_path, locations) #Modify author field using PyPDF2 final_pdf_bytes = modify_author_in_pypdf2(annotated_pdf_bytes, new_authors) return final_pdf_bytes # #Save the final modified PDF to disk # with open(output_pdf_path, "wb") as file: # file.write(final_pdf_bytes) def mainRun(schedule, plan, searcharray): #print(type(plan)) eltype = type(plan) print(f"el type beta3 variable plan:: {eltype}") len_plan = len(plan) print(f"length of the plan's array is: {len_plan}") p1_type = type(plan[0]) print(f"el mawgood fe p[0]: {p1_type}") user_input = get_user_input(searcharray) dfs = extract_tables(schedule) selected_columns_new = get_selected_columns(dfs, user_input) kelma = get_st_op_pattern(selected_columns_new, user_input) col_dict = get_similar_colors(selected_columns_new) flattened_list = get_flattened_tuples_list(col_dict) pdfs = [] for p in plan: print(f" p in plan is {type(p)}") print(p) plan_texts = read_text(p) locations, not_found = get_word_locations_plan(flattened_list,plan_texts) new_data = get_cleaned_data(locations) repeated_labels = get_repeated_labels(locations) if kelma == None: widths = get_width_info_tobeprinted(new_data) else: width_info_tobeprinted = get_width_info_tobeprinted(new_data) cleaned_width = get_cleaned_width(width_info_tobeprinted) widths = get_widths_bb_format(cleaned_width, kelma) final_pdf_bytes = process_pdf(p, "final_output_width.pdf", new_data, widths) pdfs.append(final_pdf_bytes) #plan_texts = read_text(plan) #locations, not_found = get_word_locations_plan(flattened_list,plan_texts) #new_data = get_cleaned_data(locations) #repeated_labels = get_repeated_labels(locations) #if kelma == None: # widths = get_width_info_tobeprinted(new_data) #else: # width_info_tobeprinted = get_width_info_tobeprinted(new_data) # cleaned_width = get_cleaned_width(width_info_tobeprinted) # widths = get_widths_bb_format(cleaned_width, kelma) if selected_columns_new.shape[1] == 2: widths = [] #for j in range(len(locations)): # widths.append("Dimensions not found in schedule") #final_pdf_bytes= process_pdf(plan, "final_output_width.pdf", new_data, widths) merged_pdf = merge_pdf_bytes_list(pdfs) print(f"number of pges of merged_pdf is {len(merged_pdf)} and its type is {type(merged_pdf)}") not_found = [] doc2 =fitz.open('pdf',merged_pdf) len_doc2 = len(doc2) print(f"number of pges of doc2 is {len_doc2} and its type is {type(doc2)}") page=doc2[0] pix = page.get_pixmap() # render page to an image pl=Image.frombytes('RGB', [pix.width,pix.height],pix.samples) img=np.array(pl) annotatedimg = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) list1=pd.DataFrame(columns=['content', 'id', 'subject','color']) # for page in doc: for page in doc2: # Iterate through annotations on the page for annot in page.annots(): # Get the color of the annotation annot_color = annot.colors if annot_color is not None: # annot_color is a dictionary with 'stroke' and 'fill' keys stroke_color = annot_color.get('stroke') # Border color fill_color = annot_color.get('fill') # Fill color if fill_color: v='fill' # print('fill') if stroke_color: v='stroke' x,y,z=int(annot_color.get(v)[0]*255),int(annot_color.get(v)[1]*255),int(annot_color.get(v)[2]*255) list1.loc[len(list1)] =[annot.info['content'],annot.info['id'],annot.info['subject'],[x,y,z]] return annotatedimg, doc2 , list1, repeated_labels , not_found