import sys import json import re import logging from pathlib import Path import pdfplumber from pdf2image import convert_from_path import pytesseract logging.basicConfig(level=logging.INFO, format="[%(asctime)s] [%(levelname)s] %(message)s") logger = logging.getLogger("LandRecordExtractor") TABLE_SETTINGS_PRIMARY = { "vertical_strategy": "lines", "horizontal_strategy": "lines", "snap_tolerance": 3, "join_tolerance": 3, "intersection_tolerance": 3 } TABLE_SETTINGS_FALLBACK = { "vertical_strategy": "text", "horizontal_strategy": "text", "snap_tolerance": 3, "join_tolerance": 3, } def clean_text(value): if value is None: return "" value = str(value).replace("\n", " ").replace("\r", " ") return re.sub(r"\s+", " ", value).strip() def clean_person_name(name_str): cleaned = clean_text(name_str) cleaned = re.sub(r"\bbyakti\b", "", cleaned, flags=re.IGNORECASE) cleaned = re.sub(r"\bvyakti\b", "", cleaned, flags=re.IGNORECASE) return clean_text(cleaned) def split_cell(cell): """Splits multi-line cells and cleans entity markers.""" values = [] for line in str(cell or "").split("\n"): line = clean_person_name(line) if line: values.append(line) return values def run_ocr_on_pdf(pdf_path: str) -> str: try: images = convert_from_path(pdf_path, dpi=300) return "\n".join([pytesseract.image_to_string(img) for img in images]) except Exception as e: logger.error(f"OCR failed: {str(e)}") return "" def get_raw_pdf_text(pdf_path: str, pdf_obj) -> str: full_text = "\n".join([page.extract_text() or "" for page in pdf_obj.pages]).strip() if full_text: return full_text return run_ocr_on_pdf(pdf_path) def extract_header(full_text: str): header = {"jl_no": "", "daag_no": "", "mouza": "", "block": "", "district": "", "thana": ""} m = re.search(r"J[\.\s]*[LI1|i]\.?\s*No\.?\s*[:\-]?\s*(\d+)", full_text, re.IGNORECASE) if m: header["jl_no"] = m.group(1).strip() m = re.search(r"Thana\s*[:\-]?\s*([A-Za-z0-9\-\s]+?)(?=\n|Dag|Plot|J\.|Zamir|$)", full_text, re.IGNORECASE) if m: header["thana"] = m.group(1).strip() for key in ["mouza", "block", "district"]: m = re.search(rf"{key}\s*[:\-]?\s*([A-Za-z0-9\-\s]+)", full_text, re.IGNORECASE) if m: header[key] = m.group(1).strip() return header def extract_plot_info(full_text: str, pdf): """Extracts Plot No, Land Type (Shreni), and Total Land Area.""" info = {"daag_no": "", "shreni": "", "total_land_acre": ""} # 1. Proximity Search on normalized text normalized_text = re.sub(r'[\n|]', ' ', full_text) normalized_text = re.sub(r'\s+', ' ', normalized_text) m_brute = re.search(r"(?:Dag|Plot)\s*No\.?\s*(?:Shreni|Zamir|Moat|Pariman|ekar|Dager|Myap|\s)*\s+(\d{1,5})\b", normalized_text, re.IGNORECASE) if m_brute: info["daag_no"] = m_brute.group(1) anchor_match = re.search(rf"\b{info['daag_no']}\s+([A-Za-z]+)\s+(\d+\.\d+)\b", normalized_text) if anchor_match: info["shreni"] = anchor_match.group(1) info["total_land_acre"] = anchor_match.group(2) return info # 2. Native Table Fallback for page in pdf.pages: tables = page.extract_tables(table_settings=TABLE_SETTINGS_PRIMARY) or page.extract_tables(table_settings=TABLE_SETTINGS_FALLBACK) if not tables: continue for table in tables: if not table or len(table) < 2: continue header_str = " ".join([clean_text(c).lower() for c in table[0] if c]) if "dag" in header_str or "plot" in header_str: for row in table[1:]: if not any(row): continue clean_row = [clean_text(c) for c in row if c] for item in clean_row: if re.fullmatch(r"\d{1,5}", item): info["daag_no"] = item info["shreni"] = clean_row[1] if len(clean_row) > 1 else "" info["total_land_acre"] = clean_row[2] if len(clean_row) > 2 else "" return info return info def process_khatian_entries(pdf, full_text: str): """Dual-strategy Khatian extractor: Native Table Parsing + Text Regex Fallback.""" entries = [] # STRATEGY 1: Native PDF Table Parsing (Restored from old code) for page in pdf.pages: tables = page.extract_tables(table_settings=TABLE_SETTINGS_PRIMARY) or page.extract_tables(table_settings=TABLE_SETTINGS_FALLBACK) if not tables: continue for table in tables: if not table or len(table) < 2: continue header_str = " ".join([clean_text(c).lower() for c in table[0] if c]) if not ("khatian" in header_str and ("raiter" in header_str or "owner" in header_str or "nam" in header_str)): continue for row in table[1:]: if not any(row): continue khatians = split_cell(row[0]) if len(row) > 0 else [] owners = split_cell(row[1]) if len(row) > 1 else [] fathers = split_cell(row[2]) if len(row) > 2 else [] anshas = split_cell(row[3]) if len(row) > 3 else [] areas = split_cell(row[4]) if len(row) > 4 else [] if khatians and "khatian" in khatians[0].lower(): continue max_len = max(len(khatians), len(owners), len(fathers), len(anshas), len(areas)) for i in range(max_len): k_no = khatians[i] if i < len(khatians) else (khatians[0] if len(khatians) == 1 else "") o_name = owners[i] if i < len(owners) else "" f_name = fathers[i] if i < len(fathers) else "" ansh = anshas[i] if i < len(anshas) else "" area = areas[i] if i < len(areas) else "" if k_no or o_name: entries.append({ "khatian_no": k_no, "owner_name_bn": "", "owner_name": o_name, "father_husband_name_bn": "", "father_husband_name": f_name, "ansha": ansh, "area_acres": area }) if entries: return entries # STRATEGY 2: Text Regex Fallback (For OCR/Borderless PDFs) pattern = r"([\d/]+)\s*[\s|]+\s*([A-Za-z\s]+?)(?:\s+Byakti|\s+Vyakti)?\s*[\s|]+\s*([A-Za-z\s]+?)\s*[\s|]+\s*([\d\.]+)\s*[\s|]+\s*([\d\.]+)" matches = re.findall(pattern, full_text) for k_no, o_raw, f_raw, ansh, area in matches: o_name = clean_person_name(o_raw) f_name = clean_person_name(f_raw) if k_no and o_name: entries.append({ "khatian_no": k_no.strip(), "owner_name_bn": "", "owner_name": o_name, "father_husband_name_bn": "", "father_husband_name": f_name, "ansha": ansh.strip(), "area_acres": area.strip() }) return entries def extract_land_record(pdf_path: str): with pdfplumber.open(pdf_path) as pdf: full_text = get_raw_pdf_text(pdf_path, pdf) header = extract_header(full_text) plot_info = extract_plot_info(full_text, pdf) entries = process_khatian_entries(pdf, full_text) return { "source_file": Path(pdf_path).name, "jl_no": header["jl_no"], "daag_no": plot_info["daag_no"], "mouza": header["mouza"], "block": header["block"], "district": header["district"], "thana": header["thana"], "shreni": plot_info["shreni"], "total_land_acre": plot_info["total_land_acre"], "total_entries": len(entries), "khatian_entries": entries } if __name__ == "__main__": if len(sys.argv) > 1: print(json.dumps(extract_land_record(sys.argv[1]), indent=4, ensure_ascii=False)) # import sys # import json # import re # import logging # from pathlib import Path # import pdfplumber # from pdf2image import convert_from_path # import pytesseract # logging.basicConfig(level=logging.INFO, format="[%(asctime)s] [%(levelname)s] %(message)s") # logger = logging.getLogger("LandRecordExtractor") # TABLE_SETTINGS_PRIMARY = { # "vertical_strategy": "lines", # "horizontal_strategy": "lines", # "snap_tolerance": 3, # "join_tolerance": 3, # "intersection_tolerance": 3 # } # def clean_text(value): # if value is None: # return "" # value = str(value).replace("\n", " ").replace("\r", " ") # return re.sub(r"\s+", " ", value).strip() # def clean_person_name(name_str): # cleaned = clean_text(name_str) # cleaned = re.sub(r"\bbyakti\b", "", cleaned, flags=re.IGNORECASE) # cleaned = re.sub(r"\bvyakti\b", "", cleaned, flags=re.IGNORECASE) # return clean_text(cleaned) # def run_ocr_on_pdf(pdf_path: str) -> str: # try: # images = convert_from_path(pdf_path, dpi=300) # return "\n".join([pytesseract.image_to_string(img) for img in images]) # except Exception as e: # logger.error(f"OCR failed: {str(e)}") # return "" # def get_raw_pdf_text(pdf_path: str, pdf_obj) -> str: # full_text = "\n".join([page.extract_text() or "" for page in pdf_obj.pages]).strip() # if full_text: # return full_text # return run_ocr_on_pdf(pdf_path) # def extract_header(full_text: str): # header = {"jl_no": "", "daag_no": "", "mouza": "", "block": "", "district": "", "thana": ""} # m = re.search(r"J[\.\s]*[LI1|i]\.?\s*No\.?\s*[:\-]?\s*(\d+)", full_text, re.IGNORECASE) # if m: # header["jl_no"] = m.group(1).strip() # m = re.search(r"Thana\s*[:\-]?\s*([A-Za-z0-9\-\s]+?)(?=\n|Dag|Plot|J\.|Zamir|$)", full_text, re.IGNORECASE) # if m: # header["thana"] = m.group(1).strip() # for key in ["mouza", "block", "district"]: # m = re.search(rf"{key}\s*[:\-]?\s*([A-Za-z0-9\-\s]+)", full_text, re.IGNORECASE) # if m: # header[key] = m.group(1).strip() # return header # def extract_plot_info(full_text: str, pdf): # """Brute-force keyword proximity search for Daag No.""" # info = {"daag_no": "", "shreni": "", "total_land_acre": ""} # # 1. BRUTE FORCE PROXIMITY SEARCH (Solves all OCR/Canvas spacing issues) # # Replaces all newlines and pipes with spaces, then collapses multiple spaces # normalized_text = re.sub(r'[\n|]', ' ', full_text) # normalized_text = re.sub(r'\s+', ' ', normalized_text) # # Looks for "Dag No" or "Plot No", then grabs the very next number up to 5 digits long # m_brute = re.search(r"(?:Dag|Plot)\s*No\.?\s*(?:Shreni|Zamir|Moat|Pariman|ekar|Dager|Myap|\s)*\s+(\d{1,5})\b", normalized_text, re.IGNORECASE) # if m_brute: # info["daag_no"] = m_brute.group(1) # logger.info(f"-> Daag No found via Brute Force Search: {info['daag_no']}") # # Try to grab adjacent text values using the found Daag No as an anchor # anchor_match = re.search(rf"\b{info['daag_no']}\s+([A-Za-z]+)\s+(\d+\.\d+)\b", normalized_text) # if anchor_match: # info["shreni"] = anchor_match.group(1) # info["total_land_acre"] = anchor_match.group(2) # return info # # 2. Collapsed Stream Match (e.g. "Dager Myap390Saiyam0.19Click") # m_collapsed = re.search(r"(?:Dager|Plot)\s*M[ya|a]+p\s*(\d{1,5})\s*([A-Za-z]+)\s*(\d+\.\d+)", full_text, re.IGNORECASE) # if m_collapsed: # info["daag_no"] = m_collapsed.group(1).strip() # info["shreni"] = m_collapsed.group(2).strip() # info["total_land_acre"] = m_collapsed.group(3).strip() # logger.info(f"-> Plot Info found via Collapsed Regex: {info}") # return info # # 3. Native Table Extraction Fallback # for page in pdf.pages: # tables = page.extract_tables(table_settings=TABLE_SETTINGS_PRIMARY) # if not tables: # continue # for table in tables: # if not table or len(table) < 2: # continue # header_str = " ".join([clean_text(c).lower() for c in table[0] if c]) # if "dag" in header_str or "plot" in header_str: # for row in table[1:]: # if not any(row): # continue # clean_row = [clean_text(c) for c in row if c] # for item in clean_row: # if re.fullmatch(r"\d{1,5}", item): # info["daag_no"] = item # info["shreni"] = clean_row[1] if len(clean_row) > 1 else "" # info["total_land_acre"] = clean_row[2] if len(clean_row) > 2 else "" # logger.info(f"-> Plot Info found via PDF Table: {info}") # return info # logger.warning("-> Failed to extract Plot Information!") # return info # def process_khatian_entries(full_text: str): # entries = [] # pattern = r"([\d/]+)\s*[\s|]+\s*([A-Za-z\s]+?)(?:\s+Byakti|\s+Vyakti)?\s*[\s|]+\s*([A-Za-z\s]+?)\s*[\s|]+\s*([\d\.]+)\s*[\s|]+\s*([\d\.]+)" # matches = re.findall(pattern, full_text) # for k_no, o_raw, f_raw, ansh, area in matches: # o_name = re.sub(r"\bbyakti\b", "", o_raw, flags=re.IGNORECASE).strip() # f_name = re.sub(r"\bbyakti\b", "", f_raw, flags=re.IGNORECASE).strip() # if k_no and o_name: # entries.append({ # "khatian_no": k_no.strip(), # "owner_name_bn": "", # "owner_name": o_name, # "father_husband_name_bn": "", # "father_husband_name": f_name, # "ansha": ansh.strip(), # "area_acres": area.strip() # }) # return entries # def extract_land_record(pdf_path: str): # with pdfplumber.open(pdf_path) as pdf: # full_text = get_raw_pdf_text(pdf_path, pdf) # header = extract_header(full_text) # plot_info = extract_plot_info(full_text, pdf) # entries = process_khatian_entries(full_text) # return { # "source_file": Path(pdf_path).name, # "jl_no": header["jl_no"], # "daag_no": plot_info["daag_no"], # "mouza": header["mouza"], # "block": header["block"], # "district": header["district"], # "thana": header["thana"], # "shreni": plot_info["shreni"], # "total_land_acre": plot_info["total_land_acre"], # "total_entries": len(entries), # "khatian_entries": entries # } # if __name__ == "__main__": # if len(sys.argv) > 1: # print(json.dumps(extract_land_record(sys.argv[1]), indent=4, ensure_ascii=False))